data-extractor

数据提取技能

概述

本技能能够使用 unstructured 从任何文档格式中提取结构化数据——这是一个统一的库，用于处理 PDF、Word 文档、电子邮件、HTML 等。无论输入格式如何，都能获得一致的结构化输出。

使用方法

1. 提供要处理的文档
2. 可选地指定提取选项
3. 我将提取带有元数据的结构化元素

示例提示：

• "提取此 PDF 中的所有文本和表格"
• "解析此电子邮件，获取正文、附件和元数据"
• "将此 HTML 页面转换为结构化元素"
• "从这些混合格式的文档中提取数据"

领域知识

unstructured 基础

from unstructured.partition.auto import partition

# 自动检测并处理任何文档
elements = partition("document.pdf")

# 访问提取的元素
for element in elements:
    print(f"类型: {type(element).__name__}")
    print(f"文本: {element.text}")
    print(f"元数据: {element.metadata}")

支持的格式

格式	函数	备注
PDF	partition_pdf	原生 + 扫描
Word	partition_docx	完整结构
PowerPoint	partition_pptx	幻灯片和备注
Excel	partition_xlsx	工作表和表格
电子邮件	partition_email	正文和附件
HTML	partition_html	保留标签
Markdown	partition_md	保留结构
纯文本	partition_text	基本解析
图片	partition_image	OCR 提取

元素类型

from unstructured.documents.elements import (
    Title,
    NarrativeText,
    Text,
    ListItem,
    Table,
    Image,
    Header,
    Footer,
    PageBreak,
    Address,
    EmailAddress,
)

# 元素具有一致的结构
element.text           # 原始文本内容
element.metadata       # 丰富的元数据
element.category       # 元素类型
element.id            # 唯一标识符

自动分区

from unstructured.partition.auto import partition

# 处理任何文件类型
elements = partition(
    filename="document.pdf",
    strategy="auto",          # 或 "fast", "hi_res", "ocr_only"
    include_metadata=True,
    include_page_breaks=True,
)

# 按类型过滤
titles = [e for e in elements if isinstance(e, Title)]
tables = [e for e in elements if isinstance(e, Table)]

特定格式分区

# PDF 带选项
from unstructured.partition.pdf import partition_pdf

elements = partition_pdf(
    filename="document.pdf",
    strategy="hi_res",              # 高质量提取
    infer_table_structure=True,     # 检测表格
    include_page_breaks=True,
    languages=["en"],               # OCR 语言
)

# Word 文档
from unstructured.partition.docx import partition_docx

elements = partition_docx(
    filename="document.docx",
    include_metadata=True,
)

# HTML
from unstructured.partition.html import partition_html

elements = partition_html(
    filename="page.html",
    include_metadata=True,
)

处理表格

from unstructured.partition.auto import partition

elements = partition("report.pdf", infer_table_structure=True)

# 提取表格
for element in elements:
    if element.category == "Table":
        print("找到表格：")
        print(element.text)
        
        # 访问结构化表格数据
        if hasattr(element, 'metadata') and element.metadata.text_as_html:
            print("HTML:", element.metadata.text_as_html)

元数据访问

from unstructured.partition.auto import partition

elements = partition("document.pdf")

for element in elements:
    meta = element.metadata
    
    # 常见元数据字段
    print(f"页码: {meta.page_number}")
    print(f"文件名: {meta.filename}")
    print(f"文件类型: {meta.filetype}")
    print(f"坐标: {meta.coordinates}")
    print(f"语言: {meta.languages}")

为 AI/RAG 分块

from unstructured.partition.auto import partition
from unstructured.chunking.title import chunk_by_title
from unstructured.chunking.basic import chunk_elements

# 分区文档
elements = partition("document.pdf")

# 按标题分块（语义块）
chunks = chunk_by_title(
    elements,
    max_characters=1000,
    combine_text_under_n_chars=200,
)

# 或基本分块
chunks = chunk_elements(
    elements,
    max_characters=500,
    overlap=50,
)

for chunk in chunks:
    print(f"块 ({len(chunk.text)} 字符):")
    print(chunk.text[:100] + "...")

批量处理

from unstructured.partition.auto import partition
from pathlib import Path
from concurrent.futures import ThreadPoolExecutor

def process_document(file_path):
    """处理单个文档。"""
    try:
        elements = partition(str(file_path))
        return {
            'file': str(file_path),
            'status': 'success',
            'elements': len(elements),
            'text': '\n\n'.join([e.text for e in elements])
        }
    except Exception as e:
        return {
            'file': str(file_path),
            'status': 'error',
            'error': str(e)
        }

def batch_process(input_dir, max_workers=4):
    """处理目录中的所有文档。"""
    input_path = Path(input_dir)
    files = list(input_path.glob('*'))
    
    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        results = list(executor.map(process_document, files))
    
    return results

导出格式

from unstructured.partition.auto import partition
from unstructured.staging.base import elements_to_json, elements_to_dicts

elements = partition("document.pdf")

# 转为 JSON 字符串
json_str = elements_to_json(elements)

# 转为字典列表
dicts = elements_to_dicts(elements)

# 转为 DataFrame
import pandas as pd
df = pd.DataFrame(dicts)

最佳实践

1. 明智选择策略："fast" 追求速度，"hi_res" 追求精度
2. 启用表格检测：适用于包含表格的文档
3. 指定语言：提高非英文文档的 OCR 效果
4. 为 RAG 分块：使用语义分块用于 AI 应用
5. 处理错误：某些格式可能优雅地失败

常见模式

文档转 JSON

def document_to_json(file_path, output_path=None):
    """将文档转换为结构化 JSON。"""
    from unstructured.partition.auto import partition
    from unstructured.staging.base import elements_to_json
    import json
    
    elements = partition(file_path)
    
    # 创建结构化输出
    output = {
        'source': file_path,
        'elements': []
    }
    
    for element in elements:
        output['elements'].append({
            'type': type(element).__name__,
            'text': element.text,
            'metadata': {
                'page': element.metadata.page_number,
                'coordinates': element.metadata.coordinates.to_dict() if element.metadata.coordinates else None
            }
        })
    
    if output_path:
        with open(output_path, 'w') as f:
            json.dump(output, f, indent=2)
    
    return output

电子邮件解析器

from unstructured.partition.email import partition_email

def parse_email(email_path):
    """从电子邮件中提取结构化数据。"""
    
    elements = partition_email(email_path)
    
    email_data = {
        'subject': None,
        'from': None,
        'to': [],
        'date': None,
        'body': [],
        'attachments': []
    }
    
    for element in elements:
        meta = element.metadata
        
        # 从元数据中提取头部信息
        if meta.subject:
            email_data['subject'] = meta.subject
        if meta.sent_from:
            email_data['from'] = meta.sent_from
        if meta.sent_to:
            email_data['to'] = meta.sent_to
        
        # 正文内容
        email_data['body'].append({
            'type': type(element).__name__,
            'text': element.text
        })
    
    return email_data

示例

示例 1：研究论文提取

from unstructured.partition.pdf import partition_pdf
from unstructured.chunking.title import chunk_by_title

def extract_paper(pdf_path):
    """从研究论文中提取结构化数据。"""
    
    elements = partition_pdf(
        filename=pdf_path,
        strategy="hi_res",
        infer_table_structure=True,
        include_page_breaks=True
    )
    
    paper = {
        'title': None,
        'abstract': None,
        'sections': [],
        'tables': [],
        'references': []
    }
    
    # 查找标题（通常是第一个 Title 元素）
    for element in elements:
        if element.category == "Title" and not paper['title']:
            paper['title'] = element.text
            break
    
    # 提取表格
    for element in elements:
        if element.category == "Table":
            paper['tables'].append({
                'page': element.metadata.page_number,
                'content': element.text,
                'html': element.metadata.text_as_html if hasattr(element.metadata, 'text_as_html') else None
            })
    
    # 按标题分块
    chunks = chunk_by_title(elements, max_characters=2000)
    
    current_section = None
    for chunk in chunks:
        if chunk.category == "Title":
            paper['sections'].append({
                'title': chunk.text,
                'content': ''
            })
        elif paper['sections']:
            paper['sections'][-1]['content'] += chunk.text + '\n'
    
    return paper

paper = extract_paper('research_paper.pdf')
print(f"标题: {paper['title']}")
print(f"表格数: {len(paper['tables'])}")
print(f"章节数: {len(paper['sections'])}")

示例 2：发票数据提取

from unstructured.partition.auto import partition
import re

def extract_invoice_data(file_path):
    """从发票中提取关键数据。"""
    
    elements = partition(file_path, strategy="hi_res")
    
    # 合并所有文本
    full_text = '\n'.join([e.text for e in elements])
    
    invoice = {
        'invoice_number': None,
        'date': None,
        'total': None,
        'vendor': None,
        'line_items': [],
        'tables': []
    }
    
    # 提取模式
    inv_match = re.search(r'Invoice\s*#?\s*:?\s*(\w+[-\w]*)', full_text, re.I)
    if inv_match:
        invoice['invoice_number'] = inv_match.group(1)
    
    date_match = re.search(r'Date\s*:?\s*(\d{1,2}[-/]\d{1,2}[-/]\d{2,4})', full_text, re.I)
    if date_match:
        invoice['date'] = date_match.group(1)
    
    total_match = re.search(r'Total\s*:?\s*\$?([\d,]+\.?\d*)', full_text, re.I)
    if total_match:
        invoice['total'] = float(total_match.group(1).replace(',', ''))
    
    # 提取表格
    for element in elements:
        if element.category == "Table":
            invoice['tables'].append(element.text)
    
    return invoice

invoice = extract_invoice_data('invoice.pdf')
print(f"发票号: {invoice['invoice_number']}")
print(f"总计: ${invoice['total']}")

示例 3：文档语料库构建器

from unstructured.partition.auto import partition
from unstructured.chunking.title import chunk_by_title
from pathlib import Path
import json

def build_corpus(input_dir, output_path):
    """从文档集合构建可搜索的语料库。"""
    
    input_path = Path(input_dir)
    corpus = []
    
    # 支持多种格式
    patterns = ['*.pdf', '*.docx', '*.html', '*.txt', '*.md']
    files = []
    for pattern in patterns:
        files.extend(input_path.glob(pattern))
    
    for file in files:
        print(f"正在处理: {file.name}")
        
        try:
            elements = partition(str(file))
            chunks = chunk_by_title(elements, max_characters=1000)
            
            for i, chunk in enumerate(chunks):
                corpus.append({
                    'id': f"{file.stem}_{i}",
                    'source': str(file),
                    'type': type(chunk).__name__,
                    'text': chunk.text,
                    'page': chunk.metadata.page_number if chunk.metadata.page_number else None
                })
        
        except Exception as e:
            print(f"  错误: {e}")
    
    # 保存语料库
    with open(output_path, 'w') as f:
        json.dump(corpus, f, indent=2)
    
    print(f"语料库构建完成: 来自 {len(files)} 个文件的 {len(corpus)} 个块")
    return corpus

corpus = build_corpus('./documents', 'corpus.json')

局限性

• 复杂布局可能需要人工审核
• OCR 质量取决于图像质量
• 大文件可能需要分块
• 某些专有格式不受支持
• 云处理有 API 速率限制

安装

# 基本安装
pip install unstructured

# 包含所有依赖
pip install "unstructured[all-docs]"

# 用于 PDF 处理
pip install "unstructured[pdf]"

# 用于特定格式
pip install "unstructured[docx,pptx,xlsx]"

资源

• unstructured GitHub
• 文档
• Unstructured API