"""BOM 零件牌号服务
从「生产-下料计划 TAO」目录的材料清单（.xls/.xlsx）解析零件牌号，落库到
``bom_materials`` 表，并提供按 生产令号 查询与超链接索引生成能力。
文件结构异构，解析器自适应两种常见表头：
- 12 列版（.xlsx）：序号/代号/名称/数量/单位/材料/图纸尺寸/毛坯尺寸/毛坯单量/材料型态/备注
- 4 列版（.xls）：零件名称/材料/备注/方向
通过「行内同时含 名称类列 与 材料类列」定位表头，再按列名映射取数，不依赖固定列序。
"""
from __future__ import annotations
import math
import os
import re
import unicodedata
from enum import StrEnum
from pathlib import Path
from typing import Any
from urllib.parse import quote
import pandas as pd
from sqlalchemy.orm import Session
from certflow.handlers.excel_handler import ExcelHandler
from certflow.models.bom_material import (
MATERIAL_SOURCE_AUTO_LEARN,
MATERIAL_SOURCE_CHOICES,
MATERIAL_SOURCE_IMPORT,
MATERIAL_SOURCE_MANUAL,
BomMaterial,
)
from certflow.utils.logger import logger
# 表头别名：零件名称列 / 材料牌号列 / 代号列 / 数量列 / 形态列 / 序号列
_NAME_ALIASES = {"名称", "零件名称", "零件"}
_MAT_ALIASES = {"材料", "材料牌号", "材质", "牌号"}
_CODE_ALIASES = {"代号", "图号"}
_QTY_ALIASES = {"数量"}
_FORM_ALIASES = {"材料型态", "方向", "形态"}
_SEQ_ALIASES = {"序号", "项次", "no", "项"}
# 停止读取的条件：连续多少行 零件名称为空 即判定 BOM 结束
_MAX_BLANK_ROWS = 3
# ============================================================================
# 牌号归一化 / 分类（§19-L0/L1 共享工具，供盘点、数据治理、覆盖判定复用）
# ============================================================================
# 零宽/不可见字符（归一化前先剥离）
_INVISIBLE = (
"\u200b\u200c\u200d\u200e\u200f\u00ad\u2060\ufeff"
"\u202a\u202b\u202c\u202d\u202e\u2028\u2029"
"\u3000" # 全角空格（转半角空格后再 trim）
)
_DIM_RE = re.compile(r"\(?L[=＝]?\d+\s*D?[=＝]?\d*\)?|\(?D[=＝]?\d+\)?|L\s*=\s*\d+", re.IGNORECASE)
_STATE_SUFFIX = (
"组焊件",
"组焊",
"焊后",
"管",
"法兰第一系列",
"热处理",
"调质",
"正火",
"无缝钢",
"无缝",
)
_CJK_RE = re.compile(r"[一-鿿]") # 含中文 → 描述性文字（铝合金/不锈钢…），非牌号
def split_grade_variants(norm: str) -> list[str]:
"""把斜杠分隔的「可选/双牌号」标注拆成各候选牌号，逐个 base_grade。
BOM 中常以 ``A105/45``（A105 或 45）、``CF8/304``、``35/铝合金`` 形式标注可选材料。
拆分后每个候选都视为真实牌号参与覆盖统计；含中文的候选（如「铝合金」）不是牌号，剔除。
例：
A105/45 -> ["a105", "45"] （45 号钢，碳钢）
CF8/304 -> ["cf8", "304"] （均不锈钢）
35/铝合金 -> ["35"] （铝合金 含中文剔除）
/ -> [] （空，不计）
"""
parts = re.split(r"[／/]", norm)
out: list[str] = []
for p in parts:
bg = base_grade(p.strip())
if bg and not _CJK_RE.search(bg): # 含中文的候选（如 铝合金）剔除
out.append(bg)
return out
def normalize_grade(raw: str | None) -> str:
"""归一化牌号：去不可见字符 → NFKC 全角转半角 → 折叠空白 → 去首尾空格。
用于消除「前后空格 / 全角半角 / 不可见字符」造成的同牌号多值差异。
"""
if raw is None:
return ""
if not isinstance(raw, str): # 数值型牌号（如 20）归一成字符串
raw = str(raw)
s = raw
for ch in _INVISIBLE:
s = s.replace(ch, "")
s = unicodedata.normalize("NFKC", s)
s = s.replace("（", "(").replace("）", ")")
s = re.sub(r"\s+", " ", s) # 折叠所有空白（含制表/换行）为单空格
s = s.lower() # ASCII 大小写归并（16Mn/16MN/16mn 视为同牌号）
return s.strip()
def base_grade(norm: str) -> str:
"""提取基础牌号：去括号尺寸/状态标注、去组焊等后缀，揭示真实唯一牌号数。
例：WCB(L=180 D=135)→WCB；25+20+Q235B组焊件→25+20+Q235B（复合保留，因本质是多材组焊）
注意：复合牌号（含+号）本身即一种独立「组合材料」，不强行拆解，仅去后缀/尺寸。
"""
s = norm
s = re.sub(r"[（(][^()（）]*[)）]", "", s) # 去括号内容
s = re.sub(r"[（(][^()（）]*$", "", s) # 未闭合左括号残尾
s = _DIM_RE.sub("", s) # 去尺寸标注
s = re.sub(r"\s+", " ", s).strip()
for suf in _STATE_SUFFIX: # 去状态/形态后缀
if s.endswith(suf):
s = s[: -len(suf)].strip()
s = re.sub(r"[#号]$", "", s) # 去末尾 # / 号（20#→20）
return s.strip()
def classify_grade(grade: str, category_medium: str | None = None) -> tuple[str, str]:
"""把牌号归到 §17 显示剖面四类 + 数据缺失；优先用库内 category_medium。
Returns:
(bucket, category_medium_best_effort)
bucket ∈ {P_carbon, P_stainless, P_alloy, P_alloy_special, 数据缺失(未知), 数据缺失(空)}
缺失 category_medium 时用牌号文本启发式（粗，仅用于缺库时的密度估计）。
"""
if not grade:
return "数据缺失(空)", ""
cm = (category_medium or "").strip()
if cm:
if "不锈钢" in cm:
return "P_stainless", cm
if "合金" in cm or "Cr" in cm:
return "P_alloy", cm
if "碳钢" in cm or "铸铁" in cm or "铜" in cm:
return "P_carbon", cm
g = grade.upper()
# 牌号文本启发式（仅缺库时兜底）
if any(
k in g
for k in (
"CF",
"304",
"316",
"347",
"321",
"1CR",
"2CR",
"3CR",
"06CR",
"022CR",
"0CR17NI4",
"F304",
"F316",
)
):
return "P_stainless", "奥氏体不锈钢(启发式)"
if g.startswith("F1") or g.startswith("F2") or g.startswith("F9"):
return "P_alloy", "合金钢(启发式)"
if any(k in g for k in ("WC", "WCB", "WCC", "LCB", "LCC", "A105", "A217", "A182")):
return "P_alloy", "合金钢(启发式)"
if any(
k in g
for k in (
"Q",
"HT",
"ZCU",
"QT",
"ZG",
"20",
"25",
"35",
"45",
"16MN",
"20G",
"Q235",
"Q355",
)
):
return "P_carbon", "碳钢/铸铁(启发式)"
return "数据缺失(未知)", ""
# ============================================================================
# 牌号解析层（§19-L0/L1 扩展）：别名归并 / 组合拆分 / 非牌号过滤 / 非金属豁免
# —— 配置驱动（config/grade_aliases.yaml），用户可控，不引入新标准数据 ——
# ============================================================================
# 内置默认（文件缺失或损坏时回退，保证服务始终可用）
_DEFAULT_ALIASES: dict[str, str] = {
"q335b": "q355b",
"q355bii": "q355b",
"q355": "q355b",
"20": "20#",
"16mniii": "16mn",
}
_DEFAULT_NON_GRADE_PATTERNS: list[str] = [r"^dn\d", r"^pn\d", r"尺寸", r"规格"]
_DEFAULT_EXEMPT: list[str] = [
"ptfe",
"nbr",
"epdm",
"fpm",
"fkm",
"vmq",
"橡胶",
"塑料",
"尼龙",
"四氟",
"硅胶",
]
# 运行期配置（模块加载时读一次；改配置需重启进程）
_GRADE_ALIASES: dict[str, str] = dict(_DEFAULT_ALIASES)
_NON_GRADE_RES: list[re.Pattern[str]] = [
re.compile(p, re.IGNORECASE) for p in _DEFAULT_NON_GRADE_PATTERNS
]
_NONMETALLIC_EXEMPT: set[str] = set(_DEFAULT_EXEMPT)
def _load_grade_alias_config() -> None:
"""从 config/grade_aliases.yaml 载入别名/非牌号/豁免配置，覆盖内置默认。
文件缺失或损坏时静默回退到内置默认（服务不崩）。
"""
global _GRADE_ALIASES, _NON_GRADE_RES, _NONMETALLIC_EXEMPT
cfg_path = Path(__file__).resolve().parents[3] / "config" / "grade_aliases.yaml"
try:
import yaml
except ImportError:
logger.warning("未安装 pyyaml，牌号别名配置回退内置默认")
return
try:
with cfg_path.open(encoding="utf-8") as f:
data = yaml.safe_load(f) or {}
except FileNotFoundError:
return
except Exception as e: # noqa: BLE001 - 配置错误不应中断服务
logger.warning(f"读取 grade_aliases.yaml 失败，回退默认: {e}")
return
aliases = dict(_DEFAULT_ALIASES)
if isinstance(data.get("grade_aliases"), dict):
for k, v in data["grade_aliases"].items():
nk, nv = normalize_grade(k), normalize_grade(v)
if nk and nv:
aliases[nk] = nv
non_grade = list(_DEFAULT_NON_GRADE_PATTERNS)
if isinstance(data.get("non_grade_patterns"), list):
non_grade = [str(p) for p in data["non_grade_patterns"]]
exempt = set(_DEFAULT_EXEMPT)
if isinstance(data.get("nonmetallic_exempt"), list):
exempt = {normalize_grade(x) for x in data["nonmetallic_exempt"] if x}
_GRADE_ALIASES = aliases
_NON_GRADE_RES = [re.compile(p, re.IGNORECASE) for p in non_grade]
_NONMETALLIC_EXEMPT = exempt
_load_grade_alias_config()
def canonical_grade(raw: str | None) -> str:
"""把原始牌号归并到规范牌号（别名解析，可链式）。
流程：normalize_grade → 查别名表（链式解析，防环）→ 返回规范归一化牌号。
仅做「同名异写 → 已有规范牌号」的归并；非牌号/豁免判定请用独立函数。
例：
"Q335B" -> "q355b"（笔误）
"20" -> "20#"（裸写归并）
"16MnIII"-> "16mn"（质量等级后缀）
"WCB" -> "wcb"（无别名，原样归一化）
"""
norm = normalize_grade(raw)
if not norm:
return ""
target = _GRADE_ALIASES.get(norm, norm)
seen: set[str] = set()
while target in _GRADE_ALIASES and target != norm and target not in seen:
seen.add(target)
target = _GRADE_ALIASES.get(target, target)
return target
def split_combo(norm: str) -> list[str]:
"""把 '+' 连接的复合牌号拆成各组成材料（组合件分别判覆盖）。
例：
"q355b+20" -> ["q355b", "20"]
"25+20+q235b" -> ["25", "20", "q235b"]
"WCB" -> ["wcb"] （无 +，单元素）
含中文的组分（如「铝合金」）剔除，与 split_grade_variants 一致。
"""
if "+" not in norm:
return [norm]
out: list[str] = []
for part in re.split(r"\s*\+\s*", norm):
bg = base_grade(part.strip())
if bg and not _CJK_RE.search(bg):
out.append(bg)
return out
def is_likely_grade(norm: str) -> bool:
"""该归一化串是否「像牌号」；命中非牌号模式（尺寸/规格标注）返回 False。
用于覆盖判定中过滤 dn50/pn320 这类尺寸标注，避免被误计为材质缺口。
"""
if not norm:
return False
return not any(rx.search(norm) for rx in _NON_GRADE_RES)
def is_nonmetallic_exempt(norm: str) -> bool:
"""该归一化串是否属于非金属豁免（本就无化学成分表，不算缺失）。"""
return norm in _NONMETALLIC_EXEMPT
def grade_coverage(session: Session, xlsx_grades: set[str] | None = None) -> dict[str, Any]:
"""盘点 BOM 牌号在 material_grades 的覆盖情况（§20 Step2 Diff + 数据源交叉）。
在归一化/拆分基础上叠加解析层：
- 别名归并：``q335b`` / ``20`` 等经 ``canonical_grade`` 指向库内已有规范牌号；
- 组合拆分：``q355b+20`` 拆成各组分，全部命中库才算覆盖；
- 非牌号过滤：``dn50 pn320`` 等尺寸标注不计入缺失；
- 非金属豁免：``ptfe`` 等视为已豁免，不算缺失。
Args:
session: 数据库会话（已绑定 bom_materials / material_grades）。
xlsx_grades: 可选，材质数据源.xlsx 中已有的牌号归一化集合；
传入后可区分「缺失但源已含（重导即恢复）」vs「源也无（需手填）」。
Returns:
{
"bom_total_rows", "bom_distinct_base",
"covered": [base_grade...], # 覆盖（含别名/组合命中）
"missing": [{base, count, bucket, category_medium,
in_material_db, in_xlsx_source, canon}], # 真正缺失（需补录）
"missing_count", "covered_count",
"missing_in_xlsx": [...], "missing_not_in_xlsx": [...],
"non_grade": [base...], "exempt": [base...], # 过滤/豁免（非缺口）
"coverage_map": {base: {covered, canon, resolution}},
"resolution_stats": {resolution: count}, # 各解析路径命中记录数
}
resolution ∈ {direct, alias, combo, non_grade, exempt, missing}
"""
from certflow.models.material_grade import MaterialGrade
rows = session.query(BomMaterial.material_grade, BomMaterial.production_order_no).all()
base_counter: dict[str, int] = {}
for grade, _order in rows:
norm = normalize_grade(grade)
# 斜杠分隔的「可选/双牌号」标注拆成各候选，分别计数（A105/45 → a105 + 45）
for bg in split_grade_variants(norm):
base_counter[bg] = base_counter.get(bg, 0) + 1
db_grades = {normalize_grade(g) for (g,) in session.query(MaterialGrade.grade).all() if g}
xlsx_set = {normalize_grade(g) for g in (xlsx_grades or set())}
covered: list[str] = []
missing: list[dict[str, Any]] = []
non_grade: list[str] = []
exempt: list[str] = []
coverage_map: dict[str, dict[str, Any]] = {}
resolution_stats: dict[str, int] = {}
missing_in_xlsx: list[str] = []
missing_not_in_xlsx: list[str] = []
for bg, cnt in sorted(base_counter.items(), key=lambda kv: (-kv[1], kv[0])):
# 非牌号（尺寸/规格标注）：过滤，不计入缺口
if not is_likely_grade(bg):
non_grade.append(bg)
resolution_stats["non_grade"] = resolution_stats.get("non_grade", 0) + cnt
continue
# 非金属豁免：无化学成分表，不算缺失
if is_nonmetallic_exempt(bg):
exempt.append(bg)
resolution_stats["exempt"] = resolution_stats.get("exempt", 0) + cnt
continue
# 组合拆分 + 别名归并 → 全部组分命中库才算覆盖
comps = split_combo(bg)
canons = [canonical_grade(c) for c in comps]
in_db = all(c in db_grades for c in canons)
if in_db:
covered.append(bg)
if len(comps) > 1:
resolution = "combo"
elif canons[0] != comps[0]:
resolution = "alias"
else:
resolution = "direct"
resolution_stats[resolution] = resolution_stats.get(resolution, 0) + cnt
else:
bucket, cm = classify_grade(bg)
in_xlsx = bg in xlsx_set
missing.append(
{
"base": bg,
"count": cnt,
"bucket": bucket,
"category_medium": cm,
"in_material_db": in_db,
"in_xlsx_source": in_xlsx,
"canon": canons,
}
)
(missing_in_xlsx if in_xlsx else missing_not_in_xlsx).append(bg)
resolution_stats["missing"] = resolution_stats.get("missing", 0) + cnt
coverage_map[bg] = {
"covered": in_db,
"canon": canons,
"resolution": (
"combo" if len(comps) > 1 else ("alias" if canons[0] != comps[0] else "direct")
),
}
return {
"bom_total_rows": len(rows),
"bom_distinct_base": len(base_counter),
"covered": covered,
"missing": missing,
"covered_count": len(covered),
"missing_count": len(missing),
"missing_in_xlsx": missing_in_xlsx,
"missing_not_in_xlsx": missing_not_in_xlsx,
"non_grade": non_grade,
"exempt": exempt,
"coverage_map": coverage_map,
"resolution_stats": resolution_stats,
}
def _unc_uri(p: str) -> str:
"""把本地/UNC 路径转成 Excel 可点击的 file:// 超链接目标。"""
p = p.replace("\\", "/")
if p.startswith("//"): # UNC: \\server\share\...
segs = p[2:].split("/")
return "file://" + "/".join(quote(s, safe="") for s in segs)
segs = p.split("/")
return "file:///" + "/".join(quote(s, safe="") for s in segs)
def extract_production_order_no(filename: str) -> str:
"""从材料清单文件名提取 生产令号。
生产令号形如 ``202511008-1`` / ``202410004WX-1`` / ``202412058WX250814FX251215``
（数字 + 可选字母/数字后缀 + 可选 ``-N``）。文件名中还常拼接型号、``：``、空格等，
故取「开头连续的数字/字母/连字符」段；遇到中文、全角冒号、空格即止。
无法匹配时退回「首个空白前 token」。
例：
"202511008-1 125EFvt...材料清单.xlsx" -> "202511008-1"
"20221011 20221012：EF...DN150.xls" -> "20221011"
"20220104FX251215：EFvt（fcg1780）D7PY-320…" -> "20220104FX251215"
"""
base = os.path.splitext(os.path.basename(filename))[0].strip()
m = re.match(r"[0-9A-Za-z\-]+", base)
if m:
return m.group(0)
toks = base.split()
return (toks[0] if toks else base).strip()
def _read_rows(path: str) -> list[list[Any]]:
"""读取工作表为二维列表（空单元格为 None），兼容 .xlsx 与 .xls。"""
low = path.lower()
if low.endswith(".xlsx"):
wb = ExcelHandler.load_workbook(Path(path), data_only=True, read_only=True)
ws = wb.active
rows = [[c.value for c in row] for row in ws.iter_rows()]
wb.close()
return rows
# .xls（及兜底）：pandas + calamine
df = pd.read_excel(path, engine="calamine", header=None)
data = df.where(pd.notnull(df), None).values.tolist()
# pandas 对「纯数值列」用 where(..., None) 无法真正置为 None（None 会被还原成 NaN），
# 残留的 float('nan') 后续经 str() 会变成字符串 'nan'。统一把浮点 NaN 收敛为 None。
return [[None if (isinstance(v, float) and math.isnan(v)) else v for v in row] for row in data]
def _read_all_sheets(path: str) -> list[tuple[str, list[list[Any]]]]:
"""读取文件全部工作表，返回 [(sheet_name, rows)]。
空单元格收敛为 None，并将 pandas 残留的 float('nan') 也归一为 None。
材料清单文件常含多个工作表（简表 + 正式材料申请单 + 汇总表 等），
需逐一尝试定位表头，故提供「读全部 sheet」入口。
"""
low = path.lower()
result: list[tuple[str, list[list[Any]]]] = []
if low.endswith(".xlsx"):
wb = ExcelHandler.load_workbook(Path(path), data_only=True, read_only=True)
try:
for ws in wb.worksheets:
rows = [[c.value for c in row] for row in ws.iter_rows()]
rows = [
[None if (isinstance(v, float) and math.isnan(v)) else v for v in row]
for row in rows
]
result.append((ws.title, rows))
finally:
wb.close()
return result
# .xls（及兜底）：pandas + calamine，读全部 sheet
sheets = pd.read_excel(path, engine="calamine", sheet_name=None, header=None)
for name, df in sheets.items():
data = df.where(pd.notnull(df), None).values.tolist()
rows = [
[None if (isinstance(v, float) and math.isnan(v)) else v for v in row] for row in data
]
result.append((str(name), rows))
return result
def _detect_header(rows: list[list[Any]]) -> tuple[int, dict[str, int]]: # noqa: C901
"""定位表头行并返回 列名->列索引 映射。找不到返回 (-1, {})。"""
for ri, row in enumerate(rows):
cells = {str(c).strip() for c in row if c not in (None, "")}
has_name = any(c in _NAME_ALIASES for c in cells)
has_mat = any(c in _MAT_ALIASES for c in cells)
if has_name and has_mat:
colmap: dict[str, int] = {}
for ci, c in enumerate(row):
if c in (None, ""):
continue
name = str(c).strip()
if name in _NAME_ALIASES:
colmap.setdefault("name", ci)
elif name in _MAT_ALIASES:
colmap.setdefault("material", ci)
elif name in _CODE_ALIASES:
colmap.setdefault("code", ci)
elif name in _QTY_ALIASES:
colmap.setdefault("qty", ci)
elif name in _FORM_ALIASES:
colmap.setdefault("form", ci)
elif name in _SEQ_ALIASES:
colmap.setdefault("seq", ci)
if "name" in colmap and "material" in colmap:
return ri, colmap
return -1, {}
def parse_bom_file(path: str) -> list[dict[str, Any]]: # noqa: C901
"""解析单个材料清单文件，返回该 BOM 的零件字典列表。
每个字典含：production_order_no, seq, part_no, part_name, material_grade,
quantity, material_form, source_file。解析失败返回空列表。
"""
try:
sheets = _read_all_sheets(path)
except Exception as e: # noqa: BLE001 - 单文件失败不应中断整批导入
logger.warning(f"读取失败跳过: {path} ({e})")
return []
# 文件可能含多个工作表（简表 / 正式材料申请单 / 汇总表 等）。
# 遍历所有 sheet，对每个定位表头，选取「最完整」的正式材料单解析：
# 含 代号/数量 列者优先（完整材料申请单的特征），数据行数作为次要排序键。
candidates = []
for sname, srows in sheets:
h, cmap = _detect_header(srows)
if h < 0:
continue
n = sum(
1 for r in srows[h + 1 :] if len(r) > cmap["name"] and r[cmap["name"]] not in (None, "")
)
score = 0
if "code" in cmap:
score += 3
if "qty" in cmap:
score += 2
if "seq" in cmap:
score += 1
if "form" in cmap:
score += 1
candidates.append((score, n, sname, srows, h, cmap))
if not candidates:
logger.warning(f"未定位表头，跳过: {path}")
return []
candidates.sort(key=lambda c: (c[0], c[1]), reverse=True)
_score, _n, _sname, rows, hdr_row, colmap = candidates[0]
if len(candidates) > 1:
logger.info(
f"多工作表，选用 '{_sname}'（score={_score}, 数据行={_n}）而非其余 "
f"{len(candidates) - 1} 个: {path}"
)
pon = extract_production_order_no(path)
name_c = colmap["name"]
mat_c = colmap["material"]
code_c = colmap.get("code")
qty_c = colmap.get("qty")
form_c = colmap.get("form")
seq_c = colmap.get("seq")
out: list[dict[str, Any]] = []
blank = 0
seq_counter = 0
for row in rows[hdr_row + 1 :]:
if name_c >= len(row):
blank += 1
if blank >= _MAX_BLANK_ROWS:
break
continue
part_name = row[name_c]
if part_name in (None, ""):
blank += 1
if blank >= _MAX_BLANK_ROWS:
break
continue
blank = 0
part_name = str(part_name).strip()
def _cell(col, row=row):
if col is None or col >= len(row):
return None
v = row[col]
if v is None or v == "":
return None
if isinstance(v, float) and math.isnan(v):
return None
return str(v).strip()
material = _cell(mat_c)
seq_counter += 1
seq_val = _cell(seq_c)
try:
seq_val = int(float(seq_val)) if seq_val is not None else seq_counter
except (TypeError, ValueError):
seq_val = seq_counter
out.append(
{
"production_order_no": pon,
"seq": seq_val,
"part_no": _cell(code_c),
"part_name": part_name,
"material_grade": material,
"material_source": MATERIAL_SOURCE_IMPORT,
"quantity": _cell(qty_c),
"material_form": _cell(form_c),
"source_file": path,
}
)
return out
[文档]
class BomImportMode(StrEnum):
"""BOM 导入模式（UI 四选一，决定旧数据如何处理）。"""
CLEAR = "clear" # 清空重导（默认，幂等、无重复）
APPEND = "append" # 不清空直接追加（重复导入同文件会产生重复行）
REFRESH = "refresh" # 按文件刷新：已导入文件删旧行重插，新文件追加
SKIP = "skip" # 增量跳过已导入：只导未入库文件，已存在跳过（最安全/最快）
[文档]
@classmethod
def from_clear_existing(cls, clear_existing: bool | None) -> BomImportMode:
"""兼容旧 ``clear_existing`` 布尔参数：True→CLEAR，False→APPEND。"""
return cls.CLEAR if (clear_existing is None or clear_existing) else cls.APPEND
def _ingest_file(
session: Session,
fp: str,
mode: BomImportMode,
existing_files: set[str],
) -> str:
"""判定单文件导入动作：``new`` / ``refresh`` / ``skip``。
- ``skip`` 模式且文件已入库 → 跳过；
- ``refresh`` 模式且文件已入库 → 删其旧行（返回 ``refresh``）；
- 其余 → 新写入（``new``）。
"""
if mode is BomImportMode.SKIP and fp in existing_files:
return "skip"
if mode is BomImportMode.REFRESH and fp in existing_files:
session.query(BomMaterial).filter(BomMaterial.source_file == fp).delete()
session.flush()
return "refresh"
return "new"
[文档]
def scan_and_import(
folder: str,
session: Session,
*,
clear_existing: bool | None = None,
mode: str | BomImportMode | None = None,
) -> dict[str, int]:
"""遍历 folder（含子目录）下所有 .xls/.xlsx，解析并批量入库。
Args:
folder: 根目录（如 生产-下料计划TAO）。
session: 数据库会话（本函数内 commit）。
clear_existing: [兼容旧调用] 优先使用 ``mode``；为 ``None`` 时：
``True``→CLEAR，``False``→APPEND。
mode: 导入模式（见 ``BomImportMode``）：
- ``clear``：清空全表后重导（默认，幂等、无重复）；
- ``append``：不清空，所有解析行直接插入（重复导入同文件会重复）；
- ``refresh``：不清空；已导入过的 ``source_file`` 先删其旧行再写最新内容
（如某文件 12→10 行则删 2、12→15 行则加 5），新文件追加；
- ``skip``：不清空；只导入 ``source_file`` 尚未入库的文件，已存在文件
（即便内容已更新）跳过；需刷新旧文件请清空重导。
Returns:
``{files, rows, orders, new_files, refreshed_files, skipped_files}`` 统计。
``files`` 维持旧语义 = 实际写入的文件数（new+refreshed），兼容旧调用方。
"""
if mode is None:
mode = BomImportMode.from_clear_existing(clear_existing)
mode = BomImportMode(mode)
if mode is BomImportMode.CLEAR:
session.query(BomMaterial).delete()
session.flush()
existing_files: set[str] = set()
else:
existing_files = {r[0] for r in session.query(BomMaterial.source_file).distinct().all()}
new_files = refreshed_files = skipped_files = 0
rows = 0
orders: set[str] = set()
for root, _dirs, fnames in os.walk(folder):
for fn in fnames:
low = fn.lower()
if not (low.endswith(".xlsx") or low.endswith(".xls")) or low.endswith(".tmp"):
continue
fp = os.path.join(root, fn)
parts = parse_bom_file(fp)
if not parts:
continue
action = _ingest_file(session, fp, mode, existing_files)
if action == "skip":
skipped_files += 1
continue
if action == "refresh":
refreshed_files += 1
else:
new_files += 1
orders.add(parts[0]["production_order_no"])
for p in parts:
session.add(BomMaterial(**p))
rows += 1
session.commit()
logger.info(
f"导入完成(mode={mode.value}): 新文件 {new_files}, 刷新 {refreshed_files}, "
f"跳过 {skipped_files}, 行 {rows}, 生产令号 {len(orders)}"
)
return {
"files": new_files + refreshed_files,
"rows": rows,
"orders": len(orders),
"new_files": new_files,
"refreshed_files": refreshed_files,
"skipped_files": skipped_files,
}
[文档]
def ensure_bom_material_columns(engine: Any) -> None:
"""为已存在的 bom_materials 表补齐新增列（存量库 ALTER，避免 drop/recreate 丢数据）。
``create_all`` 只能建缺失的表、不会给既有表加列，故对可能已存在的表做一次幂等 ALTER。
新增列：material_source(默认 import) / material_note / updated_at。
"""
from sqlalchemy import inspect, text
insp = inspect(engine)
if "bom_materials" not in insp.get_table_names():
return
existing = {c["name"] for c in insp.get_columns("bom_materials")}
needed = {
"material_source": "VARCHAR(16) NOT NULL DEFAULT 'import'",
"material_note": "VARCHAR(255)",
"updated_at": "DATETIME",
}
with engine.begin() as conn:
for col, ddl in needed.items():
if col not in existing:
conn.execute(text(f"ALTER TABLE bom_materials ADD COLUMN {col} {ddl}"))
logger.info(f"bom_materials 已补列: {col}")
[文档]
def query_by_production_order_no(production_order_no: str, session: Session) -> list[BomMaterial]:
"""按 生产令号 精确查询其全部零件（按 seq 排序）。"""
return (
session.query(BomMaterial)
.filter(BomMaterial.production_order_no == production_order_no)
.order_by(BomMaterial.seq)
.all()
)
[文档]
def list_production_order_nos(session: Session) -> list[str]:
"""返回库中所有不重复的 生产令号（排序）。"""
from sqlalchemy import distinct
return [
r[0]
for r in session.query(distinct(BomMaterial.production_order_no))
.order_by(BomMaterial.production_order_no)
.all()
]
[文档]
def build_hyperlink_index(out_path: str, session: Session, *, folder: str | None = None) -> int:
"""生成超链接索引 xlsx：生产令号 | 文件名 | 零件数 | 零件牌号摘要 | 完整路径。
- 「文件名」列为指向源文件的 file:// 超链接，点击即可回访材料清单；
- 「零件牌号摘要」形如「阀体:WCB; 阀杆:35#无缝钢管; …」（超长截断）。
Args:
out_path: 输出 xlsx 路径。
session: 数据库会话。
folder: 可选，仅导出源文件位于该目录（及其子目录）下的记录；默认全部。
Returns:
写入的索引行数（不含表头）。
"""
from openpyxl import Workbook
from openpyxl.styles import Alignment, Font, PatternFill
from sqlalchemy import func
# 聚合：生产令号 -> [(part_name, material_grade), ...] 与 source_file
rows = (
session.query(
BomMaterial.production_order_no,
BomMaterial.part_name,
BomMaterial.material_grade,
BomMaterial.source_file,
func.count(BomMaterial.id),
)
.group_by(
BomMaterial.production_order_no,
BomMaterial.source_file,
)
.order_by(BomMaterial.production_order_no)
.all()
)
# 关联销售计划是否存在该生产令号（快速回访时一眼看出是否已建档）
from sqlalchemy import distinct
from certflow.models.sale_plan import SalePlan
planned = {
r[0]
for r in session.query(distinct(SalePlan.production_order_no))
.filter(SalePlan.production_order_no.isnot(None))
.all()
}
agg: dict[str, dict[str, Any]] = {}
for pon, pname, grade, src, cnt in rows:
if folder and not (src or "").startswith(folder):
continue
bucket = agg.setdefault(pon, {"parts": [], "src": src, "cnt": 0, "in_plan": pon in planned})
bucket["cnt"] += cnt
if pname:
bucket["parts"].append((pname, grade))
wb = Workbook()
ws = wb.active
ws.title = "BOM索引"
hdr_fill = PatternFill("solid", fgColor="4472C4")
hdr_font = Font(color="FFFFFF", bold=True)
headers = ["生产令号", "文件名", "零件数", "在销售计划", "零件牌号摘要", "完整路径"]
ws.append(headers)
for c in range(1, len(headers) + 1):
cell = ws.cell(row=1, column=c)
cell.fill = hdr_fill
cell.font = hdr_font
cell.alignment = Alignment(horizontal="center", vertical="center")
n = 0
for pon, b in agg.items():
summary = "; ".join(f"{pn}:{g}" if g else f"{pn}" for pn, g in b["parts"])
if len(summary) > 300:
summary = summary[:297] + "…"
fname = os.path.basename(b["src"])
ws.append([pon, fname, b["cnt"], "是" if b["in_plan"] else "否", summary, b["src"]])
n += 1
link = ws.cell(row=n + 1, column=2)
link.hyperlink = _unc_uri(b["src"])
link.font = Font(color="0563C1", underline="single")
ws.column_dimensions["A"].width = 22
ws.column_dimensions["B"].width = 46
ws.column_dimensions["C"].width = 8
ws.column_dimensions["D"].width = 12
ws.column_dimensions["E"].width = 70
ws.column_dimensions["F"].width = 90
ws.freeze_panes = "A2"
ws.auto_filter.ref = f"A1:F{n + 1}"
wb.save(out_path)
logger.info(f"超链接索引已生成: {out_path}（{n} 个生产令号）")
return n
[文档]
def export_bom_table( # noqa: C901
out_path: str, session: Session, *, folder: str | None = None
) -> dict[str, int]:
"""从 bom_materials 表导出自包含索引表（无网络超链接依赖，可直接拷贝存档）。
生成两张工作表：
- 「BOM索引」：生产令号 | 源文件名 | 零件数 | 在销售计划 | 零件牌号摘要
- 「零件明细」：生产令号 | 序号 | 代号 | 零件名称 | 材料牌号 | 数量 | 材料型态 | 源文件名
Args:
out_path: 输出 xlsx 路径。
session: 数据库会话。
folder: 可选，仅导出源文件位于该目录（及其子目录）下的记录；默认全部。
Returns:
{orders, rows}：索引生产令号数 / 明细零件行数。
"""
from openpyxl import Workbook
from openpyxl.styles import Alignment, Font, PatternFill
from sqlalchemy import distinct, func
from certflow.models.sale_plan import SalePlan
planned = {
r[0]
for r in session.query(distinct(SalePlan.production_order_no))
.filter(SalePlan.production_order_no.isnot(None))
.all()
}
# —— 零件明细（全量）——
detail_q = session.query(BomMaterial).order_by(BomMaterial.production_order_no, BomMaterial.seq)
if folder:
detail_q = detail_q.filter(BomMaterial.source_file.like(f"{folder}%"))
# —— 聚合：生产令号 -> 摘要 / 零件数 / 源文件 ——
agg_q = (
session.query(
BomMaterial.production_order_no,
BomMaterial.part_name,
BomMaterial.material_grade,
BomMaterial.source_file,
func.count(BomMaterial.id),
)
.group_by(BomMaterial.production_order_no, BomMaterial.source_file)
.order_by(BomMaterial.production_order_no)
)
if folder:
agg_q = agg_q.filter(BomMaterial.source_file.like(f"{folder}%"))
agg: dict[str, dict[str, Any]] = {}
for pon, pname, grade, src, cnt in agg_q.all():
bucket = agg.setdefault(pon, {"parts": [], "src": src, "cnt": 0, "in_plan": pon in planned})
bucket["cnt"] += cnt
if pname:
bucket["parts"].append((pname, grade))
wb = Workbook()
hdr_fill = PatternFill("solid", fgColor="4472C4")
hdr_font = Font(color="FFFFFF", bold=True)
# Sheet 1：BOM 索引
ws1 = wb.active
ws1.title = "BOM索引"
h1 = ["生产令号", "源文件名", "零件数", "在销售计划", "零件牌号摘要", "源文件完整路径"]
ws1.append(h1)
for c in range(1, len(h1) + 1):
cell = ws1.cell(row=1, column=c)
cell.fill = hdr_fill
cell.font = hdr_font
cell.alignment = Alignment(horizontal="center", vertical="center")
for pon, b in agg.items():
summary = "; ".join(f"{pn}:{g}" if g else f"{pn}" for pn, g in b["parts"])
if len(summary) > 300:
summary = summary[:297] + "…"
ws1.append(
[
pon,
os.path.basename(b["src"]),
b["cnt"],
"是" if b["in_plan"] else "否",
summary,
b["src"],
]
)
for col, w in zip("ABCDEF", (22, 46, 8, 12, 70, 90), strict=True):
ws1.column_dimensions[col].width = w
ws1.freeze_panes = "A2"
ws1.auto_filter.ref = f"A1:F{len(agg) + 1}"
# Sheet 2：零件明细
ws2 = wb.create_sheet("零件明细")
h2 = ["生产令号", "序号", "代号", "零件名称", "材料牌号", "数量", "材料型态", "源文件名"]
ws2.append(h2)
for c in range(1, len(h2) + 1):
cell = ws2.cell(row=1, column=c)
cell.fill = hdr_fill
cell.font = hdr_font
cell.alignment = Alignment(horizontal="center", vertical="center")
n_rows = 0
for m in detail_q.all():
ws2.append(
[
m.production_order_no,
m.seq,
m.part_no or "",
m.part_name,
m.material_grade or "",
m.quantity or "",
m.material_form or "",
os.path.basename(m.source_file),
]
)
n_rows += 1
for col, w in zip("ABCDEFGH", (22, 6, 16, 24, 20, 8, 14, 46), strict=True):
ws2.column_dimensions[col].width = w
ws2.freeze_panes = "A2"
ws2.auto_filter.ref = f"A1:H{n_rows + 1}"
wb.save(out_path)
logger.info(f"数据库导出索引表已生成: {out_path}（{len(agg)} 生产令号 / {n_rows} 零件）")
return {"orders": len(agg), "rows": n_rows}
# 供控制器/视图从服务层间接引用模型类与来源常量，避免 controller 直连 certflow.models。
__all__ = [
"BomMaterial",
"MATERIAL_SOURCE_IMPORT",
"MATERIAL_SOURCE_MANUAL",
"MATERIAL_SOURCE_AUTO_LEARN",
"MATERIAL_SOURCE_CHOICES",
"BomImportMode",
"ensure_bom_material_columns",
"scan_and_import",
"query_by_production_order_no",
"list_production_order_nos",
"build_hyperlink_index",
"export_bom_table",
]