certflow.services package

Subpackages

Submodules

Module contents

服务层模块

提供系统核心业务服务的统一导入接口。 包含销售计划服务、证书打印服务、报告服务和查询服务等业务逻辑层组件。

分层约束：本包属 L3 服务层，须保证在无 X11 / 无 PySide6 的 headless 环境下可被导入。printer_manager 仍持有 Qt 信号壳（待阶段 B 上移至 controllers 层），因此本模块采用 PEP 562 __getattr__ 惰性导出： 仅在真正访问 PrinterManager / PrinterStatus / PrinterInfo 时才导入该子模块，避免 import certflow.services 即拉起 Qt。

注意：query_builder``（QueryBuilder/BatchUpdater/TempTableManager/ StatisticsReporter）为 Access ``SignTb 直连 SQL 的旧体系残留，仅供 scripts/sync_databases_cli.py 同步脚本专用，不在此公共导出中， 亦不纳入 ORM 版 QueryService 体系（见 BLUEPRINT §2.3 偏差 6）。

示例

>>> from certflow.services import QueryService  # 不触发 Qt 加载
>>> from certflow.services import PrinterManager  # 此时才导入 printer_manager
class certflow.services.CertificatePrintService(session, output_dir='output/certificates', printer_name='')[源代码]

基类：object

合格证打印服务（编排层）

组合编号、打印引擎、日志等子模块，提供完整的打印业务流程。

编号模式（委托 CertNumberingService）：

  • auto_number: 自动编号（正则解析 production_order_no / supply_type，最常用）

  • manual_number: 手动指定前缀和起始序号

  • copy_number: 复制已有编号（单件直接复制）

打印（委托 CertPrintEngine）：

  • print_single: 单台 GDI 精确打印

  • batch_print: 批量展开编号 → 逐台打印 → 记录日志

  • print_html_fallback: HTML 降级打印

持久化（委托 CertNumberingService）：

  • create_certificates: 将已编号的 SalePlan 写入 Certificate 表

  • export_to_access: 导出到 Access 数据库

  • export_sign_tb_csv: 导出 CSV

日志与历史：

  • record_print_log: 记录打印日志到 SQLite print_logs 表

  • get_print_history: 查询打印历史

参数:

  • session (Session)

  • output_dir (str)

  • printer_name (str)

session

SQLAlchemy 数据库会话对象

output_dir

输出目录路径

printer_name

默认打印机名称

示例

>>> service = CertificatePrintService(session)
>>> # 自动编号
>>> count = service.auto_number([1, 2, 3], prefix_override="2506")
>>> # 写入 Certificate
>>> cert_count = service.create_certificates([1, 2, 3])
>>> # 逐台打印
>>> service.batch_print(certificate_id=1, printer_name="EPSON LQ-635KII")
auto_number(ids, prefix_override=None)[源代码]

自动编号（委托 CertNumberingService）

通过正则解析 production_order_no / supply_type 自动生成前缀并编号。

参数:

  • ids (list[int]) -- SalePlan.id 列表

  • prefix_override (str | None) -- 自定义前缀覆盖（如 "2506"），None 则按规则推导

返回:

成功编号的记录数

返回类型:

int

manual_number(ids, prefix, start, quantity, suffix='')[源代码]

手动编号（委托 CertNumberingService）

参数:

  • ids (list[int]) -- SalePlan.id 列表

  • prefix (str) -- 编号前缀

  • start (int) -- 起始序号

  • quantity (int) -- 每台数量

  • suffix (str) -- 编号后缀

返回:

成功编号的记录数

返回类型:

int

copy_number(ids, source_code)[源代码]

复制编号（委托 CertNumberingService）

将已有编号直接复制到目标记录（单件直接复制）。

参数:

  • ids (list[int]) -- SalePlan.id 列表

  • source_code (str) -- 源编号（作为复制模板）

返回:

成功编号的记录数

返回类型:

int

create_certificates(ids, printer_name='', print_status='已打印')[源代码]

将已编号的记录写入 Certificate 表（委托 CertNumberingService）

参数:

  • ids (list[int]) -- SalePlan.id 列表

  • printer_name (str) -- 写入的打印机名称

  • print_status (str) -- 写入的打印状态，默认 "已打印"

返回:

成功创建的 Certificate 记录数

返回类型:

int

export_to_access(ids, mdb_path=None)[源代码]

导出到 Access 数据库（委托 CertNumberingService）

参数:

  • ids (list[int]) -- Certificate.id 列表

  • mdb_path (str | None) -- 目标 .mdb 文件路径，None 则使用默认路径

返回:

成功导出的记录数

返回类型:

int

export_sign_tb_csv(output_path=None, limit=1000)[源代码]

导出 CSV（委托 CertNumberingService）

参数:

  • output_path (str | None) -- 输出 CSV 文件路径，None 则使用默认路径

  • limit (int) -- 最大导出条数

返回:

生成的 CSV 文件路径；失败返回 None

返回类型:

str | None

generate_print_certificates(ids, prefix_override=None, print_status='待打印')[源代码]

为选中的销售计划生成合格证（去重安全）。

跳过已生成 Certificate 的计划（避免证书重复生成与唯一约束冲突）， 对剩余计划自动编号并写入 Certificate 表。编号与写库均委托 ``CertNumberingService``（数据访问层），本方法仅做编排与去重判定。

参数:

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

  • prefix_override (str | None) -- 编号前缀覆盖（None 则按规则推导）

  • print_status (str) -- 写入的打印状态

返回:

(编号数, 合格证生成数)

返回类型:

tuple[int, int]

print_single(data, serial_number, printer_name=None, print_fields=None)[源代码]

打印单台合格证（委托 CertPrintEngine）

参数:

  • data (dict[str, Any]) -- 打印数据字典（含 product_name、dn、pn 等字段）

  • serial_number (str) -- 合格证编号（SN）

  • printer_name (str | None) -- 打印机名称，None 则使用服务默认打印机

  • print_fields (list[str] | None) -- 需要打印的字段名列表，None 表示打印全部字段

返回:

是否打印成功

返回类型:

bool

property last_os_job_id: int | None

最近一次打印提交的 OS 作业 ID（委托 CertPrintEngine 捕获）。

偏差 5 作业 ID 捕获链路末端：队列的 _default_print_func 经此属性 拿到 spooler 作业 ID，写入 PrintJob.os_job_id，供 OS 级取消使用。

print_html_fallback(data, serial_number)[源代码]

HTML 降级打印（委托 CertPrintEngine）

参数:

  • data (dict[str, Any]) -- 打印数据字典

  • serial_number (str) -- 合格证编号（SN）

返回:

生成的 HTML 文件路径；失败返回 None

返回类型:

str | None

check_and_supplement_params(cert, parent=None, supplement_callback=None)[源代码]

检查 Certificate 参数完整性，必要时弹出参数补充对话框

对标 VBA print.vb 中一连串的 InputBox 调用： - 阀体材质 InputBox（dd.vb:1066） - 阀杆材质 InputBox（dd.vb:1074） - 启闭件材质 InputBox（dd.vb:1082） - 检验工号 InputBox（dd.vb:894-906） - 出厂年月修正 InputBox（dd.vb:986,1009）

参数:

  • cert (Certificate) -- Certificate ORM 对象

  • parent (object | None) -- 对话框父级窗口，服务层不感知其具体类型，仅原样透传给 supplement_callback；由 Controller/View 层传入 QWidget

  • supplement_callback (Callable | None) -- 可选回调 (model, prefill, material_grades, parent) → (confirmed, skip_all, params)，由 Controller/View 层注入

返回类型:

dict[str, Any]

batch_print(certificate_id, printer_name=None, progress_callback=None, skip_supplement=False, parent=None, supplement_callback=None)[源代码]

批量打印：参数补充 → 解析编号 → 展开逐台编号 → 打印 → 记录日志

对标 VBA print.vb 批量打印流程： 1. 检查参数完整性，缺失则弹出补充对话框（对标 InputBox） 2. 解析 Certificate 编号范围 3. 展开为逐台编号后依次打印 4. 每台记录一条 print_log

参数:

  • certificate_id (int) -- Certificate.id

  • printer_name (str | None) -- 打印机名称

  • progress_callback (Callable[[int, int], None] | None) -- 进度回调 fn(current, total)

  • skip_supplement (bool) -- 跳过参数补充对话框（批量自动化时使用）

  • parent (object | None) -- 父窗口（用于对话框模态），服务层不感知其具体类型，原样透传

  • supplement_callback (Callable | None) -- 参数补充对话框回调 (model, prefill, material_grades, parent) → (confirmed, skip_all, params)

返回:

int, "failed": int, "serials": [...], "message": str}

返回类型:

{"success"

record_print_log(certificate_id, certificate_no, cert_data=None, printer_name='', status='success', error_message='', content_summary='')[源代码]

记录打印日志到 SQLite print_logs 表（字段名与 Access SignTb 兼容）

每次打印操作（单台或批量中的每台）都写入一条记录。 字段命名与 Access SignTb 保持一致，方便铭牌刻印系统直接读取。

SN 和 KKS 赋值遵循 VBA 逻辑： - SN = 短编号（去掉中间4位年月），如 V520Y - KKS = 完整编码（含年月），如 V2604520Y - 编码模式5/6时 SN/KKS 互换 - KKS 截断到12位

字段清理对齐VBA（温度去≤/℃，PN去MPA后缀等）。

参数:

  • certificate_id (int) -- 关联合格证批次ID

  • certificate_no (str) -- 合格证完整编号（如 V2604520Y）

  • cert_data (dict[str, Any] | None) -- 证书完整数据字典

  • printer_name (str) -- 打印机名称

  • status (str) -- 打印状态 (success/failed)

  • error_message (str) -- 错误信息

  • content_summary (str) -- 打印内容摘要

返回:

PrintLog 对象或 None

返回类型:

PrintLog | None

get_print_history(limit=100, status_filter=None)[源代码]

查询打印历史（返回字段名与 Access SignTb 兼容）

参数:

  • limit (int) -- 最大返回条数

  • status_filter (str | None) -- 状态筛选 (success/failed)，None 表示全部

返回:

打印历史列表，每条记录包含 SignTb 兼容字段

返回类型:

list[dict[str, Any]]

apply_template_to_certificate(cert_id, template)[源代码]

将 templates.xlsx 中的模板参数应用到 Certificate 记录

参数:

  • cert_id (int) -- Certificate.id

  • template (dict[str, Any]) -- 从 TemplateManager.query_by_model 返回的模板 dict

返回:

是否成功

返回类型:

bool

writeback_template_to_xlsx(cert_id, xlsx_path)[源代码]

打印完成后回填到 templates.xlsx「数据库」表

参数:

  • cert_id (int) -- Certificate.id

  • xlsx_path (str) -- templates.xlsx 文件路径

返回:

是否成功

返回类型:

bool

class certflow.services.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']}")
class certflow.services.PrinterInfo(name, port='', driver='', status=PrinterStatus.UNKNOWN, is_default=False, paper_sizes=<factory>)[源代码]

基类：object

打印机信息

参数:
name: str
port: str = ''
driver: str = ''
status: PrinterStatus = '未知'
is_default: bool = False
paper_sizes: list[str]
to_dict()[源代码]

序列化为字典。

返回:

包含 name/port/driver/status/is_default/paper_sizes 的字典。

返回类型:

dict

class certflow.services.PrinterManager[源代码]

基类：QObject

打印机管理器

封装 Windows 打印子系统，提供打印机枚举、状态监控、 GDI 精确文字打印、文件打印等功能。

分层说明：本模块归属 services 层，不应被 UI 库强制绑定导入。 在无 PySide6（headless / 无 X11）环境下，信号与监控定时器自动降级为空操作， 模块仍可导入与实例化，仅丢失 GUI 事件通知能力。

Signals（仅 Qt 环境生效）:

printer_added: 新增打印机 (name) printer_removed: 移除打印机 (name) printer_status_changed: 状态变化 (name, status) print_started: 打印开始 (doc_name, job_id) print_completed: 打印完成 (doc_name, job_id) print_failed: 打印失败 (doc_name, error_msg) progress_updated: 进度更新 (current, total)

printer_added
printer_removed
printer_status_changed
print_started
print_completed
print_failed
progress_updated
printers: dict[str, PrinterInfo]
current_printer: str | None
pending_jobs: list[PrintJob]
get_printers()[源代码]

获取所有打印机。

返回:

当前已枚举的打印机信息列表。

返回类型:

list[PrinterInfo]

get_printer_names()[源代码]

获取打印机名称列表。

返回:

打印机名称列表。

返回类型:

list[str]

get_default_printer()[源代码]

获取默认打印机名称。

返回:

默认打印机名称，获取失败时回退到当前打印机。

返回类型:

str | None

set_default_printer(printer_name)[源代码]

设置默认打印机。

参数:

printer_name (str) -- 目标打印机名称。

返回:

设置是否成功。

返回类型:

bool

get_printer_status(printer_name=None)[源代码]

获取打印机状态。

参数:

printer_name (str | None) -- 打印机名称，为 None 时使用默认打印机。

返回:

对应打印机状态，未知时返回 PrinterStatus.UNKNOWN。

返回类型:

PrinterStatus

is_printer_ready(printer_name=None)[源代码]

检查打印机是否就绪。

参数:

printer_name (str | None) -- 打印机名称，为 None 时使用默认打印机。

返回:

True 表示打印机处于就绪状态。

返回类型:

bool

print_text_only(data, serial_number, printer_name=None)[源代码]

使用 win32 GDI 精确打印文字到预印卡纸

在 60mm × 100mm 预印卡纸的指定坐标位置输出文字。 卡纸已预印底板（边框、标题、品保章等），打印机只输出动态文字。 跳过空值字段，避免覆盖卡纸原有内容。

参数:

  • data (dict[str, Any]) -- 打印数据字典，包含 product_name, product_model, dn, pn, temperature, medium, check_standard, inspector_id, manufacture_date

  • serial_number (str) -- 产品编号（逐台打印时不同）

  • printer_name (str | None) -- 打印机名称，默认使用系统默认打印机

返回:

打印是否成功

返回类型:

bool

print_file(file_path, printer_name=None, copies=1)[源代码]

打印文件（PDF/图片/文本）。

参数:

  • file_path (str) -- 待打印文件的完整路径。

  • printer_name (str | None) -- 目标打印机名称，为 None 时使用默认打印机。

  • copies (int) -- 打印份数。

返回:

(是否成功, 作业 ID)，失败时作业 ID 为 None。

返回类型:

tuple[bool, int | None]

print_html(html_content, printer_name=None, copies=1)[源代码]

打印 HTML 内容（先转换为 PDF 再打印）。

参数:

  • html_content (str) -- HTML 字符串内容。

  • printer_name (str | None) -- 目标打印机名称，为 None 时使用默认打印机。

  • copies (int) -- 打印份数。

返回:

(是否成功, 作业 ID)，失败时作业 ID 为 None。

返回类型:

tuple[bool, int | None]

get_print_queue(printer_name=None)[源代码]

获取打印队列。

参数:

printer_name (str | None) -- 打印机名称，为 None 时使用默认打印机。

返回:

当前打印队列中的任务列表。

返回类型:

list[PrintJob]

cancel_job(job_id, printer_name=None)[源代码]

取消打印任务。

参数:

  • job_id (int) -- 打印作业 ID。

  • printer_name (str | None) -- 打印机名称，为 None 时使用默认打印机。

返回:

取消是否成功。

返回类型:

bool

pause_printer(printer_name=None)[源代码]

暂停打印机。

参数:

printer_name (str | None) -- 打印机名称，为 None 时使用默认打印机。

返回:

暂停是否成功。

返回类型:

bool

resume_printer(printer_name=None)[源代码]

恢复打印机。

参数:

printer_name (str | None) -- 打印机名称，为 None 时使用默认打印机。

返回:

恢复是否成功。

返回类型:

bool

save_settings(settings_file='printer_settings.json')[源代码]

保存打印机设置到 JSON 文件。

参数:

settings_file (str) -- 设置文件路径，默认为 "printer_settings.json"。

返回类型:

None

load_settings(settings_file='printer_settings.json')[源代码]

从 JSON 文件加载并设置默认打印机。

参数:

settings_file (str) -- 设置文件路径，默认为 "printer_settings.json"。

返回类型:

None

static cleanup_temp_files()[源代码]

清理系统中本模块产生的临时 HTML/PDF 文件。

返回类型:

None

staticMetaObject = PySide6.QtCore.QMetaObject("PrinterManager" inherits "QObject": Methods:   #4 type=Signal, signature=printer_added(QString), parameters=QString   #5 type=Signal, signature=printer_removed(QString), parameters=QString   #6 type=Signal, signature=printer_status_changed(QString,PyObject), parameters=QString, PyObject   #7 type=Signal, signature=print_started(QString,int), parameters=QString, int   #8 type=Signal, signature=print_completed(QString,int), parameters=QString, int   #9 type=Signal, signature=print_failed(QString,QString), parameters=QString, QString   #10 type=Signal, signature=progress_updated(int,int), parameters=int, int )
class certflow.services.PrinterStatus(*values)[源代码]

基类：Enum

打印机状态枚举

READY = '就绪'
PAUSED = '暂停'
ERROR = '错误'
PENDING = '待机'
PAPER_OUT = '缺纸'
OFFLINE = '离线'
UNKNOWN = '未知'
class certflow.services.QueryService(session)[源代码]

基类：object

销售计划查询服务

提供灵活的多条件组合查询功能: - 支持模糊匹配、精确匹配、范围查询 - 支持多字段组合查询 - 支持分页和排序 - 支持关联 Certificate 表查询

示例

>>> service = QueryService(session)
>>> results = service.query(
...     conditions={
...         "product_model": {"value": "阀门", "operator": "contains"},
...         "customer": {"value": "东方电气", "operator": "contains"},
...         "quantity": {"value": 10, "operator": "gte"}
...     }
... )
参数:

session (Session)

OPERATORS = {'contains': <function QueryService.<lambda>>, 'endswith': <function QueryService.<lambda>>, 'eq': <function QueryService.<lambda>>, 'gt': <function QueryService.<lambda>>, 'gte': <function QueryService.<lambda>>, 'in': <function QueryService.<lambda>>, 'is_not_null': <function QueryService.<lambda>>, 'is_null': <function QueryService.<lambda>>, 'lt': <function QueryService.<lambda>>, 'lte': <function QueryService.<lambda>>, 'ne': <function QueryService.<lambda>>, 'not_contains': <function QueryService.<lambda>>, 'startswith': <function QueryService.<lambda>>}
property saleplan_fields: set[str]

SalePlan 表全部列名集合。

惰性计算并缓存（_salepian_fields），用于判定查询字段归属 SalePlan 表。

返回:

SalePlan 表列名集合。

返回类型:

set[str]

property certificate_fields: set[str]

Certificate 表全部列名集合。

惰性计算并缓存（_certificate_fields），用于判定查询字段归属 Certificate 表。

返回:

Certificate 表列名集合。

返回类型:

set[str]

HYBRID_FIELDS: dict[str, Any] = {}
FIELD_ALIASES = {'cert_status': 'print_status', 'print_status': 'print_status'}
build_status_filter(status)[源代码]

构建状态筛选条件（同时匹配 production_status 和 execution_status）

当用户筛选标准化状态时，同时模糊匹配 execution_status 原始文本， 避免因原始文本包含状态关键词而遗漏记录。

关键词从 StatusInference 配置中获取（config.yaml production_status.query_keywords）。

参数:

status (str) -- 状态值，如 "已发货"、"已完成"、"生产中"

返回:

SQLAlchemy 过滤表达式

返回类型:

Any

build_query(conditions, or_filters=None)[源代码]

根据条件构建查询对象

自动识别字段所属表： - SalePlan 字段直接过滤 - Certificate 字段通过 LEFT JOIN 过滤

参数:

  • conditions (dict[str, Any]) --

    查询条件字典，格式: {

    "字段名": {

    "value": 查询值, "operator": "操作符" # 默认 "contains"

    }

    }

  • or_filters (Any) -- 可选 OR 条件列表 [{field, operator, value}, ...]， 与主 conditions（AND）以 OR 组合，用于跨字段「任一命中」检索。

返回:

SQLAlchemy查询对象（可能包含 JOIN）

返回类型:

Any

query(conditions=None, page=1, page_size=50, order_by='sort_group', order_desc=False, or_filters=None)[源代码]

执行多条件组合查询

参数:

  • conditions (dict[str, Any]) -- 查询条件字典

  • page (int) -- 页码（从1开始）

  • page_size (int) -- 每页记录数

  • order_by (str) -- 排序字段

  • order_desc (bool) -- 是否降序

  • or_filters (Any)

返回:

包含 total, page, page_size, total_pages, records, conditions

返回类型:

Dict

get_field_values(field_name, search=None, order_desc=True)[源代码]

获取某字段的所有唯一值（用于下拉框）

参数:

  • field_name (str) -- 字段名

  • search (str) -- 搜索关键字（可选）

  • order_desc (bool) -- 是否降序排序（最新在前），默认True

返回:

所有唯一值列表

返回类型:

list[str]

get_distinct_field_combos(fields, conditions=None)[源代码]

返回指定字段组合的去重值列表（全量，不受分页影响），用于快查框候选项。

跨整个结果集（尊重 conditions，但忽略分页）计算去重组合， 避免候选项仅来自当前页导致「越查越窄 / 只能选已显示项」。

参数:

  • fields (list[str]) -- 字段名列表（如 ["customer", "product_model"]）

  • conditions (dict[str, Any] | None) -- 当前筛选条件（可选），把候选项限定在已筛选范围内

返回:

去重后的组合字符串（字段值以 QUICK_SEARCH_SEP 连接，保留空段），按出现顺序

返回类型:

list[str]

search(conditions=None, page=1, page_size=50, order_by='sort_group', order_desc=False)[源代码]

执行查询并返回与 query 一致的结果字典

兼容旧测试/调用方：在 query 基础上将记录字段命名为 results （query 使用 records），其余字段（total/page/page_size/ total_pages/conditions）保持一致。

参数:

  • conditions (dict[str, Any]) -- 查询条件字典

  • page (int) -- 页码（从1开始）

  • page_size (int) -- 每页记录数

  • order_by (str) -- 排序字段

  • order_desc (bool) -- 是否降序

返回:

包含 total, page, page_size, total_pages, results, conditions

返回类型:

Dict

get_statistics(conditions=None)[源代码]

获取查询结果统计信息

参数:

conditions (dict[str, Any]) -- 查询条件

返回:

统计信息

返回类型:

Dict

get_records_by_ids(ids)[源代码]

根据 ID 列表获取 SalePlan 记录

参数:

ids (list[int]) -- 记录 ID 列表

返回:

记录列表

返回类型:

list[Any]

get_certificate_count(plan_ids)[源代码]

获取关联的合格证数量

参数:

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

返回:

关联的合格证数量

返回类型:

int

delete_records(ids)[源代码]

删除 SalePlan 记录（事务回滚）

检查是否有关联合格证，如有则拒绝删除。

参数:

ids (list[int]) -- 记录 ID 列表

返回:

(是否成功, 删除数量, 消息)

返回类型:

tuple[bool, int, str]

delete_with_certificates(ids)[源代码]

级联删除 SalePlan 记录及关联的 Certificate

参数:

ids (list[int]) -- 记录 ID 列表

返回:

(是否成功, 删除计划数, 删除合格证数, 消息)

返回类型:

tuple[bool, int, int, str]

batch_update_field(ids, field_name, value)[源代码]

批量更新 SalePlan 字段（事务回滚）

参数:

  • ids (list[int]) -- 记录 ID 列表

  • field_name (str) -- 字段名

  • value (Any) -- 要设置的值

返回:

更新记录数

返回类型:

int

batch_update_production_status(ids, status, update_relations=None)[源代码]

批量更新生产状态（支持关联字段更新）

参数:

  • ids (list[int]) -- 记录 ID 列表

  • status (str) -- 目标状态

  • update_relations (list[dict] | None) -- 关联更新规则列表

返回:

更新记录数

返回类型:

int

copy_cert_info_to_plan(ids)[源代码]

复制合同信息到合格证字段

参数:

ids (list[int]) -- 记录 ID 列表

返回:

更新记录数

返回类型:

int

fill_plan_no(ids, sequence)[源代码]

填充计划单号（批量更新版本）

使用每条记录的 plan_date 生成对应的年月格式 plan_no。 如果 plan_date 为空，则回退到当前年月。

参数:

  • ids (list[int]) -- 记录 ID 列表

  • sequence (str) -- 序号

返回:

更新记录数

返回类型:

int

get_certificates_by_plan_ids(plan_ids)[源代码]

根据销售计划 ID 查询关联的合格证记录

参数:

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

返回:

合格证记录列表

返回类型:

list[Any]

get_latest_certificate(plan_ids)[源代码]

获取最近创建的一条合格证记录（用于参数补充）

参数:

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

返回:

最近创建的合格证记录，或 None

返回类型:

Any

class certflow.services.SalePlanService(db_session)[源代码]

基类：object

销售计划服务

提供销售计划数据的完整业务流程管理，包括: - Excel文件导入(支持带格式和不带格式两种模式) - 数据清洗和校验 - 分组排序(先按业务分组，组内按产品排序) - 数据库持久化 - 查询和统计

参数:

db_session (Session)

session

SQLAlchemy数据库会话对象

excel_handler

Excel文件处理器实例

cleaner

数据清洗器实例

sorter

排序器实例

save_handler

保存处理器实例

示例

>>> from sqlalchemy import create_engine
>>> from sqlalchemy.orm import sessionmaker
>>>
>>> engine = create_engine("sqlite:///certflow.db")
>>> Session = sessionmaker(bind=engine)
>>> session = Session()
>>>
>>> service = SalePlanService(session)
>>>
>>> # 导入Excel文件
>>> result = service.import_from_excel_with_config({
...     "file_path": "sales_plan.xlsx",
...     "sheet_name": "Sheet1",
...     "preserve_formatting": True
... })
>>> print(f"导入成功: {result['new_count']}条")
static read_excel(path, sheet_name=None, **kwargs)[源代码]

读取 Excel 工作簿（委托 ExcelHandler.read_excel）。

参数:

  • path (str)

  • sheet_name (str | None)

  • kwargs (Any)

返回类型:

Any

static clean_sale_plan(df)[源代码]

清洗销售计划数据框（委托 DataCleaner.clean_sale_plan）。

参数:

df (Any)

返回类型:

Any

static detect_header_row(df_raw, keywords, max_rows=20)[源代码]

自动检测表头行（委托 ExcelHandler.detect_header_row）。

参数:
返回类型:

int

static extract_year_from_filename(file_path)[源代码]

从工作簿文件名提取年份后两位（委托 Sorter._extract_year_from_filename）。

参数:

file_path (str)

返回类型:

str

static apply_range_filter(records, range_filter, selected_rows)[源代码]

B0-8 导入范围筛选。

  • selected_rows：勾选行模式，传入 0-based 数据行序号集合，仅保留这些行。

  • range_filter：按订单筛选模式，dict 含 plan_date/customer/project_name/plan_no 中若干非空字段，仅保留这些字段**全部精确匹配**的行（空值字段视为通配）。

  • 两者均未指定：返回原 records（整表导入）。

临时字段 _src_row 用后即清，不污染落库数据。

参数:
返回类型:

list[dict[str, Any]]

record_shipment(sale_plan_id, quantity, ship_date=None, operator='', note='')[源代码]

记录一次分批发货（B0-3 分批发货数量追踪）。

将本次发货追加到 shipment_batches JSON 列表，累加 shipped_quantity； 当累计已发货量达到订单总量 quantity 时，标记 shipping_status='已发货' （仅更新发货状态，不动生产状态），并记录发货日期。

参数:

  • sale_plan_id (int) -- 销售计划主键 id。

  • quantity (int) -- 本次发货数量（正整数）。

  • ship_date (str | None) -- 发货日期 YYYY-MM-DD，默认今天。

  • operator (str) -- 操作人。

  • note (str) -- 备注。

返回:

含 id / shipped_quantity / remaining_quantity / batches。

返回类型:

dict

抛出:

ValueError -- 记录不存在或发货数量非法。

import_from_excel_with_config(config_params)[源代码]

使用自定义配置从Excel导入销售计划

根据配置参数选择带格式或不带格式的导入方式。

参数:

config_params (dict[str, Any]) -- 导入配置参数字典，包含以下字段: - file_path: Excel文件路径(必填) - sheet_name: 工作表名称或索引，默认为0 - header_row: 表头行号(可选) - skip_rows: 跳过的行数，默认为0 - column_mapping: 自定义列映射字典(可选) - preserve_formatting: 是否保留单元格格式，默认为False

返回:

导入结果字典

返回类型:

Dict[str, Any]

抛出:

Exception -- 导入失败时抛出异常

示例

>>> service = SalePlanService(session)
>>> result = service.import_from_excel_with_config({
...     "file_path": "sales.xlsx",
...     "sheet_name": "1月",
...     "header_row": 1,
...     "skip_rows": 0,
...     "preserve_formatting": True,
... })
>>> print(result["new_count"])
get_all_sale_plans()[源代码]

获取所有销售计划

返回:

按分组键排序的所有销售计划列表

返回类型:

List[SalePlan]

示例

>>> service = SalePlanService(session)
>>> plans = service.get_all_sale_plans()
>>> print(len(plans))
get_by_sort_group(sort_group)[源代码]

根据分组键获取销售计划

参数:

sort_group (str) -- 分组键值

返回:

指定分组的销售计划列表

返回类型:

List[SalePlan]

get_by_product_model(product_model)[源代码]

根据产品型号获取销售计划

参数:

product_model (str) -- 产品型号

返回:

指定产品型号的销售计划列表

返回类型:

List[SalePlan]

clear_all()[源代码]

清空所有销售计划数据

返回:

删除的记录数

返回类型:

int

get_statistics()[源代码]

获取销售计划统计信息

返回:

统计信息字典，包含:

  • total_records: 总记录数

  • group_count: 分组数

  • status_stats: 按生产状态分组统计

返回类型:

Dict[str, Any]

get_by_status(status)[源代码]

根据格式状态获取销售计划

参数:

status (str) -- 格式状态值(如"已开票"、"已发货"等)

返回:

指定格式状态的销售计划列表

返回类型:

List[SalePlan]

get_status_statistics()[源代码]

获取格式状态统计信息

返回:

各格式状态对应的记录数统计

返回类型:

Dict

get_exportable_plans()[源代码]

获取可导出的销售计划（排除已发货、外购未回、隐藏行）

对应 VBA 中导出合格证时的过滤逻辑： - 排除字体蓝色(已发货)的记录 - 排除背景橙棕色(外购未回)的记录 - 排除行高=0(隐藏/取消)的记录

返回:

可导出的销售计划列表

返回类型:

List[SalePlan]

find_by_unique_key(unique_key)[源代码]

根据唯一键查找已存在的销售计划

参数:

unique_key (str) -- 唯一键

返回:

已存在的 SalePlan 或 None

返回类型:

SalePlan | None

find_by_id(plan_id)[源代码]

按主键查询销售计划。

参数:

plan_id (int) -- 销售计划主键

返回:

命中的 SalePlan 或 None

返回类型:

SalePlan | None

find_by_production_order_no(production_order_no)[源代码]

根据生产令号查找已存在的销售计划

用于老数据模式降级去重：当唯一键匹配失败但生产令号有值时， 按生产令号查找已有记录，避免同一记录从不同来源重复导入。

参数:

production_order_no (str) -- 生产令号

返回:

已存在的 SalePlan 或 None

返回类型:

SalePlan | None

find_by_contract_and_model(contract_no, product_model)[源代码]

根据合同号 + 产品型号查找已存在的销售计划

用于跨来源补充场景的兜底匹配：当唯一键和生产令号都匹配失败时， 通过合同号 + 产品型号定位已有记录，实现缺失字段补充。

参数:

  • contract_no (str) -- 合同号

  • product_model (str) -- 产品型号

返回:

已存在的 SalePlan 或 None

返回类型:

SalePlan | None

add_sale_plan(sale_plan)[源代码]

添加销售计划到会话

参数:

sale_plan (SalePlan) -- SalePlan 对象

返回类型:

None

add_change(change)[源代码]

添加变更记录到会话

参数:

change (SalePlanChange) -- SalePlanChange 对象

返回类型:

None

flush_session()[源代码]

刷新会话

返回类型:

None

commit_session()[源代码]

提交事务

返回类型:

None

rollback_session()[源代码]

回滚事务

返回类型:

None

CERT_EDITABLE_FIELDS: tuple[str, ...] = ('cert_product_name', 'cert_product_model', 'cert_product_spec', 'certificate_remarks')
update_cert_fields(plan_id, fields)[源代码]

仅更新合格证打印字段（蓝图 §4.2.3：查询视图 cert-only 订正）

只写入 cert_product_name/model/spec 与 certificate_remarks， 绝不触碰合同字段（product_* 等）。这是合格证清单/打印的数据源， 订正即时生效于后续 create_certificates。

参数:

  • plan_id (int) -- 销售计划记录主键

  • fields (dict[str, Any]) -- 待写入字段字典；非合格证字段会被静默忽略

返回:

是否成功更新（id 不存在返回 False）

返回类型:

bool

count_sale_plans()[源代码]

统计销售计划总数

返回类型:

int

archive_to_shipped(sale_plan_ids)[源代码]

将选中的 SalePlan 归档到 sale_plans_shipped 表并从主表移除

三层模型的 Layer 1 → Layer 2: - 将记录完整拷贝到 sale_plans_shipped（设置 shipped_at / shipped_by="archive"） - 从 sale_plans 表删除对应记录 - 同时将 production_status 设为"已发货"（如果还不是）

参数:

sale_plan_ids (list[int]) -- 要归档的 SalePlan ID 列表

返回:

成功归档的记录数

返回类型:

int

export_and_delete(sale_plan_ids, output_dir)[源代码]

导出选中 SalePlan 的完整字段 Excel，归档到 shipped 表，并从主表删除

三层模型的 Layer 1 → Layer 3: - 生成完整字段 Excel 文件 - 将记录拷贝到 sale_plans_shipped（shipped_by="export_delete"） - 从 sale_plans 表删除

参数:

  • sale_plan_ids (list[int]) -- 要导出的 SalePlan ID 列表

  • output_dir (str) -- Excel 输出目录

返回:

N, "filepath": "..."}

返回类型:

{"count"

export_selected_records(records, columns, file_path)[源代码]

导出选中记录到 Excel（按列定义）

参数:

  • records (list[Any]) -- ORM 记录列表

  • columns (list[dict]) -- 列定义列表 [{"field": "...", "label": "..."}, ...]

  • file_path (str) -- 目标文件路径

返回:

导出记录数

返回类型:

int

export_query_result(records, columns, file_path)[源代码]

导出查询结果到 Excel（带 plan_date 格式化）

参数:

  • records (list[Any]) -- ORM 记录列表

  • columns (list[dict]) -- 列定义列表

  • file_path (str) -- 目标文件路径

返回:

导出记录数

返回类型:

int

class certflow.services.ScanService(session=None)[源代码]

基类：object

扫描件生成服务

封装扫描件生成的业务逻辑，支持单张与多张拼接两种模式， 并委托底层 Handler 完成画布创建、字段绘制与文件导出。

参数:

session (Session | None)

load_certificates(certificate_ids)[源代码]

加载合格证记录

参数:

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

返回:

合格证列表

返回类型:

list[Certificate]

generate_scans(certificate_ids, output_dir, template_type='全中文', output_format='JPG', rows=1, cols=1, paper_width_mm=60, paper_height_mm=100, data_overrides=None, dual_code=False, progress_callback=None)[源代码]

生成扫描件

参数:

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

  • output_dir (str) -- 输出目录

  • template_type (str) -- 模板类型 (全中文/中英文/俄英文)

  • output_format (str) -- 输出格式 (JPG/PDF)

  • rows (int) -- 每页行数

  • cols (int) -- 每页列数

  • paper_width_mm (float) -- 纸张宽度 (mm)

  • paper_height_mm (float) -- 纸张高度 (mm)

  • progress_callback (Callable[[int, int, str], None] | None) -- 进度回调函数 fn(current, total, message)

  • data_overrides (dict[int, dict] | None)

  • dual_code (bool)

返回:

{

"success": int, # 成功生成的文件数 "failed": int, # 失败的数量 "files": list[str], # 生成的文件路径列表 "message": str # 结果消息

}

返回类型:

生成结果字典

class certflow.services.WorkflowService(session)[源代码]

基类：object

工作流服务 - 管理文档完成状态

自动升级规则由 config.yaml production_status.auto_upgrade 配置驱动。 新增升级规则只需修改 YAML，无需改动代码。

参数:

session (Session)

mark_document_done(sale_plan_id, doc_type, file_path=None)[源代码]

标记文档完成

参数:

  • sale_plan_id (int) -- 销售计划 ID

  • doc_type (str) -- 文档类型 (certificate/nameplate/test_report/warranty/scan)

  • file_path (str | None) -- 文件路径（可选，预留扩展）

返回:

更新后的 SalePlan 对象，未找到则返回 None

返回类型:

SalePlan | None

mark_document_undone(sale_plan_id, doc_type)[源代码]

撤销文档完成标记

参数:

  • sale_plan_id (int) -- 销售计划 ID

  • doc_type (str) -- 文档类型

返回:

更新后的 SalePlan 对象，未找到则返回 None

返回类型:

SalePlan | None

batch_mark_done(sale_plan_ids, doc_type)[源代码]

批量标记文档完成

参数:

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

  • doc_type (str) -- 文档类型

返回:

N, "failed": M}

返回类型:

{"success"