certflow.controllers.query_controller module

查询控制器

负责查询视图的业务逻辑处理，将 UI 层与数据层解耦。

职责: - 删除记录（含级联删除、事务回滚） - 批量更新字段 - 导出选中记录到 Excel - 自动编号并创建合格证 - 查询数据 - 获取字段可选值

架构:
View (QueryView) → Controller (QueryController) → Service (QueryService/SalePlanService)

→ Handler (ShippedExportHandler)

Usage:

controller = QueryController(session) success, count, msg = controller.delete_records([{"id": 1}, {"id": 2}])

class certflow.controllers.query_controller.FieldValuesWorker(service, field_name, order_desc=True)[源代码]

基类：QThread

异步加载字段可选值的工作线程

避免同步 SQL 查询阻塞 UI 线程，通过信号将结果传回主线程。

参数:
finished

(field_name, values)

Type:

完成信号

error

error_message

Type:

错误信号

run()[源代码]

线程主执行方法（由 QThread.start() 触发）

通过 service 查询字段可选值，成功后发出 finished 信号， 失败则发出 error 信号。

返回:

None

返回类型:

None

staticMetaObject = PySide6.QtCore.QMetaObject("FieldValuesWorker" inherits "QThread": Methods:   #12 type=Signal, signature=finished(QString,QVariantList), parameters=QString, QVariantList   #13 type=Signal, signature=error(QString), parameters=QString )
class certflow.controllers.query_controller.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]