如何按命名规则分组批量读取目录下匹配模式的TXT文件
解决方案
你现有脚本的核心问题是分组key选取错误,首先我们先明确你的文件命名结构:
文件名格式为 [机构].[站点].[编号].[MTx].[年份].[年积日].[时分秒].txt,其中MTx的x为R/S/T,是同一组文件唯一的差异点,因此分组key需要取除MTx之外的其他固定字段组合。
以下是适配百万级数据集的修改后代码:
import os from collections import defaultdict def groupfiles(file_dir: str = "./", suffix: str = ".txt"): filedict = defaultdict(list) # 用os.scandir替代glob,迭代读取文件列表不会一次性加载所有文件名到内存,百万级场景内存占用更低 with os.scandir(file_dir) as entries: for entry in entries: # 只筛选符合后缀的普通文件,跳过目录、软链接等非目标内容 if entry.is_file() and entry.name.endswith(suffix): parts = entry.name.split(".") # 校验文件名格式,过滤不符合规则的脏文件避免报错 if len(parts) != 8: continue # 分组key:排除第3位MTx字段,拼接其他固定字段 group_key = ".".join([parts[0], parts[1], parts[2], parts[4], parts[5], parts[6]]) filedict[group_key].append(entry.path) # 仅返回正好包含3个文件的有效分组,可根据实际需求调整过滤规则 for filegroup in filedict.values(): if len(filegroup) == 3: # 按MT后缀排序,保证每组返回顺序固定为MTR/MTS/MTT,避免处理顺序混乱 filegroup.sort(key=lambda x: x.split(".")[3]) yield filegroup # 调用示例 for relatedfiles in groupfiles("./"): # relatedfiles即为同组的3个文件,直接在此处编写你的业务处理逻辑即可 print("当前处理分组:", relatedfiles) # 业务逻辑示例 # for filename in relatedfiles: # 单文件处理逻辑
优化说明
- 修正了分组逻辑,确保只有MT后缀不同的文件被分到同一组
- 适配百万级文件处理场景,不会一次性占用大量内存
- 增加脏文件过滤、无效分组自动跳过逻辑,鲁棒性更强
- 每组文件固定排序,处理顺序可预测
内容的提问来源于stack exchange,提问作者manas
相关产品推荐
相关产品推荐

