"""证书服务模块
提供合格证的查询、创建、更新和删除等业务服务。
"""
from __future__ import annotations
from datetime import datetime
from typing import Any
from loguru import logger
from sqlalchemy import func
from sqlalchemy.exc import SQLAlchemyError
from sqlalchemy.orm import Session
from certflow.models import Certificate, PrintLog, SalePlan
[文档]
class CertificateService:
"""证书服务类
提供合格证数据的CRUD操作服务，供控制器层调用。
封装业务逻辑，协调数据访问层完成各类证书操作。
Attributes:
_session: 数据库会话对象(当前为占位，实际实现时需注入)
Examples:
>>> service = CertificateService()
>>> # 创建证书
>>> cert = service.create_certificate({
... "certificate_no": "CERT-20231201-0001",
... "product_model": "阀门A",
... "customer": "某某公司"
... })
>>> print(cert.get("success"))
True
>>>
>>> # 查询证书
>>> cert_info = service.get_certificate(1)
>>> if cert_info:
... print(f"证书编号: {cert_info['certificate_no']}")
"""
def __init__(self, session: Session) -> None:
"""初始化证书服务
Args:
session: SQLAlchemy 数据库会话对象
"""
self.session = session
[文档]
@staticmethod
def generate_certificate_no(prefix: str = "CERT", sequence: int | None = None) -> str:
"""生成合格证编号
生成格式为 PREFIX-YYYYMMDD-NNNN 的合格证编号，序号部分自动补零至4位。
Args:
prefix: 编号前缀，默认为 "CERT"
sequence: 序号，如果为None则默认为1
Returns:
str: 生成的合格证编号字符串，如 "CERT-20231201-0001"
"""
date_str = datetime.now().strftime("%Y%m%d")
if sequence is None:
sequence = 1
return f"{prefix}-{date_str}-{sequence:04d}"
# ============================================================
# 查询方法
# ============================================================
[文档]
def get_certificate_by_sale_plan_id(self, sale_plan_id: int) -> Certificate | None:
"""根据销售计划 ID 查询已存在的合格证
Args:
sale_plan_id: 销售计划 ID
Returns:
合格证对象，未找到返回 None
"""
return (
self.session.query(Certificate).filter(Certificate.sale_plan_id == sale_plan_id).first()
)
[文档]
def get_sale_plan_by_id(self, sale_plan_id: int) -> SalePlan | None:
"""根据 ID 查询销售计划
Args:
sale_plan_id: 销售计划 ID
Returns:
销售计划对象，未找到返回 None
"""
return self.session.query(SalePlan).filter(SalePlan.id == sale_plan_id).first()
[文档]
def count_certificates_created_today(self) -> int:
"""统计当天已创建的合格证数量
Returns:
当天创建的合格证总数
"""
today = datetime.now().date()
today_start = datetime(today.year, today.month, today.day)
return self.session.query(Certificate).filter(Certificate.created_at >= today_start).count()
# 允许手工订正的合格证记录字段（白名单，蓝图 §4.3 步骤 5）
# 仅含合格证打印字段；certificate_no 是唯一键、print_status 等是系统字段，不在内。
CERT_EDITABLE_FIELDS: tuple[str, ...] = (
"product_name",
"product_model",
"product_spec",
"customer",
"project_name",
"test_standard",
"supplier",
"issue_date",
"working_temp",
"working_medium",
"remarks",
# 以下为回源改正(§10.7 保存订正)需同步的打印层字段；此前漏入白名单，
# 导致 correct_print_fields 经 update_fields 写入时被静默丢弃（仅打印路径
# update_certificate_status 才写），与 §10.7 FORM_TO_CERT_MAP 意图矛盾。
"pn_display", # 表单 pn → 公称压力显示文本
"inspector_id", # 检验工号
"template_type", # 模板语言族
)
# 表单键 → Certificate 列（蓝图 §10.7 回源改正归一化；表单用 dn/pn 键，库列 product_spec/pn_display）
FORM_TO_CERT_MAP: dict[str, str] = {
"product_name": "product_name",
"product_model": "product_model",
"dn": "product_spec",
"pn": "pn_display",
"temperature": "working_temp",
"medium": "working_medium",
"check_standard": "test_standard",
"inspector_id": "inspector_id",
"manufacture_date": "issue_date",
"template_type": "template_type",
}
# Certificate 列 → SalePlan 合格证覆盖层(cert_*)（仅产品三字段，蓝图 §10.8-6）
CERT_TO_SALEPLAN_CERT_MAP: dict[str, str] = {
"product_name": "cert_product_name",
"product_model": "cert_product_model",
"product_spec": "cert_product_spec",
}
[文档]
def update_fields(self, cert_id: int, fields: dict[str, Any]) -> bool:
"""仅更新合格证记录的白名单字段（订正合格证记录）
与 ``SalePlanService.update_cert_fields`` 对称：只写允许订正的打印字段，
非白名单字段被静默忽略，绝不触动编号/状态等系统字段。
``certificate_no`` 是唯一键，不在白名单内，避免误改导致唯一约束冲突。
Args:
cert_id: 合格证记录主键
fields: 待写入字段字典
Returns:
bool: 是否成功更新（id 不存在 / 无可写字段返回 False）
"""
if not fields:
return False
allowed = {k: v for k, v in fields.items() if k in self.CERT_EDITABLE_FIELDS}
if not allowed:
return False
cert = self.session.get(Certificate, cert_id)
if cert is None:
logger.warning(f"update_fields 跳过：cert_id={cert_id} 不存在")
return False
try:
for key, value in allowed.items():
setattr(cert, key, value)
self.session.commit()
logger.info(f"订正合格证记录 | cert_id={cert_id} | 字段={sorted(allowed.keys())}")
return True
except SQLAlchemyError as e:
self.session.rollback()
logger.error(f"订正合格证记录失败，已回滚: {e}")
raise
[文档]
def correct_print_fields(
self, cert_id: int, sale_plan_id: int | None, fields: dict[str, Any]
) -> bool:
"""回源改正（蓝图 §10.7 / §10.8-6）：白名单写 Certificate 打印层 +
SalePlan 合格证覆盖层(cert_*) 同步，并对型号/压力/标准自学习字典。
- 表单键(dn/pn/…) 归一化为 Certificate 列(product_spec/pn_display/…)；
- 产品名称/型号/规格同时落到 SalePlan.cert_product_* 覆盖层，两层副本一致；
- 绝不写 SalePlan.product_name/model/spec 合同层（参与导入唯一键）；
- 压力/温度/介质/标准额外自学习 ModelParamMapping / CaliberMapping。
Args:
cert_id: 合格证记录主键
sale_plan_id: 关联销售计划主键（None 时跳过 SalePlan 覆盖层写）
fields: 表单字段字典（键为表单键，如 product_name/dn/pn/…）
Returns:
bool: 是否执行了改正（Certificate 或 SalePlan 有写）
"""
if not fields:
return False
# 1) 表单键 → Certificate 列
cert_fields = {
self.FORM_TO_CERT_MAP[k]: v for k, v in fields.items() if k in self.FORM_TO_CERT_MAP
}
if not cert_fields:
return False
# 2) Certificate 打印层（复用 update_fields 白名单）
self.update_fields(cert_id, cert_fields)
# 3) SalePlan 合格证覆盖层（仅产品三字段）
if sale_plan_id:
sp = self.session.get(SalePlan, sale_plan_id)
if sp is not None:
sp_fields = {
sp_col: cert_fields[cert_col]
for cert_col, sp_col in self.CERT_TO_SALEPLAN_CERT_MAP.items()
if cert_col in cert_fields
}
if sp_fields:
try:
for col, val in sp_fields.items():
setattr(sp, col, val)
self.session.commit()
except SQLAlchemyError as e:
self.session.rollback()
logger.error(f"回源改正 SalePlan 覆盖层失败，已回滚: {e}")
raise
# 4) 字典自学习（型号→压力/标准，规格→口径），失败不阻断订正
try:
from certflow.services.correction_queue_service import CorrectionQueueService
sp = self.session.get(SalePlan, sale_plan_id) if sale_plan_id else None
if sp is not None:
cqs = CorrectionQueueService(self.session)
old_spec = str(getattr(sp, "product_spec", "") or "")
old_model = str(getattr(sp, "product_model", "") or "")
cqs._learn_dicts(sp, old_spec, old_model, cert_fields)
except Exception as e: # 自学习失败不应阻断订正
logger.warning(f"回源改正字典自学习失败(忽略，订正已生效): {e}")
return True
[文档]
def get_certificate_by_unique_key(self, unique_key: str) -> Certificate | None:
"""根据唯一键查询合格证（取最新一条）
Args:
unique_key: 合格证唯一键
Returns:
合格证对象，未找到返回 None
"""
return (
self.session.query(Certificate)
.filter(Certificate.unique_key == unique_key)
.order_by(Certificate.id.desc())
.first()
)
[文档]
def get_certificates_by_ids(self, certificate_ids: list[int]) -> list[Certificate]:
"""根据 ID 列表批量获取合格证
Args:
certificate_ids: 合格证 ID 列表
Returns:
合格证对象列表
"""
return self.session.query(Certificate).filter(Certificate.id.in_(certificate_ids)).all()
[文档]
def get_certificates_by_status(self, print_status: str = "待打印") -> list[Certificate]:
"""按打印状态获取合格证列表
Args:
print_status: 打印状态，默认 "待打印"
Returns:
合格证对象列表
"""
return (
self.session.query(Certificate)
.filter(Certificate.print_status == print_status)
.order_by(Certificate.created_at)
.all()
)
[文档]
def get_certificates_by_product(self, product_model: str) -> list[Certificate]:
"""按产品型号获取合格证列表
Args:
product_model: 产品型号
Returns:
合格证对象列表
"""
return (
self.session.query(Certificate).filter(Certificate.product_model == product_model).all()
)
[文档]
def get_all_certificates(self, limit: int = 100, offset: int = 0) -> list[Certificate]:
"""获取所有合格证（分页）
Args:
limit: 每页数量
offset: 偏移量
Returns:
合格证对象列表
"""
return (
self.session.query(Certificate)
.order_by(Certificate.created_at.desc())
.offset(offset)
.limit(limit)
.all()
)
[文档]
def get_ungenerated_sale_plans(self) -> list[SalePlan]:
"""获取未生成合格证的销售计划
Returns:
未生成合格证的销售计划列表
"""
generated_ids = self.session.query(Certificate.sale_plan_id).distinct().all()
generated_ids = [g[0] for g in generated_ids]
if generated_ids:
return self.session.query(SalePlan).filter(~SalePlan.id.in_(generated_ids)).all()
return self.session.query(SalePlan).all()
# ============================================================
# 写入方法
# ============================================================
[文档]
def add(self, obj: Any) -> None:
"""将对象添加到会话
Args:
obj: 要持久化的 ORM 对象。
"""
self.session.add(obj)
[文档]
def merge_certificate(self, certificate: Certificate) -> None:
"""合并更新合格证（用于 force_regenerate 场景）
Args:
certificate: 合格证对象
"""
self.session.merge(certificate)
[文档]
def flush(self) -> None:
"""刷新会话，确保数据库操作已执行"""
self.session.flush()
[文档]
def commit(self) -> None:
"""提交事务"""
self.session.commit()
# ============================================================
# 更新方法
# ============================================================
[文档]
def mark_as_engraved(self, certificate_ids: list[int]) -> dict[str, int]:
"""标记合格证为已刻印
Args:
certificate_ids: 合格证 ID 列表
Returns:
更新结果字典: {total, success, failed}
"""
result: dict[str, int] = {"total": len(certificate_ids), "success": 0, "failed": 0}
for cert_id in certificate_ids:
try:
certificate = (
self.session.query(Certificate).filter(Certificate.id == cert_id).first()
)
if certificate:
certificate.is_engraved = True
certificate.engraved_time = datetime.now()
certificate.print_status = "已打印"
certificate.engrave_status = "已刻印"
result["success"] += 1
logger.info(f"标记刻印: {certificate.certificate_no}")
else:
result["failed"] += 1
logger.warning(f"合格证不存在: id={cert_id}")
except Exception as e:
result["failed"] += 1
logger.error(f"标记刻印失败: {e}")
return result
# ============================================================
# 统计方法
# ============================================================
[文档]
def get_print_statistics(self, days: int = 7) -> dict[str, object]:
"""获取打印统计信息
Args:
days: 统计天数
Returns:
统计结果字典
"""
from datetime import timedelta
end_date = datetime.now()
start_date = end_date - timedelta(days=days)
print_count = (
self.session.query(PrintLog).filter(PrintLog.hgz_PrintTime >= start_date).count()
)
by_type = (
self.session.query(PrintLog.status, func.count(PrintLog.id))
.filter(PrintLog.hgz_PrintTime >= start_date)
.group_by(PrintLog.status)
.all()
)
cert_stats = (
self.session.query(Certificate.print_status, func.count(Certificate.id))
.group_by(Certificate.print_status)
.all()
)
return {
"period": f"{days}天",
"total_prints": print_count,
"by_type": [{"status": s, "count": c} for s, c in by_type],
"certificate_status": [{"print_status": s, "count": c} for s, c in cert_stats],
}
# ============================================================
# 原有骨架方法（保持向后兼容）
# ============================================================
[文档]
def get_certificate(self, cert_id: int) -> dict[str, Any] | None:
"""获取证书信息
根据证书ID查询并返回证书详细信息。
Args:
cert_id: 证书的唯一标识ID
Returns:
Optional[Dict[str, Any]]: 证书信息字典，包含以下字段:
- id: 证书ID
- certificate_no: 合格证编号
- product_model: 产品型号
- order_no: 订单号
- customer: 客户名称
- status: 合格证状态
- print_time: 打印时间
- is_engraved: 是否已刻印
未找到时返回 None。
Examples:
>>> service = CertificateService()
>>> cert = service.get_certificate(100)
>>> if cert:
... print(f"编号: {cert['certificate_no']}, 状态: {cert['status']}")
"""
pass
[文档]
def create_certificate(self, data: dict[str, Any]) -> dict[str, Any]:
"""创建证书
根据提供的数据创建新的证书记录。
Args:
data: 证书数据字典，应包含以下字段:
- certificate_no: 合格证编号(必填)
- sale_plan_id: 关联销售计划ID(必填)
- unique_key: 唯一键(必填)
- product_model: 产品型号
- order_no: 订单号
- customer: 客户名称
- status: 合格证状态，默认为"待打印"
Returns:
Dict[str, Any]: 创建结果字典，包含以下字段:
- success: 是否成功(布尔值)
- message: 结果消息
- certificate_id: 成功时返回新创建的证书ID
- error: 失败时返回错误信息
Examples:
>>> service = CertificateService()
>>> result = service.create_certificate({
... "certificate_no": "CERT-20231201-0001",
... "sale_plan_id": 100,
... "unique_key": "abc123",
... "product_model": "阀门A",
... "customer": "某某公司"
... })
>>> if result["success"]:
... print(f"证书创建成功，ID: {result['certificate_id']}")
... else:
... print(f"创建失败: {result['error']}")
"""
pass
[文档]
def update_certificate(self, cert_id: int, data: dict[str, Any]) -> dict[str, Any]:
"""更新证书
根据证书ID更新证书的指定字段。
Args:
cert_id: 证书的唯一标识ID
data: 需要更新的字段字典，可包含以下字段:
- certificate_no: 合格证编号
- product_model: 产品型号
- order_no: 订单号
- customer: 客户名称
- status: 合格证状态
- print_time: 打印时间
- is_engraved: 是否已刻印
- engraved_time: 刻印时间
- electronic_image_path: 电子版图片路径
Returns:
Dict[str, Any]: 更新结果字典，包含以下字段:
- success: 是否成功(布尔值)
- message: 结果消息
- error: 失败时返回错误信息
Examples:
>>> service = CertificateService()
>>> result = service.update_certificate(100, {
... "status": "已打印",
... "print_time": "2024-01-15 10:30:00"
... })
>>> if result["success"]:
... print("证书更新成功")
"""
pass
[文档]
def delete_certificate(self, cert_id: int) -> dict[str, Any]:
"""删除证书
根据证书ID删除对应的证书记录。
Args:
cert_id: 证书的唯一标识ID
Returns:
Dict[str, Any]: 删除结果字典，包含以下字段:
- success: 是否成功(布尔值)
- message: 结果消息
- error: 失败时返回错误信息
Examples:
>>> service = CertificateService()
>>> result = service.delete_certificate(100)
>>> if result["success"]:
... print("证书删除成功")
... else:
... print(f"删除失败: {result['error']}")
"""
pass