PDF 文档智能阅读助手 v1.2.1

触发条件

当用户提到 PDF阅读/读PDF/提取PDF/PDF总结/PDF转文本/PDF转Markdown/PDF摘要/PDF关键词/PDF问答/合同提取/论文摘要/识别扫描件/PDF表格/批量处理PDF/PDF导出 时自动加载。

环境要求

快速决策树

用户请求PDF操作
  ├── 纯文本PDF？→ [A] pymupdf提取 → LLM分析
  ├── 扫描件/图片PDF？→ [B] OCR识别（pytesseract优先，marker-pdf高配可选）
  ├── 表格数据？→ [C] pdfplumber提取 → 结构化输出
  ├── 加密PDF？→ [D] 完整解密流程 → 解锁后按A/B/C处理
  ├── 大文件(>50MB)？→ [E] 分页流式处理 → 增量分析
  ├── 导出Markdown？→ [F] 格式化导出 → 保留结构
  └── 批量处理？→ [G] 并发进度 → 汇总报告

路径安全校验（v1.2.1 新增）

所有涉及文件路径的操作用此函数校验，防止路径注入：

import os, shlex, re

def safe_path(filepath, base_dir=None):
    """
    校验文件路径安全性
    - 防止路径遍历攻击（../ 等）
    - 防止命令注入字符
    - 仅允许合法文件名
    """
    if not filepath or not isinstance(filepath, str):
        raise ValueError("无效的文件路径")

    # 过滤危险字符
    dangerous = [';', '|', '&', '$', '`', '(', ')', '{', '}', '<', '>', 'n', 'r']
    for ch in dangerous:
        if ch in filepath:
            raise ValueError(f"路径包含非法字符: {repr(ch)}")

    # 禁止空字节和通配符（防止 glob 注入）
    if 'x00' in filepath or '*' in filepath or '?' in filepath:
        raise ValueError("路径包含通配符或空字节")

    # 规范化路径，解析 ../
    normalized = os.path.normpath(os.path.abspath(filepath))

    # 如果指定了基准目录，确保不逃逸
    if base_dir:
        base_real = os.path.realpath(base_dir)
        file_real = os.path.realpath(normalized)
        if not file_real.startswith(base_real + os.sep) and file_real != base_real:
            raise ValueError(f"路径逃逸: {filepath} 超出允许范围 {base_dir}")

    # 验证文件扩展名（仅允许PDF相关）
    ext = os.path.splitext(normalized)[1].lower()
    if ext not in ['.pdf', '.PDF']:
        # 对于输出目录，允许无扩展名
        if ext and ext not in ['.md', '.json', '.txt']:
            raise ValueError(f"不支持的文件类型: {ext}")

    # 确保路径存在（输入文件）
    if not os.path.exists(normalized) and ext == '.pdf':
        raise FileNotFoundError(f"文件不存在: {normalized}")

    return normalized

所有工作流的文件路径入口统一调用 safe_path()。

工作流 A：纯文本PDF提取 + LLM分析

1. 安装依赖

python -c "import fitz; print('ok')" 2>/dev/null || pip install pymupdf -q

2. 提取文本

import fitz

def extract_text_pdf(filepath, max_pages=None):
    filepath = safe_path(filepath)  # 路径校验
    doc = fitz.open(filepath)
    text_parts = []
    pages = min(len(doc), max_pages or float('inf'))
    for i in range(int(pages)):
        text = doc[i].get_text()
        if text.strip():
            text_parts.append(f"[第{i+1}页]n{text}")
    doc.close()
    return "nn".join(text_parts)

text = extract_text_pdf("file.pdf")

3. LLM分析模板

全文总结：

请对以下PDF文档进行结构化总结：1.文档主题 2.核心要点(3-5个) 3.关键数据 4.结论
内容：{text[:8000]}

合同提取：

从合同PDF提取JSON：{"合同标题","签约方","合同金额","签署日期","有效期","违约责任","终止条件","保密条款","争议解决"}
内容：{text[:12000]}

论文摘要：

提取：1.标题 2.作者 3.研究问题 4.方法论 5.主要发现 6.创新点 7.局限性
内容：{text[:10000]}

发票提取（v1.2新增）：

从发票PDF提取JSON：{"发票号码","开票日期","销售方","购买方","金额合计","税额","价税合计","备注"}
内容：{text[:8000]}

简历解析（v1.2新增）：

从简历PDF提取JSON：{"姓名","性别","年龄","学历","毕业院校","专业","工作年限","当前职位","技能标签","工作经历":[],"教育经历":[]}
内容：{text[:10000]}

关键词搜索：

import re
def keyword_search(text, keywords, ctx=100):
    results = {}
    for kw in keywords:
        matches = list(re.finditer(re.escape(kw), text, re.I))
        results[kw] = [f"...{text[max(0,m.start()-ctx):m.end()+ctx]}..." for m in matches]
    return results

工作流 B：扫描件OCR（v1.2.1 优化）

OCR策略：pytesseract 首选（轻量沙箱可用），marker-pdf 可选（高配环境高质量）

B0: 判断是否需要OCR

import fitz

def needs_ocr(filepath):
    """快速判断PDF是否需要OCR"""
    filepath = safe_path(filepath)
    doc = fitz.open(filepath)
    # 检查前3页文本量
    text_len = sum(len(doc[i].get_text().strip()) for i in range(min(3, len(doc))))
    doc.close()
    return text_len < 150  # 前3页总文本<150字符判定为扫描件

B1: pytesseract OCR（推荐首选，轻量级）

# 安装（~100MB磁盘，沙箱可用）
pip install pytesseract pdf2image -q
# Windows: https://github.com/UB-Mannheim/tesseract/wiki
# Linux: apt install tesseract-ocr tesseract-ocr-chi-sim -y
# macOS: brew install tesseract tesseract-lang

def ocr_tesseract(filepath, lang='chi_sim+eng', dpi=200):
    """pytesseract OCR — 轻量级，沙箱环境可用"""
    import pytesseract
    from pdf2image import convert_from_path

    filepath = safe_path(filepath)

    # 检查 tesseract 是否可用
    try:
        pytesseract.get_tesseract_version()
    except Exception:
        raise RuntimeError(
            "tesseract-ocr 未安装。请安装：n"
            "  Linux: sudo apt install tesseract-ocr tesseract-ocr-chi-simn"
            "  macOS: brew install tesseract tesseract-langn"
            "  Windows: 下载 https://github.com/UB-Mannheim/tesseract/wiki"
        )

    images = convert_from_path(filepath, dpi=dpi)
    texts = []
    for i, img in enumerate(images):
        text = pytesseract.image_to_string(img, lang=lang)
        if text.strip():
            texts.append(f"[第{i+1}页]n{text.strip()}")
    return "nn".join(texts)

B2: marker-pdf OCR（可选，高精度但需高配环境）

⚠️ 环境要求: 8GB+ RAM，2GB+ 模型下载，首次使用需下载模型。仅推荐在服务器/高配机器上使用。

pip install marker-pdf -q

def ocr_marker(filepath, output_dir="./ocr_output"):
    """marker-pdf OCR — 高精度，需要8GB+ RAM 和 2GB 模型"""
    import subprocess, glob, os

    filepath = safe_path(filepath)
    output_dir = safe_path(output_dir, base_dir=os.getcwd())

    # 使用 shlex.quote 安全转义路径参数
    cmd = [
        "python", "-m", "marker_single",
        filepath,
        output_dir,
        "--batch_multiplier", "2"
    ]

    try:
        result = subprocess.run(
            cmd,
            capture_output=True, text=True, timeout=600
        )
    except subprocess.TimeoutExpired:
        return None, "timeout"

    if result.returncode != 0:
        return None, f"marker error: {result.stderr[:200]}"

    md_files = glob.glob(f"{output_dir}/**/*.md", recursive=True)
    if md_files:
        with open(md_files[0], 'r', encoding='utf-8') as f:
            return f.read(), "marker-pdf"
    return None, "no output"

B3: 智能OCR（自动选择最佳引擎）

def smart_ocr(filepath, output_dir="./ocr_output"):
    """
    自动选择OCR引擎：
    1. 优先 pytesseract（轻量通用）
    2. 尝试 marker-pdf（高精度，自动检测环境是否支持）
    3. 降级 pymupdf 直接提取
    """
    import os

    filepath = safe_path(filepath)

    # 1. 先试 pytesseract
    try:
        text = ocr_tesseract(filepath)
        if text and len(text) > 100:
            return text, "pytesseract"
    except Exception as e:
        pass  # tesseract 不可用，继续尝试其他方案

    # 2. 尝试 marker-pdf（检查环境）
    try:
        import marker
        text, engine = ocr_marker(filepath, output_dir)
        if text:
            return text, engine
    except (ImportError, MemoryError, OSError):
        pass  # marker 不可用或内存不足

    # 3. 最终降级：pymupdf 直接提取
    import fitz
    doc = fitz.open(filepath)
    text = "nn".join(
        f"[第{i+1}页]n{doc[i].get_text()}"
        for i in range(len(doc))
    )
    doc.close()
    return text, "pymupdf_fallback"

工作流 C：表格提取

pip install pdfplumber -q

import pdfplumber, json

def extract_tables(filepath, fmt="json"):
    filepath = safe_path(filepath)
    tables = []
    with pdfplumber.open(filepath) as pdf:
        for i, page in enumerate(pdf.pages):
            for j, table in enumerate(page.extract_tables()):
                cleaned = [[cell.strip() if cell else "" for cell in row] for row in table]
                tables.append({"page":i+1, "table":j+1, "headers":cleaned[0], "rows":cleaned[1:], "row_count":len(cleaned)-1})
    return json.dumps(tables, ensure_ascii=False, indent=2)

工作流 D：加密PDF（v1.2.1 完整版）

import fitz

def handle_encrypted(filepath, password=None):
    """
    处理加密PDF — 完整实现
    返回: (fitz.Document, status_string)
    """
    filepath = safe_path(filepath)

    # 尝试打开
    try:
        doc = fitz.open(filepath)
    except Exception as e:
        raise FileNotFoundError(f"无法打开PDF: {filepath} — {e}")

    # 不需要密码
    if not doc.needs_pass:
        return doc, "no_password"

    # 用户提供了密码
    if password:
        if doc.authenticate(password):
            return doc, "decrypted"
        doc.close()
        raise PermissionError(f"密码错误，无法解密: {filepath}")

    # 尝试常见密码
    common_passwords = ["123456", "password", "000000", "1234", "admin"]

    # 从文件名猜测密码（如 "合同_123456.pdf"）
    import re, os
    basename = os.path.basename(filepath)
    pwd_match = re.search(r'[_-](d{4,8})', basename)
    if pwd_match:
        common_passwords.insert(0, pwd_match.group(1))

    for pwd in common_passwords:
        try:
            doc2 = fitz.open(filepath)
            if doc2.authenticate(pwd):
                return doc2, f"decrypted_with_{pwd}"
            doc2.close()
        except Exception:
            continue

    # 所有密码都失败
    doc.close()
    raise PermissionError(
        f"PDF已加密，已尝试 {len(common_passwords)} 个常见密码均失败。n"
        f"文件: {os.path.basename(filepath)}n"
        f"请提供正确密码后重试。n"
        f"提示: 可以尝试从文件名中提取密码，或联系文件提供方。"
    )

def safe_open_pdf(filepath, password=None):
    """
    安全打开PDF（自动处理加密）
    用法:
      doc, status = safe_open_pdf("encrypted.pdf", password="mypassword")
      doc, status = safe_open_pdf("normal.pdf")  # 自动检测无需密码
    """
    filepath = safe_path(filepath)

    try:
        doc = fitz.open(filepath)
    except Exception as e:
        raise FileNotFoundError(f"文件无法打开: {e}")

    if not doc.needs_pass:
        return doc, "ok_no_password"

    # 需要密码 → 走完整流程
    doc.close()
    return handle_encrypted(filepath, password=password)

def extract_encrypted_pdf(filepath, password=None, max_pages=None):
    """
    一键处理加密PDF：自动尝试解密 → 提取文本
    密码优先级：用户提供 > 常见密码列表 > 文件名猜测 > 报错请求用户输入
    """
    doc, status = safe_open_pdf(filepath, password=password)

    text_parts = []
    pages = min(len(doc), max_pages or float('inf'))
    for i in range(int(pages)):
        text = doc[i].get_text()
        if text.strip():
            text_parts.append(f"[第{i+1}页]n{text}")
    doc.close()

    info = f"[解密状态: {status}]"
    return info + "nn" + "nn".join(text_parts)

工作流 E：大文件分页

def process_large(filepath, chunk_size=20):
    filepath = safe_path(filepath)
    import fitz
    doc = fitz.open(filepath)
    total = len(doc)
    for start in range(0, total, chunk_size):
        end = min(start + chunk_size, total)
        chunk = "".join(f"n[第{i+1}页]n{doc[i].get_text()}" for i in range(start, end))
        yield chunk, start, end, total
    doc.close()

工作流 F：Markdown 导出

F1: 纯文本 → Markdown

def export_markdown(filepath, output_path=None):
    """将PDF内容导出为结构化Markdown文件"""
    import fitz, os, re
    from datetime import datetime

    filepath = safe_path(filepath)
    doc = fitz.open(filepath)
    if output_path is None:
        output_path = filepath.rsplit('.', 1)[0] + '.md'

    md_lines = []
    filename = os.path.basename(filepath)
    filesize = os.path.getsize(filepath) / (1024*1024)

    md_lines.append(f"# {filename}n")
    md_lines.append(f"> 导出时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
    md_lines.append(f"> 文件大小: {filesize:.1f} MB | 页数: {len(doc)}n")
    md_lines.append("---n")

    if len(doc) > 3:
        md_lines.append("## 目录n")
        for i in range(len(doc)):
            text = doc[i].get_text().strip()
            title = text.split('n')[0][:60] if text else f"第{i+1}页（空白）"
            md_lines.append(f"- [第{i+1}页](#page-{i+1}) — {title}")
        md_lines.append("n---n")

    for i in range(len(doc)):
        page = doc[i]
        text = page.get_text().strip()
        if not text:
            md_lines.append(f"## 第{i+1}页 {{#page-{i+1}}}nn*（本页无文本内容）*n")
            continue

        md_lines.append(f"## 第{i+1}页 {{#page-{i+1}}}n")
        for line in text.split('n'):
            line = line.strip()
            if not line:
                md_lines.append("")
                continue
            if line.isupper() and len(line) < 80:
                md_lines.append(f"### {line.title()}")
            elif re.match(r'^[一二三四五六七八九十]+[、．.]', line):
                md_lines.append(f"### {line}")
            elif re.match(r'^[d]+[、．.)]', line):
                md_lines.append(f"- {line}")
            else:
                md_lines.append(line)
        md_lines.append("")

    md_lines.append("---n")
    md_lines.append(f"*由 PDF Reader v1.2.1 自动生成*")
    doc.close()

    with open(output_path, 'w', encoding='utf-8') as f:
        f.write('n'.join(md_lines))
    return output_path, len(doc)

F2: 表格 → Markdown 表格

def table_to_markdown(filepath, output_path=None):
    filepath = safe_path(filepath)
    import pdfplumber, os
    if output_path is None:
        output_path = filepath.rsplit('.', 1)[0] + '_tables.md'

    md_lines = [f"# 表格提取结果n", f"> 源文件: {os.path.basename(filepath)}n"]
    with pdfplumber.open(filepath) as pdf:
        table_count = 0
        for i, page in enumerate(pdf.pages):
            tables = page.extract_tables()
            if not tables:
                continue
            for j, table in enumerate(tables):
                table_count += 1
                md_lines.append(f"## 表格 {table_count}（第{i+1}页）n")
                if table:
                    cleaned = [[cell.strip() if cell else "" for cell in row] for row in table]
                    headers = cleaned[0]
                    md_lines.append("| " + " | ".join(headers) + " |")
                    md_lines.append("| " + " | ".join(["---"] * len(headers)) + " |")
                    for row in cleaned[1:]:
                        md_lines.append("| " + " | ".join(row) + " |")
                    md_lines.append("")

    md_lines.append(f"n*共提取 {table_count} 个表格*")
    with open(output_path, 'w', encoding='utf-8') as f:
        f.write('n'.join(md_lines))
    return output_path, table_count

F3: 一键导出

def pdf_export(filepath, fmt="auto", output_path=None):
    filepath = safe_path(filepath)
    import os, fitz, json as jmod
    base = filepath.rsplit('.', 1)[0]

    if fmt == "auto":
        import pdfplumber
        with pdfplumber.open(filepath) as pdf:
            has_table = any(page.extract_tables() for page in pdf.pages)
        fmt = "table" if has_table else "md"

    if fmt == "md":
        return export_markdown(filepath, output_path or f"{base}.md")
    elif fmt == "table":
        return table_to_markdown(filepath, output_path or f"{base}_tables.md")
    elif fmt == "json":
        doc = fitz.open(filepath)
        pages = [{"page": i+1, "text": doc[i].get_text().strip(), "char_count": len(doc[i].get_text().strip())} for i in range(len(doc))]
        doc.close()
        out = output_path or f"{base}.json"
        with open(out, 'w', encoding='utf-8') as f:
            jmod.dump({"file": os.path.basename(filepath), "total_pages": len(pages), "pages": pages}, f, ensure_ascii=False, indent=2)
        return out, len(pages)

工作流 G：批量处理

import os, glob, time, sys
from concurrent.futures import ThreadPoolExecutor, as_completed

def batch_process(folder, pattern="*.pdf", mode="extract", max_workers=2, show_progress=True):
    folder = safe_path(folder)
    files = sorted(glob.glob(os.path.join(folder, pattern)))
    total = len(files)
    if total == 0:
        return {"error": f"未找到匹配 {pattern} 的文件"}

    results = {"total": total, "success": 0, "failed": 0, "files": []}
    completed = 0
    start_time = time.time()

    def process_one(fp):
        fp = safe_path(fp)
        try:
            if mode == "extract":
                text = extract_text_pdf(fp)
                return fp, {"status": "ok", "chars": len(text), "preview": text[:150]}
            elif mode == "export":
                out, pages = export_markdown(fp)
                return fp, {"status": "ok", "output": out, "pages": pages}
            elif mode == "ocr":
                text, engine = smart_ocr(fp)
                return fp, {"status": "ok", "engine": engine, "chars": len(text) if text else 0}
            elif mode == "summary":
                text = extract_text_pdf(fp, max_pages=5)
                return fp, {"status": "ok", "summary_ready": len(text)}
        except Exception as e:
            return fp, {"status": "error", "error": str(e)}

    if show_progress:
        print(f"PDF批量处理: {total}个文件 ({mode}模式)")

    with ThreadPoolExecutor(max_workers=max_workers) as ex:
        futures = {ex.submit(process_one, f): f for f in files}
        for future in as_completed(futures):
            fp = futures[future]
            try:
                fp, info = future.result()
                results["files"].append({"path": fp, **info})
                if info["status"] == "ok":
                    results["success"] += 1
                else:
                    results["failed"] += 1
            except Exception as e:
                results["files"].append({"path": fp, "status": "error", "error": str(e)})
                results["failed"] += 1

            completed += 1
            if show_progress:
                elapsed = time.time() - start_time
                eta = (elapsed / completed) * (total - completed) if completed > 0 else 0
                pct = completed / total * 100
                bar = "#" * int(pct // 5) + "-" * (20 - int(pct // 5))
                print(f"r[{bar}] {completed}/{total} ({pct:.0f}%) | OK:{results['success']} FAIL:{results['failed']} | ETA:{eta:.0f}s", end="")

    elapsed = time.time() - start_time
    if show_progress:
        print(f"n完成! {elapsed:.1f}s | OK {results['success']}, FAIL {results['failed']}")
    results["elapsed_seconds"] = round(elapsed, 1)
    return results

一键总结

def pdf_quick_summary(filepath, max_chars=8000):
    import fitz, os
    filepath = safe_path(filepath)
    file_size_mb = os.path.getsize(filepath) / (1024*1024)
    doc = fitz.open(filepath)

    # 处理加密
    if doc.needs_pass:
        doc, status = safe_open_pdf(filepath)
        note = f"（解密: {status}）"
    else:
        note = ""

    is_scanned = len(doc[0].get_text().strip()) < 50
    if is_scanned:
        text, engine = smart_ocr(filepath)
        note += f"（OCR: {engine}）"
    else:
        pages = min(len(doc), max_chars//1000) if file_size_mb > 50 else len(doc)
        text = extract_text_pdf(filepath, max_pages=pages)
    doc.close()

    if len(text) > max_chars:
        text = text[:max_chars] + "n[已截断]"

    return f"""请分析PDF{note}：n文件大小:{file_size_mb:.0f}MBn扫描件:{is_scanned}n---n{text}"""

依赖速查

错误速查