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'發票\s*#?\s*:?\s*(\w+[-\w]*)', full_text, re.I)
    if inv_match:
        invoice['invoice_number'] = inv_match.group(1)
    
    date_match = re.search(r'日期\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'總計\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