app.py

#!/usr/bin/env python3
"""
PDF OCR & Markdown変換ツール (v5.latest)

最新のgoogle.genai SDKを使用したPDFからMarkdownへの変換アプリ
"""

import os
import socket
import tempfile
from concurrent.futures import ThreadPoolExecutor, as_completed
from pathlib import Path
from typing import List, Tuple

import gradio as gr
import fitz  # PyMuPDF
from google import genai
from google.genai import types

from presets import SYSTEM_PROMPTS, list_presets


# 環境変数から設定を読み込む
GEMINI_API_KEY = os.environ.get("GEMINI_API_KEY", "")

if not GEMINI_API_KEY:
    print("⚠️ 警告: GEMINI_API_KEYが設定されていません。")
    print("環境変数またはHugging Face Spacesのシークレットに設定してください。")


def split_pdf(pdf_path: str, output_dir: str, pages_per_chunk: int) -> List[Tuple[int, str]]:
    """
    PDFを指定ページ数ごとに分割する

    Args:
        pdf_path: 元のPDFファイルパス
        output_dir: 分割PDFの保存先ディレクトリ
        pages_per_chunk: 何ページごとに分割するか

    Returns:
        (開始ページ番号, 分割PDFパス)のリスト
    """
    pdf_document = fitz.open(pdf_path)
    total_pages = len(pdf_document)

    split_pdfs = []

    for start_page in range(0, total_pages, pages_per_chunk):
        end_page = min(start_page + pages_per_chunk - 1, total_pages - 1)

        # 新しいPDFドキュメントを作成
        output_pdf = fitz.open()

        # 指定範囲のページを新しいPDFに追加
        for page_num in range(start_page, end_page + 1):
            output_pdf.insert_pdf(pdf_document, from_page=page_num, to_page=page_num)

        # 分割したPDFを保存
        output_path = os.path.join(
            output_dir, f"split_{start_page+1}_to_{end_page+1}.pdf"
        )
        output_pdf.save(output_path)
        output_pdf.close()

        split_pdfs.append((start_page, output_path))

    pdf_document.close()
    return split_pdfs


def process_pdf_chunk(
    pdf_path: str,
    system_instruction: str,
    model_name: str,
    enable_thinking: bool,
) -> str:
    """
    1つのPDFチャンクをGemini APIで処理

    Args:
        pdf_path: PDFファイルパス
        system_instruction: システムインストラクション
        model_name: 使用するモデル名
        enable_thinking: Thinking Mode有効化

    Returns:
        生成されたMarkdown文字列
    """
    # PDFをバイナリ読み込み
    pdf_bytes = Path(pdf_path).read_bytes()

    # Geminiクライアント初期化
    client = genai.Client(api_key=GEMINI_API_KEY)

    # リクエストコンテンツを構築
    contents = [
        types.Content(
            role="user",
            parts=[
                types.Part.from_text(
                    text="添付のPDFを人間が読む順序に従ってMarkdownへ変換してください。"
                ),
                types.Part.from_bytes(
                    data=pdf_bytes,
                    mime_type="application/pdf"
                ),
            ],
        ),
    ]

    # 生成設定
    config = types.GenerateContentConfig(
        system_instruction=[
            types.Part.from_text(text=system_instruction),
        ],
    )

    # Thinking Mode設定
    if enable_thinking:
        try:
            config.thinking_config = types.ThinkingConfig()
        except (TypeError, ValueError) as thinking_error:
            print(f"⚠️ ThinkingConfig設定時にエラーが発生しました: {thinking_error}")

    # ストリーミング処理
    collected_chunks: List[str] = []

    try:
        for chunk in client.models.generate_content_stream(
            model=model_name,
            contents=contents,
            config=config,
        ):
            if hasattr(chunk, "text") and chunk.text:
                collected_chunks.append(chunk.text)

        return "".join(collected_chunks)

    except Exception as e:
        return f"<!-- エラー: {e} -->\n\n⚠️ このチャンクの処理でエラーが発生しました。"


def process_pdf(
    pdf_file,
    preset_name: str,
    custom_prompt: str,
    pages_per_chunk: int,
    model_name: str,
    enable_thinking: bool,
    progress=gr.Progress(),
) -> str:
    """
    PDFファイルを処理してMarkdownに変換

    Args:
        pdf_file: アップロードされたPDFファイル
        preset_name: プリセット名
        custom_prompt: カスタムシステムプロンプト
        pages_per_chunk: 何ページごとに分割するか
        model_name: 使用するモデル名
        enable_thinking: Thinking Mode有効化
        progress: Gradio進捗バー

    Returns:
        変換されたMarkdown文字列
    """
    if not GEMINI_API_KEY:
        return "❌ エラー: GEMINI_API_KEYが設定されていません。\n\n環境変数またはHugging Face Spacesのシークレットに設定してください。"

    if not pdf_file:
        return "❌ エラー: PDFファイルをアップロードしてください。"

    instruction_text = custom_prompt.strip()

    # システムインストラクション決定
    if preset_name == "カスタム":
        if not instruction_text:
            return "❌ エラー: カスタムプロンプトを入力してください。"
        system_instruction = instruction_text
    else:
        # プリセット選択時もテキストボックスを編集可能にする
        system_instruction = instruction_text or SYSTEM_PROMPTS[preset_name]

    progress_cb = progress if callable(progress) else (lambda *args, **kwargs: None)

    log_progress(progress_cb, 0.1, "PDFを読み込み中...")

    with tempfile.TemporaryDirectory() as temp_dir:
        # PDFファイルパス
        temp_pdf_path = resolve_pdf_path(pdf_file)
        if not temp_pdf_path or not os.path.exists(temp_pdf_path):
            return (
                "❌ エラー: アップロードされたPDFファイルを読み込めませんでした。\n"
                "再度アップロードしてから実行してください。"
            )

        # PDFを分割
        log_progress(progress_cb, 0.2, "PDFを分割中...")
        split_pdf_paths = split_pdf(temp_pdf_path, temp_dir, pages_per_chunk)

        total_chunks = len(split_pdf_paths)
        log_progress(progress_cb, 0.3, f"{total_chunks}個のチャンクに分割完了")

        # 各チャンクを最大3並列で処理
        markdown_results = {}
        max_workers = min(3, total_chunks)

        with ThreadPoolExecutor(max_workers=max_workers) as executor:
            future_map = {}
            for start_page, pdf_path in split_pdf_paths:
                future = executor.submit(
                    process_pdf_chunk,
                    pdf_path=pdf_path,
                    system_instruction=system_instruction,
                    model_name=model_name,
                    enable_thinking=enable_thinking,
                )
                future_map[future] = start_page

            completed = 0
            for future in as_completed(future_map):
                start_page = future_map[future]
                try:
                    result = future.result()
                except Exception as chunk_error:
                    result = (
                        f"<!-- エラー: {chunk_error} -->\n\n⚠️ このチャンクの処理でエラーが発生しました。"
                    )

                markdown_results[start_page] = result
                completed += 1
                progress_value = 0.3 + 0.6 * (completed / total_chunks)
                log_progress(
                    progress_cb,
                    progress_value,
                    f"チャンク {completed}/{total_chunks} を処理中...",
                )

        # 結果を結合
        log_progress(progress_cb, 0.9, "結果を結合中...")

        combined_markdown = "\n\n".join(
            markdown_results[page] for page in sorted(markdown_results.keys())
        )

        log_progress(progress_cb, 1.0, "完了！")

        return combined_markdown


def update_instruction_text(preset_name: str):
    """プリセットに対応するシステムインストラクションをテキストエリアへ反映"""

    if preset_name == "カスタム":
        return gr.update(value="")

    return gr.update(value=SYSTEM_PROMPTS[preset_name])


def log_progress(progress_cb, value: float, desc: str) -> None:
    """進捗バー更新と同時にログへも出力"""

    try:
        progress_cb(value, desc=desc)
    except TypeError:
        progress_cb(value)

    print(f"[Progress] {value * 100:.0f}% - {desc}", flush=True)


def resolve_pdf_path(uploaded_pdf) -> str:
    """
    GradioのFileコンポーネントから渡される値を安全にファイルパスへ変換
    """
    if uploaded_pdf is None:
        return ""

    if isinstance(uploaded_pdf, str):
        return uploaded_pdf

    if isinstance(uploaded_pdf, dict):
        # Gradio 5では辞書形式で渡されるケースもある
        return uploaded_pdf.get("name") or uploaded_pdf.get("path") or ""

    return getattr(uploaded_pdf, "name", "")


def create_interface():
    """Gradio UIを作成"""

    with gr.Blocks(title="PDF OCR & Markdown変換") as demo:
        gr.Markdown("# 📄 PDF OCR & Markdown変換ツール (v5.latest)")
        gr.Markdown(
            "PDFをアップロードすると、Gemini APIでOCR処理してMarkdown形式に変換します。\n\n"
            "✨ **新機能**: システムプロンプトのプリセット選択、分割単位の調整、Thinking Mode"
        )

        with gr.Row():
            with gr.Column(scale=1):
                pdf_input = gr.File(
                    label="📎 PDFファイルをアップロード",
                    file_types=[".pdf"],
                    type="filepath"
                )

                preset_dropdown = gr.Dropdown(
                    choices=list_presets() + ["カスタム"],
                    value="汎用OCR",
                    label="🎯 システムプロンプトのプリセット",
                    info="用途に応じたプリセットを選択"
                )

                custom_prompt_text = gr.Textbox(
                    label="✏️ システムインストラクション",
                    placeholder="プリセットに応じた指示が自動で入力されます。自由に編集できます。",
                    lines=10,
                    value=SYSTEM_PROMPTS["汎用OCR"],
                )

                pages_slider = gr.Slider(
                    minimum=1,
                    maximum=20,
                    value=5,
                    step=1,
                    label="📑 PDFを何ページごとに分割？",
                    info="小さいほど並列処理が増えますが、API呼び出し数も増えます"
                )

                model_dropdown = gr.Dropdown(
                    choices=[
                        "gemini-2.5-flash-preview-tts",
                        "gemini-flash-latest",
                        "gemini-pro-latest"
                    ],
                    value="gemini-flash-latest",
                    label="🤖 使用するモデル",
                    info="Flash: 高速・低コスト、Pro: 高精度"
                )

                thinking_checkbox = gr.Checkbox(
                    label="🧠 Thinking Mode有効化",
                    value=True,
                    info="モデルの思考過程を活用（より正確な結果）"
                )

                convert_btn = gr.Button(
                    "🚀 変換開始",
                    variant="primary",
                    size="lg"
                )

            with gr.Column(scale=2):
                markdown_output = gr.Textbox(
                    label="📝 変換結果（Markdown）",
                    lines=20,
                    max_lines=30,
                    show_copy_button=True
                )

                with gr.Row():
                    download_btn = gr.Button("💾 Markdownをダウンロード")

        preset_dropdown.change(
            fn=update_instruction_text,
            inputs=preset_dropdown,
            outputs=custom_prompt_text
        )

        convert_btn.click(
            fn=process_pdf,
            inputs=[
                pdf_input,
                preset_dropdown,
                custom_prompt_text,
                pages_slider,
                model_dropdown,
                thinking_checkbox,
            ],
            outputs=markdown_output
        )

        download_btn.click(
            None,
            markdown_output,
            [],
            js="""(x) => {
                const blob = new Blob([x], {type: 'text/markdown;charset=utf-8'});
                const url = URL.createObjectURL(blob);
                const a = document.createElement('a');
                a.href = url;
                a.download = 'converted.md';
                document.body.appendChild(a);
                a.click();
                document.body.removeChild(a);
                URL.revokeObjectURL(url);
            }"""
        )

        with gr.Accordion("📖 使用方法", open=False):
            gr.Markdown("""
### 基本的な使い方

1. **PDFアップロード**: 変換したいPDFファイルを選択
2. **プリセット選択**: 用途に応じたプリセットを選択
   - **汎用OCR**: 一般的な文書
   - **教育・参考書**: 教科書や参考書（右→左、上→下の読み順対応）
   - **ビジネス文書**: ビジネスレポート、提案書
   - **スクリーンショット**: ソフトウェアのUI画像
   - **カスタム**: 独自のシステムプロンプトを入力
3. **分割設定**: PDFを何ページごとに分割するか調整（デフォルト: 5ページ）
4. **モデル選択**: Flash（高速）またはPro（高精度）
5. **変換開始**: ボタンをクリックして処理開始
6. **結果確認**: 変換されたMarkdownを確認・ダウンロード

### Tips

- **大きなPDF**: ページ数を減らして分割すると並列処理で高速化
- **高精度**: Proモデル + Thinking Mode ONで最高品質
- **低コスト**: Flashモデル + Thinking Mode OFFで高速・低コスト
- **カスタムプロンプト**: 特定の用途に最適化した独自の指示を作成可能

### API Key設定

Hugging Face Spacesで使用する場合:
1. Spaceの設定で「Secrets」を開く
2. `GEMINI_API_KEY`という名前でGoogle AI StudioのAPIキーを設定
            """)

    return demo


def find_open_port() -> int:
    """利用可能なポート番号を取得"""
    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
        sock.bind(("", 0))
        return sock.getsockname()[1]


def launch_demo():
    """Gradioアプリを起動（ポート競合を自動回避）"""
    demo = create_interface()

    server_name = os.getenv("GRADIO_SERVER_NAME") or os.getenv("HOST", "0.0.0.0")
    share = os.getenv("GRADIO_SHARE", "").lower() == "true"

    candidate_ports: List[int | None] = []
    for env_var in ("GRADIO_SERVER_PORT", "PORT"):
        value = os.getenv(env_var)
        if value:
            try:
                candidate_ports.append(int(value))
            except ValueError:
                print(f"⚠️ 無効なポート番号が指定されています ({env_var}={value})")

    # デフォルト → 自動検出 → 最後にGradioのデフォルトへ
    candidate_ports.extend([7860, None])

    last_error: Exception | None = None

    for port in candidate_ports:
        kwargs = {
            "server_name": server_name,
            "share": share,
            "show_error": True,
        }

        if port is None:
            port_to_use = find_open_port()
        else:
            port_to_use = port

        kwargs["server_port"] = port_to_use

        try:
            print(f"ℹ️ Gradio UIを http://{server_name}:{port_to_use} で起動します。")
            demo.launch(**kwargs)
            return
        except OSError as error:
            print(f"⚠️ ポート {port_to_use} の起動に失敗しました: {error}")
            last_error = error

    if last_error:
        raise last_error


if __name__ == "__main__":
    launch_demo()