certflow.handlers.id_generator module¶

向后兼容重导出（handlers → utils 分层迁移）。

IDGenerator 为纯算法工具（零 Qt 依赖），已从 handlers 层迁移至 utils 层（详见阶段 B：断开 models → handlers 环形耦合）。本模块仅作兼容垫片，转发所有既有 from certflow.handlers.id_generator import IDGenerator 引用，待后续各引用点（views / services / handlers / scripts / tests）逐一迁移至 certflow.utils.id_generator 后可删除本文件。

相关改动：models/sale_plan.py 已直接改用 certflow.utils.id_generator。

class certflow.handlers.id_generator.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)[源代码]¶

生成唯一键（分层策略；身份键 = MD5 哈希）

参数:

data (dict[str, Any]) -- 数据字典，包含需要生成唯一键的字段
keys (list[str] | None) -- 指定字段列表。有值时走旧逻辑（向后兼容，测试用）

返回:

16位MD5哈希值

返回类型:

str

示例

>>> # 分层策略（有生产令号时）
>>> data = {"sales_order_no": "SO001", "production_order_no": "PO001",
...         "product_model": "阀门A", "product_spec": "DN50"}
>>> key = IDGenerator.generate_unique_key(data)
>>> len(key)
16
>>>
>>> # 向后兼容：指定 keys 走旧逻辑
>>> key2 = IDGenerator.generate_unique_key(data, keys=["contract_no", "product_model"])
>>> len(key2)
16

classmethod generate_unique_key_readable(data, keys=None)[源代码]¶

生成可读唯一键（分层策略，调试用；字段间 :: 分隔）

参数:

data (dict[str, Any])
keys (list[str] | None)

返回类型:

str

classmethod generate_identity_key(data)[源代码]¶

生成可读身份键串（§9 落地待办②：身份键显式化，用于追溯/人工核对）

与 generate_unique_key 选用同一分层与字段，仅返回人工可读的拼接串（不哈希），作为 unique_key 的人类可读形式。

参数:: data (dict[str, Any])
返回类型:: str

classmethod generate_change_signature(data)[源代码]¶

生成变更签名（§9 落地待办②：内容摘要，用于更新 vs 跳过快速判定）

对受监控字段（import_deduplication.monitored_fields）做拼接+MD5，作为该行"内容指纹"。身份键命中已存在行时，若签名一致则判定无变更、跳过逐字段 diff（save_handler._handle_existing 使用），否则跑全量 diff。

参数:: data (dict[str, Any])
返回类型:: str

classmethod normalize_spec(record, session=None)[源代码]¶

导入前口径规范化（R1 总前置）。

调 DNService.resolve_caliber 解析 product_spec，将合法数字口径写回 record["product_spec_norm"]；解析失败（含 φ50 伪归一）保留原值并标记 record["spec_needs_manual"]=True，供 R4 标黄。

参数:

record (dict[str, Any]) -- 记录字典（含 product_spec）
session (Session | None) -- SQLAlchemy 会话（DB 映射查询依赖；None 时退化为纯文本解析）

返回:

同一 record（就地补充 product_spec_norm / spec_needs_manual）

返回类型:

dict

classmethod normalize_pn(record, session=None)[源代码]¶

导入前公称压力/标准号规范化（#30 P0 DN/PN 字典 DB 化）。

调 PNService.resolve_pn 解析 product_model，将公称压力写回 record["pressure_value"]；从型号压力转换表(db) 补全省录 record["test_standard"]``（标准号导入期默认空、由合格证打印时回填，不计入黄标）；黄标仅由**公称压力无法解析**驱动——``pn_needs_manual 在 pressure_value 为空时置 True，对标 VBA 的 Stop 硬断点，改为**非中断黄标**（更优：批量导入不被单点卡死）。

参数:

record (dict[str, Any]) -- 记录字典（含 product_model / 可选 pressure / test_standard）
session (Session | None) -- SQLAlchemy 会话（DB 映射查询依赖；None 时退化为纯文本解析）

返回:

同一 record（就地补充 pressure_value / test_standard / pn_needs_manual）

返回类型:

dict

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 (DataFrame) -- 需要处理的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]