DeepSeek-OCR によるPDF・画像OCR（ソースコードと説明と利用ガイド）

REM Python をシステム領域にインストール
winget install --scope machine --id Python.Python.3.12 -e --silent
REM Python のパス設定
set "PYTHON_PATH=C:\Program Files\Python312"
set "PYTHON_SCRIPTS_PATH=C:\Program Files\Python312\Scripts"
echo "%PATH%" | find /i "%PYTHON_PATH%" >nul
if errorlevel 1 setx PATH "%PATH%;%PYTHON_PATH%" /M >nul
echo "%PATH%" | find /i "%PYTHON_SCRIPTS_PATH%" >nul
if errorlevel 1 setx PATH "%PATH%;%PYTHON_SCRIPTS_PATH%" /M >nul

winget install --scope machine Codeium.Windsurf -e --silent

"""
DeepSeek-OCR による画像・PDF OCRアプリケーション
画像ファイルまたはPDFを選択してOCR処理を実行します

事前インストールが必要なパッケージ (Windows):
pip install transformers==4.46.3 tokenizers==0.20.3 einops addict easydict gitpython PyMuPDF
pip install https://github.com/mjun0812/flash-attention-prebuild-wheels/releases/download/v0.4.15/flash_attn-2.8.3+cu126torch2.9-cp312-cp312-win_amd64.whl

flash-attnはプリビルドwhlをダウンロードしてインストール（環境に合わせて選択）:
https://github.com/mjun0812/flash-attention-prebuild-wheels/releases
"""
import warnings
warnings.filterwarnings('ignore')

import os
os.environ['TF_CPP_MIN_LOG_LEVEL'] = '3'
os.environ['TF_ENABLE_ONEDNN_OPTS'] = '0'
os.environ['TRANSFORMERS_VERBOSITY'] = 'error'

import sys
import logging

logging.getLogger('transformers').setLevel(logging.CRITICAL)
logging.getLogger('transformers.generation').setLevel(logging.CRITICAL)
logging.getLogger('transformers.modeling_utils').setLevel(logging.CRITICAL)

import tkinter as tk
from tkinter import filedialog
from transformers import AutoModel, AutoTokenizer
import torch
from pathlib import Path
import fitz
import tempfile
import shutil
from io import StringIO

SUPPORTED_IMAGE_FORMATS = ['.png', '.jpg', '.jpeg', '.bmp', '.tiff']
PDF_ZOOM_FACTOR = 150 / 72

MODE_CONFIGS = {
    "1": {"base_size": 512, "image_size": 512, "crop_mode": False, "name": "Tiny", "tokens": "64"},
    "2": {"base_size": 640, "image_size": 640, "crop_mode": False, "name": "Small", "tokens": "100"},
    "3": {"base_size": 1024, "image_size": 1024, "crop_mode": False, "name": "Base", "tokens": "256"},
    "4": {"base_size": 1280, "image_size": 1280, "crop_mode": False, "name": "Large", "tokens": "400"},
    "5": {"base_size": 1024, "image_size": 640, "crop_mode": True, "name": "Gundam", "tokens": "n×100+256 (動的)"}
}

class TeeOutput:
    """標準出力を複数のストリームに分配"""
    def __init__(self, *streams):
        self.streams = streams
    def write(self, data):
        for stream in self.streams:
            stream.write(data)
    def flush(self):
        for stream in self.streams:
            stream.flush()

class SuppressOutput:
    """出力を抑制"""
    def write(self, data):
        pass
    def flush(self):
        pass

def print_separator():
    print("=" * 60)

def run_ocr_inference(image_path, config, model, tokenizer, prompt):
    old_stdout = sys.stdout
    old_stderr = sys.stderr
    captured_out = StringIO()
    sys.stdout = TeeOutput(old_stdout, captured_out)
    sys.stderr = SuppressOutput()

    temp_output_dir = tempfile.mkdtemp()

    try:
        model.infer(
            tokenizer,
            prompt=prompt,
            image_file=image_path,
            output_path=temp_output_dir,
            base_size=config['base_size'],
            image_size=config['image_size'],
            crop_mode=config['crop_mode'],
            save_results=True,
            test_compress=False
        )
    finally:
        sys.stdout = old_stdout
        sys.stderr = old_stderr

    output = captured_out.getvalue()

    filtered_lines = []
    for line in output.split('\n'):
        if line.startswith('BASE:') or line.startswith('PATCHES:') or line.startswith('====='):
            continue
        if line.strip():
            filtered_lines.append(line)

    result_path = os.path.join(temp_output_dir, "result.md")
    result_text = None
    if os.path.exists(result_path):
        with open(result_path, 'r', encoding='utf-8') as f:
            result_text = f.read()

    shutil.rmtree(temp_output_dir, ignore_errors=True)

    return result_text if result_text else '\n'.join(filtered_lines)

def execute_ocr(image_path, filename_base, config, model, tokenizer, original_path):
    print(f"  処理中（Free OCRモード - 高精度）...")
    prompt_text = "<image>\nFree OCR."

    result_text = run_ocr_inference(image_path, config, model, tokenizer, prompt_text)

    return {
        'filename': filename_base,
        'text': result_text,
        'original': original_path
    }

def process_file(file_path, config, model, tokenizer):
    file_ext = Path(file_path).suffix.lower()

    if file_ext in SUPPORTED_IMAGE_FORMATS:
        result_dict = execute_ocr(
            file_path,
            Path(file_path).name,
            config,
            model,
            tokenizer,
            file_path
        )
        return [result_dict]
    elif file_ext == '.pdf':
        doc = fitz.open(file_path)
        total_pages = len(doc)
        print(f"  ページ数: {total_pages}")
        results = []

        for page_num in range(total_pages):
            page = doc[page_num]
            mat = fitz.Matrix(PDF_ZOOM_FACTOR, PDF_ZOOM_FACTOR)
            pix = page.get_pixmap(matrix=mat)

            tmp_path = None
            try:
                with tempfile.NamedTemporaryFile(suffix='.png', delete=False) as tmp_file:
                    tmp_path = tmp_file.name
                pix.save(tmp_path)

                filename_base = f"{Path(file_path).stem}_page{page_num + 1}.png"
                result_dict = execute_ocr(
                    tmp_path,
                    filename_base,
                    config,
                    model,
                    tokenizer,
                    file_path
                )
                result_dict['filename'] = f"{Path(file_path).name} Page {page_num + 1}/{total_pages}"

                results.append(result_dict)
            finally:
                if tmp_path and os.path.exists(tmp_path):
                    os.unlink(tmp_path)

        doc.close()
        return results
    else:
        print(f"未対応形式: {file_path}")
        return []

def select_files_dialog():
    root = tk.Tk()
    root.withdraw()
    root.update()
    user_files = filedialog.askopenfilenames(
        title="ファイルを選択してください（複数選択可）",
        filetypes=[
            ("対応ファイル", "*.pdf *.png *.jpg *.jpeg *.bmp *.tiff"),
            ("すべてのファイル", "*.*")
        ]
    )
    root.destroy()
    return user_files

print_separator()
print("DeepSeek-OCR: 画像・PDFからMarkdown変換")
print("\n【モデル選択】")
print("1. Tiny (512×512, 64トークン) - 最速")
print("2. Small (640×640, 100トークン) - 高速")
print("3. Base (1024×1024, 256トークン) - 推奨・バランス型")
print("4. Large (1280×1280, 400トークン) - 高精度")
print("5. Gundam (動的解像度) - 最高精度")

choice = input("\nモード選択 (1-5, デフォルト=3): ").strip() or "3"
config = MODE_CONFIGS.get(choice, MODE_CONFIGS["3"])
print(f"\n選択: {config['name']}モード")

print("\n【モデルについて】")
print("""DeepSeek-OCRは2つのコンポーネントで構成:
  1. DeepEncoder: 画像を圧縮しビジョントークンに変換（SAM+CLIP構造）
  2. DeepSeek3B-MoE: ビジョントークンからテキストを復元（570M活性パラメータ）
  特徴: 7-20倍トークン削減、97%精度で10倍圧縮達成（Fox benchmark）""")

print("\nモデルロード中...")
if not torch.cuda.is_available():
    print_separator()
    print("【エラー】NVIDIA GPUが検出されませんでした")
    print("DeepSeek-OCRはNVIDIA GPU専用モデルです")
    print_separator()
    raise RuntimeError("GPU required for DeepSeek-OCR")

print(f"デバイス: cuda")

tokenizer = AutoTokenizer.from_pretrained('deepseek-ai/DeepSeek-OCR', trust_remote_code=True)
model = AutoModel.from_pretrained(
    'deepseek-ai/DeepSeek-OCR',
    _attn_implementation='flash_attention_2',
    trust_remote_code=True,
    use_safetensors=True
).eval().cuda().to(torch.bfloat16)
print("ロード完了\n")

print()
print_separator()
print("【ファイル選択】")
print_separator()
print("複数の画像・PDFファイルをアップロード可能です")
print("対応形式: PDF, PNG, JPG, JPEG, BMP, TIFF")
print(f"\nOCR方式: Free OCRモード（高精度、座標なし、斜め文字対応）\n")

user_files = select_files_dialog()

if user_files:
    all_results = []

    for file_path in user_files:
        print(f"\n処理中: {file_path}")
        file_results = process_file(file_path, config, model, tokenizer)
        all_results.extend(file_results)

    print()
    print_separator()
    print("【OCR結果一覧】")
    print_separator()
    for result in all_results:
        print(f"\n--- {result['filename']} ---")
        print(f"元のファイル: {result['original']}")
        print()
        print(result['text'])
        print_separator()

    print("\n【検証データ】")
    total_files = len(user_files)
    total_pages = len(all_results)
    print(f"  処理ファイル数: {total_files}")
    print(f"  総ページ/画像数: {total_pages}")
    print(f"  モード: {config['name']} (base_size={config['base_size']}, image_size={config['image_size']}, crop_mode={config['crop_mode']})")
    print(f"  ビジョントークン数: {config['tokens']}")
else:
    print("ファイル未選択")

print("\n処理完了")