"""合格证控制器模块
处理合格证的生成、打印、刻印标记及统计查询等业务逻辑.
支持批量操作和按产品型号分组打印.
"""
from __future__ import annotations
from datetime import datetime
from typing import Any
from loguru import logger
from sqlalchemy.orm import Session
from certflow.config.settings import CERTIFICATE_PREFIX
from certflow.controllers.base_controller import BaseController
from certflow.services.cert_numbering_policy import get_number_model
from certflow.services.certificate_print_service import CertificatePrintService
from certflow.services.certificate_service import Certificate, CertificateService, SalePlan
[文档]
class CertificateController(BaseController):
"""合格证控制器 - 处理合格证打印、刻印等业务
提供合格证的生成、批量打印、刻印标记、状态查询及统计等功能.
支持按产品型号分组打印，自动处理合格证编号生成和打印日志记录.
Attributes:
certificate_cache: 合格证编号缓存字典，用于避免重复生成编号
"""
def __init__(self, db_session: Session | None = None, db_manager: Any = None) -> None:
"""初始化合格证控制器
Args:
db_session: 数据库会话对象，如果为None则自动创建或由 db_manager 派生
db_manager: 数据库管理器实例；优先以它派生会话，统一会话生命周期
"""
super().__init__(db_session, db_manager=db_manager)
self.certificate_cache: dict[str, str] = {} # 缓存合格证编号
self._cert_service = self.get_service(CertificateService)
[文档]
def generate_certificates_for_sale_plans(
self, sale_plan_ids: list[int], force_regenerate: bool = False
) -> dict[str, object]:
"""为销售计划批量生成合格证
遍历指定的销售计划ID列表，为每条销售计划生成对应的合格证记录.
当force_regenerate为True时，即使合格证已存在也会重新生成.
Args:
sale_plan_ids: 销售计划ID列表
force_regenerate: 是否强制重新生成已存在的合格证，默认为False
Returns:
Dict[str, object]: 生成结果字典，包含以下键:
- total: 处理总数
- generated: 成功生成数
- failed: 失败数
- certificates: 生成的合格证对象列表
Examples:
>>> controller = CertificateController()
>>> result = controller.generate_certificates_for_sale_plans([1, 2, 3])
>>> print(f"生成成功: {result['generated']}/{result['total']}")
"""
result = {"total": len(sale_plan_ids), "generated": 0, "failed": 0, "certificates": []}
for sale_plan_id in sale_plan_ids:
try:
certificate = self._generate_single_certificate(sale_plan_id, force_regenerate)
if certificate:
result["generated"] += 1
result["certificates"].append(certificate)
else:
result["failed"] += 1
except Exception as e:
logger.error(f"生成合格证失败 (sale_plan_id={sale_plan_id}): {e}")
result["failed"] += 1
self.commit()
logger.info(f"合格证生成完成: {result['generated']}/{result['total']}")
return result
[文档]
def get_history_last_values(self, model: str | None) -> dict[str, str] | None:
"""取同型号上次实际打印值（供统一护栏 ④ diff 闸）。
收口 print_view 原先直连 ``HistoryBackfillService(self.session)`` 的写法，
视图层不再直接持有 Service，统一经 Controller 访问。
Args:
model: 产品型号字符串
Returns:
dict[str, str] | None: 历史候选值；无历史或异常时返回 None
"""
try:
from certflow.services.print_history_service import HistoryBackfillService
return HistoryBackfillService(self.session).candidates(model)
except Exception as exc: # noqa: BLE001
logger.debug(f"[CertificateController] 历史候选获取失败({model}): {exc}")
return None
def _generate_single_certificate(
self, sale_plan_id: int, force_regenerate: bool = False
) -> Certificate | None:
"""生成单个合格证
根据销售计划ID查找对应数据，生成合格证编号并创建合格证记录.
如果合格证已存在且不强制重新生成，则直接返回已有记录.
Args:
sale_plan_id: 销售计划ID
force_regenerate: 是否强制重新生成，默认为False
Returns:
Optional[Certificate]: 生成的合格证对象，失败时返回None
"""
cert_svc = self._cert_service
# 检查是否已存在
existing = cert_svc.get_certificate_by_sale_plan_id(sale_plan_id)
if existing and not force_regenerate:
logger.debug(f"合格证已存在: {existing.certificate_no}")
return existing
# 获取销售计划
sale_plan = cert_svc.get_sale_plan_by_id(sale_plan_id)
if not sale_plan:
logger.error(f"销售计划不存在: id={sale_plan_id}")
return None
# 生成新的合格证编号
certificate_no = self._generate_certificate_number(sale_plan)
# 创建合格证记录
certificate = Certificate(
certificate_no=certificate_no,
sale_plan_id=sale_plan.id,
unique_key=sale_plan.unique_key,
product_model=sale_plan.product_model,
sales_order_no=sale_plan.sales_order_no,
production_order_no=sale_plan.production_order_no,
customer=sale_plan.customer,
print_status="待打印",
engrave_status="未刻印",
)
# 用型号压力对照字典补全技术参数（标准号/公称压力/温度/介质）
self._fill_cert_from_dict(certificate, sale_plan)
if existing and force_regenerate:
# 更新已存在的记录
existing.certificate_no = certificate_no
existing.product_model = sale_plan.product_model
existing.sales_order_no = sale_plan.sales_order_no
existing.production_order_no = sale_plan.production_order_no
existing.customer = sale_plan.customer
existing.print_status = "待打印"
existing.engrave_status = "未刻印"
self._fill_cert_from_dict(existing, sale_plan)
cert_svc.merge_certificate(existing)
certificate = existing
else:
cert_svc.add(certificate)
cert_svc.flush()
logger.info(f"生成合格证: {certificate_no} for {sale_plan.product_model}")
return certificate
def _fill_cert_from_dict(self, cert: Certificate, sale_plan: SalePlan) -> None:
"""用型号压力对照字典(model_param_mappings)补全合格证技术参数。
仅填充字典可确定的字段：公称压力(pn_value/pn_display/pn_unit)、
试压标准、工作温度、工作介质。不覆盖合格证已手工维护的值，
也不伪造字典中没有的信息（如产品名称/材质牌号）。
"""
model = (sale_plan.product_model or "").strip()
if not model:
return
try:
from certflow.services.pn_service import PNService
pn = PNService(self.session).resolve_pn(model)
db = PNService(self.session).lookup_pn_from_db(model)
except Exception as e:
logger.debug(f"字典补全合格证技术参数失败({model}): {e}")
return
if pn.get("pn_value"):
cert.pn_value = pn["pn_value"]
if pn.get("pn_display"):
cert.pn_display = pn["pn_display"]
if pn.get("pn_unit"):
cert.pn_unit = pn["pn_unit"]
if db:
if db.get("test_standard"):
cert.test_standard = db["test_standard"]
if db.get("working_temp"):
cert.working_temp = db["working_temp"]
if db.get("working_medium"):
cert.working_medium = db["working_medium"]
def _generate_certificate_number(self, sale_plan: SalePlan) -> str:
"""生成合格证编号
基于当天已生成的合格证数量生成唯一编号，格式为 CERT-YYYYMMDD-NNNN.
Args:
sale_plan: 销售计划对象，用于获取产品信息
Returns:
str: 生成的合格证编号字符串，格式如"CERT-20231201-0001"
"""
count_today = self._cert_service.count_certificates_created_today()
# 生成编号: CERT-20231201-0001
return CertificateService.generate_certificate_no(
prefix=CERTIFICATE_PREFIX, sequence=count_today + 1
)
# ============================================================
# 编号续号（打印视图「目标年月 / 起始流水手填覆盖」后台支撑）
# ============================================================
[文档]
def get_next_serial(self, prefix: str) -> int:
"""只读预瞄某前缀下一个可用流水号（不消耗计数器，供 UI 自动续号/预览）。
Args:
prefix: 编号前缀（如 ``V2604``）
Returns:
int: 下一可用流水号（跨 SalePlan + Certificate 取该月最大 + 1）
"""
from certflow.services.certificate_number_service import CertificateNumberService
return CertificateNumberService().peek_next_serial(self.session, prefix)
[文档]
def sync_serial_counter(self, prefix: str, end_serial: int) -> None:
"""手动编号写库后，将计数器推进到 ``end_serial``，保证后续自动续号不回退。
Args:
prefix: 编号前缀（如 ``V2604``）
end_serial: 本次手动编号覆盖到的最大流水号
"""
from certflow.services.certificate_number_service import CertificateNumberService
CertificateNumberService()._bump_serial(self.session, prefix, end_serial)
self.session.flush()
[文档]
def batch_print_certificates(
self, certificate_ids: list[int], printer_name: str = "Default Printer", copies: int = 1
) -> dict[str, object]:
"""批量打印合格证(相同产品会分组打印)
按产品型号对合格证进行分组，逐组执行打印操作，
并为每张合格证记录打印日志.
Args:
certificate_ids: 合格证ID列表
printer_name: 打印机名称，默认为"Default Printer"
copies: 每张合格证的打印份数，默认为1
Returns:
Dict[str, object]: 打印结果字典，包含以下键:
- total: 总处理数
- success: 成功数
- failed: 失败数
- groups: 各产品组的打印详情列表
- logs: 打印日志列表
Examples:
>>> controller = CertificateController()
>>> result = controller.batch_print_certificates([1, 2, 3], "HP LaserJet", 2)
>>> print(f"打印成功: {result['success']}/{result['total']}")
"""
result = {
"total": len(certificate_ids),
"success": 0,
"failed": 0,
"groups": [],
"logs": [],
}
# 获取合格证并按产品分组
certificates = self._cert_service.get_certificates_by_ids(certificate_ids)
# 按产品型号分组
groups = {}
for cert in certificates:
product = cert.product_model
if product not in groups:
groups[product] = []
groups[product].append(cert)
logger.info(f"打印分组: {len(groups)} 个产品组")
# 逐组打印
for product_model, certs in groups.items():
group_result = {
"product_model": product_model,
"count": len(certs),
"certificate_nos": [c.certificate_no for c in certs],
"status": "success",
}
try:
# 使用 CertificatePrintService 的 record_print_log 记录日志
print_service = CertificatePrintService(self.session)
for cert in certs:
cert.print_time = datetime.now()
cert.printer_name = printer_name
cert.print_status = "已打印"
# 构建打印数据字典（字段对齐 batch_print 中的 cert_data）
cert_data = {
# Br(1) ~ Br(9): 基础信息
"plan_date": cert.plan_date or "",
"customer": cert.customer or "",
"project_name": cert.project_name or "",
"product_name": cert.product_name or "",
"product_model": cert.product_model or "",
"dn": cert.product_spec or "",
"pn": cert.pn_display or cert.pn_value or "",
"medium": cert.working_medium or "",
"temperature": cert.working_temp or "",
# Br(10) ~ Br(11): 出厂日期年月（从 product_code_ym 提取）
"manufacture_year": int(cert.product_code_ym[:4])
if cert.product_code_ym and len(cert.product_code_ym) >= 4
else 0,
"manufacture_month": int(cert.product_code_ym[4:6])
if cert.product_code_ym and len(cert.product_code_ym) >= 6
else 0,
# Br(25) ~ Br(27): 合格证版型/图片/编码模式
"template_style": 1,
"pic_style": 0,
# 步骤 6.2：编码模式由 certificate.numbering.modes 配置判定（默认 0）
"number_model": get_number_model(cert),
# Br(28): 合格证模板名称
"template_name": cert.product_name or "",
# Br(35) ~ Br(45): 材质/技术要求/销售信息
"body_material": "",
"stem_material": "",
"disc_material": "",
"tech_requirements": cert.tech_requirements,
"equipment_no": "",
"sub_project": "",
"sales_sn": "",
"category": "",
"weight": "",
"seq_no": "",
"requisition_no": "",
# Br(49): 打印模式
"print_mode": "1",
# Br(50) ~ Br(51): 水头
"max_inlet_head": "",
"max_boost_head": "",
# hgz_ShuLiang
"copies": copies,
}
log = print_service.record_print_log(
certificate_id=cert.id,
certificate_no=cert.certificate_no or "",
cert_data=cert_data,
printer_name=printer_name,
status="success",
content_summary=f"产品: {cert.product_model}",
)
if log:
self._cert_service.add(log)
result["success"] += 1
self.commit()
logger.info(f"打印成功: {product_model} - {len(certs)} 张")
except Exception as e:
group_result["status"] = "failed"
group_result["error"] = str(e)
result["failed"] += len(certs)
logger.error(f"打印失败: {product_model} - {e}")
result["groups"].append(group_result)
return result
[文档]
def get_certificates_by_status(self, print_status: str = "待打印") -> list[dict[str, str]]:
"""按打印状态获取合格证列表
查询指定打印状态的所有合格证，按创建时间升序排序.
Args:
print_status: 合格证打印状态，可选值: "待打印", "已打印", "打印失败"等，默认为"待打印"
Returns:
List[Dict[str, str]]: 合格证信息字典列表，每个字典包含id、编号、产品型号等字段
"""
certificates = self._cert_service.get_certificates_by_status(print_status)
return [
{
"id": c.id,
"certificate_no": c.certificate_no,
"product_model": c.product_model,
"order_no": c.order_no,
"customer": c.customer,
"print_status": c.print_status,
"engrave_status": c.engrave_status,
"created_at": c.created_at.strftime("%Y-%m-%d %H:%M:%S") if c.created_at else "",
}
for c in certificates
]
[文档]
def get_certificates_by_product(self, product_model: str) -> list[dict[str, str]]:
"""按产品型号获取合格证列表
查询指定产品型号的所有合格证.
Args:
product_model: 产品型号，支持完全匹配
Returns:
List[Dict[str, str]]: 合格证信息字典列表，每个字典包含id、编号、订单号和状态
"""
certificates = self._cert_service.get_certificates_by_product(product_model)
return [
{
"id": c.id,
"certificate_no": c.certificate_no,
"order_no": c.order_no,
"print_status": c.print_status,
"engrave_status": c.engrave_status,
}
for c in certificates
]
[文档]
def mark_as_engraved(self, certificate_ids: list[int]) -> dict[str, int]:
"""标记合格证为已刻印(铭牌刻印完成后调用)
将指定合格证的状态更新为"已刻印"，并记录刻印时间.
Args:
certificate_ids: 合格证ID列表
Returns:
Dict[str, int]: 更新结果字典，包含以下键:
- total: 总处理数
- success: 成功数
- failed: 失败数
"""
result = self._cert_service.mark_as_engraved(certificate_ids)
self._cert_service.commit()
return result
[文档]
def get_print_statistics(self, days: int = 7) -> dict[str, object]:
"""获取打印统计信息
统计指定天数内的打印数量、按类型统计及合格证状态分布.
Args:
days: 统计天数，默认为7天
Returns:
Dict[str, object]: 统计结果字典
"""
return self._cert_service.get_print_statistics(days)
# ============================================================
# 视图层收口（阶段 C 补丁 58）：取代 print_view 直连 Service
# ============================================================
[文档]
def get_sale_plan_by_id(self, sale_plan_id: int) -> Any:
"""按 ID 取销售计划（收口 print_view 原直连 CertificateService）。"""
return self._cert_service.get_sale_plan_by_id(sale_plan_id)
[文档]
def get_certificate_by_unique_key(self, unique_key: str) -> Any:
"""按 unique_key 取最新合格证（收口 print_view 原直连 CertificateService）。"""
return self._cert_service.get_certificate_by_unique_key(unique_key)
[文档]
def render_certificate_bytes(
self,
config: Any,
data: dict[str, Any],
copies: int = 1,
x_offsets: dict[str, float] | None = None,
printer_name: str | None = None,
) -> bytes | None:
"""渲染合格证 ESC/P-K 原始字节（仅预览/落盘，不真正出纸）。
收口 print_view 原直连 ``PrinterService.render_certificate_bytes`` 的写法，
视图层不再直接持有 Service。
"""
from certflow.services.printer.printer_service import PrinterService
try:
svc = PrinterService(printer_name)
return svc.render_certificate_bytes(config, data, copies=copies, x_offsets=x_offsets)
except Exception as exc: # noqa: BLE001
logger.debug(f"[CertificateController] 原始字节渲染失败: {exc}")
return None
[文档]
def auto_save_log(
self,
serials: list[str],
data: dict[str, Any],
printer_name: str,
certificate_id: int | None = None,
) -> None:
"""自动保存打印记录（委托 CertLogService，收口 print_view 原直连）。"""
from certflow.services.cert_log_service import CertLogService
try:
log_service = CertLogService(self.session, certificate_id)
log_service.auto_save(serials, data, printer_name)
except Exception as exc: # noqa: BLE001
logger.error(f"[CertificateController] 打印记录自动保存失败: {exc}")
[文档]
def get_certificate_by_id(self, cert_id: int) -> Any:
"""按 ID 取合格证（收口 print_view 原直连 session.get(Certificate, ...)）。"""
from certflow.models.certificate import Certificate
return self.session.get(Certificate, cert_id)
[文档]
def get_job_certificate_id(self, job_id: int) -> int | None:
"""按打印任务 ID 取关联合格证 ID（H2 事件驱动重载用）。
视图层不直连 models，经本门面查 ``print_jobs.certificate_id``。
任务不存在或 certificate_id 为空时返回 None。
"""
from certflow.models.print_job import PrintJob
job = self.session.query(PrintJob).filter(PrintJob.id == job_id).first()
if job is None:
return None
return job.certificate_id
[文档]
def save_correction(self, cert_id: int, data: dict[str, Any], template_key: str) -> None:
"""保存订正写回数据库（收口 print_view 原直连 ORM + commit）。
视图层只负责采集表单数据 / 触发护栏，不再直接持有 session 或提交事务。
"""
from certflow.models.certificate import Certificate
from certflow.services.pn_service import PNService
from certflow.services.printer.template_field_formatter import family_of
from certflow.utils.database import DatabaseManager
cert = self.session.get(Certificate, cert_id)
if cert is None:
raise RuntimeError("Certificate 不存在，无法保存订正")
cert.product_name = data["product_name"]
cert.product_model = data["product_model"]
cert.product_spec = data["dn"]
family = family_of(template_key)
pn_raw = (data.get("pn") or "").strip()
cert.pn_display, cert.pn_value, cert.pn_unit = PNService.derive_pn_fields(pn_raw, family)
cert.working_temp = data["temperature"]
cert.working_medium = data["medium"]
cert.test_standard = data["check_standard"]
cert.inspector_id = data["inspector_id"]
cert.issue_date = self._format_issue_ym(data["manufacture_date"])
cert.template_type = data["template_type"]
prefix = data["prefix"] or "V"
ym = data["year_month"]
start = data["start_number"]
quantity = data["quantity"] or 1
suffix = data["suffix"]
start_str = f"{start:03d}" if start < 1000 else str(start)
end = start + quantity - 1
end_str = f"{end:03d}" if end < 1000 else str(end)
cert.product_code_prefix = prefix
cert.product_code_ym = ym
cert.product_code_seq_start = start
cert.product_code_seq_end = end
cert.quantity = quantity
cert.product_code_range = (
f"{prefix}{ym}{start_str}{suffix}"
if quantity == 1
else f"{prefix}{ym}{start_str}{suffix}---{end_str}{suffix}"
)
# 续号保序：手动编号写库后推进计数器（异常不影响本次写库）
try:
self.sync_serial_counter(prefix, end)
except Exception as exc: # noqa: BLE001
logger.warning(f"[CertificateController] 续号计数器同步失败: {exc}")
DatabaseManager.commit_with_retry(self.session)
logger.info(
f"[CertificateController] 订正已写库 cert_id={cert.id} | "
f"{cert.product_name} {cert.product_model}"
)
@staticmethod
def _format_issue_ym(issue_date: str | None) -> str:
"""出厂日期格式化为 YYYY.MM（与 BUG-006 J 展示口径一致）。"""
if not issue_date:
return datetime.now().strftime("%Y.%m")
s = str(issue_date).replace("/", "-")
parts = s.split("-")
if len(parts) >= 2:
return f"{parts[0]}.{parts[1]}"
return str(issue_date)
[文档]
def get_ungenerated_sale_plans(self) -> list[SalePlan]:
"""获取未生成合格证的销售计划
查询所有尚未关联合格证记录的销售计划.
Returns:
List[SalePlan]: 未生成合格证的销售计划对象列表
"""
return self._cert_service.get_ungenerated_sale_plans()
[文档]
def get_all_certificates(self, limit: int = 100, offset: int = 0) -> list[Certificate]:
"""获取所有合格证(分页)
按创建时间倒序分页查询合格证列表.
Args:
limit: 每页数量，默认为100
offset: 偏移量，默认为0
Returns:
List[Certificate]: 合格证对象列表
"""
return self._cert_service.get_all_certificates(limit, offset)