"""报告填充服务（批次4 / B4-3+B4-7 集成层）。
把报告生成从「读 xlsx 数据源」改为「读 certflow.db（SalePlan / Certificate）」，
复用已迁入 ``certflow.services.report_migration`` 的泛型填充引擎，并接下游
``ReportOutputService`` 完成双存根(xlsx+pdf) / 印章 / 文件夹命名 / 报告目录。
数据流：
SalePlan(+关联 Certificate)
→ build_product_records() → list[ProductRecord]
→ ReportFillEngine.fill() 填充 12 模板之一（openpyxl，分页）
→ ReportOutputService.emit() 落盘到 VBA 同款项目文件夹
材质块优先取 Certificate 的 阀体/阀杆/启闭件 材质；缺失时按生产令号查 BOM 兜底牌号
（配置 ``report.material_bom_fallback``，默认开启），仍缺则留空（绝不把型号当牌号回填）。
试压压力优先取 Certificate 已算值；缺失则经 PressureCalcService 计算。
报告的名称/型号/规格/编号/标准号一律取「合格证打印信息」（非合同值）：
优先 Certificate 打印列 → 其次 SalePlan.cert_product_*（合格证打印列）
→ 最后兜底合同列 product_*（仅 legacy 缺打印列时）。
"""
from __future__ import annotations
import contextlib
import re
import tempfile
from datetime import date, datetime
from pathlib import Path
from typing import Any
import openpyxl
from loguru import logger
from sqlalchemy.orm import Session
from certflow.config import settings as app_settings
from certflow.config.paths_override import cfg
from certflow.config.settings import _resolve_path
from certflow.handlers.excel_handler import ExcelHandler
from certflow.models import Certificate, SalePlan
from certflow.services.bom_part_classifier import DEFAULT_BOM_REPORT_CFG
from certflow.services.certificate_number_service import CertificateNumberService
from certflow.services.material_grade_service import MaterialGradeService
from certflow.services.pressure_calc_service import (
LB_TO_MPA,
compute_pressure_params,
)
from certflow.services.report_bom_enricher import enrich
from certflow.services.report_migration import (
ReportFillEngine,
caizhi_1hang,
caizhi_2hang,
caizhi_3hang,
shiyan_report_shao,
shiyan_report_zhong,
shiyan_report_zhongwen_shao,
shiyan_report_zhongying,
zhibaoshu_5hang,
zhibaoshu_6hang,
zhibaoshu_9hang_bianhao_biaozhun,
zhibaoshu_9hang_bianhao_biaozhun_wendu,
zhibaoshu_9hang_biaozhunhao,
)
from certflow.services.report_migration.material_provider import (
FunctionMaterialProvider,
MaterialPart,
MaterialProvider,
StaticMaterialProvider,
XlsxMaterialProvider,
get_material_function,
random_material_resolver,
)
from certflow.services.report_migration.report_fill_engine import ZhiBaoShuSpec
from certflow.services.report_output_service import ReportOutputRequest, ReportOutputService
# --------------------------------------------------------------------------- #
# 模板规格解析
# --------------------------------------------------------------------------- #
_SPEC_FACTORIES = {
"zhibaoshu_5hang": zhibaoshu_5hang,
"zhibaoshu_6hang": zhibaoshu_6hang,
"zhibaoshu_9hang_biaozhunhao": zhibaoshu_9hang_biaozhunhao,
"zhibaoshu_9hang_bianhao_biaozhun": zhibaoshu_9hang_bianhao_biaozhun,
"zhibaoshu_9hang_bianhao_biaozhun_wendu": zhibaoshu_9hang_bianhao_biaozhun_wendu,
"shiyan_report_zhong": shiyan_report_zhong,
"shiyan_report_shao": shiyan_report_shao,
"shiyan_report_zhongwen_shao": shiyan_report_zhongwen_shao,
"shiyan_report_zhongying": shiyan_report_zhongying,
"caizhi_1hang": caizhi_1hang,
"caizhi_2hang": caizhi_2hang,
"caizhi_3hang": caizhi_3hang,
}
# report_kind → (模板类型中文标签, 默认 spec key)
_KIND_LABEL = {
"quality": "质量证明书",
"pressure": "试压报告",
"material": "材质报告",
}
[文档]
def resolve_spec_key(report_kind: str, language: str = "zh", records_per_page: int = 6) -> str:
"""根据 UI 选择（报告类型 + 语言 + 每页记录数）解析出模板 spec key。
Args:
report_kind: ``quality`` / ``pressure`` / ``material``。
language: ``zh`` / ``en``。
records_per_page: UI 微调框值，用于挑选模板变体（行数/双语）。
Returns:
str: ``_SPEC_FACTORIES`` 中的 key。
"""
if report_kind == "quality":
if records_per_page >= 9:
return "zhibaoshu_9hang_bianhao_biaozhun"
if records_per_page <= 5:
return "zhibaoshu_5hang"
return "zhibaoshu_6hang"
if report_kind == "pressure":
if language == "en":
return (
"shiyan_report_zhongying"
if records_per_page > 13
else "shiyan_report_zhongwen_shao"
)
return "shiyan_report_zhong" if records_per_page > 13 else "shiyan_report_shao"
if report_kind == "material":
n = max(1, min(3, records_per_page))
return f"caizhi_{n}hang"
raise ValueError(f"未知报告类型: {report_kind}")
[文档]
def spec_for_key(key: str):
"""返回对应模板的 ZhiBaoShuSpec 实例。"""
factory = _SPEC_FACTORIES.get(key)
if factory is None:
raise KeyError(f"未知模板规格 key: {key}")
return factory()
def _kind_from_spec_key(spec_key: str) -> str:
"""从模板 spec key 反推报告类型（``quality`` / ``pressure`` / ``material``）。"""
if spec_key.startswith("zhibaoshu"):
return "quality"
if spec_key.startswith("shiyan"):
return "pressure"
if spec_key.startswith("caizhi"):
return "material"
raise ValueError(f"无法从 spec key 推断报告类型: {spec_key}")
# 模板 spec key → 中文变体标签（用于报告目录索引展示）
_VARIANT_LABEL: dict[str, str] = {
"zhibaoshu_5hang": "5行",
"zhibaoshu_6hang": "6行",
"zhibaoshu_9hang_biaozhunhao": "9行·标准号",
"zhibaoshu_9hang_bianhao_biaozhun": "9行·编号标准",
"zhibaoshu_9hang_bianhao_biaozhun_wendu": "9行·编号标准·温度",
"shiyan_report_zhong": "中文·多",
"shiyan_report_shao": "中文·少",
"shiyan_report_zhongwen_shao": "中英·少",
"shiyan_report_zhongying": "中英·多",
"caizhi_1hang": "1行",
"caizhi_2hang": "2行",
"caizhi_3hang": "3行",
}
def variant_label_for(spec_key: str) -> str:
"""模板 spec key → 中文变体标签（用于报告目录 §7.4 索引展示）。"""
return _VARIANT_LABEL.get(spec_key, spec_key)
def _resolve_sheet_name(
report_type: str,
language: str,
product_no: str,
plan_date: date | str | None,
report_date: date | str | None,
serial: str = "",
) -> str:
"""按 ``report.sheet_name_format`` 组装 Excel 工作表(数据表)标签名。
默认 ``{report_type}_{plan_date_short}_{report_date}``：自动带计划日期与制作日期，
避免千篇一律的模板变体名（如「质保书6行」）。占位符缺失→空；非法字符→下划线。
"""
import re
from string import Formatter
def _to_date(v):
if isinstance(v, (date, datetime)):
return v
if isinstance(v, str) and v:
try:
return datetime.strptime(v[:10], "%Y-%m-%d").date()
except ValueError:
return None
return None
pd = _to_date(plan_date)
rd = _to_date(report_date)
fmt = app_settings.get_report_sheet_name_format()
ctx: dict[str, Any] = {
"report_type": report_type or "",
"language": language or "",
"product_no": product_no or "",
"customer": "",
"project": "",
"date": pd.strftime("%Y%m%d") if pd else "",
"plan_date_short": pd.strftime("%y%m%d") if pd else "",
"report_date": rd.strftime("%y%m%d") if rd else "",
"serial": serial or "",
}
used = {fn for _, fn, _, _ in Formatter().parse(fmt) if fn}
for k in used:
ctx.setdefault(k, "")
try:
name = fmt.format(**ctx)
except (KeyError, IndexError, ValueError):
name = f"{report_type}_{language}" if report_type else "Sheet"
return re.sub(r'[\\/:*?"<>|]', "_", name).strip("_ ") or "Sheet"
def _build_material_provider() -> MaterialProvider:
"""按配置 ``report.material_provider`` 选择材质解析器（可插拔 / 自定义函数）。
- ``static``（默认）：``StaticMaterialProvider``，无真实牌号时留空（绝不伪造）。
- ``random``：内置 ``random_material_resolver``，返回随机真实牌号（排版压测用）。
- ``xlsx``：``XlsxMaterialProvider``，读取「加工进度跟踪单」离线表按生产令号解析
真实零件牌号（DingTalk 落地前的真实数据源）；路径取自 ``report.material_xlsx_path``
（空则回退 static，绝不伪造）。
- ``function:<name>``：取回 ``register_material_function`` 注册的自定义函数。
未知值回退 ``static``，保证默认行为不被破坏。
"""
kind = str(app_settings.cfg("report.material_provider", "static")).strip().lower()
if kind in ("random", "rand", "随机"):
return FunctionMaterialProvider(random_material_resolver)
if kind in ("xlsx", "excel"):
path = (app_settings.cfg("report.material_xlsx_path", "") or "").strip()
if not path:
logger.warning("material_provider=xlsx 但 report.material_xlsx_path 为空，回退 static")
return StaticMaterialProvider()
sheet = app_settings.cfg("report.material_xlsx_sheet", "加工进度跟踪单") or "加工进度跟踪单"
from certflow.config.settings import _resolve_path
resolved = _resolve_path(path)
if not resolved.exists():
logger.warning(f"材质 xlsx 不存在: {resolved}，回退 static")
return StaticMaterialProvider()
return XlsxMaterialProvider(str(resolved), sheet=sheet)
if kind.startswith("function:") or kind.startswith("func:"):
name = kind.split(":", 1)[1].strip()
fn = get_material_function(name)
if fn is None:
logger.warning(f"未注册材质函数 '{name}'，回退 static")
return StaticMaterialProvider()
return FunctionMaterialProvider(fn)
return StaticMaterialProvider()
# --------------------------------------------------------------------------- #
# 取数：SalePlan(+Certificate) → ProductRecord
# --------------------------------------------------------------------------- #
_KNOWN_STANDARDS = [
"API 598",
"GB/T 13927",
"GB/T 26480",
"JB/T 9092",
"JB/T 3595",
"GB/T 12465",
"GB/T 12245",
"GB/T 14382",
"NB/T 47044",
"GB/T 14478",
"GB/T 12251",
"GB 12242",
"SL 553",
"DL/T 5190.5",
"JB/T 7387",
]
_DN_RE = re.compile(r"DN\s*(\d+)", re.IGNORECASE)
_PN_RE = re.compile(r"PN\s*(\d+(?:\.\d+)?)", re.IGNORECASE)
_MPA_RE = re.compile(r"(\d+(?:\.\d+)?)\s*MPa", re.IGNORECASE)
_LB_RE = re.compile(r"(\d+)\s*LB", re.IGNORECASE)
_STAR_RE = re.compile(r"\*\s*(\d+(?:\.\d+)?)")
_SUFFIX_RE = re.compile(r"-(\d+(?:\.\d+)?)\s*[A-Za-z]?")
def _parse_standard(text: str) -> str:
if text:
u = text.upper()
for s in _KNOWN_STANDARDS:
if s.upper() in u:
return s
return "GB/T 13927"
def _parse_mpa(raw: str) -> float | None:
if not raw:
return None
m = _LB_RE.search(raw)
if m:
return LB_TO_MPA.get(int(m.group(1)))
m = _MPA_RE.search(raw)
if m:
return float(m.group(1))
m = _PN_RE.search(raw)
if m:
return round(float(m.group(1)) / 10.0, 2)
m = _STAR_RE.search(raw)
if m:
return float(m.group(1))
m = _SUFFIX_RE.search(raw)
if m:
return round(float(m.group(1)) / 10.0, 2)
return None
def _parse_caliber(*texts: str) -> float:
for t in texts:
if not t:
continue
m = _DN_RE.search(t)
if m:
return float(m.group(1))
return 0.0
def _lookup_model_dict(session: Session, model: str) -> dict[str, str]:
"""从型号压力对照字典(model_param_mappings)查型号对应的标准号/压力/温度/介质。
供报告取数在 SalePlan / Certificate 字段为空时兜底补全，让已导入的字典底座真正生效。
无字典命中或异常时返回空 dict（调用方需自行兜底，绝不伪造）。
"""
if not model or session is None:
return {}
try:
from certflow.services.pn_service import PNService
db = PNService(session).lookup_pn_from_db(model.strip())
except Exception:
return {}
if not db:
return {}
out: dict[str, str] = {}
for key in ("test_standard", "pressure_value", "working_temp", "working_medium"):
val = (db.get(key) or "").strip()
if val:
out[key] = val
return out
def _resolve_pressures(
plan: SalePlan, cert: Certificate | None, session: Session | None = None
) -> dict[str, str]:
"""返回 6 个试压压力/时间字段（壳体/上密封/密封 × 压力/时间）。
标准号/公称压力在 plan / cert 字段为空时，回退型号压力对照字典(model_param_mappings)。
"""
if cert and cert.strength_test_pressure:
return {
"壳体压力": cert.strength_test_pressure or "",
"壳体时间": cert.strength_test_time or "",
"上密封压力": cert.upper_seal_pressure or "",
"上密封时间": cert.upper_seal_time or "",
"密封压力": cert.main_seal_pressure or "",
"密封时间": cert.main_seal_time or "",
}
# 字典兜底：型号→标准号/压力
d = _lookup_model_dict(session, plan.product_model or "") if session else {}
# 标准号：plan/tech > cert > 字典；三者皆无则保留 _parse_standard 的默认兜底
explicit_std = (
plan.tech_requirements or plan.test_standard or (cert.test_standard if cert else "")
)
std = (
_parse_standard(explicit_std)
if explicit_std
else d.get("test_standard") or _parse_standard("")
)
# 公称压力：cert.pn_display > plan.pressure_value > cert.pn_value > 字典 > 型号兜底
nominal = (
(cert.pn_display if cert and cert.pn_display else "")
or plan.pressure_value
or (cert.pn_value if cert else "")
or d.get("pressure_value", "")
or plan.product_model
or ""
)
mpa = _parse_mpa(nominal)
if mpa is None:
return {
k: ""
for k in ("壳体压力", "壳体时间", "上密封压力", "上密封时间", "密封压力", "密封时间")
}
caliber = _parse_caliber(plan.product_spec, plan.product_model)
name_blob = f"{plan.product_name or ''} {plan.product_model or ''}"
is_check = "止回" in name_blob
is_butterfly = "蝶" in name_blob
is_imperial = bool(_LB_RE.search(plan.product_model or ""))
try:
r = compute_pressure_params(
standard=std,
nominal_pressure_mpa=mpa,
caliber_mm=caliber,
is_check_valve=is_check,
is_butterfly_valve=is_butterfly,
is_imperial=is_imperial,
)
except Exception as e: # 计算异常不阻断整批，留空由模板占位
logger.warning(f"压力计算失败({plan.id}): {e}")
return {
k: ""
for k in ("壳体压力", "壳体时间", "上密封压力", "上密封时间", "密封压力", "密封时间")
}
return {
"壳体压力": r.get("strength_test_pressure", ""),
"壳体时间": r.get("strength_hold_time", ""),
"上密封压力": r.get("upper_seal_pressure", ""),
"上密封时间": r.get("upper_seal_hold_time", ""),
"密封压力": r.get("main_seal_pressure", ""),
"密封时间": r.get("main_seal_hold_time", ""),
}
def _bom_reportcfg() -> dict:
"""读取阶段 C 的 BOM 归类配置（config/report.yaml 的 ``report.bom_report``）。
缺省回退 ``DEFAULT_BOM_REPORT_CFG``（配置缺失/损坏时分类器仍可用）。
"""
return app_settings.cfg("report.bom_report", {}) or dict(DEFAULT_BOM_REPORT_CFG)
# BOM 分类槽位 → 报告材质块主材位（阀体/阀杆/启闭件）映射（C2/C4）。
_BOM_TO_REPORT = {
"body": "阀体",
"stem": "阀杆",
"disc": "启闭件",
"gate": "启闭件",
"seat": "启闭件",
"bolt": "启闭件",
}
# 报告「启闭件」位在 BOM 槽位中的优先取用顺序（首个命中即取，其余进待核对）。
_启闭件_BOM_PRIORITY = ["disc", "gate", "seat", "bolt"]
# BOM 槽位固定顺序（用于空行补位时优先取真实承压件）。
_BOM_SLOT_ORDER = ["body", "stem", "disc", "gate", "seat", "bolt"]
# 报告材质块主材位固定顺序（与模板 阀体/阀杆/启闭件 行序一致）。
_REPORT_SLOT_ORDER = ["阀体", "阀杆", "启闭件"]
def _resolve_materials( # noqa: C901
cert: Certificate | None,
plan: SalePlan,
composition_svc: MaterialGradeService | None = None,
session: Session | None = None,
bom_fallback: bool = True,
spec: ZhiBaoShuSpec | None = None,
bom_detail: list[dict[str, str]] | None = None,
) -> list[MaterialPart]:
"""质保书/材质报告材质块：取 Certificate 三零件材质，缺失时按生产令号查 BOM 回填。
阶段 C 改造（vs 旧硬编码 ``_match_material_slot``）：
- BOM 归类改为配置驱动（``BomPartClassifier``），零件名 → 主材位（body/stem/disc/...）；
- 多对一 / 未命中 / 牌号缺 material_grades 的零件一律进 ``bom_detail``（待核对明细），
不写 sale_plan、不写超链接（原则 3）；
- 命中的主材位按 ``证书优先 > BOM 兜底`` 填入报告材质块（阀体/阀杆/启闭件），
并经 ``generate_composition_values`` 产生化学/力学性能虚拟实测值。
Args:
bom_detail: 可选收集器，传入时把待核对明细条目追加进去（供报告页展示/计数）。
Returns:
list[MaterialPart]：按 阀体/阀杆/启闭件 顺序的材质块零件（空则整组留空）。
"""
slots_grade = {
"阀体": (cert.fati_material if cert else None),
"阀杆": (cert.fagan_material if cert else None),
"启闭件": (cert.qibijian_material if cert else None),
}
# C1：按生产令号查 BOM 并富集（命中主材位 + 待核对明细）
bom_cfg = _bom_reportcfg()
res: Any = None
bom_classified: dict[str, MaterialPart] = {}
if bom_fallback and session and plan.production_order_no:
try:
res = enrich(plan.production_order_no, session, bom_cfg)
bom_classified = res.classified
if bom_detail is not None:
bom_detail.extend(res.detail)
except Exception: # BOM 查询/富集异常不阻断报告，降级为留空
logger.exception("BOM 材质富集失败，材质块降级为留空")
# 哪些 BOM 槽位被报告「启闭件」位取用（其余进待核对明细）
consumed_bom: set[str] = set()
# 模板表头元素键集：用于把化学成分拆成「表头内」与「表头外」（向后兼容）
header_keys = set(spec.material_cols.get("化学成分", {}).keys()) if spec else None
def _fill(part_name: str, grade: str, source: str, source_no: str = "") -> MaterialPart:
part = MaterialPart(
零件名称=part_name,
材料=grade,
来源=source,
is_virtual=True,
来源令号=source_no,
)
if composition_svc is not None and grade:
comp = composition_svc.generate_composition_values(grade)
chem = comp.get("化学成分", {})
mech = comp.get("力学性能", {})
if header_keys is not None:
# 不在表头集的元素挪入 表头外元素（值保留模拟值），供批注 B / 溢出列消费
part.表头外元素 = {k: v for k, v in chem.items() if k not in header_keys}
chem = {k: v for k, v in chem.items() if k in header_keys}
part.化学成分 = chem
part.力学性能 = mech
return part
parts: list[MaterialPart] = []
for report_slot in _REPORT_SLOT_ORDER:
grade = (slots_grade.get(report_slot) or "").strip()
source = "cert"
no = ""
name = report_slot
if not grade:
# 找映射到该报告位的 BOM 槽位（启闭件按优先级取首个，其余进待核对）
if report_slot == "启闭件":
cands = [s for s in _启闭件_BOM_PRIORITY if s in bom_classified]
else:
cands = [
s
for s, rl in _BOM_TO_REPORT.items()
if rl == report_slot and s in bom_classified
]
if cands:
s = cands[0]
bp = bom_classified[s]
grade = bp.材料
# 启闭件 是泛化主材位（阀瓣/阀座/闸板/螺栓 等共用一个报告位），
# 若由 BOM 命中具体零件，则把行标签刷新为真实零件名（如「阀瓣」），
# 更贴近实际；阀体/阀杆 仅在 BOM 零件名与报告位一致时才同名，保留原标签。
name = bp.零件名称 if report_slot == "启闭件" else report_slot
source = "bom"
no = plan.production_order_no or ""
consumed_bom.add(s)
if not grade:
continue
parts.append(_fill(name, grade, source, no))
# 空行补位（阶段 C 补丁 15 延伸）：主材位仍有空行时，用「其它已找到的 BOM 零件」
# 补满材质块空行——保留真实零件名，不冒充阀体/阀杆/启闭件；补位的零件仍保留在
# 待核对明细（res.detail）供人工核对。开关：config report.bom_report.fill_empty_rows。
cap = spec.单条行数 if spec else 3
if cap and len(parts) < cap and res is not None and bom_cfg.get("fill_empty_rows", True):
# 候选池优先：未消费的 BOM 主材位（真实承压件，如 阀座）；
# 其次：未命中关键词 / 多对一的零件（leftover_parts）。
for s in _BOM_SLOT_ORDER:
if len(parts) >= cap:
break
if s in bom_classified and s not in consumed_bom:
parts.append(bom_classified[s])
consumed_bom.add(s)
for bp in getattr(res, "leftover_parts", []) or []:
if len(parts) >= cap:
break
parts.append(bp)
# 未被报告位取用的 BOM 主材位 → 待核对明细（多零件被合并 / 合格证已覆盖）
if bom_detail is not None:
for s, bp in bom_classified.items():
if s in consumed_bom:
continue
bom_detail.append(
{
"part_name": bp.零件名称,
"material_grade": bp.材料,
"reason": "BOM 主材位未填入报告（多零件合并/合格证已覆盖）",
"note": "",
}
)
return parts
def _resolve_print(
plan: SalePlan, cert: Certificate | None, session: Session | None = None
) -> dict[str, str]:
"""报告展示的名称/型号/规格/编号/标准号：统一取「合格证打印信息」。
优先级：合格证表(Certificate 打印列) → 销售计划合格证打印列(cert_product_*)
→ 销售计划合同列(product_* 兜底)。以上皆空时，标准号回退型号压力对照字典。
"""
name = (cert.product_name if cert else "") or plan.cert_product_name or plan.product_name or ""
model = (
(cert.product_model if cert else "") or plan.cert_product_model or plan.product_model or ""
)
spec = (cert.product_spec if cert else "") or plan.cert_product_spec or plan.product_spec or ""
code = (cert.product_code_range if cert else "") or plan.product_code or ""
# 标准号：cert > plan > 字典兜底
explicit_std = (
(cert.test_standard if cert else "")
or plan.test_standard
or plan.tech_requirements
or (cert.tech_requirements if cert else "")
)
if explicit_std:
std = _parse_standard(explicit_std)
else:
d = _lookup_model_dict(session, model) if session else {}
std = d.get("test_standard") or _parse_standard("")
return {"name": name, "model": model, "spec": spec, "code": code, "std": std}
[文档]
def build_product_records(
session: Session, sale_plan_ids: list[int], report_kind: str = "quality", spec=None
) -> list[Any]:
"""从 certflow.db 构建报告所需的产品记录列表。
名称/型号/规格/编号/标准号一律取合格证打印信息（详见 ``_resolve_print``）。
Args:
session: SQLAlchemy 会话。
sale_plan_ids: 选中的销售计划 ID 列表。
report_kind: ``quality`` / ``pressure`` / ``material``（影响材质块取舍）。
Returns:
list[ProductRecord]: 填充引擎可直接消费的记录列表。
"""
from certflow.services.report_migration.report_fill_engine import ProductRecord
if not sale_plan_ids:
return []
plans = (
session.query(SalePlan).filter(SalePlan.id.in_(sale_plan_ids)).order_by(SalePlan.id).all()
)
# 批量预载关联证书（按 sale_plan_id 建索引），避免逐行查询的会话副作用
cert_map = {
c.sale_plan_id: c
for c in session.query(Certificate)
.filter(Certificate.sale_plan_id.in_(sale_plan_ids))
.all()
}
# 材质块化学/力学取数服务（按牌号查 material_grades），quality/material 报告需要
composition_svc = (
MaterialGradeService(session) if report_kind in ("quality", "material") else None
)
out: list[Any] = []
for i, p in enumerate(plans):
cert = cert_map.get(p.id)
pr = _resolve_pressures(p, cert, session)
info = _resolve_print(p, cert, session)
rec = ProductRecord(
序号=i + 1,
产品名称=info["name"],
产品型号=info["model"],
规格=info["spec"],
数量=p.quantity or 1,
产品编号=info["code"],
试验标准=info["std"],
壳体压力=pr["壳体压力"],
壳体时间=pr["壳体时间"],
上密封压力=pr["上密封压力"],
上密封时间=pr["上密封时间"],
密封压力=pr["密封压力"],
密封时间=pr["密封时间"],
结果="合格",
适用温度=(cert.working_temp if cert else "") or "",
生产令号=p.production_order_no or "",
)
if report_kind in ("quality", "material"):
bom_detail: list[dict[str, str]] = []
rec.parts = _resolve_materials(
cert,
p,
composition_svc,
session=session,
bom_fallback=app_settings.cfg("report.material_bom_fallback", True),
spec=spec,
bom_detail=bom_detail,
)
# 阶段 C：待核对明细（未命中/多对一/缺数据）随记录传递，供报告页展示与计数
rec.待核对明细 = bom_detail or None
out.append(rec)
return out
# --------------------------------------------------------------------------- #
# 主流程：填充 + 落盘
# --------------------------------------------------------------------------- #
_TEMPLATE_NAME = "各种报告的xlsx文件模板.xlsx"
def _template_path() -> Path:
"""报告母版工作簿路径（配置驱动, 见 paths.templates.report_master）。
旧实现硬编码 ``BASE_DIR / "data" / "xlsx" / _TEMPLATE_NAME``（目录与文件名双重硬编码,
不经 ``_cfg()``）。现改为读取 ``paths.templates.report_master``, 回退到原硬编码默认值,
与 paths.templates.* 体系保持一致。
"""
rel = cfg("paths.templates.report_master", f"data/xlsx/{_TEMPLATE_NAME}")
return _resolve_path(rel)
# 报告单号「前缀 + 版本号(记录单序号)」由**模板工作表内容**驱动（配置驱动）：
# 解析模板编号格已有的完整示例号（如 "C/1 编号：9.1-13-170331001"），提取前缀与
# 版本号，使渲染严格贴合工作表实际写法（C/0/C/1、编号/N0、空格数均一致）；
# 模板编号格缺失或不可解析时回退 spec 的 number_prefix / report_code。
_REPORT_NO_PAT = re.compile(
r"^(?P<prefix>.*?)(?P<code>\d+\.\d+-\d+)-(?P<ymd>\d{6})(?P<serial>\d+)\s*$"
)
def _resolve_report_prefix_code(template_ws, spec, spec_key: str | None = None) -> tuple[str, str]:
"""解析报告单号 (前缀, 版本号)；解析优先级：配置覆盖 > 模板工作表内容 > spec 兜底。
1) 配置 ``report.number_prefix_map[spec_key]``（集中管理 3-4 种前缀/版本号，优先级最高）；
2) 模板工作表内容（解析编号格已有的完整示例号，自动贴合实际写法）；
3) spec 的 ``number_prefix`` / ``report_code``（最终兜底）。
Args:
template_ws: 已加载的模板工作表（含示例编号）。
spec: 当前模板规格（提供兜底 number_prefix / report_code）。
spec_key: 模板 spec key（用于查配置 ``number_prefix_map``）；为空则跳过配置层。
"""
# 1) 配置覆盖：集中管理前缀/版本号，优先级最高
cfg_map = app_settings.cfg("report.number_prefix_map", {}) or {}
if spec_key and isinstance(cfg_map, dict):
entry = cfg_map.get(spec_key)
if isinstance(entry, dict) and entry.get("prefix") and entry.get("code"):
return str(entry["prefix"]), str(entry["code"])
# 2) 模板工作表内容驱动（解析编号格示例号）
cell = template_ws.cell(row=spec.number_row, column=spec.number_col).value
if isinstance(cell, str):
m = _REPORT_NO_PAT.match(cell.strip())
if m:
return m.group("prefix").rstrip(), m.group("code")
# 3) spec 兜底
return spec.number_prefix, spec.report_code
[文档]
def generate( # noqa: C901
session: Session,
sale_plan_ids: list[int],
report_kind: str = "quality",
language: str = "zh",
records_per_page: int = 6,
report_date: date | str | None = None,
export_pdf: bool | None = None,
spec_key: str | None = None,
variant: str = "",
plan_date: date | str | None = None,
) -> dict[str, Any]:
"""生成一份报告：填充 12 模板之一并落盘到 VBA 同款项目文件夹。
Args:
session: SQLAlchemy 会话。
sale_plan_ids: 选中的销售计划 ID 列表。
report_kind: ``quality`` / ``pressure`` / ``material``。
language: ``zh`` / ``en``。
records_per_page: 用于挑选模板变体（行数/双语）。
report_date: 报告日期（默认今天）。
export_pdf: 是否同时派生 PDF（None→读配置，默认 False）。
spec_key: 直接指定模板 spec key（覆盖 ``report_kind/language/records_per_page``
的自动解析）。用于「一次性跑全部 12 个模板」等场景。
Returns:
dict: 含 ``success`` / ``folder`` / ``xlsx`` / ``pdf`` / ``sealed`` 等键。
"""
if not sale_plan_ids:
return {"success": False, "error": "请选择要生成报告的数据"}
if spec_key:
if spec_key not in _SPEC_FACTORIES:
return {"success": False, "error": f"未知模板 spec key: {spec_key}"}
report_kind = _kind_from_spec_key(spec_key)
spec_key = (
resolve_spec_key(report_kind, language, records_per_page) if spec_key is None else spec_key
)
# 单模板生成未显式传 variant 时，自动用 spec_key 作为变体标签，
# 避免 --kind/--rpp 单次生成的文件名退化为「质量证明书_.xlsx」而与同类型互相覆盖。
if not variant:
variant = spec_key
spec = spec_for_key(spec_key)
records = build_product_records(session, sale_plan_ids, report_kind, spec=spec)
if not records:
return {"success": False, "error": "未从数据库取到任何产品记录"}
# 报告前数据完整性闸门（配置驱动，对齐 output.targets.scan.required_fields 契约）。
# gate_mode: off(默认，不校验) / warn(仅记清单不阻断) / block(缺失即中断)。
gate_mode = str(app_settings.cfg("report.gate_mode", "off")).strip().lower()
if gate_mode in ("block", "warn"):
chk = check_report_readiness(session, sale_plan_ids, report_kind)
if not chk["passed"]:
summary = "; ".join(
f"#{m['id']}({m['product_model'] or '?'}) 缺: {','.join(m['fields'])}"
for m in chk["missing"]
)
if gate_mode == "block":
return {
"success": False,
"error": f"报告数据不完整（已阻断，请补全后重试）：{summary}",
"incomplete": chk["missing"],
}
logger.warning(f"报告数据不完整（warn，仍生成）：{summary}")
tpl = _template_path()
if not tpl.exists():
return {"success": False, "error": f"模板文件不存在: {tpl}"}
wb_tpl = ExcelHandler.load_workbook(tpl, data_only=False)
if spec.name not in wb_tpl.sheetnames:
return {"success": False, "error": f"模板缺少工作表: {spec.name}"}
template_ws = wb_tpl[spec.name]
rd = report_date or date.today()
rd = rd if isinstance(rd, date) else datetime.strptime(rd, "%Y-%m-%d").date()
# 项目级信息（计划日期/客户/产品号/表名）取首批 SalePlan
first = session.query(SalePlan).filter(SalePlan.id == sale_plan_ids[0]).first()
eff_plan_date = (
plan_date if plan_date is not None else (first.plan_date if first else rd.isoformat())
)
report_type = _KIND_LABEL.get(report_kind, report_kind)
product_no = first.product_code or first.production_order_no or "" if first else ""
# 页数：每页 spec.k 条记录（不足补一页）；一页一个编号，工作表名取最小号。
pages = (len(records) + spec.k - 1) // spec.k if records else 1
# 输出服务（双存根/印章/目录），流水号由其统一自动编号服务分配
output_service = ReportOutputService()
# 统一自动编号服务（与合格证 V+YYMM 编号同源的 AutoNumberCounter 计数基础设施）
# 按制作日期 YYMMDD 归组，分配连续流水号，每页一个。
num_svc = CertificateNumberService()
serials = num_svc.allocate_report_serials(session, rd, pages)
min_serial = serials[0] if serials else 0
serial = f"{min_serial:03d}"
ymd = rd.strftime("%y%m%d")
# 每页编号：前缀 + 版本号(记录单序号) + "-" + 制作日期 YYMMDD + 流水号。
# 前缀/版本号由模板工作表内容驱动（见 _resolve_report_prefix_code），
# 例如「C/1 编号：9.1-13-260726001」「C/0 N0：8.2-20-260726002」。
prefix, code = _resolve_report_prefix_code(template_ws, spec, spec_key)
page_numbers = [f"{prefix}{code}-{ymd}{s:03d}" for s in serials]
sheet_name = _resolve_sheet_name(report_type, language, product_no, eff_plan_date, rd, serial)
wb_out = openpyxl.Workbook()
engine = ReportFillEngine(
_build_material_provider(),
material_annotation=app_settings.cfg("report.material_annotation", True),
)
engine.fill(
wb_out,
template_ws,
spec,
records,
sheet_name=sheet_name,
report_date=rd,
page_numbers=page_numbers,
)
# 移除 openpyxl 默认空 Sheet
if "Sheet" in wb_out.sheetnames:
wb_out.remove(wb_out["Sheet"])
tmp = (
Path(cfg("paths.temp.base", tempfile.gettempdir()))
/ f"_cf_report_{datetime.now():%Y%m%d%H%M%S}.xlsx"
)
wb_out.save(tmp)
req = ReportOutputRequest(
report_type=report_type,
xlsx_source=tmp,
plan_date=eff_plan_date,
customer=first.customer if first else "",
project=first.project_name if first else "",
language=language,
product_no=first.product_code or first.production_order_no or "",
variant=variant,
report_date=rd,
serial=serial,
sales_order_no=first.sales_order_no if first else "",
records_per_page=records_per_page,
variant_label=variant_label_for(spec_key),
)
result = output_service.emit(req, export_pdf=export_pdf)
# warn 模式：把完整性缺失清单随结果透出（不阻断），供 UI 提示用户核对。
if gate_mode == "warn" and not chk["passed"]:
result["incomplete"] = chk["missing"]
# 阶段 C：聚合 BOM 材质命中/待核对计数，供报告页结果提示（C4.3）
bom_matched = 0
bom_detail_count = 0
for r in records:
for p in r.parts or []:
if getattr(p, "来源", None) == "bom":
bom_matched += 1
if getattr(r, "待核对明细", None):
bom_detail_count += len(r.待核对明细)
result["bom_matched"] = bom_matched
result["bom_detail_count"] = bom_detail_count
# 输出工作表实际使用的（动态）名称：由 report.sheet_name_format 生成，
# 供调用方/测试按真实表名取表，而非假设固定模板 spec 名。
result["sheet_name"] = sheet_name
with contextlib.suppress(OSError):
tmp.unlink()
return result
# --------------------------------------------------------------------------- #
# 数据完备性诊断（用于「完善数据库、避免报告缺数据」）
# --------------------------------------------------------------------------- #
# 各报告类型判定「可生成」所需的字段；值为取数/兜底表达式，空/假即视为缺失。
# 与 generate() 的容错逻辑保持一致：公称压力/标准号缺列时可从型号、技术要求兜底。
# 诊断 lambda 签名为 (plan, cert, session)，需查 material_grades 的字段可借 session。
def _grade_coverage(session: Session, grade: str) -> tuple[bool, bool]:
"""某材质牌号在 material_grades（启用）中是否同时有化学成分与力学性能记录。
与 ``MaterialGradeService.generate_composition`` 的查表口径一致（按 grade 精确匹配）。
"""
from certflow.models.material_grade import MaterialGrade
grade = (grade or "").strip()
if not grade:
return (False, False)
rows = (
session.query(MaterialGrade)
.filter(MaterialGrade.grade == grade, MaterialGrade.is_active.is_(True))
.all()
)
return (
any(r.kind == "chemical" for r in rows),
any(r.kind == "mechanical" for r in rows),
)
def _materials_resolvable(session: Session, cert: Certificate | None) -> bool:
"""合格证各零件材质是否都能在 material_grades 查到化学成分+力学性能。
- 无合格证(cert=None)：返回 True（缺证本身由「关联合格证」字段报缺）；
- 合格证无材质文本：返回 True（由「阀体/阀杆/启闭件材质」字段报缺）。
"""
if cert is None:
return True
grades = [
g
for g in (cert.fati_material, cert.fagan_material, cert.qibijian_material)
if g and g.strip()
]
if not grades:
return True
return all(all(_grade_coverage(session, g)) for g in grades)
_REQUIRED: dict[str, dict[str, Any]] = {
"quality": {
"产品名称": lambda p, c, s: bool(
p.product_name or p.cert_product_name or (c and c.product_name)
),
"产品型号": lambda p, c, s: bool(p.product_model or (c and c.product_model)),
"产品编号": lambda p, c, s: bool(p.product_code or (c and c.product_code_range)),
"生产令号": lambda p, c, s: bool(p.production_order_no),
"试验标准": lambda p, c, s: bool(
p.test_standard
or (c and c.test_standard)
or _lookup_model_dict(s, p.product_model).get("test_standard")
),
"材质化学成分/力学性能可查表": lambda p, c, s: _materials_resolvable(s, c),
},
"pressure": {
"产品型号": lambda p, c, s: bool(p.product_model),
"公称压力": lambda p, c, s: bool(
p.pressure_value
or _parse_mpa(p.product_model or "")
or _lookup_model_dict(s, p.product_model).get("pressure_value")
),
"试验标准": lambda p, c, s: bool(
p.test_standard
or p.tech_requirements
or _lookup_model_dict(s, p.product_model).get("test_standard")
),
"口径规格": lambda p, c, s: bool(p.product_spec or _parse_caliber(p.product_model or "")),
},
"material": {
"关联合格证": lambda p, c, s: c is not None,
"阀体/阀杆/启闭件材质": lambda p, c, s: bool(
c and (c.fati_material or c.fagan_material or c.qibijian_material)
),
"材质化学成分/力学性能可查表": lambda p, c, s: _materials_resolvable(s, c),
},
}
[文档]
def check_report_readiness(
session: Session,
sale_plan_ids: list[int],
report_kind: str = "quality",
) -> dict[str, Any]:
"""报告生成前的完整性预检（阻断式闸门的数据来源）。
按 ``report.required_fields[report_kind]``（配置驱动，对齐
``output.targets.scan.required_fields`` 契约）逐条核对所选 SalePlan 的必检字段；
quality/material 报告额外校验材质牌号能否在 material_grades 查表。
Args:
session: SQLAlchemy 会话。
sale_plan_ids: 待生成报告的销售计划 ID 列表。
report_kind: ``quality`` / ``pressure`` / ``material``。
Returns:
dict: ``{"passed": bool, "missing": list[dict]}``，
``missing`` 每项含 ``id`` / ``product_model`` / ``fields``（缺失字段列表）。
"""
fields = app_settings.cfg(f"report.required_fields.{report_kind}", []) or []
if not fields:
return {"passed": True, "missing": []}
plans = (
session.query(SalePlan).filter(SalePlan.id.in_(sale_plan_ids)).order_by(SalePlan.id).all()
)
cert_map = {
c.sale_plan_id: c
for c in session.query(Certificate)
.filter(Certificate.sale_plan_id.in_(sale_plan_ids))
.all()
}
missing: list[dict[str, Any]] = []
for p in plans:
cert = cert_map.get(p.id)
bad: list[str] = []
for f in fields:
val = getattr(p, f, None)
if val is None or (isinstance(val, str) and not val.strip()):
bad.append(f)
continue
# 材质类字段：额外校验是否能在 material_grades 查到化学/力学性能
if f in (
"fati_material",
"fagan_material",
"qibijian_material",
) and not _materials_resolvable(session, cert):
bad.append(f"{f}(材质不可查表)")
if bad:
missing.append({"id": p.id, "product_model": p.product_model, "fields": bad})
return {"passed": not missing, "missing": missing}
[文档]
def report_readiness(
session: Session,
report_kind: str = "quality",
limit: int = 50,
) -> dict[str, Any]:
"""诊断 sale_plans 中能支撑指定类型报告的完备度。
仅统计「已可进入报告流程」的销售计划（有生产令号，作为可报告数据的最小门槛）。
返回聚合统计与少量缺失样例，便于定位需要补全的字段。
对 ``quality`` / ``material`` 报告，额外判定合格证各零件材质能否在
``material_grades`` 查到化学成分+力学性能（供材质块/RandBetween 实测值使用）。
Args:
session: SQLAlchemy 会话。
report_kind: ``quality`` / ``pressure`` / ``material``。
limit: 返回缺失样例的最大条数。
Returns:
dict: 含 ``total`` / ``ready`` / ``missing_field_counts`` / ``samples``。
"""
req = _REQUIRED.get(report_kind)
if req is None:
raise ValueError(f"未知报告类型: {report_kind}")
plans = (
session.query(SalePlan)
.filter(SalePlan.production_order_no.isnot(None), SalePlan.production_order_no != "")
.order_by(SalePlan.id)
.all()
)
# 批量预载关联证书（按 sale_plan_id 建索引），避免逐行查询的会话副作用
cert_map = {
c.sale_plan_id: c
for c in session.query(Certificate)
.filter(Certificate.sale_plan_id.in_([p.id for p in plans]))
.all()
}
total = len(plans)
ready = 0
field_counts: dict[str, int] = {f: 0 for f in req}
samples: list[dict[str, Any]] = []
for p in plans:
cert = cert_map.get(p.id)
missing = [f for f, fn in req.items() if not fn(p, cert, session)]
if not missing:
ready += 1
continue
for f in missing:
field_counts[f] += 1
if len(samples) < limit:
samples.append({"id": p.id, "product_model": p.product_model, "missing": missing})
return {
"report_kind": report_kind,
"total": total,
"ready": ready,
"missing_field_counts": field_counts,
"samples": samples,
}
__all__ = [
"resolve_spec_key",
"spec_for_key",
"build_product_records",
"generate",
"report_readiness",
"check_report_readiness",
]