certflow.services.certificate_service module

证书服务模块

提供合格证的查询、创建、更新和删除等业务服务。

class certflow.services.certificate_service.CertificateService(session)

基类：object

证书服务类

提供合格证数据的CRUD操作服务，供控制器层调用。 封装业务逻辑，协调数据访问层完成各类证书操作。

参数:

session (Session)

_session

数据库会话对象(当前为占位，实际实现时需注入)

示例

>>> 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']}")

static generate_certificate_no(prefix='CERT', sequence=None)

生成合格证编号

生成格式为 PREFIX-YYYYMMDD-NNNN 的合格证编号，序号部分自动补零至4位。

参数:

• prefix (str) -- 编号前缀，默认为 "CERT"

• sequence (int | None) -- 序号，如果为None则默认为1

返回:

生成的合格证编号字符串，如 "CERT-20231201-0001"

返回类型:

str

get_certificate_by_sale_plan_id(sale_plan_id)

根据销售计划 ID 查询已存在的合格证

参数:

sale_plan_id (int) -- 销售计划 ID

返回:

合格证对象，未找到返回 None

返回类型:

Certificate | None

get_sale_plan_by_id(sale_plan_id)

根据 ID 查询销售计划

参数:

sale_plan_id (int) -- 销售计划 ID

返回:

销售计划对象，未找到返回 None

返回类型:

SalePlan | None

count_certificates_created_today()

统计当天已创建的合格证数量

返回:

当天创建的合格证总数

返回类型:

int

CERT_EDITABLE_FIELDS: tuple[str, ...] = ('product_name', 'product_model', 'product_spec', 'customer', 'project_name', 'test_standard', 'supplier', 'issue_date', 'working_temp', 'working_medium', 'remarks', 'pn_display', 'inspector_id', 'template_type')

FORM_TO_CERT_MAP: dict[str, str] = {'check_standard': 'test_standard', 'dn': 'product_spec', 'inspector_id': 'inspector_id', 'manufacture_date': 'issue_date', 'medium': 'working_medium', 'pn': 'pn_display', 'product_model': 'product_model', 'product_name': 'product_name', 'temperature': 'working_temp', 'template_type': 'template_type'}

CERT_TO_SALEPLAN_CERT_MAP: dict[str, str] = {'product_model': 'cert_product_model', 'product_name': 'cert_product_name', 'product_spec': 'cert_product_spec'}

update_fields(cert_id, fields)

仅更新合格证记录的白名单字段（订正合格证记录）

与 SalePlanService.update_cert_fields 对称：只写允许订正的打印字段， 非白名单字段被静默忽略，绝不触动编号/状态等系统字段。 certificate_no 是唯一键，不在白名单内，避免误改导致唯一约束冲突。

参数:

• cert_id (int) -- 合格证记录主键

• fields (dict[str, Any]) -- 待写入字段字典

返回:

是否成功更新（id 不存在 / 无可写字段返回 False）

返回类型:

bool

correct_print_fields(cert_id, sale_plan_id, fields)

回源改正（蓝图 §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。

参数:

• cert_id (int) -- 合格证记录主键

• sale_plan_id (int | None) -- 关联销售计划主键（None 时跳过 SalePlan 覆盖层写）

• fields (dict[str, Any]) -- 表单字段字典（键为表单键，如 product_name/dn/pn/…）

返回:

是否执行了改正（Certificate 或 SalePlan 有写）

返回类型:

bool

get_certificate_by_unique_key(unique_key)

根据唯一键查询合格证（取最新一条）

参数:

unique_key (str) -- 合格证唯一键

返回:

合格证对象，未找到返回 None

返回类型:

Certificate | None

get_certificates_by_ids(certificate_ids)

根据 ID 列表批量获取合格证

参数:

certificate_ids (list[int]) -- 合格证 ID 列表

返回:

合格证对象列表

返回类型:

list[Certificate]

get_certificates_by_status(print_status='待打印')

按打印状态获取合格证列表

参数:

print_status (str) -- 打印状态，默认 "待打印"

返回:

合格证对象列表

返回类型:

list[Certificate]

get_certificates_by_product(product_model)

按产品型号获取合格证列表

参数:

product_model (str) -- 产品型号

返回:

合格证对象列表

返回类型:

list[Certificate]

get_all_certificates(limit=100, offset=0)

获取所有合格证（分页）

参数:

• limit (int) -- 每页数量

• offset (int) -- 偏移量

返回:

合格证对象列表

返回类型:

list[Certificate]

get_ungenerated_sale_plans()

获取未生成合格证的销售计划

返回:

未生成合格证的销售计划列表

返回类型:

list[SalePlan]

add(obj)

将对象添加到会话

参数:

obj (Any) -- 要持久化的 ORM 对象。

返回类型:

None

merge_certificate(certificate)

合并更新合格证（用于 force_regenerate 场景）

参数:

certificate (Certificate) -- 合格证对象

返回类型:

None

flush()

刷新会话，确保数据库操作已执行

返回类型:

None

commit()

提交事务

返回类型:

None

mark_as_engraved(certificate_ids)

标记合格证为已刻印

参数:

certificate_ids (list[int]) -- 合格证 ID 列表

返回:

{total, success, failed}

返回类型:

更新结果字典

get_print_statistics(days=7)

获取打印统计信息

参数:

days (int) -- 统计天数

返回:

统计结果字典

返回类型:

dict[str, object]

get_certificate(cert_id)

获取证书信息

根据证书ID查询并返回证书详细信息。

参数:

cert_id (int) -- 证书的唯一标识ID

返回:

证书信息字典，包含以下字段:

• id: 证书ID

• certificate_no: 合格证编号

• product_model: 产品型号

• order_no: 订单号

• customer: 客户名称

• status: 合格证状态

• print_time: 打印时间

• is_engraved: 是否已刻印

未找到时返回 None。

返回类型:

Optional[Dict[str, Any]]

示例

>>> service = CertificateService()
>>> cert = service.get_certificate(100)
>>> if cert:
...     print(f"编号: {cert['certificate_no']}, 状态: {cert['status']}")

create_certificate(data)

创建证书

根据提供的数据创建新的证书记录。

参数:

data (dict[str, Any]) -- 证书数据字典，应包含以下字段: - certificate_no: 合格证编号(必填) - sale_plan_id: 关联销售计划ID(必填) - unique_key: 唯一键(必填) - product_model: 产品型号 - order_no: 订单号 - customer: 客户名称 - status: 合格证状态，默认为"待打印"

返回:

创建结果字典，包含以下字段:

• success: 是否成功(布尔值)

• message: 结果消息

• certificate_id: 成功时返回新创建的证书ID

• error: 失败时返回错误信息

返回类型:

Dict[str, Any]

示例

>>> 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']}")

update_certificate(cert_id, data)

更新证书

根据证书ID更新证书的指定字段。

参数:

• cert_id (int) -- 证书的唯一标识ID

• data (dict[str, Any]) -- 需要更新的字段字典，可包含以下字段: - certificate_no: 合格证编号 - product_model: 产品型号 - order_no: 订单号 - customer: 客户名称 - status: 合格证状态 - print_time: 打印时间 - is_engraved: 是否已刻印 - engraved_time: 刻印时间 - electronic_image_path: 电子版图片路径

返回:

更新结果字典，包含以下字段:

• success: 是否成功(布尔值)

• message: 结果消息

• error: 失败时返回错误信息

返回类型:

Dict[str, Any]

示例

>>> service = CertificateService()
>>> result = service.update_certificate(100, {
...     "status": "已打印",
...     "print_time": "2024-01-15 10:30:00"
... })
>>> if result["success"]:
...     print("证书更新成功")

delete_certificate(cert_id)

删除证书

根据证书ID删除对应的证书记录。

参数:

cert_id (int) -- 证书的唯一标识ID

返回:

删除结果字典，包含以下字段:

• success: 是否成功(布尔值)

• message: 结果消息

• error: 失败时返回错误信息

返回类型:

Dict[str, Any]

示例

>>> service = CertificateService()
>>> result = service.delete_certificate(100)
>>> if result["success"]:
...     print("证书删除成功")
... else:
...     print(f"删除失败: {result['error']}")