"""质保书/报告中「主要零件名称 + 零件牌号」的可插拔解析器。
设计取向（来自用户，已记入长期记忆）：
- 主键 = 生产令号（公司通用追踪 ID）。
- 权威来源 = 钉钉在线表「加工进度跟踪单」（复合键 生产令号+型号+规格），
每条记录自带 零件名称 / 材质（及尺寸等）。
- 完工产品的记录会被从钉钉源表剪切走，故 provider 必须「自学习即时 DB 化」：
按生产令号查询后把结果落库缓存，保证已完工令号仍可回溯。
- 优先级：低于报告填充引擎主线，列为后续任务。短期用静态 material_grades 兜底。
"""
from __future__ import annotations
import abc
import random
from collections.abc import Callable
from dataclasses import dataclass, field
from pathlib import Path
from certflow.handlers.excel_handler import ExcelHandler
[文档]
@dataclass
class MaterialPart:
"""一个产品的一个主要零件的材质信息（写入质保书材质块）。"""
零件名称: str = ""
材料: str = ""
化学成分: dict[str, float] = field(default_factory=dict) # C/Mn/Si/S/P/Cr/Ni/Mo/V/Cu
力学性能: dict[str, float] = field(default_factory=dict) # ReL/Rm/A(%)/Z(%)
结果: str = "合格"
# —— 溯源字段（REPORT_MATERIAL_BLUEPRINT §2，向后兼容默认）——
来源: str = "cert" # "cert"(合格证手工) / "bom"(BOM派生) / "unknown"
is_virtual: bool = True # 化学成分/力学性能是否虚拟示意值（当前恒 True）
表头外元素: dict[str, float] = field(default_factory=dict) # 不在模板列集、被静默丢弃的元素
来源令号: str = "" # BOM 派生时附生产令号（供批注 [溯源] 使用）
[文档]
class MaterialProvider(abc.ABC):
"""按生产令号解析该产品的主要零件 + 牌号列表。"""
[文档]
@abc.abstractmethod
def resolve(self, 生产令号: str, *, 型号: str = "", 规格: str = "") -> list[MaterialPart]:
"""返回该产品的主要零件材质列表（按报告所需顺序）。"""
raise NotImplementedError
[文档]
class StaticMaterialProvider(MaterialProvider):
"""短期兜底：无真实数据源时的「留空」策略（绝不伪造）。
设计原则（report_fill_service._resolve_materials 注释）：**型号本身不是材质
牌号，绝不把型号填进材质块**。当前没有可靠的「型号→牌号」查表，故无真实牌号时
一律返回空列表，交由填充引擎在该组写「以下空白」，而不是把型号当牌号回填。
"""
def __init__(self, material_grade_service=None):
self._svc = material_grade_service
[文档]
def resolve(
self, 生产令号: str, *, 型号: str = "", 规格: str = "", 零件数: int = 1
) -> list[MaterialPart]:
# 短期兜底：无真实牌号时不伪造（型号不是牌号）。返回空列表 → 材质块留空。
# 若未来 material_grades 提供「型号→牌号」可靠查表，可在此接入，但绝不可回退到型号。
return []
[文档]
class XlsxMaterialProvider(MaterialProvider):
"""（短期真实来源）直接读取「材料进度跟踪单」xlsx 按生产令号解析主要零件+牌号。
对标用户指定的 reports_migration/scratch/一个钉钉文档在线表格的酷应用.js 的
读取逻辑（源表 '加工进度跟踪单'，复合键 生产令号+型号+规格）。本类用离线 xlsx
替代钉钉实时接口，作为 DingTalkMaterialProvider 落地前的真实数据源：
- 精确匹配 生产令号；未命中则退化按 (型号, 规格) 查。
- 返回 MaterialPart 列表（零件名称 + 材料=材质）；质保书材质块只写首零件。
- 未命中返回空列表（引擎留空，绝不伪造）。
"""
def __init__(self, xlsx_path: str, sheet: str = "加工进度跟踪单"):
self._xlsx_path = xlsx_path
self._sheet = sheet
self._by_order: dict[str, list] = {}
self._by_model: dict[str, list] = {}
self._loaded = False
def _ensure_loaded(self) -> None: # noqa: C901
if self._loaded:
return
wb = ExcelHandler.load_workbook(Path(self._xlsx_path), data_only=True, read_only=False)
ws = wb[self._sheet]
rows = list(ws.iter_rows(values_only=True))
# 真实表头不在第1行（第1行是标题），定位含“生产令号”的行作为表头
def norm(s):
return str(s).replace("\n", "").replace(" ", "")
header = None
for r in rows[:10]:
if any(h and "生产令号" in str(h) for h in r):
header = r
break
if header is None:
header = rows[0] if rows else []
idx = {}
for col in ("生产令号", "型号", "规格", "零件名称", "材质"):
for i, h in enumerate(header):
if h and norm(col) in norm(h):
idx[col] = i
break
start = (rows.index(header) + 1) if header in rows else 1
for row in rows[start:]:
order = (
str(row[idx["生产令号"]]).strip()
if "生产令号" in idx and len(row) > idx["生产令号"]
else ""
)
model = (
str(row[idx["型号"]]).strip() if "型号" in idx and len(row) > idx["型号"] else ""
)
spec = str(row[idx["规格"]]).strip() if "规格" in idx and len(row) > idx["规格"] else ""
pname = (
str(row[idx["零件名称"]]).strip()
if "零件名称" in idx and len(row) > idx["零件名称"]
else ""
)
mat = str(row[idx["材质"]]).strip() if "材质" in idx and len(row) > idx["材质"] else ""
if not pname and not mat:
continue
part = MaterialPart(零件名称=pname, 材料=mat)
if order:
self._by_order.setdefault(order, []).append(part)
key = (model, spec)
if model:
self._by_model.setdefault(key, []).append(part)
self._loaded = True
[文档]
def resolve(
self, 生产令号: str, *, 型号: str = "", 规格: str = "", 零件数: int = 1
) -> list[MaterialPart]:
self._ensure_loaded()
if 生产令号 and 生产令号 in self._by_order:
return self._by_order[生产令号][: max(1, 零件数)]
key = (型号 or "", 规格 or "")
if key in self._by_model:
return self._by_model[key][: max(1, 零件数)]
return []
[文档]
class DingTalkMaterialProvider(MaterialProvider):
"""（后续任务，低优先级）钉钉生产进度表驱动 + 自学习 DB 化。
实现要点：
1. 用 reports_migration/scratch/一个钉钉文档在线表格的酷应用.js 的读取逻辑
（源表 '加工进度跟踪单'，COMPOSITE_KEY=['生产令号','型号','规格']）查询某 生产令号；
2. 提取每个零件的 零件名称 + 材质（及尺寸等），构成 MaterialPart 列表；
3. 首次查询后写入本地 DB（按 生产令号 缓存），已完工被剪切的记录仍可回溯；
4. 后续查询优先命中本地缓存，未命中再走钉钉实时查并回写。
当前为桩，调用即抛 NotImplementedError。
"""
def __init__(self, db_session=None, dingtalk_client=None):
self._db = db_session
self._client = dingtalk_client
[文档]
def resolve(self, 生产令号: str, *, 型号: str = "", 规格: str = "") -> list[MaterialPart]:
raise NotImplementedError(
"DingTalkMaterialProvider 为后续任务：需接入钉钉读取 + 自学习 DB 化"
)
# --------------------------------------------------------------------------- #
# 自定义函数机制：把任意 callable 注册为材质解析器
# --------------------------------------------------------------------------- #
#
# 设计目的：材质块的取值来源是可插拔的（钉钉 / xlsx / 函数）。当标准数据源缺失、
# 或需要「动态请求得到随机/计算函数值」来压测排版时，可用自定义函数驱动。
#
# 注册表用法：
# from certflow.services.report_migration.material_provider import (
# register_material_function, FunctionMaterialProvider,
# )
# def my_resolver(生产令号, *, 型号="", 规格="", 零件数=1):
# return [MaterialPart(零件名称="阀体", 材料="WCB")]
# register_material_function("my", my_resolver)
# # 配置 report.material_provider: function:my 即可启用
#
# 函数签名契约：``(生产令号, *, 型号="", 规格="", 零件数=1) -> list[MaterialPart]``
# （零件数缺省时返回至少 1 个零件；返回空列表 = 该组留空，由引擎写「以下空白」）。
_MATERIAL_FUNCTIONS: dict[str, Callable] = {}
[文档]
def register_material_function(name: str, func: Callable) -> None:
"""注册一个具名材质解析函数，供配置 ``report.material_provider: function:<name>`` 调用。"""
_MATERIAL_FUNCTIONS[name] = func
[文档]
def get_material_function(name: str) -> Callable | None:
"""按名取回已注册的材质解析函数；未注册返回 ``None``。"""
return _MATERIAL_FUNCTIONS.get(name)
# 内置示例：随机真实牌号解析器（用于排版压测 / 原型验证「动态随机值」）。
_RANDOM_GRADES = [
"WCB",
"CF8",
"CF8M",
"LCB",
"A105",
"25",
"1Cr13",
"2Cr13",
"304",
"316",
"F304",
"F316",
"QT450",
"ZG230-450",
"20",
"Q235",
]
_RANDOM_PART_NAMES = ["阀体", "阀杆", "启闭件"]
[文档]
def random_material_resolver(
生产令号: str, *, 型号: str = "", 规格: str = "", 零件数: int = 1
) -> list[MaterialPart]:
"""返回随机真实牌号的材质零件列表（仅用于排版压测，不入正式报告）。
每个零件名取自 阀体/阀杆/启闭件（超过 3 个则补 零件N），牌号随机选取。
"""
n = max(1, int(零件数))
parts: list[MaterialPart] = []
for i in range(n):
name = _RANDOM_PART_NAMES[i] if i < len(_RANDOM_PART_NAMES) else f"零件{i + 1}"
parts.append(MaterialPart(零件名称=name, 材料=random.choice(_RANDOM_GRADES)))
return parts
[文档]
class FunctionMaterialProvider(MaterialProvider):
"""把任意已注册/传入的 callable 当作材质解析器（可返回随机/计算值）。
用于「附加自定义函数动态驱动材质块」：例如 ``random_material_resolver``
或用户自注册的 ``function:<name>``。
"""
def __init__(self, func: Callable):
self._func = func
[文档]
def resolve(
self, 生产令号: str, *, 型号: str = "", 规格: str = "", 零件数: int = 1
) -> list[MaterialPart]:
result = self._func(生产令号, 型号=型号, 规格=规格, 零件数=零件数)
return result or []