certflow.handlers.id_generator module

唯一键生成器 - 分层策略（配置驱动）

新数据（有生产令号）：用配置的稳定字段 老数据（无生产令号）：用配置的内容字段

class IDGenerator

基类：object

分层唯一键生成器（配置驱动）

提供合格证编号生成、基于数据内容的唯一键生成及批量生成功能。 唯一键采用分层策略，通过配置文件 id_generator 节点驱动字段选择。

分层策略:

• 新数据模式：当配置的 new_data_fields 字段全部有值时使用 （默认: sales_order_no + production_order_no）

• 老数据模式：当 new_data_fields 任一为空时降级使用 （默认: 9个内容字段）

配置节点:

id_generator.new_data_fields: 新数据稳定字段列表 id_generator.old_data_fields: 老数据内容匹配字段列表 id_generator.empty_marker_prefix: 空值标记前缀（默认 "【空白"）

向后兼容:

generate_unique_key(data, keys) 指定 keys 参数时走旧逻辑

主要功能:

• generate_certificate_no(): 生成合格证编号

• generate_unique_key(): 生成MD5唯一键（支持分层策略）

• generate_unique_key_readable(): 生成可读唯一键（调试用）

• generate_batch_unique_keys(): 批量生成唯一键

• get_all_unique_key_fields(): 获取所有唯一键字段（供去重配置使用）

• normalize_empty_values(): 空值规范化

• normalize_dataframe_empty_values(): DataFrame空值规范化

static generate_certificate_no(prefix='CERT', sequence=None)

生成合格证编号

生成格式为 CERT-YYYYMMDD-NNNN 的合格证编号， 序号部分自动补零至4位。

参数:

• prefix (str) -- 编号前缀，默认为"CERT"

• sequence (int | None) -- 序号，如果为None则默认为1

返回:

生成的合格证编号字符串

返回类型:

str

示例

>>> # 生成默认格式的编号
>>> cert_no = IDGenerator.generate_certificate_no()
>>> print(cert_no)  # 输出: CERT-20231201-0001
>>>
>>> # 生成自定义前缀和序号
>>> cert_no = IDGenerator.generate_certificate_no(prefix="QT", sequence=5)
>>> print(cert_no)  # 输出: QT-20231201-0005

classmethod generate_unique_key(data, keys=None)

生成唯一键（分层策略）

参数:

• data (dict[str, Any]) -- 数据字典，包含需要生成唯一键的字段

• keys (list[str] | None) -- 指定字段列表。有值时走旧逻辑（向后兼容）， None 时走分层策略

返回:

16位MD5哈希值

返回类型:

str

classmethod generate_unique_key_readable(data, keys=None)

生成可读唯一键（分层策略，调试用）

参数:

• data (dict[str, Any]) -- 数据字典

• keys (list[str] | None) -- 指定字段列表。有值时走旧逻辑，None 时走分层策略

返回:

可读的唯一键字符串，字段值之间用"::"分隔

返回类型:

str

static generate_batch_unique_keys(data_list, keys=None)

批量生成唯一键

为数据列表中的每条记录生成唯一键，并将键值写入"unique_key"字段。

参数:

• data_list (list[dict[str, Any]]) -- 数据字典列表

• keys (list[str] | None) -- 用于生成唯一键的字段列表，如果为None则使用所有字段

返回:

添加了unique_key字段的数据列表

返回类型:

List[Dict[str, Any]]

示例

>>> data_list = [
...     {"contract_no": "PO-001", "product_model": "阀门A"},
...     {"contract_no": "PO-002", "product_model": "阀门B"}
... ]
>>> result = IDGenerator.generate_batch_unique_keys(data_list, ["contract_no"])
>>> for item in result:
...     print(item["unique_key"])

classmethod normalize_empty_values(data, keys=None)

规范化字典中的空值

将 None 或空字符串转换为 【空白中文字段名】 格式。 用于统一空值表示，便于后续处理和识别。

参数:

• data (dict[str, Any]) -- 原始数据字典

• keys (list[str] | None) -- 需要处理的字段列表，如果为None则处理所有字段

返回:

空值规范化后的字典

返回类型:

Dict[str, Any]

示例

>>> data = {"contract_no": "", "product_model": "阀门A", "quantity": None}
>>> normalized = IDGenerator.normalize_empty_values(data, ["contract_no", "quantity"])
>>> print(normalized["contract_no"])  # 输出: "【空白合同号】"
>>> print(normalized["quantity"])    # 输出: "【空白数量】"

classmethod normalize_dataframe_empty_values(df, fields)

规范化DataFrame中指定字段的空值

将DataFrame中指定字段的空值(NaN、None、空字符串)统一替换为格式化的空值标记。 适用于批量数据处理场景。

参数:

• df (<MagicMock name='mock.DataFrame' id='2619415774032'>) -- 需要处理的DataFrame

• fields (list[str]) -- 需要处理的字段列表

返回:

空值规范化后的DataFrame副本

返回类型:

pd.DataFrame

示例

>>> import pandas as pd
>>> df = pd.DataFrame({
...     "contract_no": ["PO-001", "", None],
...     "product_model": ["阀门A", "阀门B", "阀门C"]
... })
>>> cleaned = IDGenerator.normalize_dataframe_empty_values(df, ["contract_no"])
>>> print(cleaned["contract_no"][1])  # 输出: "【空白合同号】"

classmethod get_all_unique_key_fields()

获取所有唯一键字段（去重合并，供 import_deduplication 使用）

返回:

去重后的唯一键字段列表

返回类型:

list[str]