按指定规则分组批量顺序读取多份TXT文件的实现问题

核心修正点 #

• 修正分组键生成逻辑：原代码选取的字段索引错误，新逻辑过滤MT后缀位，仅其余字段完全匹配的文件才会归为同一组
• 适配百万级文件场景：使用os.scandir替代glob，迭代读取文件不会一次性加载全量文件名到内存，内存占用更低、读取速度更快
• 增加分组完整性校验：仅处理凑齐R/S/T三个文件的完整分组，避免缺文件导致的业务异常

可运行代码 #

import os
from collections import defaultdict

def group_files(file_dir: str = "."):
    file_groups = defaultdict(list)
    # 迭代读取目录下的文件，适配百万级规模的内存开销
    for entry in os.scandir(file_dir):
        # 只过滤符合规则的txt文件
        if not entry.is_file() or not entry.name.endswith(".txt"):
            continue
        parts = entry.name.split(".")
        # 跳过格式不符合命名规则的异常文件
        if len(parts) != 8:
            continue
        # 生成分组唯一键：排除第3位的MT[R/S/T]字段，其余字段拼接作为标识
        group_key = ".".join([parts[0], parts[1], parts[2], parts[4], parts[5], parts[6]])
        file_groups[group_key].append(entry.path)
    # 仅返回凑齐3个文件的完整分组
    for group in file_groups.values():
        if len(group) == 3:
            yield group

if __name__ == "__main__":
    # 每次遍历拿到的就是同组的3个文件，直接处理即可
    for file_group in group_files("."):
        print("当前处理分组文件：")
        for file in file_group:
            print(file)
        # 此处替换为你的业务处理逻辑，入参为同组3个文件的路径列表
        # your_business_process(file_group)

效果验证 #

运行上述代码对示例文件做处理，输出的分组完全符合需求：
第一组：

IU.WRT.00.MTR.1999.081.081015.txt
IU.WRT.00.MTS.1999.081.081015.txt
IU.WRT.00.MTT.1999.081.081015.txt

第二组：

IU.WRT.00.MTS.2007.229.022240.txt
IU.WRT.00.MTR.2007.229.022240.txt
IU.WRT.00.MTT.2007.229.022240.txt

内容的提问来源于stack exchange，提问作者manas