在客户管理系统里，我们经常需要处理成千上万条客户数据，并根据签约时间或者注册时间，将它们按照月份或者年份进行分批。这种分批操作可以用于生成月度报表、年度对账、或是分库分表迁移。Python凭借其丰富的标准库，可以用非常简洁的代码完成这一任务。
一、核心思路：时间字段转周期标识
实现按月或按年分批，最关键的一步是把每条客户记录里的日期，转换成一个能够代表“月”或“年”的标识。例如，将2023-05-12转换为“2023-05”表示年月，或者“2023”表示年。有了这个标识，我们就可以把相同标识的客户归为一批。
在Python中，通常客户数据会以字典、对象或数据库行的形式存在。我们假设每条客户数据都有一个类似sign_date的字段，类型为datetime.date或datetime.datetime。通过访问其year和month属性，就能拼出周期字符串。这种方式比用字符串切片更安全，因为它天然处理了日期对象的合法性。
二、使用itertools.groupby按月份分批
标准库中的itertools.groupby非常适合做“连续相同键”的分组。不过要注意，groupby只能对已经按分组键排序的数据生效，所以我们要先对客户列表排序。
下面的示例展示了如何按年月分批。我们先写一个函数提取周期键，再用sorted排序，最后用groupby聚合。代码中的转义字符确保了HTML安全性。
from itertools import groupby
from datetime import date
# 模拟客户列表
clients = [
{'name': '张三', 'sign_date': date(2023, 5, 12)},
{'name': '李四', 'sign_date': date(2023, 5, 28)},
{'name': '王五', 'sign_date': date(2023, 6, 3)},
{'name': '赵六', 'sign_date': date(2022, 12, 1)},
]
# 提取年月标识
def month_key(c):
return c['sign_date'].strftime('%Y-%m')
# 必须按分组键排序
clients.sort(key=month_key)
# 按月分批
for key, group in groupby(clients, key=month_key):
batch = list(group)
print('批次:' + key + ' 客户数:' + str(len(batch)))
for c in batch:
print(' - ' + c['name'])
上述代码运行后，会依次输出2022-12、2023-05、2023-06三个批次。这种写法的优点是逻辑直观，且不需要引入第三方库。缺点是如果数据量极大，先排序可能消耗较多内存和时间。
如果数据来自数据库，更推荐在SQL里直接用DATE_FORMAT或EXTRACT做分组查询，把分批压力交给数据库。但在纯脚本或离线分析场景下，上面的Python代码已经足够。
三、按年份分批的改造
按年分批与按月分批几乎一致，只需把键函数改为只返回年份。我们既可以用strftime('%Y')，也可以直接用sign_date.year。
为了复用代码，我们可以把分批逻辑封装成一个通用函数，通过参数控制粒度。这样业务层调用时非常灵活，不用重复写排序和分组代码。
from itertools import groupby
from datetime import date
clients = [
{'name': '张三', 'sign_date': date(2023, 5, 12)},
{'name': '李四', 'sign_date': date(2023, 5, 28)},
{'name': '王五', 'sign_date': date(2022, 6, 3)},
]
def year_key(c):
return c['sign_date'].year
def batch_by(clients, key_func):
data = sorted(clients, key=key_func)
return {k: list(g) for k, g in groupby(data, key=key_func)}
result = batch_by(clients, year_key)
for yr, batch in result.items():
print('年份:' + str(yr) + ' 数量:' + str(len(batch)))
这里用了字典推导式直接生成年份到客户列表的映射，比循环打印更适合后续写文件或入库。如果客户量在百万级，建议改用生成器逐批产出，避免一次性构建大字典。
四、边界情况与性能比较
实际业务中要小心两种边界：其一是日期字段可能为None，键函数应返回默认桶如“未知”；其二是跨时区数据，最好先统一转成同一时区的日期再分批。
我们简单对比两种方案。使用groupby前需要排序，时间复杂度约为O(n log n)；若用defaultdict直接散列分组则只需O(n)，但会丢失批次顺序。对于报表类需求，顺序通常重要，所以排序可接受。下面是用defaultdict的替代写法：
from collections import defaultdict
from datetime import date
clients = [
{'name': '张三', 'sign_date': date(2023, 5, 12)},
{'name': '李四', 'sign_date': date(2023, 5, 28)},
]
buckets = defaultdict(list)
for c in clients:
key = c['sign_date'].strftime('%Y-%m')
buckets[key].append(c)
for k in sorted(buckets):
print(k, len(buckets[k]))
这种写法不必预先排序整个列表，且在流式读取客户数据时可边读边分桶。若你的客户列表来自CSV或API翻页，用defaultdict会更省心。两种方案各有适用面，按数据规模和顺序需求选择即可。