🛠️ 技術調査 - Unstructured

from unstructured.partition.auto import partition

elements = partition(filename="document.pdf")

from unstructured.partition.pdf import partition_pdf
from unstructured.partition.strategies import PartitionStrategy

elements = partition_pdf(
    filename="document.pdf",
    strategy=PartitionStrategy.HI_RES
)

elements = partition(
    filename="document.pdf",
    chunking_strategy="by_title"
)

from unstructured.chunking.title import chunk_by_title

elements = partition(filename="document.pdf")
chunks = chunk_by_title(
    elements,
    max_characters=1024, # 1チャンクあたりの最大文字数
    new_after_n_chars=700, # この文字数を超えたら新しいチャンクを開始
)

from unstructured_client import UnstructuredClient
from unstructured_client.models import shared
from unstructured_client.models.errors import SDKError

client = UnstructuredClient(api_key_auth="YOUR_API_KEY")

with open("document.pdf", "rb") as f:
    files=shared.Files(
        content=f.read(),
        file_name="document.pdf",
    )

req = shared.PartitionParameters(files=files, strategy="hi_res")

try:
    res = client.general.partition(req)
    for element in res.elements:
        print(element)
except SDKError as e:
    print(e)

from unstructured.partition.pdf import partition_pdf
from unstructured.partition.strategies import PartitionStrategy

# FAST戦略：速度を重視し、基本的なテキストを抽出
elements_fast = partition_pdf(
    filename="document.pdf",
    strategy=PartitionStrategy.FAST
)

# HI_RES戦略：精度を重視し、レイアウト検出モデルを使用
elements_hi_res = partition_pdf(
    filename="document.pdf",
    strategy=PartitionStrategy.HI_RES
)

# AUTO戦略：文書の特性に応じて戦略を自動で選択
elements_auto = partition_pdf(
    filename="document.pdf",
    strategy=PartitionStrategy.AUTO
)

import multiprocessing
from concurrent.futures import ProcessPoolExecutor
from unstructured.partition.auto import partition

def process_document(filename):
    try:
        # 並列処理ではリソース消費の少ないFAST戦略が適している場合がある
        return partition(filename=filename, strategy="fast")
    except Exception as e:
        print(f"Error processing {filename}: {e}")
        return None

# CPUコア数に応じてワーカー数を設定
with ProcessPoolExecutor(max_workers=multiprocessing.cpu_count()) as executor:
    filenames = ["doc1.pdf", "doc2.docx", "doc3.html"]
    results = list(executor.map(process_document, filenames))

from unstructured.partition.auto import partition
from unstructured.partition.pdf import partition_pdf
from unstructured.partition.docx import partition_docx
from unstructured.file_utils.filetype import detect_filetype, FileType

def robust_partition(filepath):
    try:
        # まずは自動検出を試みる
        return partition(filepath)
    except Exception as e:
        print(f"Initial partition failed: {e}. Falling back to extension-based routing.")
        # 拡張子に基づいて手動でパーサーを選択
        try:
            if filepath.endswith('.pdf'):
                return partition_pdf(filepath, strategy="fast")
            elif filepath.endswith('.docx'):
                return partition_docx(filepath)
            else:
                raise ValueError(f"Unsupported file extension for fallback: {filepath}")
        except Exception as fallback_e:
            print(f"Fallback processing failed for {filepath}: {fallback_e}")
            return []

def process_with_fallback(filename):
    try:
        print("Attempting HI_RES strategy...")
        return partition_pdf(filename, strategy="hi_res")
    except Exception as hi_res_error:
        print(f"HI_RES failed: {hi_res_error}. Falling back to OCR_ONLY.")
        try:
            return partition_pdf(filename, strategy="ocr_only")
        except Exception as ocr_error:
            print(f"OCR_ONLY failed: {ocr_error}. Falling back to FAST.")
            return partition_pdf(filename, strategy="fast")

import pypdf
import tempfile
from pathlib import Path

def process_large_pdf_in_chunks(filename, chunk_size=20):
    reader = pypdf.PdfReader(filename)
    total_pages = len(reader.pages)
    all_elements = []

    for start_page in range(0, total_pages, chunk_size):
        end_page = min(start_page + chunk_size, total_pages)
        writer = pypdf.PdfWriter()
        
        with tempfile.NamedTemporaryFile(suffix=".pdf", delete=False) as tmp:
            chunk_filename = tmp.name
            
        try:
            for i in range(start_page, end_page):
                writer.add_page(reader.pages[i])
            writer.write(chunk_filename)
            
            print(f"Processing pages {start_page + 1}-{end_page}...")
            chunk_elements = partition_pdf(chunk_filename)
            # 元のページ番号をメタデータに反映
            for elem in chunk_elements:
                elem.metadata.page_number += start_page
            all_elements.extend(chunk_elements)
            
        except Exception as e:
            print(f"Error processing pages {start_page + 1}-{end_page}: {e}")
        finally:
            Path(chunk_filename).unlink()
            
    return all_elements

import Levenshtein

def evaluate_text_extraction(extracted_text, ground_truth_text):
    distance = Levenshtein.distance(extracted_text, ground_truth_text)
    max_len = max(len(extracted_text), len(ground_truth_text))
    # 1 - 正規化された編集距離 = 類似度
    similarity = 1 - (distance / max_len) if max_len > 0 else 1
    return {"levenshtein_similarity": similarity}

# 使用例
extracted = "This is a extracted text."
ground_truth = "This is an extracted text."
metrics = evaluate_text_extraction(extracted, ground_truth)
print(f"Text Similarity: {metrics['levenshtein_similarity']:.3f}")

from sklearn.metrics import classification_report

def evaluate_element_classification(predicted_elements, ground_truth_elements):
    # 要素のIDやテキストで対応付けをすると仮定
    y_true = [elem.category for elem in ground_truth_elements]
    y_pred = [elem.category for elem in predicted_elements]
    
    # 要素数が違う場合は、アライメント処理が必要だが、ここでは数が同じと仮定
    if len(y_true) != len(y_pred):
        print("Warning: Number of predicted and ground truth elements differ.")
        return None
        
    report = classification_report(y_true, y_pred, zero_division=0)
    return report

# 使用例
# ground_truth_elements = load_ground_truth() # 正解データの読み込み（要実装）
# predicted_elements = partition("document.pdf")
# report = evaluate_element_classification(predicted_elements, ground_truth_elements)
# print(report)

def calculate_iou(boxA, boxB):
    # box = (x1, y1, x2, y2)
    xA = max(boxA[0], boxB[0])
    yA = max(boxA[1], boxB[1])
    xB = min(boxA[2], boxB[2])
    yB = min(boxA[3], boxB[3])

    interArea = max(0, xB - xA) * max(0, yB - yA)
    boxAArea = (boxA[2] - boxA[0]) * (boxA[3] - boxA[1])
    boxBArea = (boxB[2] - boxB[0]) * (boxB[3] - boxB[1])
    
    iou = interArea / float(boxAArea + boxBArea - interArea)
    return iou

# 使用例
# pred_tables = [elem for elem in elements if elem.category == "Table"]
# gt_tables = load_ground_truth_tables() # 正解データの読み込み
# iou_score = calculate_iou(
#     pred_tables[0].metadata.coordinates.points, 
#     gt_tables[0].metadata.coordinates.points
# )
# print(f"Table Detection IoU: {iou_score:.3f}")