certflow.handlers.sorter module¶

多键排序处理器模块

提供基于多字段组合的分组、排序和序号生成功能，用于销售计划数据先按业务键分组，组内按产品键排序，并为每条记录生成全局唯一的序号。

class certflow.handlers.sorter.Sorter[源代码]¶

基类：object

分组排序处理器

提供两级分组排序功能： 1. 第一层：按业务分组键分组（如订单信息） 2. 第二层：每组内按产品排序键排序（如产品名称、型号、规格） 3. 生成全局唯一的序号（分组前缀 + 组内序号）

示例

>>> sorter = Sorter()
>>> data = [
...     {"plan_date": "2024-01-01", "customer": "客户A", "product_name": "阀门B", "product_model": "DN80"},
...     {"plan_date": "2024-01-01", "customer": "客户A", "product_name": "阀门A", "product_model": "DN100"},
...     {"plan_date": "2024-01-01", "customer": "客户A", "product_name": "阀门A", "product_model": "DN80"},
... ]
>>> result = sorter.group_and_sort(
...     data,
...     group_keys=["plan_date", "customer"],
...     sort_keys=["product_name", "product_model"]
... )
>>> for item in result:
...     print(f"{item['full_seq']}: {item['product_name']}")
G001-001: 阀门A
G001-002: 阀门A
G001-003: 阀门B

static group_and_sort(data, group_keys, sort_keys=None, keep_original_order=True, generate_seq=True, use_global_prefix=True, prefix_format='G{:03d}', seq_format='{:03d}', seq_field='group_seq', group_key_field='_group_key', group_index_field='_group_index', source_file='', source_sheet='')[源代码]¶

两级分组排序

先按业务分组键分组，再在每组内按指定排序键排序，并可生成组内序号。

参数:

data (list[dict[str, Any]]) -- 原始数据列表，每个元素为字典格式的记录
group_keys (list[str]) -- 分组键列表，如["plan_date", "customer", "project_name", "sales_order_no"]
sort_keys (list[str] | None) -- 组内排序键列表，如["product_name", "product_model", "product_spec"]， None表示保持原始顺序
keep_original_order (bool) -- 是否保留原始顺序标记，默认为True
generate_seq (bool) -- 是否生成组内序号，默认为True
use_global_prefix (bool) -- 是否使用全局分组前缀（用于区分不同分组），默认为True
prefix_format (str) -- 分组前缀格式，默认"G{:03d}"，生成G001, G002...
seq_format (str) -- 组内序号格式，默认"{:03d}"，生成001,002...
seq_field (str) -- 序号存储的字段名，默认"group_seq"
group_key_field (str) -- 分组键存储的字段名，默认"_group_key"
group_index_field (str) -- 分组索引存储的字段名，默认"_group_index"
source_file (str) -- 源文件路径，用于提取年份生成_original_order，如"2026年销售计划.xlsx"
source_sheet (str) -- 工作表名称，用于提取月份生成_original_order，如"1月"

返回:

处理后的数据列表，每条记录包含:

_group_key: 分组键值（由group_keys组合而成）
_group_index: 分组索引(1-based)
group_seq: 组内序号（如"001"）
full_seq: 完整序号（如"G001-001"），仅当use_global_prefix=True时
_original_order: 原始顺序键(YYMMxxxx)，仅当keep_original_order=True时

返回类型:

list[dict[str, Any]]

示例

>>> data = [
...     {"order_no": "SO001", "product": "阀门B", "spec": "DN80"},
...     {"order_no": "SO001", "product": "阀门A", "spec": "DN100"},
...     {"order_no": "SO002", "product": "阀门C", "spec": "DN50"}
... ]
>>> result = Sorter.group_and_sort(
...     data,
...     group_keys=["order_no"],
...     sort_keys=["product", "spec"]
... )
>>> for item in result:
...     print(f"{item['full_seq']}: {item['product']}")
G001-001: 阀门A
G001-002: 阀门B
G002-001: 阀门C

static get_group_info(data, group_index=None, group_key=None)[源代码]¶

获取指定分组的记录

参数:

data (list[dict[str, Any]]) -- 已分组的数据列表(由group_and_sort生成)
group_index (int | None) -- 分组索引(1-based)
group_key (str | None) -- 分组键值

返回:

指定分组的记录列表

返回类型:

list[dict[str, Any]]

示例

>>> result = Sorter.group_and_sort(data, group_keys=["order_no"], sort_keys=["product"])
>>> # 按索引获取
>>> group1 = Sorter.get_group_info(result, group_index=1)
>>> # 按键值获取
>>> group = Sorter.get_group_info(result, group_key="SO001")

static get_all_groups(data, seq_field='group_seq', group_key_field='_group_key')[源代码]¶

获取所有分组的信息

参数:

data (list[dict[str, Any]]) -- 已分组的数据列表(由group_and_sort生成)
seq_field (str) -- 序号字段名，与group_and_sort中的seq_field一致，默认"group_seq"
group_key_field (str) -- 分组键字段名，与group_and_sort中的group_key_field一致，默认"_group_key"

返回:

分组信息字典，键为分组索引，值为分组信息

返回类型:

dict[int, dict[str, Any]]

示例

>>> result = Sorter.group_and_sort(data, group_keys=["order_no"], sort_keys=["product"])
>>> groups = Sorter.get_all_groups(result)
>>> for idx, info in groups.items():
...     print(f"分组{idx}: {info['count']}条记录")

static restore_original_order(data)[源代码]¶

恢复原始顺序

根据"_original_order"字段恢复数据的原始顺序，并清理所有临时字段。

参数:: data (list[dict[str, Any]]) -- 需要恢复顺序的数据列表，应包含"_original_order"字段 (由group_and_sort方法生成)
返回:: 恢复原始顺序并清理临时字段后的数据列表
返回类型:: list[dict[str, Any]]

备注

如果数据中没有"_original_order"字段，会记录警告并返回原数据
恢复顺序后会删除所有临时字段，保持数据干净

示例

>>> sorted_data = Sorter.group_and_sort(data, group_keys=["order_no"], sort_keys=["product"])
>>> # 执行某些处理后...
>>> original_data = Sorter.restore_original_order(sorted_data)