上一篇文章介绍了 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_namepath.name, file_typepath.suffix.lower().lstrip(.) or unknown, source_pathstr(path), errorf读取失败: {exc}, ) return ExtractedDocument.unsupported( path, f暂不支持文件格式: {path.suffix or none}, )这里最关键的是try...except的位置。它包围的是单个文件的读取操作。如果某个读取器抛出异常程序会把失败转换成一个ExtractedDocument返回。调用方拿到的仍然是统一的数据结构只不过其中的error字段记录了失败信息。例如一个文件读取失败后结果会包含ExtractedDocument( file_namepath.name, file_typepath.suffix.lower().lstrip(.) or unknown, source_pathstr(path), errorf读取失败: {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_namepath.name, file_typeunsupported, source_pathstr(path), errormessage, )这样做是为了回答一个问题文件夹里明明有这个文件为什么输出中看不到它的内容如果完全跳过使用者很难判断是程序漏扫了文件还是格式本身不支持。保留记录之后就可以区分文件被扫描到了但当前不支持解析文件格式受支持但读取过程中失败文件读取成功获得了内容。这三种状态对后续排查和展示都有帮助。五、目录错误仍然需要明确抛出单个文件失败可以记录下来但目标目录本身有问题时继续执行就没有意义。因此扫描入口保留了明确的路径检查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( encodingutf-8-sig, errorsreplace, )遇到无法解码的字节时它会使用替代字符读取过程可能仍然成功。因此“没有抛出异常”不能直接理解成“文件内容没有问题”。Excel 读取也有类似边界当前版本只预览前 50 行并且读取公式本身不负责计算公式结果。这些属于解析范围限制需要在使用和文档中明确说明。八、后续准备怎样改进当前版本把失败原因放进了error字符串已经能够完成基础定位。随着功能增加我准备进一步区分错误类型例如文件损坏、权限不足和格式不支持。测试方面目前仓库已有“不支持的文件仍保留在扫描结果中”的用例。接下来还可以补充一个更直接的异常隔离测试在同一目录中放入正常文本和损坏的.docx检查正常文件是否读取成功以及损坏文件是否留下错误记录。如果后续处理的文件更多还可以在输出中增加单独的失败清单让使用者修复文件后再次处理。这次实现让我更明确地意识到批量任务除了需要一条成功路径还需要为局部失败设计返回结果。对于 Work Inbox Agent我希望每个文件都有可以追踪的处理结果。正常文件继续向后流转失败文件留下名称、路径和原因使用者才能判断本次整理覆盖了哪些资料还有哪些需要补充处理。项目源码Work Inbox Agent 
长河君 · 2026/10/27 1:53:03 · 阅读 1.2k 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_namepath.name, file_typepath.suffix.lower().lstrip(.) or unknown, source_pathstr(path), errorf读取失败: {exc}, ) return ExtractedDocument.unsupported( path, f暂不支持文件格式: {path.suffix or none}, )这里最关键的是try...except的位置。它包围的是单个文件的读取操作。如果某个读取器抛出异常程序会把失败转换成一个ExtractedDocument返回。调用方拿到的仍然是统一的数据结构只不过其中的error字段记录了失败信息。例如一个文件读取失败后结果会包含ExtractedDocument( file_namepath.name, file_typepath.suffix.lower().lstrip(.) or unknown, source_pathstr(path), errorf读取失败: {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_namepath.name, file_typeunsupported, source_pathstr(path), errormessage, )这样做是为了回答一个问题文件夹里明明有这个文件为什么输出中看不到它的内容如果完全跳过使用者很难判断是程序漏扫了文件还是格式本身不支持。保留记录之后就可以区分文件被扫描到了但当前不支持解析文件格式受支持但读取过程中失败文件读取成功获得了内容。这三种状态对后续排查和展示都有帮助。五、目录错误仍然需要明确抛出单个文件失败可以记录下来但目标目录本身有问题时继续执行就没有意义。因此扫描入口保留了明确的路径检查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( encodingutf-8-sig, errorsreplace, )遇到无法解码的字节时它会使用替代字符读取过程可能仍然成功。因此“没有抛出异常”不能直接理解成“文件内容没有问题”。Excel 读取也有类似边界当前版本只预览前 50 行并且读取公式本身不负责计算公式结果。这些属于解析范围限制需要在使用和文档中明确说明。八、后续准备怎样改进当前版本把失败原因放进了error字符串已经能够完成基础定位。随着功能增加我准备进一步区分错误类型例如文件损坏、权限不足和格式不支持。测试方面目前仓库已有“不支持的文件仍保留在扫描结果中”的用例。接下来还可以补充一个更直接的异常隔离测试在同一目录中放入正常文本和损坏的.docx检查正常文件是否读取成功以及损坏文件是否留下错误记录。如果后续处理的文件更多还可以在输出中增加单独的失败清单让使用者修复文件后再次处理。这次实现让我更明确地意识到批量任务除了需要一条成功路径还需要为局部失败设计返回结果。对于 Work Inbox Agent我希望每个文件都有可以追踪的处理结果。正常文件继续向后流转失败文件留下名称、路径和原因使用者才能判断本次整理覆盖了哪些资料还有哪些需要补充处理。项目源码Work Inbox Agent