按用户指定数量批量读取文件夹文件并分组处理的技术问题

需求：

• 文件夹内有数十万数据文件，需按输入的数量n将文件分组
• 依次读取每组文件并处理（如计算均值），处理完一组后从断点继续处理下一组（示例：n=5时，先处理第1-5个文件，再处理第6-10个，依此类推）
• 尝试过用for循环+yield实现但出现生成器错误，希望改用其他基于for循环的方法

初代代码（存在问题） #

def fileavg(path,n):
  import numpy as np
  import xlsxwriter
  from glob import glob

  workbook = xlsxwriter.Workbook('Test.xlsx')
  worksheet = workbook.add_worksheet()
  row=0

  b=glob.iglob(path) #输入路径时请以r'开头并以'结尾
  for i in range(0,len(b),n):
      f=yield [i:i +n]
      A=np.mean(f(1),axis=1)
      for col, data in enumerate(A):
          worksheet.write_column(row, col, data)
      row +=1

初代代码问题分析 #

1. glob.iglob()返回生成器对象，无法直接用len(b)获取长度，会触发类型错误
2. yield用法错误：此处不需要生成器逻辑，反而打乱了分组取文件的流程
3. f=yield [i:i +n]后调用f(1)，但f是生成器产出的列表，并非可调用函数，会直接报错

更新后代码（仍有问题） #

def fileavg(path,n):
 import numpy as np
 import xlsxwriter
 import glob


 workbook = xlsxwriter.Workbook('Test.xlsx')
 worksheet = workbook.add_worksheet()
 row=0
 more_files=True
 b=glob.iglob(path) #输入路径时请以r'开头并以'结尾


 while more_files:
     for i in range(n):
         try:
             next_file=next(b)
             print(row,next_file)
             A=np.mean(next_file(1))
         except StopIteration:
             more_files=False
             break
        
         for col, data in enumerate(A):
             worksheet.write_column(row, col, data)
         row +=1

更新后代码问题分析 #

1. next_file是文件路径字符串，不是可调用对象，next_file(1)会触发调用错误
2. 逻辑为逐个处理文件，未实现需求中的分组批量处理（即先收集n个文件，统一计算均值后写入）
3. 无断点续接逻辑：程序中断后无法从上次处理的位置继续

正确实现方案 #

核心思路 #

1. 用glob.glob()获取所有文件路径（若文件量极大，可改用iglob并调整分组逻辑）
2. 基于for循环按步长n分组，每次批量取n个文件
3. 加入断点续接：通过记录已处理的文件索引，下次启动时自动从该位置恢复
4. 正确读取文件数据并计算均值（假设文件为numpy可读取格式，如txt、npy等）

最终代码 #

import numpy as np
import xlsxwriter
import glob
import os

def fileavg(path, n, checkpoint_file="checkpoint.txt"):
    # 初始化Excel工作簿
    workbook = xlsxwriter.Workbook('Test.xlsx')
    worksheet = workbook.add_worksheet()
    row = 0

    # 获取所有目标文件路径
    all_files = glob.glob(path)
    total_files = len(all_files)
    if total_files == 0:
        print("未找到目标文件")
        workbook.close()
        return

    # 读取断点记录，确定起始处理位置
    start_idx = 0
    try:
        with open(checkpoint_file, "r") as f:
            start_idx = int(f.read().strip())
            if start_idx >= total_files:
                print("所有文件已处理完成")
                workbook.close()
                return
            print(f"从第 {start_idx+1} 个文件开始处理")
    except FileNotFoundError:
        print("无断点记录，从第一个文件开始处理")

    # 按步长n分组处理文件
    for idx in range(start_idx, total_files, n):
        # 获取当前组的文件列表
        group_files = all_files[idx:idx + n]
        current_end = min(idx + n, total_files)
        print(f"处理第 {idx+1}-{current_end} 个文件")

        # 读取组内所有文件的数据
        group_data = []
        for file_path in group_files:
            # 根据实际文件格式调整读取方式，示例为txt文件
            data = np.loadtxt(file_path)
            group_data.append(data)
        
        # 计算组内均值（按示例需求调整axis参数）
        if group_data:
            group_array = np.array(group_data)
            A = np.mean(group_array, axis=0)

            # 将结果写入Excel
            for col, val in enumerate(A):
                worksheet.write(row, col, val)
            row += 1

        # 更新断点记录，写入当前处理到的最后一个文件索引
        with open(checkpoint_file, "w") as f:
            f.write(str(idx + n))

    # 处理完成后删除断点文件（可选操作）
    if os.path.exists(checkpoint_file):
        os.remove(checkpoint_file)
    
    workbook.close()
    print("所有文件处理完成")

# 调用示例
# fileavg(r"path/to/your/files/*.txt", 5)

代码说明 #

1. 断点续接：通过checkpoint.txt记录已处理的最后一个文件索引，程序重启时自动读取并恢复流程
2. 分组逻辑：用range(start_idx, total_files, n)实现按步长n批量取文件，避免逐个处理的低效
3. 文件读取：根据实际文件格式替换np.loadtxt，比如用np.load读取npy文件
4. 均值计算：通过调整np.mean的axis参数，可实现行均值、列均值或全局均值
5. Excel写入：每组处理结果对应Excel的一行，确保数据规整

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