certflow.controllers package

Submodules

Module contents

控制器模块

提供系统核心业务逻辑控制器的统一导入接口. 包含控制器基类及各业务领域的具体控制器实现.

class certflow.controllers.BaseController(db_session=None, db_manager=None, use_shared=True)[源代码]

基类：object

控制器基类

所有业务控制器的基类，提供统一的数据库会话管理和事务处理方法. 支持上下文管理器协议，可自动管理会话生命周期. 支持共享数据库管理器，实现全局WebDAV配置统一管理.

参数:
db_manager

数据库管理器实例，负责创建和管理数据库连接

session

SQLAlchemy数据库会话对象，用于执行数据库操作

_owns_session

是否拥有会话的所有权，决定close()时是否关闭会话

示例

>>> # 方式1: 使用上下文管理器(推荐)
>>> with BaseController() as controller:
...     result = controller.session.query(User).all()
...     controller.commit()
>>>
>>> # 方式2: 手动管理会话
>>> controller = BaseController()
>>> try:
...     result = controller.session.query(User).all()
...     controller.commit()
... finally:
...     controller.close()
>>>
>>> # 方式3: 共享数据库管理器（支持WebDAV）
>>> db_manager = DatabaseManager(webdav_config=config)
>>> controller = BaseController(db_manager=db_manager)
>>>
>>> # 方式4: 使用全局共享管理器
>>> BaseController.set_shared_db_manager(db_manager)
>>> controller = BaseController()  # 自动使用共享管理器
classmethod set_shared_db_manager(db_manager)[源代码]

设置共享的数据库管理器（用于全局WebDAV配置）

设置后，所有新创建的控制器如果没有显式传入db_manager， 都会使用这个共享实例。

参数:

db_manager (DatabaseManager) -- 共享的数据库管理器实例

返回:

None

返回类型:

None

示例

>>> # 在应用启动时设置
>>> db_manager = DatabaseManager(webdav_config=config)
>>> BaseController.set_shared_db_manager(db_manager)
classmethod get_shared_db_manager()[源代码]

获取共享的数据库管理器

返回:

共享的数据库管理器实例，如果未设置则返回None

返回类型:

DatabaseManager | None

classmethod clear_shared_db_manager()[源代码]

清除共享的数据库管理器

将类级别的共享管理器置空，后续控制器将回退到独立管理器。

返回:

None

返回类型:

None

__enter__()[源代码]

上下文管理器入口

支持with语句，返回控制器实例自身.

返回:

控制器实例自身

返回类型:

BaseController

示例

>>> with BaseController() as controller:
...     # 使用controller进行操作
...     pass
__exit__(exc_type, exc_val, exc_tb)[源代码]

上下文管理器出口

退出with代码块时自动关闭数据库会话.

参数:

  • exc_type (type[BaseException] | None) -- 异常类型，如果发生异常则为异常类

  • exc_val (BaseException | None) -- 异常实例，如果发生异常则为异常对象

  • exc_tb (TracebackType | None) -- 异常追踪信息，如果发生异常则为traceback对象

返回:

None

返回类型:

None

close()[源代码]

关闭数据库会话

仅当控制器拥有会话所有权(即通过无参构造函数自动创建会话)时， 才会关闭会话.外部传入的会话不会被关闭.

返回:

None

返回类型:

None

commit()[源代码]

提交数据库事务

提交当前会话中的所有更改.如果提交过程中发生异常， 会自动回滚事务并重新抛出异常.

返回:

None

抛出:

Exception -- 事务提交失败时抛出原始异常

返回类型:

None

rollback()[源代码]

回滚数据库事务

撤销当前会话中所有未提交的更改.

返回:

None

返回类型:

None

get_session()[源代码]

获取当前会话

返回:

当前使用的数据库会话

返回类型:

Session

get_service(service_cls, *args, **kwargs)[源代码]

以当前控制器会话构造一个 Service 实例（收口 Service 构造来源）

统一替代 XxxService(self.session) 的散落写法：session 来源经 BaseController 收口，调用方不直接触碰 self.session。

仅适用于构造签名首个位置参数为 Session 的 Service；首个参数非 会话的 Service（如 PrinterService(printer_name)）不应走此工厂。

参数:

  • service_cls (type) -- Service 类（如 CertificateService）

  • *args (object) -- 透传给 Service 构造的额外位置参数（排在 session 之后）

  • **kwargs (object) -- 透传给 Service 构造的关键字参数

返回:

构造好的 Service 实例

返回类型:

object

static resolve_dict_csv_path(dict_key)[源代码]

解析字典 CSV 导出/导入路径（委托 dict_csv_sync.resolve_dict_csv_path）。

供 LookupTableManagerView 等查表视图替代对 certflow.services. dict_csv_sync 的直接依赖，统一经控制器门面取路径。

参数:

dict_key (str)

返回类型:

Any

refresh_session()[源代码]

刷新会话（如果会话已关闭，重新创建）

用于在长时间运行的应用中恢复会话。

返回:

None

返回类型:

None

is_session_active()[源代码]

检查会话是否活跃

返回:

会话是否存在且活跃

返回类型:

bool

class certflow.controllers.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。

参数:
返回类型:

bytes | None

auto_save_log(serials, data, printer_name, certificate_id=None)[源代码]

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

参数:
返回类型:

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 或提交事务。

参数:
返回类型:

None

get_ungenerated_sale_plans()[源代码]

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

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

返回:

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

返回类型:

List[SalePlan]

get_all_certificates(limit=100, offset=0)[源代码]

获取所有合格证(分页)

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

参数:

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

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

返回:

合格证对象列表

返回类型:

List[Certificate]

class certflow.controllers.ImportController(db_session=None, db_manager=None, use_shared=True)[源代码]

基类：BaseController

导入控制器 — 导入界面的业务调度层（薄层）

参数:

use_shared (bool)

run_import(config_params)[源代码]

执行销售计划导入，返回统计 dict。

仅做「请求透传 + 调 Service + 收结果」，事务由 Service 内部持有 （与历史 run_import 行为一致，Controller 不额外 commit/rollback）。

参数:

config_params (dict)

返回类型:

dict

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

导入范围筛选（预览用，不落库）。

委托 ``SalePlanService.apply_range_filter``（M5 已提升为公开静态方法）。

参数:
返回类型:

list[dict[str, Any]]

static read_and_clean_sheet(file_path, sheet_name, header_row, skip_rows, column_mapping=None)[源代码]

读取工作表并经清洗管线转为记录列表（与真实导入同款读取）。

委托 ExcelHandler.read_excel + DataCleaner.clean_sale_plan； pandas 在 header= 下自动对重名列加 .1/.2 后缀去重，避免 df[col] 因重名列返回 DataFrame。

参数:
返回类型:

list[dict[str, Any]]

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

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

参数:
返回类型:

int | None

static extract_year_from_filename(file_path)[源代码]

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

参数:

file_path (str)

返回类型:

int | None

static normalize_record_for_key(rec, session)[源代码]

预览/扫描阶段与入库同款清洗：口径/压力规范化后返回新 dict（不污染入参）。

委托 IDGenerator.normalize_spec / IDGenerator.normalize_pn。

参数:
返回类型:

dict[str, Any]

static generate_unique_key(rec)[源代码]

生成记录唯一键（委托 IDGenerator.generate_unique_key）。

参数:

rec (dict[str, Any])

返回类型:

str

static generate_change_signature(rec)[源代码]

生成记录变更签名（委托 IDGenerator.generate_change_signature）。

参数:

rec (dict[str, Any])

返回类型:

str

static find_existing_by_unique_key(session, unique_key)[源代码]

按唯一键查询已存在记录（委托 SalePlanService.find_by_unique_key）。

参数:

  • session (Any)

  • unique_key (str)

返回类型:

Any

class certflow.controllers.OutputController(session=None, printer_name=None, printer_controller=None, scan_controller=None)[源代码]

基类：object

打印 / 扫描统一调度控制器（§11.8 阶段4）。

参数:
run(target, action='print', **kwargs)[源代码]

按 (target, action) 显式 dispatch 到对应控制器方法。

参数:

  • target (str) -- "print" / "scan"

  • action (str) -- 目标内的动作键（见 _dispatch 表）

  • **kwargs (Any) -- 透传给具体控制器方法

返回:

被调方法的返回值（generate_scans 返回 task_id；print_certificate 为 None）

抛出:

KeyError -- target/action 不在 dispatch 表中

返回类型:

Any

property printer_controller: PrinterController
property scan_controller: ScanController
get_controller(target)[源代码]

按目标取对应控制器（统一出口）。

参数:

target (str)

返回类型:

Any

get_printers()[源代码]
返回类型:

list[str]

property current_printer: str
get_printer_status()[源代码]
返回类型:

bool

class certflow.controllers.QueryController(session, db_manager=None, parent=None)[源代码]

基类：QObject

查询控制器 — 业务逻辑层

参数:

  • session (Session)

  • db_manager (Any)

  • parent (QObject | None)

data_updated

数据变更信号（删除/更新后触发，视图应刷新）

operation_completed

操作完成信号，携带状态消息

operation_failed

操作失败信号，携带错误信息

field_values_loaded

(field_name, values)

Type:

字段可选值加载完成信号

field_values_load_failed

(field_name, error_message)

Type:

字段可选值加载失败信号

query(conditions, page, page_size, order_by, order_desc, or_filters=None)[源代码]

执行分页查询

参数:

  • conditions (dict[str, Any]) -- 筛选条件字典

  • page (int) -- 页码

  • page_size (int) -- 每页条数

  • order_by (str) -- 排序字段

  • order_desc (bool) -- 是否降序

  • or_filters (Any)

返回:

查询结果字典，包含 records、page、total_pages、total

返回类型:

dict[str, Any]

示例

>>> controller = QueryController(session)
>>> result = controller.query(
...     conditions={"product_model": "DN100"},
...     page=1,
...     page_size=20,
...     order_by="id",
...     order_desc=True,
... )
>>> print(result["total"], result["records"])
get_field_values(field_name, order_desc=True)[源代码]

获取字段的所有可选值（同步，会阻塞 UI 线程）

推荐使用 load_field_values_async() 异步加载。

参数:

  • field_name (str) -- 字段名

  • order_desc (bool) -- 是否按数量降序排列

返回:

字段值列表

返回类型:

list[Any]

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

获取字段组合的去重候选项（全量，跨所有分页），用于快查框。

参数:

  • fields (list[str]) -- 字段名列表

  • conditions (dict[str, Any] | None) -- 当前筛选条件（可选）

返回:

去重组合字符串列表

返回类型:

list[str]

load_field_values_async(field_name, order_desc=True)[源代码]

异步加载字段可选值（推荐）

通过后台线程加载，完成后通过 field_values_loaded 信号通知。

参数:

  • field_name (str) -- 字段名

  • order_desc (bool) -- 是否按数量降序排列

返回:

FieldValuesWorker 实例（调用方需保持引用避免 GC）

返回类型:

FieldValuesWorker

get_records_by_ids(ids)[源代码]

根据 ID 列表获取记录

参数:

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

返回:

记录列表

返回类型:

list[Any]

get_statistics(conditions)[源代码]

获取统计信息

参数:

conditions (dict[str, Any]) -- 筛选条件

返回:

统计信息字典

返回类型:

dict[str, Any]

delete_records(records)[源代码]

删除记录（事务回滚）

参数:

records (list[dict]) -- 要删除的记录列表，每项包含 id

返回:

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

返回类型:

tuple[bool, int, str]

delete_with_certificates(records)[源代码]

级联删除记录及关联的合格证

参数:

records (list[dict]) -- 要删除的记录列表，每项包含 id

返回:

(是否成功, 删除销售计划数量, 消息)

返回类型:

tuple[bool, int, str]

check_certificate_exists(ids)[源代码]

检查记录是否有关联合格证

参数:

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

返回:

关联的合格证数量

返回类型:

int

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

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

参数:

  • 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

export_selected(ids, columns, file_path)[源代码]

导出选中记录到 Excel

参数:

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

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

  • file_path (str) -- 导出文件路径

返回:

导出记录数

返回类型:

int

archive_to_shipped(ids)[源代码]

将选中记录归档到已发货表（Layer 1 → Layer 2）

参数:

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

返回:

归档记录数

返回类型:

int

export_shipped_and_delete(ids, output_dir)[源代码]

导出完整字段 Excel 并删除（Layer 1 → Layer 3）

参数:

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

  • output_dir (str) -- 导出目录

返回:

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

返回类型:

{"count"

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

统一「打印合格证」入口：去重安全地编号并生成合格证。

合并原 auto_number_and_create_certs``（查询页「编号/自动编号」）与 ``import_cert_list_from_sales_plan``（「导入合格证清单」）为单一入口。 业务编排在此层，实际生成委托 ``CertificatePrintService： 跳过已生成合格证的计划（避免重复生成 / 唯一约束冲突）， 自动编号并写入 Certificate（待打印），返回 (编号数, 合格证数)。

分层调用链： View(_goto_print) → Controller(本方法) → CertificatePrintService(编排) → CertNumberingService(数据访问) → Certificate(Model)。

参数:

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

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

  • print_status (str) -- 生成的合格证打印状态，默认 "待打印"

返回:

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

返回类型:

tuple[int, int]

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

手动编号并创建合格证

参数:

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

  • prefix (str) -- 编号前缀

  • start (int) -- 起始序号

  • quantity (int) -- 每件数量

  • suffix (str) -- 后缀

返回:

更新记录数

返回类型:

int

auto_number(ids, prefix_override=None)[源代码]

自动编号（正则解析生产令号/供货类型，委托 CertificatePrintService.auto_number）。

为选中且 product_code 为空/占位符、needs_numbering!=False 的记录生成产品编号， 写入 SalePlan 并即时刷新。对应工具栏「🔢 自动编号」按钮（修复「点击打印提示需先编号 却没有自动编号按钮」的体验缺口）。

参数:

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

  • prefix_override (str | None) -- 手动指定的年月前缀（YYMM），None 取当前年月

返回:

成功编号的记录数

返回类型:

int

generate_batch(sale_plan_id, batch_qty, ym=None, start_override=None)[源代码]

按「本批台数」为单个销售计划生成一批产品编号（G4 入口）。

委托 CertificateNumberService.generate_batch：合同总数从 SalePlan.quantity 读取，用 shipped_quantity 判定剩余与满发，超额直接抛 ``ValueError``（调用方捕获 并转用户提示）。对应工具栏「🔢 自动编号」弹窗输入本批台数的路径。

参数:

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

  • batch_qty (int) -- 本批发货台数（> 0 且 <= 剩余可发）

  • ym (str | None) -- 目标编号年月（YYMM），None 取当前月

  • start_override (int | None) -- 起始流水手填覆盖，None 表示自动续号

返回:

含 product_code / seq_start / seq_end / remaining / fully_shipped 等

返回类型:

dict

create_certificates(ids)[源代码]

生成合格证记录

参数:

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

返回:

生成的合格证数量

返回类型:

int

copy_cert_info_to_plan(ids)[源代码]

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

参数:

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

返回:

更新记录数

返回类型:

int

update_cert_fields(plan_id, fields)[源代码]

订正单条记录的合格证打印字段（仅 cert_* + certificate_remarks）

委托 SalePlanService.update_cert_fields，仅更新合格证字段， 不动合同字段。成功后在查询视图即时刷新。

参数:

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

  • fields (dict[str, Any]) -- 待写入字段字典（非合格证字段被服务层忽略）

返回:

是否更新成功

返回类型:

bool

update_certificate_fields(cert_id, fields)[源代码]

订正单条合格证记录字段（白名单，对称于 update_cert_fields）

委托 CertificateService.update_fields，仅更新合格证打印字段， 不动编号/状态等系统字段。成功后触发查询视图刷新。

参数:

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

  • fields (dict[str, Any]) -- 待写入字段字典（非白名单字段被服务层忽略）

返回:

是否更新成功

返回类型:

bool

copy_number(ids, source_number)[源代码]

复制编号到选中行

参数:

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

  • source_number (str) -- 源编号

返回:

更新记录数

返回类型:

int

fill_plan_no(ids, sequence)[源代码]

填充计划单号

参数:

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

  • sequence (str) -- 序号

返回:

更新记录数

返回类型:

int

hide_records(ids)[源代码]

隐藏记录

参数:

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

返回:

更新记录数

返回类型:

int

staticMetaObject = PySide6.QtCore.QMetaObject("QueryController" inherits "QObject": Methods:   #4 type=Signal, signature=data_updated()   #5 type=Signal, signature=operation_completed(QString), parameters=QString   #6 type=Signal, signature=operation_failed(QString), parameters=QString   #7 type=Signal, signature=field_values_loaded(QString,QVariantList), parameters=QString, QVariantList   #8 type=Signal, signature=field_values_load_failed(QString,QString), parameters=QString, QString )
unhide_records(ids)[源代码]

取消隐藏记录

参数:

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

返回:

更新记录数

返回类型:

int

mark_document_done(ids, doc_type)[源代码]

标记选中行的文档完成

参数:

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

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

返回:

操作结果字典，包含 success 和 msg

返回类型:

dict[str, Any]

export_to_access(ids)[源代码]

导出到 Access 数据库

参数:

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

返回:

导出记录数

返回类型:

int

get_number_prefix(prefix_override=None)[源代码]

获取编号前缀

参数:

prefix_override (str | None) -- 前缀覆盖值

返回:

编号前缀字符串

返回类型:

str

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

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

补充合格证参数（委托 Service 层）

参数:

  • cert (Any) -- 合格证记录

  • parent (Any) -- 父窗口（用于对话框）

  • supplement_callback (Any) -- 参数补充回调（UI 对话框）

返回:

补充结果字典

返回类型:

dict[str, Any]

class certflow.controllers.ReportController(db_session=None)[源代码]

基类：BaseController

报告控制器 - 处理报告生成的业务逻辑

提供质保书 / 试压报告 / 材质报告的生成，以及模板查询、文件打开等功能。

generate_report(sale_plan_ids, report_kind='quality', language='zh', records_per_page=6, export_pdf=None)[源代码]

生成报告（统一入口）。

参数:

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

  • report_kind (str) -- ``quality``(质保书) / ``pressure``(试压报告) / ``material``(材质报告)。

  • language (str) -- zh / en。

  • records_per_page (int) -- 每页记录数（挑选模板变体：行数/双语）。

  • export_pdf (bool | None) -- 是否同时派生 PDF（None→读配置，默认 False）。

返回:

ReportOutputService.emit 的结果（含 success/folder/xlsx/pdf/sealed）。

返回类型:

dict

generate_pressure_test_report(sale_plan_ids, template='standard', language='zh', records_per_page=5)[源代码]

生成试压报告（兼容旧签名）。

参数:

  • sale_plan_ids (list[int])

  • template (str)

  • language (str)

  • records_per_page (int)

返回类型:

dict

generate_material_report(data_ids, report_type='material', language='zh', records_per_page=6)[源代码]

生成材质报告或质保书（兼容旧签名）。

参数:

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

  • report_type (str) -- ``material``(材质报告) / ``quality``(质保书)。

  • language (str)

  • records_per_page (int)

返回类型:

dict

get_available_templates()[源代码]

获取可用的报告模板列表（用于 UI 展示）。

返回类型:

list[dict]

open_report_file(filepath)[源代码]

打开生成的报告文件（系统默认程序）。

参数:

filepath (str)

返回类型:

bool

open_report_folder(folder)[源代码]

打开报告目录（文件夹）；不存在时惰性创建，确保可打开。

参数:

folder (str)

返回类型:

bool

get_report_statistics()[源代码]

获取报告统计信息（轻量占位，历史记录由报告目录.xlsx 承载）。

返回类型:

dict

query(conditions=None, page=1, page_size=200, or_filters=None)[源代码]

复用查询引擎对全库检索（报告数据选择）。

收口 report_view 对 QueryService 的直连调用，使取数统一经 Controller。

参数:

  • conditions (dict | None) -- 过滤条件字典（同 QueryService.query）

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

  • page_size (int) -- 每页条数

  • or_filters (Any) -- 或条件（同 QueryService.query）

返回:

含 records/total 等键的查询结果

返回类型:

dict

get_distinct_field_combos(fields)[源代码]

查询可选的字段组合（供报告数据选择下拉去重）。

收口 report_view 对 QueryService 的直连调用。

参数:

fields (list[str]) -- 需要去重组合的字段名列表

返回:

去重后的字段值组合

返回类型:

list[tuple]

enrich_bom(pno)[源代码]

按产品型号富化 BOM 材质信息。

收口 report_view 对 report_bom_enricher.enrich 的直连调用， 避免 View 直接持有 controller.session 喂给 Service 层函数。

参数:

pno (str) -- 产品型号

返回:

富化后的 BOM 材质字典

返回类型:

dict

class certflow.controllers.ScanController(session=None, db_manager=None)[源代码]

基类：QObject

扫描件控制器

管理扫描件生成任务，通过信号槽实现 UI 与业务逻辑的解耦。

Signals:

progress: 进度信号 (current, total, message) completed: 完成信号 (result: dict) error: 错误信号 (error_message: str)

参数:

  • session (Any)

  • db_manager (Any)

progress
completed
error
strip_unit(pn_display)[源代码]

从展示用压力串拆出 (数值, 单位)

收口 ScanView 原先直连 PNService.strip_unit 的静态调用， 使压力单位解析逻辑的唯一入口收敛到 Controller 层。

参数:

pn_display (str) -- 打印界面展示的压力串（如 25MPa / 150Lb）

返回:

(数值, 单位)；无法拆分时返回 ("", "")

返回类型:

tuple[str, str]

static family_of(template_key)[源代码]

由模板键推导语言族（委托 template_field_formatter.family_of）。

参数:

template_key (str | None)

返回类型:

str

static scan_required_fields(family, base)[源代码]

按语言族计算扫描必填字段（委托 output_guard.scan_required_fields_for）。

参数:
返回类型:

list[str]

static family_requires_pn_unit(family)[源代码]

判断语言族是否要求压力单位（委托 output_guard.family_requires_pn_unit）。

参数:

family (str)

返回类型:

bool

static collect_incomplete(cdata, required)[源代码]

收集缺失必填字段（委托 output_guard.collect_incomplete）。

参数:
返回类型:

list[str]

static field_label(field)[源代码]

字段中文标签（委托 output_guard.FIELD_LABELS）。

参数:

field (str)

返回类型:

str

static list_enabled_templates()[源代码]

列出已启用模板（委托 template_manager.get_template_manager）。

返回类型:

list[tuple[str, str, str]]

static load_dual_defaults(session, sale_plan_id)[源代码]

读取关联 SalePlan 的 SN/KKS 基准值（委托 SalePlanService.find_by_id）。

查询失败或不存在时返回空串字典，不影响基础字段。

参数:

  • session (Any)

  • sale_plan_id (Any)

返回类型:

dict[str, str]

individual_codes(cert, override=None)[源代码]

展开逐台编号（委托 ScanService._individual_codes）。

参数:

  • cert (Certificate)

  • override (dict[str, Any] | None)

返回类型:

list[str]

load_coordinates(template_type)[源代码]

加载模板坐标（委托 ScanService._load_coordinates）。

参数:

template_type (str)

返回类型:

dict[str, Any]

build_cert_data(cert, template_type, serial_number=None, overrides=None, dual_code=False)[源代码]

构建单台合格证字段数据（委托 ScanService._build_cert_data）。

参数:

  • cert (Certificate)

  • template_type (str)

  • serial_number (str | None)

  • overrides (dict[str, Any] | None)

  • dual_code (bool)

返回类型:

dict[str, Any]

render_preview_png(cert, template_type, width_mm, height_mm, serial_number=None, overrides=None, dual_code=False)[源代码]

渲染扫描件预览并返回 PNG 字节流（框架无关，不含 PySide6）。

收口 ScanView 原先对 ScanService``（坐标/数据）与 ``ScanImageHandler （画布/绘制）的直连：预览 == 产出，统一经 Controller 管道。

参数:

  • cert (Certificate)

  • template_type (str)

  • width_mm (float)

  • height_mm (float)

  • serial_number (str | None)

  • overrides (dict[str, Any] | None)

  • dual_code (bool)

返回类型:

bytes

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)[源代码]

异步生成扫描件

参数:

  • 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)

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

  • dual_code (bool)

返回:

任务 ID

返回类型:

int

示例

>>> controller = ScanController(session)
>>> from certflow.config.paths_override import cfg
>>> task_id = controller.generate_scans([1, 2, 3], cfg("paths.output.scans", "output/scans"), "全中文")
>>> print(task_id)
cancel()[源代码]

取消当前生成任务

返回类型:

None

is_running()[源代码]

检查是否有任务正在运行

返回:

是否有任务正在运行

返回类型:

bool

staticMetaObject = PySide6.QtCore.QMetaObject("ScanController" inherits "QObject": Methods:   #4 type=Signal, signature=progress(int,int,QString), parameters=int, int, QString   #5 type=Signal, signature=completed(QVariantMap), parameters=QVariantMap   #6 type=Signal, signature=error(QString), parameters=QString )
class certflow.controllers.SettingsController(parent=None)[源代码]

基类：QObject

设置控制器 — 集中管理用户偏好的读写与运行时生效

架构层次:
View (SettingsView) → Controller (SettingsController) → Handler (ThemeManager/I18nManager)

→ Service (CertLogService 自动保存)

参数:

parent (QObject | None)

_auto_save_timer

自动保存定时器

_auto_save_callback

定时器到期时的回调函数

theme_changed

主题变更信号 (light / dark)

language_changed

语言变更信号 (zh_CN / en_US)

auto_save_triggered

自动保存定时器触发信号

settings_saved

设置保存完成信号，携带完整偏好字典

get_preference(key, default=None)[源代码]

读取用户偏好（代理 settings.get_user_preference）

参数:

  • key (str) -- 偏好键名

  • default (Any) -- 键不存在时返回的默认值，默认为 None

返回:

偏好值；键不存在时返回 default

返回类型:

Any

save_all(preferences)[源代码]

保存全部偏好设置并应用运行时变更

参数:

preferences (dict[str, Any]) -- 偏好设置字典，可包含: - auto_save: bool - auto_save_interval: int (分钟) - confirm_before_overwrite: bool - language: str - theme: str - max_recent_files: int - printer_offsets: dict {"x_offset": int, "y_offset": int}

返回:

None

返回类型:

None

示例

>>> controller = SettingsController()
>>> controller.save_all({
...     "theme": "dark",
...     "language": "zh_CN",
...     "auto_save": True,
...     "auto_save_interval": 5,
... })
set_auto_save_callback(callback)[源代码]

设置自动保存定时器到期时的回调

参数:

callback (callable | None) -- 回调函数，签名为 () -> None，传 None 禁用

返回:

None

返回类型:

None

staticMetaObject = PySide6.QtCore.QMetaObject("SettingsController" inherits "QObject": Methods:   #4 type=Signal, signature=theme_changed(QString), parameters=QString   #5 type=Signal, signature=language_changed(QString), parameters=QString   #6 type=Signal, signature=auto_save_triggered()   #7 type=Signal, signature=settings_saved(QVariantMap), parameters=QVariantMap )