# src/certflow/services/cert_numbering.py
"""合格证编号与数据生成服务
负责编号模式（自动/手动/复制）和 Certificate 表写入。
"""
from __future__ import annotations
from datetime import datetime
from sqlalchemy.orm import Session
from certflow.models.certificate import Certificate
from certflow.models.sale_plan import SalePlan
from certflow.services.cert_numbering_policy import get_sn_kks_position
from certflow.services.dn_service import DNService
from certflow.services.pn_service import PNService
from certflow.utils.logger import logger
def _default_template_type_for(supply_type: str | None) -> str:
"""由供货类型推导默认模板语言族（A4）
自产 → 全中文；外购/采购/进口 → 中英文；含「俄」→ 俄英文；其余 → 全中文。
与 vba_mapping.get_style_code 的语言族取值保持一致。
"""
st = (supply_type or "").strip()
if "俄" in st:
return "俄英文"
if any(k in st for k in ("外购", "采购", "进口")):
return "中英文"
return "全中文"
[文档]
class CertNumberingService:
"""合格证编号与 Certificate 写入服务
提供三种编号模式 + Certificate 表写入 + Access/CSV 导出。
**编号模式**：
- auto_number: 自动编号（正则解析 production_order_no / supply_type，最常用）
- manual_number: 手动指定前缀和起始序号
- copy_number: 复制已有编号（单件直接复制）
**持久化**：
- create_certificates: 将已编号的 SalePlan 写入 Certificate 表
- export_to_access: 导出到 Access 数据库
- export_sign_tb_csv: 导出 CSV
"""
def __init__(self, session: Session) -> None:
self.session = session
# ============================================================
# 编号模式
# ============================================================
[文档]
def auto_number(self, ids: list[int], prefix_override: str | None = None) -> int:
"""自动编号（正则解析 production_order_no / supply_type，最常用）
根据生产令号和供货类型自动判定后缀（L/B/G/Y/XL），
按 sort_group 排序后依次编号，同一批次序列号连续递增。
Args:
ids: 需要编号的 SalePlan ID 列表
prefix_override: 手动指定的年月前缀（格式 YYMM），为 None 则使用当前年月
Returns:
int: 成功编号的记录数
"""
from certflow.services.certificate_number_service import CertificateNumberService
service = CertificateNumberService(prefix_override)
return service.generate_numbers(self.session, ids)
[文档]
def manual_number(
self, ids: list[int], prefix: str, start: int, quantity: int, suffix: str = ""
) -> int:
"""手动编号
手动指定编号前缀、起始序号和数量后缀，为选中的记录生成编号。
只对尚未编号的记录（product_code 为空）进行操作。
Args:
ids: 需要编号的 SalePlan ID 列表
prefix: 编号前缀（如 "V2506"）
start: 起始序列号
quantity: 每台设备的数量
suffix: 编号后缀（如 "Y"、"L"），默认为空
Returns:
int: 成功编号的记录数
"""
records = (
self.session.query(SalePlan)
.filter(
SalePlan.id.in_(ids),
(SalePlan.product_code.is_(None)) | (SalePlan.product_code.like("【空白%")),
# H(维度16)：跳过不需要编号的记录（needs_numbering=False）；
# NULL 视为需要编号，兼容旧数据。
(SalePlan.needs_numbering.is_(True) | (SalePlan.needs_numbering.is_(None))),
)
.order_by(SalePlan.sort_group)
.all()
)
serial = start
last_end = start - 1
# 编号阶段分隔符（默认 "---"，§5.7.2-B）；与自动编号一致
from certflow.services.cert_numbering_policy import get_number_model
from certflow.services.certificate_number_service import CertificateNumberService
sep = CertificateNumberService.stage_separator()
for record in records:
# 军工三位流水（mode:7）：无 V+YYMM，前缀已是军工前缀（如 JG），
# 用清单分隔符 "-" 连接逐台独立码（对齐 VBA getKKS 单 '-' 形态）
is_military = get_number_model(record) == 7
mil_sep = CertificateNumberService.list_separator() if is_military else sep
end = serial + (record.quantity or quantity) - 1
if serial == end:
record.product_code = f"{prefix}{serial:03d}{suffix}"
else:
record.product_code = f"{prefix}{serial:03d}{suffix}{mil_sep}{end:03d}{suffix}"
serial = end + 1
last_end = end
# 手动编号后同步计数器，避免后续自动编号回退/碰撞（BUG-005 计数器为唯一权威源）
if records:
from certflow.services.certificate_number_service import CertificateNumberService
CertificateNumberService()._bump_serial(self.session, prefix, last_end)
self.session.commit()
logger.info(f"手动编号完成 | 数量={len(records)}")
return len(records)
[文档]
def copy_number(self, ids: list[int], source_code: str) -> int:
"""复制编号（单件直接复制）
将指定编号直接复制给选中的所有记录，适用于同一产品重复使用的场景。
Args:
ids: 需要编号的 SalePlan ID 列表
source_code: 要复制的编号字符串
Returns:
int: 成功编号的记录数
"""
records = self.session.query(SalePlan).filter(SalePlan.id.in_(ids)).all()
for record in records:
record.product_code = source_code
self.session.commit()
logger.info(f"复制编号完成 | 数量={len(records)} | 编号={source_code}")
return len(records)
# ============================================================
# 生成 Certificate 并写入 Access
# ============================================================
[文档]
def create_certificates(
self, ids: list[int], printer_name: str = "", print_status: str = "已打印"
) -> int:
"""将已编号的记录写入 Certificate 表
从 SalePlan 表中读取已编号记录，创建对应的 Certificate 记录。
会自动合并 cert_product_name/model/spec 字段（优先使用 cert\\_ 前缀字段，回退到原始字段）。
同时从产品型号解析公称压力(PN)信息，并标准化口径(DN)。
当 ``certificate.numbering.timing == "on_import"`` 时，先对未编号记录
自动编号（步骤 6.1），使写入 Certificate 即带编号。``certificate_no``
由组A 服务 ``generate_certificate_no`` 生成合格证号
``CERT-YYYYMMDD-NNNN``（逐台一个 id，G1）；产品编号区间串逐台拆填写入
``Certificate.product_code_range``（G2）。默认 ``on_demand`` 保持现状
（仅收已编号记录）。
合格证按「实际发货台数」逐台生成（BUG-006：合格证数量 = 发货台数总和，
而非销售计划记录数）。一个 SalePlan 的 ``product_code`` 区间串会展开为
多台 Certificate，每台独立 ``product_code_range`` / ``certificate_no`` /
审计字段（G5）。
Args:
ids: 已编号的 SalePlan ID 列表
printer_name: 打印机名称，默认为空
print_status: 打印状态，默认"已打印"，查询页自动编号时传"待打印"
Returns:
int: 成功创建的 Certificate 记录数
"""
from certflow.services.cert_numbering_policy import (
get_issue_date_source,
get_numbering_timing,
parse_military_code_meta,
parse_product_code_meta,
)
# 步骤 6.1：on_import 时机——先给未编号记录自动编号
if get_numbering_timing() == "on_import":
self.auto_number(ids)
records = (
self.session.query(SalePlan)
.filter(
SalePlan.id.in_(ids),
SalePlan.product_code.is_not(None),
SalePlan.product_code != "",
SalePlan.product_code.not_like("【空白%"),
)
.all()
)
# === 组A：合格证编号 CERT-YYYYMMDD-NNNN（G1，与文档 §11.4.2 一致）===
# certificate_no 为「逐台一个」的唯一合格证号（格式 CERT-日期-序号），
# 与组B 产品编号区间串 (Certificate.product_code_range，G2 填充) 解耦；
# 序号按当天已建数 base_seq 递增，保证全局唯一。产品编号单码只存于
# product_code_range（逐台），绝不写入 certificate_no（避免 UNIQUE 冲突）。
from certflow.services.certificate_number_service import CertificateNumberService
from certflow.services.certificate_service import CertificateService
cert_service = CertificateService(self.session)
base_seq = cert_service.count_certificates_created_today()
pn_service = PNService(self.session)
dn_service = DNService(self.session)
# === G5：预统计每个 SalePlan 已存在的 Certificate 数量与累计台数 ===
# 供 batch_seq(第几批) / cumulative_quantity(累计台数) 计算（审计用）。
from sqlalchemy import func
existing_rows = (
self.session.query(
Certificate.sale_plan_id,
func.count(Certificate.id),
func.coalesce(func.sum(Certificate.quantity), 0),
func.coalesce(func.max(Certificate.batch_seq), 0),
)
.filter(Certificate.sale_plan_id.in_([r.id for r in records]))
.group_by(Certificate.sale_plan_id)
.all()
)
existing_map = {row[0]: (row[1], int(row[2]), int(row[3])) for row in existing_rows}
# === 逐台展开：一个 SalePlan 的 product_code 区间串 → 多台 Certificate ===
# 合格证按「实际发货台数」逐台生成（BUG-006：合格证数量 = 发货台数总和，
# 而非销售计划记录数）。每台独立 product_code_range / certificate_no / 审计字段。
created_units = 0
for sp in records:
# 解析产品编号区间串 → 逐台完整编码列表
# 军工三位流水（mode:7，无 V+YYMM）：走军工专用解析，逐台独立码
mil_meta = parse_military_code_meta(sp.product_code)
meta = mil_meta or parse_product_code_meta(sp.product_code)
is_military = mil_meta is not None
if meta:
if is_military:
# 军工逐台独立码：前缀 + 三位流水 + 后缀（无年月）
prefix = meta["prefix"]
ym = "" # 军工无 V+YYMM 年月
suffix = meta["suffix"]
unit_codes = [
f"{prefix}{seq:03d}{suffix}"
for seq in range(meta["seq_start"], meta["seq_end"] + 1)
]
else:
prefix, ym, suffix = meta["prefix"], meta["ym"], meta["suffix"]
# 单台码用前缀+yy+mm+流水 (无分隔符)，与 _format_code 单台形态一致
unit_codes = [
f"{prefix}{ym}{CertificateNumberService._format_serial(seq)}{suffix}"
for seq in range(meta["seq_start"], meta["seq_end"] + 1)
]
else:
# 非标准编号（单台/特殊码）：整体作为一台的产品编号
unit_codes = [sp.product_code]
# 出厂日期年月来源（配置驱动）：默认同步编号年月，可选当前年月
# 军工三位流水（mode:7）无 V+YYMM → meta 无 "ym"，落当前年月
pc_ym = meta.get("ym", "") if meta else ""
if get_issue_date_source() == "current_ym" or not pc_ym:
issue_date = datetime.now().strftime("%Y-%m-%d")
else:
# pc_ym 形如 "2607" → 取该年月首日，保证与编号年月一致
issue_date = f"20{pc_ym[:2]}-{pc_ym[2:4]}-01"
# G5：本 SalePlan 既有批次数 / 累计台数（本次调用 = 该计划的一批发货）
prev_count, prev_cum, prev_batch = existing_map.get(sp.id, (0, 0, 0))
local_batch = prev_batch + 1
created_units += self._expand_plan_to_certificates(
sp=sp,
unit_codes=unit_codes,
meta=meta,
prefix=prefix,
ym=ym,
issue_date=issue_date,
local_batch=local_batch,
prev_cum=prev_cum,
base_seq=base_seq,
created_so_far=created_units,
print_status=print_status,
printer_name=printer_name,
pn_service=pn_service,
dn_service=dn_service,
)
self.session.commit()
logger.info(f"写入 Certificate 完成 | 数量={created_units}")
return created_units
def _expand_plan_to_certificates(
self,
sp: SalePlan,
unit_codes: list[str],
meta: dict | None,
prefix: str,
ym: str,
issue_date: str,
local_batch: int,
prev_cum: int,
base_seq: int,
created_so_far: int,
print_status: str,
printer_name: str,
pn_service: PNService,
dn_service: DNService,
) -> int:
"""逐台展开一个 SalePlan 的 product_code 区间串为 Certificate 记录。
抽出 create_certificates 内层「组B 逐台拆填」循环，降低主函数圈复杂度
（C901）。逻辑与原实现逐行一致，仅改变作用域；合格证号 ``certificate_no``
由组A 服务 ``CertificateService.generate_certificate_no`` 生成唯一
``CERT-YYYYMMDD-NNNN`` 号（G1，与文档 §11.4.2 一致），序号按
``base_seq + created_so_far + created + 1`` 递增，每台独立；产品编号单码
仅存入 ``product_code_range``（逐台），二者解耦。
"""
from certflow.config.settings import CERTIFICATE_PREFIX
from certflow.services.certificate_service import CertificateService
created = 0
for k, code in enumerate(unit_codes):
# 幂等：同一 SalePlan 内相同 product_code_range（逐台单码）已建则跳过，
# 避免重跑 create_certificates 产生重复合格证。certificate_no 为全局
# 唯一 CERT- 号，不参与去重判定。
dup = (
self.session.query(Certificate)
.filter_by(sale_plan_id=sp.id, product_code_range=code)
.first()
)
if dup is not None:
continue
cert_no = CertificateService.generate_certificate_no(
prefix=CERTIFICATE_PREFIX,
sequence=base_seq + created_so_far + created + 1,
)
# === 字典兜底（烘焙非空合格证）：sp 的温度/介质/标准为空时，
# 用 model_param_mappings[型号] 补齐，避免生成后 9 字段空白。
# 优先级：SalePlan 合同值 > 型号字典（仅补空，不覆盖已有值）。
model = (sp.cert_product_model or sp.product_model or "").strip()
_mpm = None
if model:
from certflow.models.model_param_mapping import ModelParamMapping
_mpm = self.session.query(ModelParamMapping).filter_by(product_model=model).first()
# 注：SalePlan 无 working_temp/working_medium 列，仅 Certificate 与
# model_param_mappings 持有；此处以 getattr 容错，温度/介质的唯一来源即字典。
_wt = getattr(sp, "working_temp", None) or (_mpm.working_temp if _mpm else None)
_wm = getattr(sp, "working_medium", None) or (_mpm.working_medium if _mpm else None)
_ts = sp.test_standard or (_mpm.test_standard if _mpm else None)
cert = Certificate(
sale_plan_id=sp.id,
unique_key=sp.unique_key,
production_order_no=sp.production_order_no,
sales_order_no=sp.sales_order_no,
certificate_no=cert_no,
product_name=sp.cert_product_name or sp.product_name,
product_model=sp.cert_product_model or sp.product_model,
product_spec=sp.cert_product_spec or sp.product_spec,
plan_date=sp.plan_date,
customer=sp.customer,
project_name=sp.project_name,
supplier=getattr(sp, "cert_supplier", None) or getattr(sp, "supplier", None),
test_standard=_ts,
working_temp=_wt,
working_medium=_wm,
print_status=print_status,
issue_date=issue_date,
printer_name=printer_name,
print_time=datetime.now(),
)
# === 组B：逐台产品编号拆填（G2） ===
cert.product_code_range = code
if meta:
cert.product_code_prefix = prefix
cert.product_code_ym = ym
cert.product_code_seq_start = meta["seq_start"] + k
cert.product_code_seq_end = meta["seq_start"] + k
else:
cert.product_code_prefix = ""
cert.product_code_ym = ""
cert.product_code_seq_start = 0
cert.product_code_seq_end = 0
cert.quantity = 1
# === 双编码位置约定（mode 5/6，对齐 VBA ar(2,8)）===
# 由编号模式规则命中时一并确定（电建三双编码客户 → 产品编号/KKS 显示框互换）。
# 落库后供打印/扫描派生 SN/KKS 时读取（替代原先永远空串导致 mode 5/6 失效）。
cert.sn_kks_position = get_sn_kks_position(sp)
# === G5：批次审计字段（逐台） ===
cert.batch_seq = local_batch
cert.cumulative_quantity = prev_cum + k + 1
# === PN 解析 ===
model = sp.cert_product_model or sp.product_model or ""
if model:
pn_info = pn_service.resolve_pn(model=model)
if pn_info.get("pn_value"):
cert.pn_value = pn_info["pn_value"]
cert.pn_unit = pn_info["pn_unit"]
cert.pn_display = pn_info["pn_display"]
# === DN 标准化 ===
spec = sp.cert_product_spec or sp.product_spec or ""
if spec:
dn_result = dn_service.resolve_caliber(spec, model=model)
if dn_result.get("caliber_value"):
cert.product_spec = dn_result["caliber_value"]
# A4：supply_type → 默认模板（自产全中文 / 外购中英文 / 俄单俄英文）
cert.template_type = _default_template_type_for(getattr(sp, "supply_type", None))
# 检验工号：按模板语言族烘焙默认（全中文→检5 / 其他→No5），
# 打印页可覆盖（与 print.guard 的 _apply_inspector_default 同源）。
from certflow.services.cert_log_service import _default_inspector_id
cert.inspector_id = _default_inspector_id(cert.template_type)
self.session.add(cert)
created += 1
return created
[文档]
def export_to_access(self, ids: list[int], mdb_path: str | None = None) -> int:
"""导出到 Access 数据库（委托 CertExportService）
Args:
ids: Certificate.id 列表
mdb_path: 目标 .mdb 文件路径，None 则使用默认路径
Returns:
int: 成功导出的记录数
"""
from certflow.services.cert_export_service import CertExportService
return CertExportService(self.session).export_to_access(ids, mdb_path)
[文档]
def export_sign_tb_csv(self, output_path: str | None = None, limit: int = 1000) -> str | None:
"""导出 CSV（委托 CertExportService）
Args:
output_path: 输出 CSV 文件路径，None 则使用默认路径
limit: 最大导出条数
Returns:
str | None: 生成的 CSV 文件路径；失败返回 None
"""
from certflow.services.cert_export_service import CertExportService
return CertExportService(self.session).export_sign_tb_csv(output_path, limit)