certflow.services.bom_part_classifier 源代码

"""BOM 零件归类器(阶段 C:报告生成时按生产令号带出主材)

把 ``list[BomMaterial]`` 按「零件名称 → 报告主材位」关键词映射归类为
``{主材位: [BomMaterial, ...]}`` + 待核对明细 ``[BomMaterial, ...]``。

纯逻辑、无 IO、可单测;关键词映射由 config 驱动(见 ``report.bom_report``),
不在代码里硬编码,便于用户维护(阶段 C 原则 5 / §3)。

归类规则(详见 docs/PATCH_LOG.md 补丁 14「阶段 C:BOM 材质集成到报告生成」):
- 遍历 part_name,按 ``part_keywords`` 优先级(自上而下)匹配**首个命中**关键词
  → 归入对应主材位;优先级保证「阀体盖」不会误归 body(C2.3)。
- 同一主材位命中多个零件(多对一)时按 ``multi_match_strategy``:
  - ``first``     → 取 seq 最小(首条)落主材位;
  - ``detail``    → 全部进待核对明细,不自动填;
  - ``same_grade``→ 牌号完全一致(含均无牌号,无歧义)取首条落主材位,
    牌号不一致(有歧义)才全部进待核对明细(默认,兼顾「避免错误回填」
    与「同牌号零件正常填入」,如阀体+阀盖均为 35 应填阀体而非丢明细)。
- 未命中任何关键词的零件 → 进待核对明细(``fallback_to_detail`` 开关)。
"""

from __future__ import annotations

from dataclasses import dataclass, field
from typing import Any


[文档] @dataclass class BomClassifyResult: """BOM 归类结果。 Attributes: slots: 主材位 → 命中该位的 BOM 零件列表(已按 seq 排序,每个位 0~N 条)。 detail: 待核对明细(未命中关键词 + 多对一被剔除的零件),不落报告主材位。 """ slots: dict[str, list[Any]] = field(default_factory=dict) detail: list[Any] = field(default_factory=list)
# 默认关键词映射(config 缺失/损坏时回退,保证分类器始终可用)。 # key 为报告主材位标识;value 为该位的关键词列表(命中其一即归类,自上而下优先)。 DEFAULT_BOM_REPORT_CFG: dict[str, Any] = { "part_keywords": { "body": ["阀体", "阀盖", "body"], "stem": ["阀杆", "stem"], "disc": ["阀瓣", "阀芯", "disc"], "gate": ["闸板", "闸阀板", "gate"], "seat": ["阀座", "seat"], "bolt": ["螺栓", "螺柱", "bolt"], }, # 未命中关键词的零件 → 进待核对明细(不强行落主材位) "fallback_to_detail": True, # 同主材位多对一策略:first(取首条) / detail(全部进待核对) / # same_grade(牌号一致取首条,不一致进待核对,默认) "multi_match_strategy": "same_grade", # 报告材质块有空行时,是否用「其它已找到的 BOM 零件」补满空行(保留真实零件名, # 不冒充阀体/阀杆/启闭件),默认开启。补位的零件仍保留在待核对明细供人工核对。 "fill_empty_rows": True, } def _part_seq(p: Any) -> int: """安全取零件 seq(缺省 0),用于稳定排序。""" return getattr(p, "seq", 0) or 0 def _part_name(p: Any) -> str: """安全取零件名称(去空白)。""" return (getattr(p, "part_name", None) or "").strip() def _part_grade(p: Any) -> str: """安全取零件牌号(去空白)。""" return (getattr(p, "material_grade", None) or "").strip()
[文档] def classify(parts: list[Any], cfg: dict[str, Any] | None = None) -> BomClassifyResult: # noqa: C901 """把 BOM 零件列表按 config 关键词映射归类为主材位 + 待核对明细。 入参 ``parts`` 为鸭子类型:只需具备 ``part_name``(str)、``seq``(int,可选)、 ``material_grade``(可选)属性即可(``BomMaterial`` 或测试桩均可)。 Args: parts: BOM 零件列表(同生产令号,已按 seq 排序或无序均可)。 cfg: 归类配置(含 ``part_keywords`` / ``fallback_to_detail`` / ``multi_match_strategy``);缺省回退 ``DEFAULT_BOM_REPORT_CFG``。 Returns: BomClassifyResult:``slots`` 为主材位→零件列表(每个位已 <=1 条,除非 strategy=first),``detail`` 为待核对零件列表。 Examples: >>> class P: # 测试桩 ... def __init__(self, name, seq=0, grade=""): ... self.part_name = name; self.seq = seq; self.material_grade = grade >>> r = classify([P("阀体A", 1, "WCB"), P("法兰盘", 2, "Q235")]) >>> r.slots["body"][0].material_grade 'WCB' >>> [p.part_name for p in r.detail] ['法兰盘'] """ cfg = cfg or DEFAULT_BOM_REPORT_CFG part_keywords = cfg.get("part_keywords") or DEFAULT_BOM_REPORT_CFG["part_keywords"] fallback = cfg.get("fallback_to_detail", True) strategy = str(cfg.get("multi_match_strategy") or "detail").strip().lower() slots: dict[str, list[Any]] = {k: [] for k in part_keywords} detail: list[Any] = [] for p in sorted(parts, key=_part_seq): name = _part_name(p) matched_slot: str | None = None # 优先级自上而下:首个命中关键词即归类(避免「阀体盖」误归 body) for slot, keywords in part_keywords.items(): for kw in keywords: if kw and kw in name: matched_slot = slot break if matched_slot: break if matched_slot: slots[matched_slot].append(p) elif fallback: detail.append(p) # 多对一处理: # detail → 同主材位多零件全部进待核对明细(不自动填); # first → 仅取 seq 最小(首条)落主材位,其余静默丢弃(不进明细); # same_grade → 牌号完全一致(含均无牌号,无歧义)取首条落主材位, # 牌号不一致(有歧义)才全部进待核对明细。 if strategy == "detail": for slot, lst in list(slots.items()): if len(lst) > 1: detail.extend(lst) slots[slot] = [] elif strategy == "first": for slot, lst in list(slots.items()): if len(lst) > 1: slots[slot] = [lst[0]] elif strategy == "same_grade": for slot, lst in list(slots.items()): if len(lst) > 1: grades = {_part_grade(p) for p in lst} non_empty = [g for g in grades if g] if len(non_empty) <= 1: # 牌号一致(或无牌号)→ 无歧义,取首条落主材位, # 其余零件进明细(可被报告空行补位 / 待核对) slots[slot] = [lst[0]] detail.extend(lst[1:]) else: # 牌号不一致 → 存在歧义,全部进待核对明细 detail.extend(lst) slots[slot] = [] return BomClassifyResult(slots=slots, detail=detail)