Python批量处理Excel文件遇异常如何跳过错误文件继续执行

AI 小北

2026-8-26

问题成因

Python的异常默认遵循向上冒泡的传播规则：只要某段代码抛出的异常没有被就近捕获，就会直接中断当前整个执行链路，不会自动跳过错误继续执行后续代码。
你遇到的问题核心原因有两个：

整个for循环内部没有任何异常拦截逻辑，所有不同前缀文件的处理分支都共享同一个循环执行上下文。只要任意一个分支里的校验、处理函数抛出异常，会直接击穿当前循环，终止整个函数的运行，根本不会进入下一轮遍历处理剩余文件——这个逻辑和异常出现在哪个分支、你主观判断的遍历位置没有关系。
os.listdir()返回的文件列表顺序不保证按文件名字典序排列，具体顺序由操作系统、文件系统实现决定，你以为的文件遍历顺序和实际运行顺序很可能不一致，这也是你觉得执行逻辑不符合预期的常见原因。
另外你贴的原代码本身存在一个隐藏bug：validation_and_processing函数定义的形参是file_list，但循环里写的是for file in files，如果函数作用域内找不到files变量，本身就会直接触发名称错误中断运行。

解决方案

核心实现思路是在循环内部为单个文件的完整处理流程添加异常捕获，单个文件处理出错时记录错误信息，直接跳过当前文件进入下一轮循环，不让异常向上冒泡打断整体批量处理流程。
修正后的可运行代码如下：

import os
import shutil
import logging

# 配置错误日志，记录出错文件和具体错误原因
logging.basicConfig(
    level=logging.ERROR,
    format="%(asctime)s - 文件处理失败 - %(message)s"
)

# 可根据需要配置错误文件存放目录，单独存放处理失败的文件
ERROR_DIR = "./error_files/"
os.makedirs(ERROR_DIR, exist_ok=True)

def validation_and_processing(file_list):
    for file in file_list:
        try:
            # 单个文件从校验到归档的全流程都放在try块内
            if file.startswith('Nomura'):
                check_filename_format(input_file = file)
                nomura = nomura_validation(input_dir = input_dir, file = file)
                nomura_df = nomura_processing(nomura)
                write_output(path=output_dir, filename=file, dataframe=nomura_df)
                shutil.move(os.path.join(input_dir, file), os.path.join(archive_dir, file))   
                
            elif file.startswith('MSTN'):
                check_filename_format(input_file = file)
                morgan = morgan_validation(input_dir = input_dir, file = file) 
                morgan_df = morgan_processing(morgan)
                write_output(path=output_dir, filename=file, dataframe=morgan_df)
                shutil.move(os.path.join(input_dir, file), os.path.join(archive_dir, file))    
                
            elif file.startswith('JPM'):
                check_filename_format(input_file = file)
                jpm, input_file = jpm_validation(input_dir = input_dir, file = file)
                jpm = jpm_processing(jpm, input_file)
                write_output(path=output_dir, filename=file, dataframe=jpm)
                shutil.move(os.path.join(input_dir, file), os.path.join(archive_dir, file))
                    
            elif file.startswith('BofA'):
                check_filename_format(input_file = file)
                bofa_not_ginnie, bofa_ginnie = bofa_validation(input_dir=input_dir, file=file)
                bofa_final = bofa_processing(df1=bofa_not_ginnie, df2=bofa_ginnie, input_file=file)       
                write_output(path=output_dir, filename=file, dataframe=bofa_final)
                shutil.move(os.path.join(input_dir, file), os.path.join(archive_dir, file))
            
            elif file.startswith('CITI'):
                check_filename_format(input_file = file)
                citi = citi_validation(input_dir=input_dir, file=file)
                citi_final = citi_processing(df=citi, input_file=file)    
                write_output(path=output_dir, filename=file, dataframe=citi_final)
                shutil.move(os.path.join(input_dir, file), os.path.join(archive_dir, file))

        except Exception as e:
            # 捕获当前文件的所有处理异常，记录完整错误栈
            logging.error(f"文件名：{file}，错误信息：{str(e)}", exc_info=True)
            # 将出错文件移动到错误目录，避免下次运行重复处理
            shutil.move(os.path.join(input_dir, file), os.path.join(ERROR_DIR, file))
            # 跳过当前文件，直接进入下一轮循环
            continue

if __name__ == "__main__":
    files = os.listdir(input_dir)
    validation_and_processing(file_list = files)

代码里额外用os.path.join拼接路径，避免不同操作系统下路径分隔符不兼容的问题，比直接字符串拼接路径更稳妥。

实现注意事项

绝对不能把整个for循环套在try/except块里，否则一旦某个文件出错还是会直接跳出循环，无法实现跳过错误文件继续处理的效果。
不建议用裸except:捕获所有异常，用except Exception as e即可覆盖绝大多数业务运行时异常，同时不会拦截键盘中断、系统退出等特殊信号。
异常捕获时必须记录完整错误栈和对应文件名，不要静默吞掉异常，否则后续排查问题文件会非常困难。
只有文件全流程处理成功后再移动到归档目录，避免处理中途出错把原文件错误归档。

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