终结 PDF 解析噩梦:深度体验高性能 Python 文档智能库 Kreuzberg

graph TD    A[收到一份待处理文档] --> B{传统解析逻辑: 判断文件类型}    B -- .pdf后缀 --> C[调用 PyPDF2/pdfplumber]    B -- .docx后缀 --> D[调用 python-docx]    B -- 图片格式 --> E[调用 OCR 引擎]    C --> F[各种复杂的错误捕获与重试]    D --> F    E --> F    F --> G[拼凑成最终文本]    A --> H{Kreuzberg 解析逻辑}    H --> I[直接调用 extract_file 接口]    I --> J[底层 Rust 引擎自动识别并处理]    J --> K[返回结构化文本结果]    style B stroke-dasharray: 5 5    style F stroke-dasharray: 5 5

pip install kreuzberg

import asynciofrom pathlib import Pathfrom kreuzberg import extract_fileasync def process_local_documents():    # 假设我们有两个不同格式的文件    pdf_path = Path("sample_report.pdf")    word_path = Path("company_policy.docx")    print("开始解析文档...")    # 异步并发执行，互不阻塞    try:        pdf_result, word_result = await asyncio.gather(            extract_file(pdf_path),            extract_file(word_path)        )        # 返回的结果不仅有文本，还包含了 mime 类型等元数据        print(f"PDF 提取完成，MIME类型: {pdf_result.mime_type}")        print(f"PDF 文本前50个字符: {pdf_result.content[:50]}...")        print(f"\nWord 提取完成，MIME类型: {word_result.mime_type}")        print(f"Word 文本前50个字符: {word_result.content[:50]}...")    except Exception as e:        print(f"解析过程中翻车了: {e}")if __name__ == "__main__":    asyncio.run(process_local_documents())

import asynciofrom kreuzberg import extract_bytesasync def process_memory_data(file_bytes: bytes, file_name: str):    """    模拟处理来自 Web 框架的文件上传    """    print(f"接收到文件: {file_name}，大小: {len(file_bytes)} bytes")    # 直接丢进引擎，不需要落盘    result = await extract_bytes(file_bytes)    # 甚至我们可以根据它自动识别的类型做点业务逻辑    if "pdf" in result.mime_type:        print("这是一个 PDF，准备灌入向量数据库...")    elif "image" in result.mime_type:        print("这是一张图片，文本内容如下：")    print(result.content.strip())    return result.content# 模拟一段读取async def main():    with open("sample_report.pdf", "rb") as f:        raw_bytes = f.read()    await process_memory_data(raw_bytes, "upload_temp.pdf")if __name__ == "__main__":    asyncio.run(main())

import asynciofrom kreuzberg import extract_stringasync def clean_html_content():    # 模拟一段从网页抓取下来的脏数据    dirty_html = """    <html>        <body>            <header>导航栏忽略</header>            <div class="main-content">                <h1>这是我们真正需要的新闻标题</h1>                <p>今天发布了一个新版本的 Python 库。</p>                <script>console.log("这堆 JS 代码我们绝对不想要");</script>            </div>        </body>    </html>    """    # 直接提取    result = await extract_string(dirty_html)    print("清洗后的纯净文本：")    print(result.content)if __name__ == "__main__":    asyncio.run(clean_html_content())