certflow.controllers.certificate_controller module

合格证控制器模块

处理合格证的生成、打印、刻印标记及统计查询等业务逻辑. 支持批量操作和按产品型号分组打印.

class certflow.controllers.certificate_controller.CertificateController(db_session=None, db_manager=None)

基类：BaseController

合格证控制器 - 处理合格证打印、刻印等业务

提供合格证的生成、批量打印、刻印标记、状态查询及统计等功能. 支持按产品型号分组打印，自动处理合格证编号生成和打印日志记录.

参数:

• db_session (Session | None)

• db_manager (Any)

certificate_cache

合格证编号缓存字典，用于避免重复生成编号

generate_certificates_for_sale_plans(sale_plan_ids, force_regenerate=False)

为销售计划批量生成合格证

遍历指定的销售计划ID列表，为每条销售计划生成对应的合格证记录. 当force_regenerate为True时，即使合格证已存在也会重新生成.

参数:

• sale_plan_ids (list[int]) -- 销售计划ID列表

• force_regenerate (bool) -- 是否强制重新生成已存在的合格证，默认为False

返回:

生成结果字典，包含以下键:

• total: 处理总数

• generated: 成功生成数

• failed: 失败数

• certificates: 生成的合格证对象列表

返回类型:

Dict[str, object]

示例

>>> controller = CertificateController()
>>> result = controller.generate_certificates_for_sale_plans([1, 2, 3])
>>> print(f"生成成功: {result['generated']}/{result['total']}")

get_history_last_values(model)

取同型号上次实际打印值（供统一护栏 ④ diff 闸）。

收口 print_view 原先直连 HistoryBackfillService(self.session) 的写法， 视图层不再直接持有 Service，统一经 Controller 访问。

参数:

model (str | None) -- 产品型号字符串

返回:

历史候选值；无历史或异常时返回 None

返回类型:

dict[str, str] | None

get_next_serial(prefix)

只读预瞄某前缀下一个可用流水号（不消耗计数器，供 UI 自动续号/预览）。

参数:

prefix (str) -- 编号前缀（如 V2604）

返回:

下一可用流水号（跨 SalePlan + Certificate 取该月最大 + 1）

返回类型:

int

sync_serial_counter(prefix, end_serial)

手动编号写库后，将计数器推进到 end_serial，保证后续自动续号不回退。

参数:

• prefix (str) -- 编号前缀（如 V2604）

• end_serial (int) -- 本次手动编号覆盖到的最大流水号

返回类型:

None

batch_print_certificates(certificate_ids, printer_name='Default Printer', copies=1)

批量打印合格证(相同产品会分组打印)

按产品型号对合格证进行分组，逐组执行打印操作， 并为每张合格证记录打印日志.

参数:

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

• printer_name (str) -- 打印机名称，默认为"Default Printer"

• copies (int) -- 每张合格证的打印份数，默认为1

返回:

打印结果字典，包含以下键:

• total: 总处理数

• success: 成功数

• failed: 失败数

• groups: 各产品组的打印详情列表

• logs: 打印日志列表

返回类型:

Dict[str, object]

示例

>>> controller = CertificateController()
>>> result = controller.batch_print_certificates([1, 2, 3], "HP LaserJet", 2)
>>> print(f"打印成功: {result['success']}/{result['total']}")

get_certificates_by_status(print_status='待打印')

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

查询指定打印状态的所有合格证，按创建时间升序排序.

参数:

print_status (str) -- 合格证打印状态，可选值: "待打印", "已打印", "打印失败"等，默认为"待打印"

返回:

合格证信息字典列表，每个字典包含id、编号、产品型号等字段

返回类型:

List[Dict[str, str]]

get_certificates_by_product(product_model)

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

查询指定产品型号的所有合格证.

参数:

product_model (str) -- 产品型号，支持完全匹配

返回:

合格证信息字典列表，每个字典包含id、编号、订单号和状态

返回类型:

List[Dict[str, str]]

mark_as_engraved(certificate_ids)

标记合格证为已刻印(铭牌刻印完成后调用)

将指定合格证的状态更新为"已刻印"，并记录刻印时间.

参数:

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

返回:

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

• total: 总处理数

• success: 成功数

• failed: 失败数

返回类型:

Dict[str, int]

get_print_statistics(days=7)

获取打印统计信息

统计指定天数内的打印数量、按类型统计及合格证状态分布.

参数:

days (int) -- 统计天数，默认为7天

返回:

统计结果字典

返回类型:

Dict[str, object]

get_sale_plan_by_id(sale_plan_id)

按 ID 取销售计划（收口 print_view 原直连 CertificateService）。

参数:

sale_plan_id (int)

返回类型:

Any

get_certificate_by_unique_key(unique_key)

按 unique_key 取最新合格证（收口 print_view 原直连 CertificateService）。

参数:

unique_key (str)

返回类型:

Any

render_certificate_bytes(config, data, copies=1, x_offsets=None, printer_name=None)

渲染合格证 ESC/P-K 原始字节（仅预览/落盘，不真正出纸）。

收口 print_view 原直连 PrinterService.render_certificate_bytes 的写法， 视图层不再直接持有 Service。

参数:

• config (Any)

• data (dict[str, Any])

• copies (int)

• x_offsets (dict[str, float] | None)

• printer_name (str | None)

返回类型:

bytes | None

auto_save_log(serials, data, printer_name, certificate_id=None)

自动保存打印记录（委托 CertLogService，收口 print_view 原直连）。

参数:

• serials (list[str])

• data (dict[str, Any])

• printer_name (str)

• certificate_id (int | None)

返回类型:

None

get_certificate_by_id(cert_id)

按 ID 取合格证（收口 print_view 原直连 session.get(Certificate, ...)）。

参数:

cert_id (int)

返回类型:

Any

get_job_certificate_id(job_id)

按打印任务 ID 取关联合格证 ID（H2 事件驱动重载用）。

视图层不直连 models，经本门面查 print_jobs.certificate_id。 任务不存在或 certificate_id 为空时返回 None。

参数:

job_id (int)

返回类型:

int | None

save_correction(cert_id, data, template_key)

保存订正写回数据库（收口 print_view 原直连 ORM + commit）。

视图层只负责采集表单数据 / 触发护栏，不再直接持有 session 或提交事务。

参数:

• cert_id (int)

• data (dict[str, Any])

• template_key (str)

返回类型:

None

get_ungenerated_sale_plans()

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

查询所有尚未关联合格证记录的销售计划.

返回:

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

返回类型:

List[SalePlan]

get_all_certificates(limit=100, offset=0)

获取所有合格证(分页)

按创建时间倒序分页查询合格证列表.

参数:

• limit (int) -- 每页数量，默认为100

• offset (int) -- 偏移量，默认为0

返回:

合格证对象列表

返回类型:

List[Certificate]