certflow.services.query_service module

查询服务模块

提供销售计划的多条件组合查询功能，支持参数化查询和分页。 支持关联 Certificate 表进行联合查询。

class certflow.services.query_service.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