全自动文档清洗工具：从读取到生成报告

我给你逐行精简讲解，只讲核心作用，不啰嗦，方便快速看懂这段代码是干嘛的。

def run_pipeline(input_dir: str = "Data", output_dir: str = "CleanedData"):

定义一个文档清洗流水线函数，默认从 Data 文件夹读，输出到 CleanedData。

    """
    执行端到端清洗流水线
    支持：基础正则、PII 脱敏、LLM 布局剥离、结构增强、表格描述、图像内容识别、质量评估。
    """

注释：说明这个函数是全自动文档清洗，支持脱敏、格式清理、内容增强、质量打分。

    # 0. 环境准备
    attachments_dir = os.path.join(output_dir, "Attachments")
    os.makedirs(output_dir, exist_ok=True)
    os.makedirs(attachments_dir, exist_ok=True)

创建输出文件夹和附件文件夹，保证目录存在。

    # 预设 API Key (LongCat)
    os.environ["LONGCAT_API_KEY"] = "ak_2FX4eV42e8LS79D32K7Js0B995U5O"

设置第三方服务的 API 密钥，用于调用模型能力。

    # 1. 初始化核心组件
    reader = DocumentReader(attachments_dir=attachments_dir)
    engine = CleanerFactory.default()
    processor = DocumentProcessor(engine)

初始化三个工具：

• reader：读取各种格式文档（Word/PDF/图片等）
• engine：清洗规则引擎
• processor：执行清洗流程

    # 2. 收集支持的文件
    if not os.path.exists(input_dir):
        logger.error(f"输入目录 {input_dir} 不存在！")
        return

检查输入文件夹是否存在，不存在直接报错退出。

    files = []
    for root, _, filenames in os.walk(input_dir):
        for f in filenames:
            if reader.is_supported(f):
                files.append(os.path.join(root, f))

遍历文件夹，只收集能识别的文档。

    if not files:
        logger.warning(f"在 {input_dir} 中未找到受支持的文件。")
        return

没找到文件就警告并退出。

    logger.info(f"找到 {len(files)} 个待处理文档。正在启动管道...")

打印找到多少文件，开始处理。

    # 3. 循环处理
    report_rows = []
    for i, filepath in enumerate(files, 1):

逐个处理文档，并记录报告数据。

        filename = os.path.basename(filepath)
        logger.info("-" * 50)
        logger.info(f"[{i}/{len(files)}] 正在处理: {filename}")

打印当前处理进度。

        # A. 读取文档
        doc_res = reader.read(filepath)
        if not doc_res.success:
            logger.error(f"读取失败: {doc_res.error}")
            continue

读取文档内容，读失败就跳过。

        # B. 构造 LangChain Document 并注入元数据
        images = [att["path"] for att in (doc_res.attachments or []) if att["type"] == "image"]

提取文档里的图片附件路径。

        doc = LCDocument(
            page_content=doc_res.content,
            metadata={...}
        )

把文档内容包装成标准格式，方便后续处理。

        # C. 配置清洗选项 (5 大 LLM 增强任务)
        options = {
            "llm_tasks": [],  # 完全禁用 LLM 任务，避免幻觉内容
            "max_hierarchy_depth": 3,
            "skip_pii": False,        # 开启 PII 脱敏
            "deep_pii_scan": True
        }

清洗设置：

• 关闭 LLM（防止瞎编内容）
• 开启隐私信息脱敏（手机号/身份证等）
• 只保留到三级标题结构

        # D. 调用处理器执行清洗
        start_time = time.time()
        try:
            result = processor.process(doc, options=options)
            duration = time.time() - start_time

执行清洗，计算耗时。

            if not result:
                logger.warning(f"文档已被过滤规则拦截: {filename}")
                continue

无效/垃圾内容直接过滤掉。

            # E. 保存结果 — 纯文本
            output_filename = filename.rsplit('.', 1)[0] + "_cleaned.txt"
            save_path = os.path.join(output_dir, output_filename)
            
            with open(save_path, "w", encoding="utf-8") as f:
                f.write(result.content)

把清洗后的内容保存为 txt 文件。

            # F. 收集结构化报告数据
            report_rows.append({...})

记录处理结果：长度、压缩率、隐私检出数、质量得分、耗时等。

            logger.info(f"清洗完成! 耗时: {duration:.2f}s")
            ... 打印各种信息

打印处理结果日志。

        except Exception as e:
            logger.error(f"处理文档 {filename} 时发生异常: {e}", exc_info=True)

捕获异常，单个文档报错不影响整体流程。

    # 4. 生成结构化汇总报告 (Markdown 表格)
    _generate_report(report_rows, output_dir)

所有文档处理完，生成一份汇总报告。

    logger.info("-" * 50)
    logger.info("所有处理任务已完成。")

结束，打印完成提示。

总结（超精简）

这段代码是一个全自动文档清洗工具：

1. 读文件夹里的文档
2. 提取内容 + 图片
3. 清理格式、脱敏隐私、过滤垃圾
4. 输出干净文本
5. 生成质量报告
6. 稳定、报错不崩溃、可批量处理