AIclinicalresearch/docs/02-通用能力层/02-文档处理引擎/01-文档处理引擎设计方案.md

文档处理引擎设计方案

文档版本： v1.1
创建日期： 2026-01-20
最后更新： 2026-01-20
文档目的： 定义统一的文档处理策略，将各类文档转换为 LLM 友好的 Markdown 格式
适用范围： PKB 知识库、ASL 智能文献、DC 数据清洗、AIA 附件处理
核心原则： 极轻量、零 OCR、聚焦核心格式

---

📋 概述

设计理念

构建一个 "极轻量、零 OCR、LLM 友好" 的文档解析微服务。

核心原则（适合 2 人小团队）：

• 抓大放小 - 确保 PDF/Word/Excel 的绝对准确，冷门格式按需扩展
• 零 OCR - 只处理电子版文档，放弃扫描件支持，换取极致部署速度
• 容错优雅 - 解析失败时返回 LLM 友好的提示，不中断流程

设计目标

1. 聚焦核心格式 - PDF、Word、Excel、PPT 覆盖 95% 使用场景
2. LLM 友好输出 - 统一转换为结构化 Markdown，包含上下文信息
3. 表格保真 - 完整保留文献中的表格信息（临床试验核心数据）
4. 极致轻量 - Docker 镜像 < 300MB，资源占用 < 512MB 内存

架构概览

┌─────────────────────────────────────────────────────────────┐
│                   DocumentProcessor                          │
│  (统一入口：自动检测文件类型，调用对应处理器)                    │
├─────────────────────────────────────────────────────────────┤
│  ┌───────────┐ ┌───────────┐ ┌───────────┐ ┌───────────┐   │
│  │    PDF    │ │   Word    │ │    PPT    │ │   Excel   │   │
│  │ Processor │ │ Processor │ │ Processor │ │ Processor │   │
│  │pymupdf4llm│ │  mammoth  │ │python-pptx│ │  pandas   │   │
│  └───────────┘ └───────────┘ └───────────┘ └───────────┘   │
│  ┌───────────┐ ┌───────────┐ ┌───────────┐ ┌───────────┐   │
│  │    CSV    │ │   HTML    │ │  文献引用  │ │   医学    │   │
│  │ Processor │ │ Processor │ │ Processor │ │ Processor │   │
│  │  pandas   │ │markdownify│ │bibtexparser│ │  pydicom  │   │
│  └───────────┘ └───────────┘ └───────────┘ └───────────┘   │
├─────────────────────────────────────────────────────────────┤
│                    输出: 统一 Markdown 格式                   │
└─────────────────────────────────────────────────────────────┘

---

📄 支持格式与工具选型

格式覆盖矩阵

分类	格式	推荐工具	优先级	状态
文档类	PDF (.pdf)	pymupdf4llm	P0	✅ 推荐
	Word (.docx)	mammoth	P0	✅ 推荐
	PowerPoint (.pptx)	python-pptx	P1	✅ 推荐
	纯文本 (.txt/.md)	直接读取	P0	✅ 内置
	富文本 (.rtf)	striprtf	P2	🔜 待实现
表格类	Excel (.xlsx)	pandas + openpyxl	P0	✅ 推荐
	CSV (.csv)	pandas	P0	✅ 推荐
	SAS (.sas7bdat)	pandas + sas7bdat	P2	🔜 待实现
	SPSS (.sav)	pandas + pyreadstat	P2	🔜 待实现
	Stata (.dta)	pandas.read_stata()	P2	🔜 待实现
网页类	HTML (.html)	beautifulsoup4 + markdownify	P1	✅ 推荐
	电子书 (.epub)	ebooklib	P2	🔜 待实现
引用类	BibTeX (.bib)	bibtexparser	P1	✅ 推荐
	RIS (.ris)	rispy	P1	✅ 推荐
	EndNote (.enw)	自定义解析	P2	🔜 待实现
医学类	DICOM (.dcm)	pydicom	P2	🔜 待实现
	HL7/FHIR	hl7 / fhirclient	P3	📋 规划中
数据类	JSON (.json/.jsonl)	json 标准库	P1	✅ 内置
	XML (.xml)	lxml	P1	✅ 推荐

---

🔧 详细实现方案

1. PDF 文档处理

工具选择：pymupdf4llm

关键决策：只保留 pymupdf4llm，移除独立的 PyMuPDF 和 Nougat

对比项	PyMuPDF (旧)	pymupdf4llm (新)	Nougat
表格提取	基础文本	✅ Markdown 表格	✅ LaTeX 表格
图片处理	提取二进制	✅ 自动 base64	✅ 支持
数学公式	❌	✅ 保留 LaTeX	✅ 原生 LaTeX
多栏布局	需手动处理	✅ 自动重排	✅ 支持
处理速度	快	快	慢（GPU）
依赖复杂度	低	低	高
扫描版 PDF	❌	❌	✅

说明：

• pymupdf4llm 是 PyMuPDF 的上层封装，安装时自动包含 pymupdf 依赖
• 对于普通 PDF（文本型），pymupdf4llm 已完全满足需求
• 扫描版 PDF 策略：检测后返回友好提示，不阻断流程（零 OCR 原则）

代码实现

# pdf_processor.py
import pymupdf4llm
import logging
from pathlib import Path
from typing import Optional, List, Dict, Any

logger = logging.getLogger(__name__)

class PdfProcessor:
    """PDF 文档处理器 - 基于 pymupdf4llm（仅支持电子版）"""
    
    # 扫描件检测阈值：提取文本少于此字符数视为扫描件
    MIN_TEXT_THRESHOLD = 50
    
    def __init__(self, image_dir: str = "./images"):
        self.image_dir = image_dir
    
    def to_markdown(
        self,
        pdf_path: str,
        page_chunks: bool = False,
        extract_images: bool = True,
        dpi: int = 150
    ) -> str:
        """
        PDF 转 Markdown（仅支持电子版）
        
        Args:
            pdf_path: PDF 文件路径
            page_chunks: 是否按页分块
            extract_images: 是否提取图片
            dpi: 图片分辨率
        
        Returns:
            Markdown 格式的文本
        
        Note:
            扫描版 PDF 会返回友好提示，不会抛出异常
        """
        try:
            md_text = pymupdf4llm.to_markdown(
                pdf_path,
                page_chunks=page_chunks,
                write_images=extract_images,
                image_path=self.image_dir,
                dpi=dpi,
                show_progress=False
            )
            
            # 如果返回的是列表（page_chunks=True），合并为字符串
            if isinstance(md_text, list):
                md_text = "\n\n---\n\n".join([
                    f"## Page {i+1}\n\n{page['text']}" 
                    for i, page in enumerate(md_text)
                ])
            
            # 质量检查：检测是否为扫描件
            if len(md_text.strip()) < self.MIN_TEXT_THRESHOLD:
                logger.warning(f"PDF 文本过少 ({len(md_text.strip())} 字符)，可能为扫描件: {pdf_path}")
                return self._scan_pdf_hint(pdf_path, len(md_text.strip()))
            
            return md_text
            
        except Exception as e:
            logger.error(f"PDF 解析失败: {pdf_path}, 错误: {e}")
            raise ValueError(f"PDF 解析失败: {str(e)}")
    
    def _scan_pdf_hint(self, pdf_path: str, char_count: int) -> str:
        """生成扫描件友好提示（让 LLM 知道文件无法读取）"""
        filename = Path(pdf_path).name
        return f"""> **系统提示**：文档 `{filename}` 似乎是扫描件（图片型 PDF）。
> 
> - 提取文本量：{char_count} 字符
> - 本系统暂不支持扫描版 PDF 的文字识别
> - 建议：请上传电子版 PDF，或将扫描件转换为可编辑格式后重新上传"""
    
    def extract_tables(self, pdf_path: str) -> List[Dict[str, Any]]:
        """
        提取 PDF 中的所有表格
        
        Returns:
            表格列表，每个表格包含页码和 Markdown 格式内容
        """
        import fitz  # pymupdf
        
        tables = []
        doc = fitz.open(pdf_path)
        
        for page_num, page in enumerate(doc, 1):
            # pymupdf 4.x 原生表格提取
            page_tables = page.find_tables()
            for idx, table in enumerate(page_tables):
                df = table.to_pandas()
                tables.append({
                    "page": page_num,
                    "table_index": idx,
                    "markdown": df.to_markdown(index=False),
                    "rows": len(df),
                    "cols": len(df.columns)
                })
        
        doc.close()
        return tables
    
    def get_metadata(self, pdf_path: str) -> Dict[str, Any]:
        """提取 PDF 元数据"""
        import fitz
        
        doc = fitz.open(pdf_path)
        metadata = doc.metadata
        metadata["page_count"] = len(doc)
        doc.close()
        
        return metadata

配置依赖

# requirements.txt
pymupdf4llm>=0.0.10  # 自动包含 pymupdf 依赖

---

2. Word 文档处理 (.docx)

工具选择：mammoth（推荐）

对比项	python-docx	mammoth
输出格式	需手动转换	✅ 直接 Markdown/HTML
表格处理	精确控制	✅ 自动转换
样式保留	完整	基础样式
复杂度	高	✅ 低
适用场景	需要精细控制	✅ 快速转换

建议：主用 mammoth，python-docx 作为备选（复杂文档）

代码实现

# docx_processor.py
import mammoth
import logging
from pathlib import Path
from typing import Optional, Dict, Any

logger = logging.getLogger(__name__)

class DocxProcessor:
    """Word 文档处理器 - 基于 mammoth"""
    
    def to_markdown(self, docx_path: str) -> str:
        """
        Word 转 Markdown
        
        Args:
            docx_path: Word 文件路径
        
        Returns:
            Markdown 格式的文本
        
        Note:
            空文档会返回友好提示
        """
        try:
            with open(docx_path, "rb") as f:
                result = mammoth.convert_to_markdown(f)
            
            # 记录转换警告
            if result.messages:
                for msg in result.messages:
                    logger.warning(f"[Word 转换警告] {msg.message}")
            
            # 空文档检测
            if not result.value.strip():
                filename = Path(docx_path).name
                return f"> **系统提示**：Word 文档 `{filename}` 内容为空或无法识别。"
            
            return result.value
            
        except Exception as e:
            logger.error(f"Word 解析失败: {docx_path}, 错误: {e}")
            raise ValueError(f"Word 解析失败: {str(e)}")
    
    def to_html(self, docx_path: str) -> str:
        """Word 转 HTML（保留更多样式）"""
        with open(docx_path, "rb") as f:
            result = mammoth.convert_to_html(f)
        return result.value
    
    def extract_images(self, docx_path: str, output_dir: str) -> list:
        """提取 Word 中的图片"""
        from docx import Document
        import os
        
        doc = Document(docx_path)
        images = []
        
        for idx, rel in enumerate(doc.part.rels.values()):
            if "image" in rel.target_ref:
                image_data = rel.target_part.blob
                ext = rel.target_ref.split(".")[-1]
                image_path = os.path.join(output_dir, f"image_{idx}.{ext}")
                
                with open(image_path, "wb") as f:
                    f.write(image_data)
                images.append(image_path)
        
        return images

配置依赖

# requirements.txt
mammoth>=1.6.0
python-docx>=0.8.11  # 备选，用于复杂文档

---

3. PowerPoint 文档处理 (.pptx)

工具选择：python-pptx

代码实现

# pptx_processor.py
from pptx import Presentation
from pptx.enum.shapes import MSO_SHAPE_TYPE
from typing import List, Dict, Any
import os

class PptxProcessor:
    """PowerPoint 文档处理器 - 基于 python-pptx"""
    
    def to_markdown(
        self,
        pptx_path: str,
        extract_images: bool = False,
        image_dir: str = "./images"
    ) -> str:
        """
        PPT 转 Markdown
        
        Args:
            pptx_path: PPT 文件路径
            extract_images: 是否提取图片
            image_dir: 图片保存目录
        
        Returns:
            Markdown 格式的文本
        """
        prs = Presentation(pptx_path)
        md_parts = []
        image_count = 0
        
        for slide_num, slide in enumerate(prs.slides, 1):
            md_parts.append(f"## Slide {slide_num}")
            
            # 提取幻灯片标题
            if slide.shapes.title:
                md_parts.append(f"### {slide.shapes.title.text}")
            
            # 遍历所有形状
            for shape in slide.shapes:
                # 文本框
                if shape.has_text_frame:
                    for para in shape.text_frame.paragraphs:
                        text = para.text.strip()
                        if text and text != slide.shapes.title.text if slide.shapes.title else True:
                            # 根据层级添加缩进
                            level = para.level
                            prefix = "  " * level + "- " if level > 0 else ""
                            md_parts.append(f"{prefix}{text}")
                
                # 表格
                if shape.has_table:
                    md_parts.append(self._table_to_markdown(shape.table))
                
                # 图片
                if extract_images and shape.shape_type == MSO_SHAPE_TYPE.PICTURE:
                    image_count += 1
                    image_path = os.path.join(image_dir, f"slide{slide_num}_img{image_count}.png")
                    self._save_image(shape, image_path)
                    md_parts.append(f"![Image]({image_path})")
            
            md_parts.append("")  # 空行分隔
        
        return "\n".join(md_parts)
    
    def _table_to_markdown(self, table) -> str:
        """表格转 Markdown"""
        rows = []
        
        for row_idx, row in enumerate(table.rows):
            cells = [cell.text.strip() for cell in row.cells]
            rows.append("| " + " | ".join(cells) + " |")
            
            # 表头分隔行
            if row_idx == 0:
                rows.append("| " + " | ".join(["---"] * len(cells)) + " |")
        
        return "\n".join(rows)
    
    def _save_image(self, shape, output_path: str):
        """保存图片"""
        os.makedirs(os.path.dirname(output_path), exist_ok=True)
        with open(output_path, "wb") as f:
            f.write(shape.image.blob)
    
    def get_outline(self, pptx_path: str) -> List[Dict[str, Any]]:
        """获取 PPT 大纲"""
        prs = Presentation(pptx_path)
        outline = []
        
        for slide_num, slide in enumerate(prs.slides, 1):
            slide_info = {
                "slide_number": slide_num,
                "title": slide.shapes.title.text if slide.shapes.title else None,
                "text_count": sum(
                    len(shape.text_frame.text) 
                    for shape in slide.shapes 
                    if shape.has_text_frame
                )
            }
            outline.append(slide_info)
        
        return outline

配置依赖

# requirements.txt
python-pptx>=0.6.23

---

4. Excel 文档处理 (.xlsx)

工具选择：pandas + openpyxl

代码实现

# excel_processor.py
import pandas as pd
import logging
from pathlib import Path
from typing import List, Dict, Any, Optional

logger = logging.getLogger(__name__)

class ExcelProcessor:
    """Excel 文档处理器 - 基于 pandas + openpyxl"""
    
    def to_markdown(
        self,
        xlsx_path: str,
        sheet_names: Optional[List[str]] = None,
        max_rows: int = 200
    ) -> str:
        """
        Excel 转 Markdown（包含丰富上下文信息）
        
        Args:
            xlsx_path: Excel 文件路径
            sheet_names: 指定 Sheet 名称列表，None 表示全部
            max_rows: 最大行数限制（防止超大文件，默认 200 行）
        
        Returns:
            Markdown 格式的文本（包含文件名、行列数等上下文）
        """
        filename = Path(xlsx_path).name
        md_parts = []
        
        try:
            xlsx = pd.ExcelFile(xlsx_path, engine='openpyxl')
            sheets_to_process = sheet_names or xlsx.sheet_names
            
            for sheet_name in sheets_to_process:
                if sheet_name not in xlsx.sheet_names:
                    continue
                
                df = pd.read_excel(xlsx, sheet_name=sheet_name)
                total_rows = len(df)
                
                # 添加数据来源上下文（LLM 友好）
                md_parts.append(f"## 数据来源: {filename} - {sheet_name}")
                md_parts.append(f"- **行列**: {total_rows} 行 × {len(df.columns)} 列")
                
                # 截断提示
                if total_rows > max_rows:
                    md_parts.append(f"> ⚠️ 数据量较大，仅显示前 {max_rows} 行（共 {total_rows} 行）")
                    df = df.head(max_rows)
                
                md_parts.append("")
                
                # 处理空值，避免 NaN 显示
                df = df.fillna('')
                md_parts.append(df.to_markdown(index=False))
                md_parts.append("\n---\n")
            
            return "\n".join(md_parts)
            
        except Exception as e:
            logger.error(f"Excel 解析失败: {xlsx_path}, 错误: {e}")
            return f"> **系统提示**：Excel 文件 `{filename}` 解析失败: {str(e)}"
    
    def get_sheet_info(self, xlsx_path: str) -> List[Dict[str, Any]]:
        """获取 Excel 所有 Sheet 信息"""
        xlsx = pd.ExcelFile(xlsx_path, engine='openpyxl')
        sheets = []
        
        for sheet_name in xlsx.sheet_names:
            df = pd.read_excel(xlsx, sheet_name=sheet_name)
            sheets.append({
                "name": sheet_name,
                "rows": len(df),
                "columns": len(df.columns),
                "column_names": df.columns.tolist()
            })
        
        return sheets
    
    def extract_sheet(
        self,
        xlsx_path: str,
        sheet_name: str,
        as_dict: bool = False
    ) -> Any:
        """提取单个 Sheet 数据"""
        df = pd.read_excel(xlsx_path, sheet_name=sheet_name, engine='openpyxl')
        
        if as_dict:
            return df.to_dict(orient='records')
        return df

配置依赖

# requirements.txt
pandas>=2.0.0
openpyxl>=3.1.2
tabulate>=0.9.0  # pandas.to_markdown() 依赖

---

5. CSV 文件处理

代码实现

# csv_processor.py
import pandas as pd
import logging
from pathlib import Path
from typing import Optional, List
import chardet

logger = logging.getLogger(__name__)

class CsvProcessor:
    """CSV 文件处理器 - 基于 pandas"""
    
    def to_markdown(
        self,
        csv_path: str,
        encoding: Optional[str] = None,
        max_rows: int = 200,
        delimiter: str = ','
    ) -> str:
        """
        CSV 转 Markdown（包含丰富上下文信息）
        
        Args:
            csv_path: CSV 文件路径
            encoding: 文件编码（自动检测）
            max_rows: 最大行数限制（默认 200 行）
            delimiter: 分隔符
        
        Returns:
            Markdown 格式的文本
        """
        filename = Path(csv_path).name
        
        try:
            # 自动检测编码
            if encoding is None:
                encoding = self._detect_encoding(csv_path)
            
            df = pd.read_csv(csv_path, encoding=encoding, delimiter=delimiter)
            total_rows = len(df)
            
            md_parts = [
                f"## 数据来源: {filename}",
                f"- **行列**: {total_rows} 行 × {len(df.columns)} 列",
                f"- **编码**: {encoding}",
            ]
            
            # 截断提示
            if total_rows > max_rows:
                md_parts.append(f"> ⚠️ 数据量较大，仅显示前 {max_rows} 行（共 {total_rows} 行）")
                df = df.head(max_rows)
            
            md_parts.append("")
            df = df.fillna('')
            md_parts.append(df.to_markdown(index=False))
            
            return "\n".join(md_parts)
            
        except Exception as e:
            logger.error(f"CSV 解析失败: {csv_path}, 错误: {e}")
            return f"> **系统提示**：CSV 文件 `{filename}` 解析失败: {str(e)}"
    
    def _detect_encoding(self, file_path: str) -> str:
        """自动检测文件编码"""
        with open(file_path, 'rb') as f:
            raw_data = f.read(10000)  # 读取前 10KB
        
        result = chardet.detect(raw_data)
        encoding = result['encoding']
        
        # 常见编码映射
        encoding_map = {
            'GB2312': 'gbk',
            'gb2312': 'gbk',
            'GBK': 'gbk',
            'GB18030': 'gb18030',
        }
        
        return encoding_map.get(encoding, encoding or 'utf-8')

配置依赖

# requirements.txt
pandas>=2.0.0
chardet>=5.0.0

---

6. HTML 文档处理

工具选择：beautifulsoup4 + markdownify

代码实现

# html_processor.py
from bs4 import BeautifulSoup
from markdownify import markdownify as md
from typing import Optional

class HtmlProcessor:
    """HTML 文档处理器 - 基于 beautifulsoup4 + markdownify"""
    
    def to_markdown(
        self,
        html_content: str,
        strip_tags: Optional[list] = None
    ) -> str:
        """
        HTML 转 Markdown
        
        Args:
            html_content: HTML 内容
            strip_tags: 要移除的标签列表
        
        Returns:
            Markdown 格式的文本
        """
        # 预处理：移除脚本和样式
        soup = BeautifulSoup(html_content, 'html.parser')
        
        for tag in soup(['script', 'style', 'nav', 'footer', 'header']):
            tag.decompose()
        
        if strip_tags:
            for tag_name in strip_tags:
                for tag in soup(tag_name):
                    tag.decompose()
        
        # 转换为 Markdown
        markdown = md(str(soup), heading_style="ATX", bullets="-")
        
        # 清理多余空行
        lines = markdown.split('\n')
        cleaned_lines = []
        prev_empty = False
        
        for line in lines:
            is_empty = not line.strip()
            if is_empty and prev_empty:
                continue
            cleaned_lines.append(line)
            prev_empty = is_empty
        
        return '\n'.join(cleaned_lines)
    
    def from_file(self, html_path: str, encoding: str = 'utf-8') -> str:
        """从文件读取 HTML 并转换"""
        with open(html_path, 'r', encoding=encoding) as f:
            html_content = f.read()
        return self.to_markdown(html_content)
    
    def extract_text(self, html_content: str) -> str:
        """仅提取纯文本"""
        soup = BeautifulSoup(html_content, 'html.parser')
        return soup.get_text(separator='\n', strip=True)

配置依赖

# requirements.txt
beautifulsoup4>=4.12.0
markdownify>=0.11.6
lxml>=4.9.0  # BeautifulSoup 的高性能解析器

---

7. 文献引用格式处理

代码实现

# reference_processor.py
import bibtexparser
import rispy
from typing import List, Dict, Any

class ReferenceProcessor:
    """文献引用格式处理器"""
    
    def bib_to_markdown(self, bib_path: str) -> str:
        """
        BibTeX 转 Markdown
        
        Args:
            bib_path: .bib 文件路径
        
        Returns:
            Markdown 格式的参考文献列表
        """
        with open(bib_path, 'r', encoding='utf-8') as f:
            bib_database = bibtexparser.load(f)
        
        md_parts = ["# 参考文献\n"]
        
        for idx, entry in enumerate(bib_database.entries, 1):
            # 格式化为引用格式
            authors = entry.get('author', 'Unknown')
            title = entry.get('title', 'No title')
            year = entry.get('year', 'N/A')
            journal = entry.get('journal', entry.get('booktitle', ''))
            
            citation = f"{idx}. {authors}. **{title}**. "
            if journal:
                citation += f"*{journal}*. "
            citation += f"({year})"
            
            md_parts.append(citation)
            md_parts.append("")
        
        return "\n".join(md_parts)
    
    def ris_to_markdown(self, ris_path: str) -> str:
        """
        RIS 转 Markdown
        
        Args:
            ris_path: .ris 文件路径
        
        Returns:
            Markdown 格式的参考文献列表
        """
        with open(ris_path, 'r', encoding='utf-8') as f:
            entries = rispy.load(f)
        
        md_parts = ["# 参考文献\n"]
        
        for idx, entry in enumerate(entries, 1):
            authors = ', '.join(entry.get('authors', ['Unknown']))
            title = entry.get('title', entry.get('primary_title', 'No title'))
            year = entry.get('year', entry.get('publication_year', 'N/A'))
            journal = entry.get('journal_name', entry.get('secondary_title', ''))
            
            citation = f"{idx}. {authors}. **{title}**. "
            if journal:
                citation += f"*{journal}*. "
            citation += f"({year})"
            
            md_parts.append(citation)
            md_parts.append("")
        
        return "\n".join(md_parts)
    
    def parse_bib(self, bib_path: str) -> List[Dict[str, Any]]:
        """解析 BibTeX 返回结构化数据"""
        with open(bib_path, 'r', encoding='utf-8') as f:
            bib_database = bibtexparser.load(f)
        return bib_database.entries

配置依赖

# requirements.txt
bibtexparser>=1.4.0
rispy>=0.7.0

---

8. 医学数据格式处理（扩展）

DICOM 元数据提取

# dicom_processor.py
import pydicom
from typing import Dict, Any

class DicomProcessor:
    """DICOM 医学影像元数据处理器"""
    
    def extract_metadata(self, dcm_path: str) -> Dict[str, Any]:
        """
        提取 DICOM 元数据
        
        Args:
            dcm_path: DICOM 文件路径
        
        Returns:
            元数据字典
        """
        dcm = pydicom.dcmread(dcm_path)
        
        # 提取关键元数据
        metadata = {
            "patient_name": str(dcm.PatientName) if hasattr(dcm, 'PatientName') else None,
            "patient_id": dcm.PatientID if hasattr(dcm, 'PatientID') else None,
            "study_date": dcm.StudyDate if hasattr(dcm, 'StudyDate') else None,
            "modality": dcm.Modality if hasattr(dcm, 'Modality') else None,
            "study_description": dcm.StudyDescription if hasattr(dcm, 'StudyDescription') else None,
            "series_description": dcm.SeriesDescription if hasattr(dcm, 'SeriesDescription') else None,
            "institution_name": dcm.InstitutionName if hasattr(dcm, 'InstitutionName') else None,
            "manufacturer": dcm.Manufacturer if hasattr(dcm, 'Manufacturer') else None,
        }
        
        return {k: v for k, v in metadata.items() if v is not None}
    
    def to_markdown(self, dcm_path: str) -> str:
        """DICOM 元数据转 Markdown"""
        metadata = self.extract_metadata(dcm_path)
        
        md_parts = ["# DICOM 影像信息\n"]
        
        for key, value in metadata.items():
            label = key.replace('_', ' ').title()
            md_parts.append(f"- **{label}**: {value}")
        
        return "\n".join(md_parts)

统计软件数据格式

# stats_data_processor.py
import pandas as pd
from typing import Optional

class StatsDataProcessor:
    """统计软件数据格式处理器（SAS/SPSS/Stata）"""
    
    def sas_to_markdown(
        self,
        sas_path: str,
        max_rows: int = 1000
    ) -> str:
        """SAS 数据转 Markdown"""
        df = pd.read_sas(sas_path)
        return self._df_to_markdown(df, "SAS", sas_path, max_rows)
    
    def spss_to_markdown(
        self,
        sav_path: str,
        max_rows: int = 1000
    ) -> str:
        """SPSS 数据转 Markdown"""
        import pyreadstat
        df, meta = pyreadstat.read_sav(sav_path)
        
        md = self._df_to_markdown(df, "SPSS", sav_path, max_rows)
        
        # 添加变量标签信息
        if meta.column_labels:
            md += "\n\n## 变量标签\n\n"
            for col, label in zip(meta.column_names, meta.column_labels):
                if label:
                    md += f"- **{col}**: {label}\n"
        
        return md
    
    def stata_to_markdown(
        self,
        dta_path: str,
        max_rows: int = 1000
    ) -> str:
        """Stata 数据转 Markdown"""
        df = pd.read_stata(dta_path)
        return self._df_to_markdown(df, "Stata", dta_path, max_rows)
    
    def _df_to_markdown(
        self,
        df: pd.DataFrame,
        source_type: str,
        file_path: str,
        max_rows: int
    ) -> str:
        """DataFrame 转 Markdown 通用方法"""
        if len(df) > max_rows:
            df = df.head(max_rows)
            truncated = True
        else:
            truncated = False
        
        md_parts = [
            f"# {source_type} 数据\n",
            f"**文件**: {file_path}",
            f"**行数**: {len(df)} | **列数**: {len(df.columns)}",
        ]
        
        if truncated:
            md_parts.append(f"**注意**: 数据已截断，仅显示前 {max_rows} 行")
        
        md_parts.extend(["", df.to_markdown(index=False)])
        
        return "\n".join(md_parts)

配置依赖

# requirements.txt
pydicom>=2.4.0        # DICOM
pyreadstat>=1.2.0     # SPSS/SAS/Stata
sas7bdat>=2.2.3       # SAS 格式支持

---

🏗️ 统一处理器架构

主入口类

# document_processor.py
from pathlib import Path
from typing import Optional, Dict, Any
import mimetypes

class DocumentProcessor:
    """
    统一文档处理器
    
    自动检测文件类型，调用对应处理器
    """
    
    # 文件扩展名与处理器映射
    PROCESSOR_MAP = {
        '.pdf': 'pdf',
        '.docx': 'docx',
        '.doc': 'docx',
        '.pptx': 'pptx',
        '.xlsx': 'excel',
        '.xls': 'excel',
        '.csv': 'csv',
        '.txt': 'text',
        '.md': 'text',
        '.html': 'html',
        '.htm': 'html',
        '.bib': 'bibtex',
        '.ris': 'ris',
        '.json': 'json',
        '.jsonl': 'jsonl',
        '.xml': 'xml',
        '.dcm': 'dicom',
        '.sas7bdat': 'sas',
        '.sav': 'spss',
        '.dta': 'stata',
    }
    
    def __init__(self, config: Optional[Dict[str, Any]] = None):
        self.config = config or {}
        self._init_processors()
    
    def _init_processors(self):
        """初始化各处理器"""
        from .pdf_processor import PdfProcessor
        from .docx_processor import DocxProcessor
        from .pptx_processor import PptxProcessor
        from .excel_processor import ExcelProcessor
        from .csv_processor import CsvProcessor
        from .html_processor import HtmlProcessor
        from .reference_processor import ReferenceProcessor
        
        self.processors = {
            'pdf': PdfProcessor(),
            'docx': DocxProcessor(),
            'pptx': PptxProcessor(),
            'excel': ExcelProcessor(),
            'csv': CsvProcessor(),
            'html': HtmlProcessor(),
            'reference': ReferenceProcessor(),
        }
    
    def to_markdown(self, file_path: str, **kwargs) -> str:
        """
        将任意文档转换为 Markdown
        
        Args:
            file_path: 文件路径
            **kwargs: 传递给具体处理器的参数
        
        Returns:
            Markdown 格式的文本
        
        Raises:
            ValueError: 不支持的文件格式
        """
        ext = Path(file_path).suffix.lower()
        processor_type = self.PROCESSOR_MAP.get(ext)
        
        if not processor_type:
            raise ValueError(f"不支持的文件格式: {ext}")
        
        # 纯文本文件直接读取
        if processor_type == 'text':
            return self._read_text(file_path)
        
        # JSON 文件
        if processor_type in ('json', 'jsonl'):
            return self._read_json(file_path, processor_type == 'jsonl')
        
        # 引用文件
        if processor_type in ('bibtex', 'ris'):
            ref_processor = self.processors['reference']
            if processor_type == 'bibtex':
                return ref_processor.bib_to_markdown(file_path)
            else:
                return ref_processor.ris_to_markdown(file_path)
        
        # 其他格式
        processor = self.processors.get(processor_type)
        if processor:
            return processor.to_markdown(file_path, **kwargs)
        
        raise ValueError(f"处理器未实现: {processor_type}")
    
    def _read_text(self, file_path: str) -> str:
        """读取纯文本文件"""
        with open(file_path, 'r', encoding='utf-8') as f:
            return f.read()
    
    def _read_json(self, file_path: str, is_jsonl: bool = False) -> str:
        """读取 JSON 文件并转为 Markdown"""
        import json
        
        with open(file_path, 'r', encoding='utf-8') as f:
            if is_jsonl:
                data = [json.loads(line) for line in f]
            else:
                data = json.load(f)
        
        # 格式化为 Markdown 代码块
        return f"```json\n{json.dumps(data, ensure_ascii=False, indent=2)}\n```"
    
    def get_supported_formats(self) -> list:
        """获取所有支持的格式"""
        return list(self.PROCESSOR_MAP.keys())
    
    def is_supported(self, file_path: str) -> bool:
        """检查文件是否支持"""
        ext = Path(file_path).suffix.lower()
        return ext in self.PROCESSOR_MAP

---

📦 依赖清单

核心依赖（极简版）

# requirements.txt - 文档处理引擎（极简版）
# 体积预估：Docker 镜像压缩后 200-300MB

# ===== 核心解析库 =====
pymupdf4llm>=0.0.17      # PDF（自动包含 pymupdf）
mammoth>=1.8.0            # Word
python-pptx>=1.0.2        # PPT
pandas>=2.2.0             # Excel/CSV
openpyxl>=3.1.5           # Excel 引擎
tabulate>=0.9.0           # Markdown 表格输出

# ===== 基础工具 =====
chardet>=5.2.0            # 编码检测

# ===== Web 服务 =====
fastapi>=0.109.0          # API 框架
uvicorn>=0.27.0           # ASGI 服务器
python-multipart>=0.0.9   # 文件上传

扩展依赖（按需安装）

# ===== HTML 处理（P1）=====
beautifulsoup4>=4.12.0
markdownify>=0.11.6
lxml>=4.9.0

# ===== 文献引用（P1）=====
bibtexparser>=1.4.0
rispy>=0.7.0

# ===== 医学数据（P2，可选）=====
# pydicom>=2.4.0          # DICOM
# pyreadstat>=1.2.0       # SPSS/SAS

依赖体积对比

方案	镜像大小	说明
极简版（推荐）	~200-300MB	核心依赖，覆盖 95% 场景
完整版	~400-500MB	包含 HTML、引用、医学格式
带 OCR	1.5GB+	❌ 不推荐，放弃扫描件支持

---

🚀 部署建议

Docker 配置

# Dockerfile
FROM python:3.11-slim

WORKDIR /app

# 安装依赖
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .

EXPOSE 8000
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]

资源配置

配置项	推荐值	说明
CPU	0.5 核	极简版资源占用低
内存	512MB	足够处理常规文档
磁盘	1GB	镜像 + 临时文件

用户引导

💡 前端上传界面建议添加提示：

"目前仅支持电子版 PDF，暂不支持扫描件或图片型文档"

这比在后端搞复杂的 OCR 性价比高得多。

---

🎯 使用示例

基础使用

from document_processor import DocumentProcessor

# 创建处理器
processor = DocumentProcessor()

# 转换 PDF
md = processor.to_markdown("research_paper.pdf")

# 转换 Word
md = processor.to_markdown("report.docx")

# 转换 Excel（指定 Sheet）
md = processor.to_markdown("data.xlsx", sheet_names=["Sheet1", "Results"])

# 检查支持格式
if processor.is_supported("unknown.xyz"):
    md = processor.to_markdown("unknown.xyz")
else:
    print("格式不支持")

批量处理

from pathlib import Path
from document_processor import DocumentProcessor

processor = DocumentProcessor()

def batch_convert(input_dir: str, output_dir: str):
    """批量转换目录下的所有文档"""
    input_path = Path(input_dir)
    output_path = Path(output_dir)
    output_path.mkdir(parents=True, exist_ok=True)
    
    for file in input_path.iterdir():
        if processor.is_supported(str(file)):
            try:
                md = processor.to_markdown(str(file))
                
                # 保存为 .md 文件
                output_file = output_path / f"{file.stem}.md"
                output_file.write_text(md, encoding='utf-8')
                
                print(f"✅ {file.name} -> {output_file.name}")
            except Exception as e:
                print(f"❌ {file.name}: {e}")

# 使用
batch_convert("./documents", "./markdown_output")

---

📊 与 PKB 知识库集成

文档入库流程

用户上传文档
    ↓
DocumentProcessor.to_markdown()
    ↓
文本分块 (ChunkService)
    ↓
向量化 (EmbeddingService)
    ↓
存储到 PostgreSQL + pgvector

示例代码

async def ingest_document(file_path: str, knowledge_base_id: str):
    """文档入库完整流程"""
    
    # 1. 转换为 Markdown
    processor = DocumentProcessor()
    markdown_content = processor.to_markdown(file_path)
    
    # 2. 分块
    chunks = chunk_service.split_text(
        markdown_content,
        chunk_size=512,
        overlap=50
    )
    
    # 3. 向量化
    embeddings = await embedding_service.embed_batch(
        [chunk.text for chunk in chunks]
    )
    
    # 4. 存储
    for chunk, embedding in zip(chunks, embeddings):
        await prisma.ekbChunk.create({
            "knowledgeBaseId": knowledge_base_id,
            "content": chunk.text,
            "embedding": embedding,
            "metadata": {
                "source_file": file_path,
                "chunk_index": chunk.index
            }
        })

---

📅 更新日志

v1.1 (2026-01-20)

吸收同事建议，优化设计：

• 🔄 设计理念更新：强调"极轻量、零 OCR、聚焦核心格式"
• ✅ PDF 扫描件检测：添加字符数阈值检测，返回 LLM 友好提示
• ✅ Word 空文档检测：空内容返回友好提示
• ✅ Excel 上下文增强：添加文件名、行列数、截断提示
• ✅ 空值处理：fillna('') 避免 NaN 显示
• 📦 依赖版本更新：pymupdf4llm 0.0.17, mammoth 1.8.0 等
• 🚀 部署建议：添加资源配置、镜像大小估算
• 💡 用户引导：前端提示不支持扫描件

v1.0 (2026-01-20)

• 🆕 初始版本
• 🆕 PDF 处理：pymupdf4llm 替代 PyMuPDF + Nougat
• 🆕 Word 处理：mammoth
• 🆕 PPT 处理：python-pptx
• 🆕 Excel/CSV 处理：pandas
• 🆕 HTML 处理：beautifulsoup4 + markdownify
• 🆕 文献引用处理：bibtexparser + rispy
• 🆕 统一处理器架构

---

维护人： 技术架构师
相关文档：

• PKB 个人知识库
• Dify 替换为 pgvector 开发计划