Python 批量处理文件，如何避免一个坏文件让整个任务中断？我的异常隔离实践

上一篇文章介绍了 Work Inbox Agent 如何统一读取 Word、Excel、Markdown 和 TXT 文件。

读取器搭好之后，还需要考虑一个实际问题：

如果文件夹里有一个损坏的 Word 文件，其他文件还要不要继续处理？

对于批量整理工作资料的工具来说，我希望程序尽可能处理正常文件，同时明确记录失败原因。因此，我在文件读取入口做了一层异常隔离。

这篇文章结合项目中的实现，介绍文件级异常处理、不支持格式的记录，以及这套设计目前的边界。

项目地址：Work Inbox Agent

一、批量处理需要区分两类失败

假设待处理的文件夹中有这些资料：

work-records/ ├── weekly-note.md ├── meeting.docx ├── tasks.xlsx └── archive.zip

其中，meeting.docx可能已经损坏，archive.zip则不属于当前支持的格式。

面对这种情况，需要先区分两个层面的问题。

第一类是整个任务无法开始，例如目标目录不存在，或者传入的路径其实是一个普通文件。

第二类是某个文件无法处理，例如 Word 文件损坏、Excel 文件无法打开，或者文件格式暂不支持。

我的处理方式是：

• 目标目录有问题：抛出异常，让调用方知道任务无法开始。
• 单个文件读取失败：保留错误结果，继续处理其他文件。
• 文件格式不支持：记录原因，让使用者知道它没有被解析。

这样，程序能够区分“任务根本没有执行”和“任务执行了，但部分文件失败”。

二、在文件读取入口隔离异常

项目中，所有文件都通过extract_file()进入读取流程。

它会先找到支持该格式的读取器，再调用对应的read()方法：

def extract_file(path: Path) -> ExtractedDocument: for reader in READERS: if reader.supports(path): try: return reader.read(path) except Exception as exc: return ExtractedDocument( file_name=path.name, file_type=path.suffix.lower().lstrip(".") or "unknown", source_path=str(path), error=f"读取失败: {exc}", ) return ExtractedDocument.unsupported( path, f"暂不支持文件格式: {path.suffix or '<none>'}", )

这里最关键的是try...except的位置。

它包围的是单个文件的读取操作。如果某个读取器抛出异常，程序会把失败转换成一个ExtractedDocument返回。

调用方拿到的仍然是统一的数据结构，只不过其中的error字段记录了失败信息。

例如，一个文件读取失败后，结果会包含：

ExtractedDocument( file_name=path.name, file_type=path.suffix.lower().lstrip(".") or "unknown", source_path=str(path), error=f"读取失败: {exc}", )

文件名、路径和错误原因被保留下来，后续可以据此定位问题。

三、为什么不能只在整个循环外面捕获？

下面是一段用于说明区别的简化代码：

try: for path in files: documents.append(reader.read(path)) except Exception as exc: print(f"处理失败：{exc}")

这段代码虽然捕获了异常，但如果循环中间的某个文件读取失败，程序会直接跳到except，后面的文件就不会继续处理。

因此，“捕获了异常”并不等于“实现了文件级隔离”。

如果希望一个文件失败后继续处理下一个文件，异常捕获就需要落在每个文件的处理范围内。

在我的项目中，扫描函数负责遍历，extract_file()负责处理单个文件：

for path in sorted(folder.rglob("*")): if path.is_file() and not any( part.startswith(".") for part in path.parts ): documents.append(extract_file(path))

当读取器内部出现普通异常时，extract_file()会返回带错误信息的结果，循环便可以继续执行。

这种职责划分也让代码更容易理解：扫描逻辑负责“找到文件”，读取入口负责“尝试解析并记录结果”。

四、不支持的格式，也应该留下记录

对于.zip等暂不支持的格式，项目没有直接跳过，而是返回一个明确标记为unsupported的结果。

对应的构造方法如下：

@classmethod def unsupported(cls, path: Path, message: str) -> "ExtractedDocument": return cls( file_name=path.name, file_type="unsupported", source_path=str(path), error=message, )

这样做是为了回答一个问题：

文件夹里明明有这个文件，为什么输出中看不到它的内容？

如果完全跳过，使用者很难判断是程序漏扫了文件，还是格式本身不支持。

保留记录之后，就可以区分：

• 文件被扫描到了，但当前不支持解析；
• 文件格式受支持，但读取过程中失败；
• 文件读取成功，获得了内容。

这三种状态对后续排查和展示都有帮助。

五、目录错误仍然需要明确抛出

单个文件失败可以记录下来，但目标目录本身有问题时，继续执行就没有意义。

因此，扫描入口保留了明确的路径检查：

def scan_folder(folder: Path) -> list[ExtractedDocument]: if not folder.exists(): raise FileNotFoundError(f"目录不存在: {folder}") if not folder.is_dir(): raise NotADirectoryError(f"不是目录: {folder}") documents: list[ExtractedDocument] = [] for path in sorted(folder.rglob("*")): if path.is_file() and not any( part.startswith(".") for part in path.parts ): documents.append(extract_file(path)) return documents

例如，用户把目录名称输错，如果程序直接返回空列表，就容易被误解成“目录中没有资料”。

抛出FileNotFoundError能更准确地表达：输入路径无效，扫描尚未正常开始。

这里的判断原则是：任务级错误交给调用方处理，文件级读取错误进入结果列表。

六、记录错误之后，还要能检查错误

异常隔离的目的，是让正常文件继续得到处理，同时让失败可见。

调用方可以通过error字段检查结果。下面是一个基于现有接口的使用示例：

from pathlib import Path from work_inbox.readers.registry import scan_folder documents = scan_folder(Path("work-records")) for document in documents: if document.error is not None: print( f"[未成功解析] {document.file_name}：" f"{document.error}" ) else: print(f"[读取成功] {document.file_name}")

也可以统计读取结果：

success_count = sum( document.error is None for document in documents ) problem_count = sum( document.error is not None for document in documents ) print(f"扫描文件数：{len(documents)}") print(f"读取成功：{success_count}") print(f"读取失败或格式不支持：{problem_count}")

需要注意，成功返回结果并不意味着内容一定完整、准确。这里统计的是“读取过程是否报告错误”，还不能代替内容质量检查。

七、当前实现有哪些边界？

这层异常处理解决了部分文件读取失败的问题，但还不能覆盖所有异常情况。

首先，try...except包围的是reader.read()。目录遍历、路径检查等操作发生在其他位置，因此不能据此认为整个扫描流程中的所有错误都会被隔离。

其次，当前捕获的是Exception，没有捕获KeyboardInterrupt等中断信号。用户主动停止程序时，中断可以继续向外传播。

另外，有些数据问题不会触发异常。

例如，文本读取器使用了：

path.read_text( encoding="utf-8-sig", errors="replace", )

遇到无法解码的字节时，它会使用替代字符，读取过程可能仍然成功。因此，“没有抛出异常”不能直接理解成“文件内容没有问题”。

Excel 读取也有类似边界：当前版本只预览前 50 行，并且读取公式本身，不负责计算公式结果。这些属于解析范围限制，需要在使用和文档中明确说明。

八、后续准备怎样改进？

当前版本把失败原因放进了error字符串，已经能够完成基础定位。随着功能增加，我准备进一步区分错误类型，例如文件损坏、权限不足和格式不支持。

测试方面，目前仓库已有“不支持的文件仍保留在扫描结果中”的用例。接下来还可以补充一个更直接的异常隔离测试：在同一目录中放入正常文本和损坏的.docx，检查正常文件是否读取成功，以及损坏文件是否留下错误记录。

如果后续处理的文件更多，还可以在输出中增加单独的失败清单，让使用者修复文件后再次处理。

这次实现让我更明确地意识到：批量任务除了需要一条成功路径，还需要为局部失败设计返回结果。

对于 Work Inbox Agent，我希望每个文件都有可以追踪的处理结果。正常文件继续向后流转，失败文件留下名称、路径和原因，使用者才能判断本次整理覆盖了哪些资料，还有哪些需要补充处理。

项目源码：Work Inbox Agent