"""多键排序处理器模块
提供基于多字段组合的分组、排序和序号生成功能，
用于销售计划数据先按业务键分组，组内按产品键排序，
并为每条记录生成全局唯一的序号。
"""
from __future__ import annotations
from collections import defaultdict
from typing import Any
from loguru import logger
[文档]
class Sorter:
"""分组排序处理器
提供两级分组排序功能：
1. 第一层：按业务分组键分组（如订单信息）
2. 第二层：每组内按产品排序键排序（如产品名称、型号、规格）
3. 生成全局唯一的序号（分组前缀 + 组内序号）
Examples:
>>> 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
"""
[文档]
@staticmethod
def group_and_sort(
data: list[dict[str, Any]],
group_keys: list[str],
sort_keys: list[str] | None = None,
keep_original_order: bool = True,
generate_seq: bool = True,
use_global_prefix: bool = True,
prefix_format: str = "G{:03d}",
seq_format: str = "{:03d}",
seq_field: str = "group_seq",
group_key_field: str = "_group_key",
group_index_field: str = "_group_index",
source_file: str = "",
source_sheet: str = "",
) -> list[dict[str, Any]]:
"""两级分组排序
先按业务分组键分组，再在每组内按指定排序键排序，并可生成组内序号。
Args:
data: 原始数据列表，每个元素为字典格式的记录
group_keys: 分组键列表，如["plan_date", "customer", "project_name", "sales_order_no"]
sort_keys: 组内排序键列表，如["product_name", "product_model", "product_spec"]，
None表示保持原始顺序
keep_original_order: 是否保留原始顺序标记，默认为True
generate_seq: 是否生成组内序号，默认为True
use_global_prefix: 是否使用全局分组前缀（用于区分不同分组），默认为True
prefix_format: 分组前缀格式，默认"G{:03d}"，生成G001, G002...
seq_format: 组内序号格式，默认"{:03d}"，生成001,002...
seq_field: 序号存储的字段名，默认"group_seq"
group_key_field: 分组键存储的字段名，默认"_group_key"
group_index_field: 分组索引存储的字段名，默认"_group_index"
source_file: 源文件路径，用于提取年份生成_original_order，如"2026年销售计划.xlsx"
source_sheet: 工作表名称，用于提取月份生成_original_order，如"1月"
Returns:
list[dict[str, Any]]: 处理后的数据列表，每条记录包含:
- _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时
Examples:
>>> 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
"""
import copy
result = copy.deepcopy(data)
# 1. 保留原始顺序（使用年月+流水号生成排序键）
if keep_original_order:
for idx, item in enumerate(result):
order_val = Sorter._make_original_order(
record=item, idx=idx, source_file=source_file, source_sheet=source_sheet
)
item["_original_order"] = order_val
# 2. 构建分组
groups = defaultdict(list)
for item in result:
# 构建分组键
key_parts = [str(item.get(k, "")) for k in group_keys]
group_key = "_".join(key_parts)
item[group_key_field] = group_key
groups[group_key].append(item)
# 3. 对分组进行排序（按分组键排序，保证输出顺序一致）
sorted_group_keys = sorted(groups.keys())
logger.info(f"分组完成: {len(groups)} 个分组 (分组键: {group_keys[:3]}...)")
# 4. 处理每个分组
sorted_result = Sorter._process_groups(
groups=groups,
sorted_group_keys=sorted_group_keys,
sort_keys=sort_keys,
generate_seq=generate_seq,
use_global_prefix=use_global_prefix,
prefix_format=prefix_format,
seq_format=seq_format,
seq_field=seq_field,
group_index_field=group_index_field,
)
logger.info(f"分组排序完成: {len(result)} 条记录, {len(groups)} 个分组")
return sorted_result
@staticmethod
def _process_groups(
groups: dict[str, list[dict[str, Any]]],
sorted_group_keys: list[str],
sort_keys: list[str] | None,
generate_seq: bool,
use_global_prefix: bool,
prefix_format: str,
seq_format: str,
seq_field: str,
group_index_field: str,
) -> list[dict[str, Any]]:
"""处理每个分组：设置索引、组内排序、生成序号"""
sorted_result: list[dict[str, Any]] = []
for group_idx, group_key in enumerate(sorted_group_keys, start=1):
group_items = groups[group_key]
# 设置分组索引和前缀
for item in group_items:
item[group_index_field] = group_idx
if use_global_prefix:
item["_group_prefix"] = prefix_format.format(group_idx)
# 组内排序
if sort_keys:
group_items.sort(key=lambda x: tuple(str(x.get(k, "")) for k in sort_keys))
logger.debug(
f"分组 {group_idx}: 组内按 {sort_keys[:3]}... 排序, {len(group_items)} 条记录"
)
# 生成组内序号
if generate_seq:
for seq, item in enumerate(group_items, start=1):
item[seq_field] = seq_format.format(seq)
if use_global_prefix:
item["full_seq"] = f"{item['_group_prefix']}-{item[seq_field]}"
sorted_result.extend(group_items)
return sorted_result
@staticmethod
def _make_original_order(
record: dict[str, Any],
idx: int,
source_file: str,
source_sheet: str,
) -> int:
"""根据源文件和工作表生成原始顺序键
格式: YYMMxxxx (6位数字)
- YY: 从文件名提取年份后两位（如 2026 -> 26）
- MM: 从工作表名提取月份（如 "1月" -> "01"）
- xxxx: 4位流水号
Args:
record: 记录字典（保留参数用于扩展）
idx: 流水号索引（从0开始）
source_file: 源文件路径，如 "D:/links/Hard/2026年销售计划.xlsx"
source_sheet: 工作表名称，如 "1月" 或 "Sheet1"
Returns:
int: 排序键，格式 YYMMxxxx，提取失败时使用 0000xxxx
"""
year = Sorter._extract_year_from_filename(source_file)
month = Sorter._extract_month_from_sheetname(source_sheet)
yymm = f"{year:02d}{month:02d}" if year and month else "0000"
return int(f"{yymm}{idx + 1:04d}")
@staticmethod
def _extract_year_from_filename(file_path: str) -> int | None:
"""从文件名提取年份后两位
支持格式:
- "2026年销售计划.xlsx" -> 26
- "sales_plan_2026.xlsx" -> 26
- "2026销售计划.xlsx" -> 26
Args:
file_path: 文件路径
Returns:
int | None: 年份后两位(0-99)，提取失败返回None
"""
import re
from pathlib import Path
if not file_path:
return None
filename = Path(file_path).stem
patterns = [
r"(\d{4})年", # "2026年"
r"_(\d{4})", # "_2026"
r"(\d{4})", # "2026"
]
for pattern in patterns:
match = re.search(pattern, filename)
if match:
year = int(match.group(1))
if 2000 <= year <= 2100:
return year % 100
return None
@staticmethod
def _extract_month_from_sheetname(sheet_name: str) -> int | None:
"""从工作表名提取月份
支持格式:
- "1月" -> 1
- "01月" -> 1
- "一月" -> 1
Args:
sheet_name: 工作表名称
Returns:
int | None: 月份(1-12)，提取失败返回None
"""
import re
if not sheet_name:
return None
# 匹配数字月份: "1月", "01月"
match = re.search(r"(\d{1,2})月", sheet_name)
if match:
month = int(match.group(1))
if 1 <= month <= 12:
return month
# 中文月份映射
chinese_months = {
"一月": 1,
"二月": 2,
"三月": 3,
"四月": 4,
"五月": 5,
"六月": 6,
"七月": 7,
"八月": 8,
"九月": 9,
"十月": 10,
"十一月": 11,
"十二月": 12,
}
for cn, num in chinese_months.items():
if sheet_name == cn or sheet_name.startswith(cn):
return num
return None
[文档]
@staticmethod
def get_group_info(
data: list[dict[str, Any]], group_index: int | None = None, group_key: str | None = None
) -> list[dict[str, Any]]:
"""获取指定分组的记录
Args:
data: 已分组的数据列表(由group_and_sort生成)
group_index: 分组索引(1-based)
group_key: 分组键值
Returns:
list[dict[str, Any]]: 指定分组的记录列表
Examples:
>>> 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")
"""
if group_index is not None:
return [item for item in data if item.get("_group_index") == group_index]
if group_key is not None:
return [item for item in data if item.get("_group_key") == group_key]
return []
[文档]
@staticmethod
def get_all_groups(
data: list[dict[str, Any]],
seq_field: str = "group_seq",
group_key_field: str = "_group_key",
) -> dict[int, dict[str, Any]]:
"""获取所有分组的信息
Args:
data: 已分组的数据列表(由group_and_sort生成)
seq_field: 序号字段名，与group_and_sort中的seq_field一致，默认"group_seq"
group_key_field: 分组键字段名，与group_and_sort中的group_key_field一致，默认"_group_key"
Returns:
dict[int, dict[str, Any]]: 分组信息字典，键为分组索引，值为分组信息
Examples:
>>> 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']}条记录")
"""
groups = {}
for item in data:
group_idx = item.get("_group_index")
if group_idx and group_idx not in groups:
groups[group_idx] = {
"index": group_idx,
"key": item.get(group_key_field),
"count": 0,
"first_seq": item.get(seq_field),
"records": [],
}
if group_idx in groups:
groups[group_idx]["count"] += 1
groups[group_idx]["records"].append(item)
return groups
[文档]
@staticmethod
def restore_original_order(data: list[dict[str, Any]]) -> list[dict[str, Any]]:
"""恢复原始顺序
根据"_original_order"字段恢复数据的原始顺序，
并清理所有临时字段。
Args:
data: 需要恢复顺序的数据列表，应包含"_original_order"字段
(由group_and_sort方法生成)
Returns:
list[dict[str, Any]]: 恢复原始顺序并清理临时字段后的数据列表
Notes:
- 如果数据中没有"_original_order"字段，会记录警告并返回原数据
- 恢复顺序后会删除所有临时字段，保持数据干净
Examples:
>>> sorted_data = Sorter.group_and_sort(data, group_keys=["order_no"], sort_keys=["product"])
>>> # 执行某些处理后...
>>> original_data = Sorter.restore_original_order(sorted_data)
"""
if not data or "_original_order" not in data[0]:
logger.warning("没有原始顺序信息，无法恢复")
return data
sorted_data = sorted(data, key=lambda x: x.get("_original_order", 0))
# 清理临时字段
temp_fields = [
"_original_order",
"_group_key",
"_group_index",
"_group_prefix",
"group_seq",
"full_seq",
]
for item in sorted_data:
for field in temp_fields:
item.pop(field, None)
logger.info(f"恢复原始顺序完成: {len(sorted_data)} 条记录")
return sorted_data