"""唯一键生成器 - 分层策略（配置驱动）
新数据（有生产令号）：用配置的稳定字段
老数据（无生产令号）：用配置的内容字段
"""
from __future__ import annotations
import hashlib
import json
import re
from datetime import datetime
from typing import Any
import pandas as pd
from loguru import logger
from sqlalchemy.orm import Session
[文档]
class IDGenerator:
"""分层唯一键生成器（配置驱动）
提供合格证编号生成、基于数据内容的唯一键生成及批量生成功能。
唯一键采用分层策略，通过配置文件 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空值规范化
"""
_field_name_cn_cache: dict[str, str] | None = None
_new_data_fields_cache: list[str] | None = None
_old_data_fields_cache: list[str] | None = None
_empty_marker_prefix_cache: str | None = None
_tiers_cache: dict | None = None
@classmethod
def _get_new_data_fields(cls) -> list[str]:
"""获取新数据模式字段（从配置读取，带缓存）"""
if cls._new_data_fields_cache is not None:
return cls._new_data_fields_cache
try:
from certflow.config.settings import cfg
cls._new_data_fields_cache = cfg(
"id_generator.new_data_fields",
["sales_order_no", "production_order_no", "product_model", "product_spec"],
) # noqa
except Exception as e:
logger.warning(f"读取 new_data_fields 配置失败: {e}，使用默认值")
cls._new_data_fields_cache = [
"sales_order_no",
"production_order_no",
"product_model",
"product_spec",
]
return cls._new_data_fields_cache
@classmethod
def _get_old_data_fields(cls) -> list[str]:
"""获取老数据模式字段（从配置读取，带缓存）"""
if cls._old_data_fields_cache is not None:
return cls._old_data_fields_cache
try:
from certflow.config.settings import cfg
cls._old_data_fields_cache = cfg(
"id_generator.old_data_fields",
[
"plan_date",
"contract_no",
"customer",
"project_name",
"product_name",
"product_model",
"product_spec",
],
)
except Exception as e:
logger.warning(f"读取 old_data_fields 配置失败: {e}，使用默认值")
cls._old_data_fields_cache = [
"plan_date",
"contract_no",
"customer",
"project_name",
"product_name",
"product_model",
"product_spec",
]
return cls._old_data_fields_cache
@classmethod
def _get_empty_marker_prefix(cls) -> str:
"""获取空值标记前缀（从配置读取，带缓存）
用于识别被 normalize 处理过的空值标记，如"【空白合同号】"。
当分层策略中 new_data_fields 的值为该前缀开头时，视为空值。
Returns:
str: 空值标记前缀，默认 "【空白"
"""
if cls._empty_marker_prefix_cache is not None:
return cls._empty_marker_prefix_cache
try:
from certflow.config.settings import cfg
cls._empty_marker_prefix_cache = cfg("id_generator.empty_marker_prefix", "【空白")
except Exception as e:
logger.warning(f"读取 empty_marker_prefix 配置失败: {e}，使用默认值")
cls._empty_marker_prefix_cache = "【空白"
return cls._empty_marker_prefix_cache
@classmethod
def _get_field_name_cn(cls) -> dict[str, str]:
"""获取英文字段名到中文字段名的映射
从配置文件的 sales_plan.column_mapping.direct 反向构建映射关系。
用于生成空值标记时的中文显示。
Returns:
Dict[str, str]: 英文字段名到中文字段名的映射字典
Examples:
>>> mapping = IDGenerator._get_field_name_cn()
>>> print(mapping.get("contract_no"))
"合同号"
"""
if cls._field_name_cn_cache is not None:
return cls._field_name_cn_cache
try:
from certflow.config.settings import cfg
# 从 direct 映射反向构建
direct_mapping = cfg("sales_plan.column_mapping.direct", {})
# direct_mapping 格式: {"合同号": "contract_no", "计划日期": "plan_date", ...}
# 反向映射: 英文字段名 -> 中文字段名
reverse_mapping = {}
for cn_name, en_field in direct_mapping.items():
reverse_mapping[en_field] = cn_name
cls._field_name_cn_cache = reverse_mapping
logger.debug(f"加载字段中文名映射: {len(reverse_mapping)} 个字段")
except Exception as e:
logger.warning(f"加载字段中文名映射失败: {e}，将使用英文字段名")
cls._field_name_cn_cache = {}
return cls._field_name_cn_cache
@classmethod
def _get_empty_marker(cls, field_name: str) -> str:
"""获取空值标记
生成统一格式的空值标记，用于在数据中标识缺失字段。
格式为 "【空白中文字段名】"。
Args:
field_name: 英文字段名
Returns:
str: 空值标记字符串，如"【空白合同号】"
Examples:
>>> marker = IDGenerator._get_empty_marker("contract_no")
>>> print(marker)
"【空白合同号】"
"""
cn_map = cls._get_field_name_cn()
cn_name = cn_map.get(field_name, field_name)
return f"【空白{cn_name}】"
[文档]
@staticmethod
def generate_certificate_no(prefix: str = "CERT", sequence: int | None = None) -> str:
"""生成合格证编号
生成格式为 CERT-YYYYMMDD-NNNN 的合格证编号，
序号部分自动补零至4位。
Args:
prefix: 编号前缀，默认为"CERT"
sequence: 序号，如果为None则默认为1
Returns:
str: 生成的合格证编号字符串
Examples:
>>> # 生成默认格式的编号
>>> 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
"""
date_str = datetime.now().strftime("%Y%m%d")
if sequence is None:
sequence = 1
return f"{prefix}-{date_str}-{sequence:04d}"
@classmethod
def _get_tiers(cls) -> dict | None:
"""获取去重身份键分层配置（从 unique_key.tiers 读取，带缓存）
单一真相源：modern_2025 / legacy 两层的字段列表（§9 落地待办①）。
缺失时返回 None，由 _tier_fields 回退 id_generator.new_data_fields/old_data_fields。
"""
if cls._tiers_cache is not None:
return cls._tiers_cache
try:
from certflow.config.settings import cfg
raw = cfg("unique_key.tiers", None)
cls._tiers_cache = raw if isinstance(raw, dict) and raw else None
except Exception as e: # noqa: BLE001
logger.warning(f"读取 unique_key.tiers 配置失败: {e}，使用回退")
cls._tiers_cache = None
return cls._tiers_cache
@classmethod
def _select_tier(cls, data: dict[str, Any]) -> str:
"""根据数据选择身份键分层：双号齐全→modern_2025，否则→legacy。"""
stable_ids = ["sales_order_no", "production_order_no"]
trigger_values = [str(data.get(f, "")).strip() for f in stable_ids]
trigger_ok = all(trigger_values) and all(
not v.startswith(cls._get_empty_marker_prefix()) for v in trigger_values
)
return "modern_2025" if trigger_ok else "legacy"
@classmethod
def _tier_fields(cls, tier_name: str) -> list[str]:
"""获取某层的身份键字段列表（tiers 优先，缺失回退 id_generator 字段）。"""
tiers = cls._get_tiers()
if tiers and tier_name in tiers:
tier = tiers[tier_name]
fields = tier.get("fields") if isinstance(tier, dict) else getattr(tier, "fields", None)
if fields:
return list(fields)
# 回退：保持旧 new/old_data_fields 行为（向后兼容）
return (
cls._get_new_data_fields() if tier_name == "modern_2025" else cls._get_old_data_fields()
)
@classmethod
def _build_key_string(cls, data: dict[str, Any], fields: list[str]) -> str:
"""按字段列表拼接身份键串（product_spec 走 product_spec_norm 归一）。"""
prefix = cls._get_empty_marker_prefix()
parts = []
for f in fields:
if f == "product_spec":
# R1：口径规范化后参与建键，避免 φ50 / 50 / DN50 被判为不同
v = str(data.get("product_spec_norm", data.get("product_spec", ""))).strip()
else:
v = str(data.get(f, "")).strip()
if v.startswith(prefix):
v = ""
parts.append(v if v else cls._get_empty_marker(f))
return cls._separator().join(parts)
[文档]
@classmethod
def generate_unique_key(cls, data: dict[str, Any], keys: list[str] | None = None) -> str:
"""生成唯一键（分层策略；身份键 = MD5 哈希）
Args:
data: 数据字典，包含需要生成唯一键的字段
keys: 指定字段列表。有值时走旧逻辑（向后兼容，测试用）
Returns:
str: 16位MD5哈希值
Examples:
>>> # 分层策略（有生产令号时）
>>> 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
"""
# 指定 keys：走旧逻辑（向后兼容 test_quantity_in_key 等）
if keys:
filtered_data = {k: data.get(k, "") for k in keys}
exclude_keys = ["_original_order", "_sort_group"]
filtered_data = {k: v for k, v in filtered_data.items() if k not in exclude_keys}
processed_data = {}
for k, v in filtered_data.items():
if v is None or v == "":
processed_data[k] = cls._get_empty_marker(k)
else:
processed_data[k] = str(v).strip()
key_str = json.dumps(processed_data, sort_keys=True, ensure_ascii=False)
return hashlib.md5(key_str.encode("utf-8")).hexdigest()[:16]
# 分层策略（§9 落地待办①：字段来自 unique_key.tiers，缺失回退 id_generator 字段）
tier_name = cls._select_tier(data)
fields = cls._tier_fields(tier_name)
key_str = cls._build_key_string(data, fields)
return hashlib.md5(key_str.encode("utf-8")).hexdigest()[:16]
[文档]
@classmethod
def generate_unique_key_readable(
cls, data: dict[str, Any], keys: list[str] | None = None
) -> str:
"""生成可读唯一键（分层策略，调试用；字段间 :: 分隔）"""
if keys:
filtered_data = {k: data.get(k, "") for k in keys}
exclude_keys = ["_original_order", "_sort_group"]
filtered_data = {k: v for k, v in filtered_data.items() if k not in exclude_keys}
key_parts = []
for k in keys:
v = filtered_data.get(k, "")
if v is None or v == "":
key_parts.append(cls._get_empty_marker(k))
else:
key_parts.append(str(v).strip())
return "::".join(key_parts)
tier_name = cls._select_tier(data)
fields = cls._tier_fields(tier_name)
readable = cls._build_key_string(data, fields)
return readable.replace(cls._separator(), "::")
[文档]
@classmethod
def generate_identity_key(cls, data: dict[str, Any]) -> str:
"""生成可读身份键串（§9 落地待办②：身份键显式化，用于追溯/人工核对）
与 generate_unique_key 选用同一分层与字段，仅返回人工可读的拼接串
（不哈希），作为 unique_key 的人类可读形式。
"""
tier_name = cls._select_tier(data)
fields = cls._tier_fields(tier_name)
return cls._build_key_string(data, fields)
[文档]
@classmethod
def generate_change_signature(cls, data: dict[str, Any]) -> str:
"""生成变更签名（§9 落地待办②：内容摘要，用于更新 vs 跳过快速判定）
对受监控字段（import_deduplication.monitored_fields）做拼接+MD5，
作为该行"内容指纹"。身份键命中已存在行时，若签名一致则判定无变更、
跳过逐字段 diff（save_handler._handle_existing 使用），否则跑全量 diff。
"""
try:
from certflow.config.settings import MONITORED_FIELDS
except Exception: # noqa: BLE001
MONITORED_FIELDS = None # noqa: N806 刻意沿用 settings 常量名
fields = MONITORED_FIELDS or [
"product_name",
"product_model",
"product_spec",
"quantity",
"customer",
"project_name",
"plan_date",
"supply_type",
]
prefix = cls._get_empty_marker_prefix()
parts = []
for f in fields:
# product_spec 以归一化口径(product_spec_norm)为内容指纹，
# 与 _create_sale_plan 中 readable_record 的取值保持一致，
# 否则插入用 norm、比对用 raw 会导致签名不一致、假阳性触发逐字段 diff。
if f == "product_spec":
v = str(data.get("product_spec_norm") or data.get("product_spec", "")).strip()
else:
v = str(data.get(f, "")).strip()
if v.startswith(prefix):
v = ""
parts.append(v if v else cls._get_empty_marker(f))
return hashlib.md5(cls._separator().join(parts).encode("utf-8")).hexdigest()[:16]
# ============================================================
# 口径规范化（R1 总前置）
# ============================================================
_NUMERIC_RE = re.compile(r"^\d+(\.\d+)?$")
# 合法口径标识：纯数字 或 数字*数字（如 100 / 100*80），不含字母/汉字
_CALIBER_TOKEN_RE = re.compile(r"^\d+(?:\.\d+)?(?:\*\d+(?:\.\d+)?)?$")
@classmethod
def _is_numeric_mm(cls, v: str | None) -> bool:
"""口径值是否为合法纯数字口径（如 50 / 100.0）"""
return bool(cls._NUMERIC_RE.match(v or ""))
@classmethod
def _is_valid_caliber_token(cls, v: str | None) -> bool:
"""口径值是否为合法口径标识（纯数字或 数字*数字，如 100 / 100*80）"""
return bool(cls._CALIBER_TOKEN_RE.match(v or ""))
[文档]
@classmethod
def normalize_spec(
cls, record: dict[str, Any], session: Session | None = None
) -> dict[str, Any]:
"""导入前口径规范化（R1 总前置）。
调 DNService.resolve_caliber 解析 product_spec，将合法数字口径写回
record["product_spec_norm"]；解析失败（含 φ50 伪归一）保留原值并标记
record["spec_needs_manual"]=True，供 R4 标黄。
Args:
record: 记录字典（含 product_spec）
session: SQLAlchemy 会话（DB 映射查询依赖；None 时退化为纯文本解析）
Returns:
dict: 同一 record（就地补充 product_spec_norm / spec_needs_manual）
"""
from certflow.services.model_spec_parser import ModelSpecParser
model = str(record.get("product_model", "") or "")
raw = str(record.get("product_spec", "") or "")
if not raw:
# 空白口径：兜底从型号内嵌 DNxxx 反推（#1000192，对标 VBA inputbox 补录）。
# 命中即写 product_spec_norm（不伪造 product_spec 合同口径）；未命中不标黄
# （保持「空口径不新增黄标」的既有口径）。
derived = ModelSpecParser.parse_caliber(model)
if derived and cls._is_valid_caliber_token(derived):
record["product_spec_norm"] = derived
record["spec_needs_manual"] = False
else:
record["product_spec_norm"] = ""
record["spec_needs_manual"] = False
return record
from certflow.services.dn_service import DNService
svc = DNService(session)
res = svc.resolve_caliber(raw)
val = res.get("caliber_value") or ""
if cls._is_numeric_mm(val):
record["product_spec_norm"] = val
record["spec_needs_manual"] = False
else:
# 主口径解析失败 / 伪归一（φ50 等）→ 兜底从型号内嵌 DNxxx 反推
derived = ModelSpecParser.parse_caliber(model)
if derived and cls._is_valid_caliber_token(derived):
record["product_spec_norm"] = derived
record["spec_needs_manual"] = False
else:
# 仍无解 → 保留原值，标黄待人工复核，不伪造
record["product_spec_norm"] = raw
record["spec_needs_manual"] = True
return record
[文档]
@classmethod
def normalize_pn(cls, record: dict[str, Any], session: Session | None = None) -> dict[str, Any]:
"""导入前公称压力/标准号规范化（#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`` 硬断点，
改为**非中断黄标**（更优：批量导入不被单点卡死）。
Args:
record: 记录字典（含 product_model / 可选 pressure / test_standard）
session: SQLAlchemy 会话（DB 映射查询依赖；None 时退化为纯文本解析）
Returns:
dict: 同一 record（就地补充 pressure_value / test_standard / pn_needs_manual）
"""
model = str(record.get("product_model", "") or "").strip()
if not model:
record.setdefault("pressure_value", "")
record.setdefault("test_standard", "")
record["pn_needs_manual"] = False
return record
from certflow.services.pn_service import PNService
svc = PNService(session)
template_pn = str(
record.get("pressure", "") or record.get("pressure_value", "") or ""
).strip()
res = svc.resolve_pn(model, template_pn=template_pn or None)
# 解析到的公称压力（显示形式，如 2.5MPa / 150Lb）；未解析出则留空
record["pressure_value"] = res.get("pn_display") or ""
# 兜底：resolve_pn 未解析出压力时，从型号内嵌 PNxx/CLASSxx/嵌入压力编码反推
# （#1000192，如 KHBF-25-PN320 → 32MPa / CLASS1500 → 1500Lb / BQ647Y-25P → 2.5MPa）。
# 命中即写 pressure_value；未命中仍留空 → 由下方 pn_needs_manual 标黄，不伪造。
if not record["pressure_value"]:
from certflow.services.model_spec_parser import ModelSpecParser
derived = ModelSpecParser.parse_pressure(model)
if derived.get("pn_display"):
record["pressure_value"] = derived["pn_display"]
# 标准号：字典(db) 命中的优先；否则保留 Excel 已填值
db_standard = (res.get("test_standard", "") or "").strip()
existing_standard = str(record.get("test_standard", "") or "").strip()
record["test_standard"] = db_standard or existing_standard
# 黄标仅由公称压力(PN)无法解析驱动；标准号(test_standard)导入期默认空、
# 由合格证打印时回填，不计入黄标。
record["pn_needs_manual"] = not bool(record.get("pressure_value"))
return record
@classmethod
def _separator(cls) -> str:
"""唯一键字段拼接符（C3：统一取代硬编码 "|"）"""
try:
from certflow.config.settings import UNIQUE_KEY_SEPARATOR
return UNIQUE_KEY_SEPARATOR
except Exception:
return "|"
[文档]
@staticmethod
def generate_batch_unique_keys(
data_list: list[dict[str, Any]], keys: list[str] | None = None
) -> list[dict[str, Any]]:
"""批量生成唯一键
为数据列表中的每条记录生成唯一键，并将键值写入"unique_key"字段。
Args:
data_list: 数据字典列表
keys: 用于生成唯一键的字段列表，如果为None则使用所有字段
Returns:
List[Dict[str, Any]]: 添加了unique_key字段的数据列表
Examples:
>>> 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"])
"""
result = []
for data in data_list:
unique_key = IDGenerator.generate_unique_key(data, keys)
data["unique_key"] = unique_key
result.append(data)
return result
[文档]
@classmethod
def normalize_empty_values(
cls, data: dict[str, Any], keys: list[str] | None = None
) -> dict[str, Any]:
"""规范化字典中的空值
将 None 或空字符串转换为 【空白中文字段名】 格式。
用于统一空值表示，便于后续处理和识别。
Args:
data: 原始数据字典
keys: 需要处理的字段列表，如果为None则处理所有字段
Returns:
Dict[str, Any]: 空值规范化后的字典
Examples:
>>> data = {"contract_no": "", "product_model": "阀门A", "quantity": None}
>>> normalized = IDGenerator.normalize_empty_values(data, ["contract_no", "quantity"])
>>> print(normalized["contract_no"]) # 输出: "【空白合同号】"
>>> print(normalized["quantity"]) # 输出: "【空白数量】"
"""
if keys is None:
keys = list(data.keys())
normalized = {}
for k in keys:
v = data.get(k, "")
if v is None or v == "":
normalized[k] = cls._get_empty_marker(k)
else:
normalized[k] = str(v).strip()
return normalized
[文档]
@classmethod
def normalize_dataframe_empty_values(cls, df: pd.DataFrame, fields: list[str]) -> pd.DataFrame:
"""规范化DataFrame中指定字段的空值
将DataFrame中指定字段的空值(NaN、None、空字符串)统一替换为格式化的空值标记。
适用于批量数据处理场景。
Args:
df: 需要处理的DataFrame
fields: 需要处理的字段列表
Returns:
pd.DataFrame: 空值规范化后的DataFrame副本
Examples:
>>> 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]) # 输出: "【空白合同号】"
"""
df_clean = df.copy()
for field in fields:
if field not in df_clean.columns:
continue
# 获取该字段的空值标记
empty_marker = cls._get_empty_marker(field)
# 替换空值
df_clean[field] = df_clean[field].fillna(empty_marker)
df_clean[field] = df_clean[field].replace(["", "nan", "None"], empty_marker)
# 确保转换为字符串
df_clean[field] = df_clean[field].astype(str).str.strip()
return df_clean
[文档]
@classmethod
def get_all_unique_key_fields(cls) -> list[str]:
"""获取所有唯一键字段（去重合并，供 import_deduplication 使用）
Returns:
list[str]: 去重后的唯一键字段列表
"""
new_fields = cls._get_new_data_fields()
old_fields = cls._get_old_data_fields()
seen = set()
result = []
for f in new_fields + old_fields:
if f not in seen:
seen.add(f)
result.append(f)
return result