如何避免读取损坏文件？htm2txt读取超时问题求助

一、不读取完整文件的损坏文件预检测 #

针对百万级文件的处理场景，先通过轻量级规则过滤明显损坏的文件，减少后续处理的超时概率：

• 文件大小过滤：先统计正常HTML文件的大小范围（比如多数在1KB-10MB区间），将远超出上限（如50MB）或明显过小（如小于100字节）的文件直接标记为可疑损坏文件，跳过处理。
• 头尾特征校验：快速读取文件的前100字节和后100字节，检查是否符合HTML基本结构：
  • 头部是否包含<!DOCTYPE html>或<html>标签
  • 尾部是否包含</html>标签
    不符合的直接跳过，无需完整读取内容。
• 编码合法性检测：用chardet或cchardet库检测文件编码，无法识别有效编码的文件大概率已损坏，直接过滤。

二、读取超时强制跳过的实现（保留gettxt()） #

如果预检测无法覆盖所有损坏场景，用线程池实现超时控制，确保单个文件处理超时后自动跳过，不阻塞整体循环：

代码示例 #

import os
import htm2txt
from concurrent.futures import ThreadPoolExecutor, TimeoutError

def process_single_file(file_path):
    """处理单个文件的核心逻辑"""
    try:
        # 可添加多编码兼容逻辑，比如先尝试utf-8，失败再试gbk
        with open(file_path, 'r', encoding='utf-8') as f:
            html_content = f.read()
        # 调用gettxt提取文本
        extracted_text = htm2txt.gettxt(html_content)
        # 此处添加提取后文本的保存逻辑（写入文件/数据库等）
        return True, file_path, None
    except Exception as e:
        return False, file_path, str(e)

# 配置参数
target_dir = "你的文件目录路径"
timeout_seconds = 10  # 10秒超时阈值
file_list = [os.path.join(target_dir, f) for f in os.listdir(target_dir) if f.endswith('.txt')]

# 批量处理文件
for file_path in file_list:
    with ThreadPoolExecutor(max_workers=1) as executor:
        future = executor.submit(process_single_file, file_path)
        try:
            success, path, error_msg = future.result(timeout=timeout_seconds)
            if success:
                print(f"已完成: {path}")
            else:
                print(f"处理失败: {path} | 错误: {error_msg}")
        except TimeoutError:
            print(f"超时跳过: {path}")
        except Exception as e:
            print(f"未知错误: {path} | 异常: {str(e)}")

注意事项 #

• 若文件编码混杂，可在process_single_file的文件读取部分添加多编码尝试逻辑，避免因编码错误导致的异常。
• 建议将超时、失败的文件路径写入日志文件，方便后续排查或二次处理。
• 先选取小批量文件测试规则，调整大小阈值、头尾检测字节数等参数，优化检测准确率。

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