Gemini File APIのファイルが「PROCESSING」のまま止まる：タイムアウト処理とリトライ設計

import google.generativeai as genai
 
genai.configure(api_key="YOUR_GEMINI_API_KEY")
 
# ファイルをアップロード
uploaded_file = genai.upload_file("sample_video.mp4", mime_type="video/mp4")
print(uploaded_file.state)  # → FileState.PROCESSING

# ❌ 悪い例：一度確認して諦める
file = genai.get_file(uploaded_file.name)
if file.state == genai.protos.File.State.PROCESSING:
    print("まだ処理中")
    # ここで止まってしまう

import google.generativeai as genai
import time
import math
 
def wait_for_file_active(
    file_name: str,
    max_wait_seconds: int = 300,  # 最大5分待機
    initial_interval: float = 5.0,  # 最初の待機間隔
    max_interval: float = 30.0,    # 最大待機間隔
) -> genai.protos.File:
    """
    ファイルがACTIVE状態になるまで待機する。
    
    Args:
        file_name: アップロードしたファイルのname（例: "files/abc123"）
        max_wait_seconds: この時間を超えたらTimeoutErrorを投げる
        initial_interval: 最初のポーリング間隔（秒）
        max_interval: 最大ポーリング間隔（秒）
    
    Returns:
        ACTIVE状態のFileオブジェクト
    
    Raises:
        TimeoutError: max_wait_seconds以内にACTIVEにならなかった場合
        RuntimeError: ファイルがFAILED状態になった場合
    """
    start_time = time.time()
    interval = initial_interval
    attempt = 0
    
    while True:
        elapsed = time.time() - start_time
        
        if elapsed > max_wait_seconds:
            raise TimeoutError(
                f"ファイル {file_name} が {max_wait_seconds}秒以内にACTIVEになりませんでした"
            )
        
        file = genai.get_file(file_name)
        attempt += 1
        
        print(f"  [{attempt}] {file.name}: {file.state} ({elapsed:.0f}秒経過)")
        
        if file.state == genai.protos.File.State.ACTIVE:
            print(f"✅ ACTIVE になりました（{elapsed:.0f}秒）")
            return file
        
        if file.state == genai.protos.File.State.FAILED:
            raise RuntimeError(
                f"ファイル {file_name} の処理が失敗しました: {getattr(file, 'error', '不明なエラー')}"
            )
        
        # 指数バックオフ：間隔を徐々に広げる
        time.sleep(interval)
        interval = min(interval * 1.5, max_interval)

# アップロード
uploaded = genai.upload_file("document.pdf", mime_type="application/pdf")
print(f"アップロード完了: {uploaded.name}")
 
try:
    # ACTIVE になるまで待機
    active_file = wait_for_file_active(uploaded.name, max_wait_seconds=180)
    
    # モデルにリクエスト
    model = genai.GenerativeModel("gemini-2.5-flash")
    response = model.generate_content(["この書類を要約してください。", active_file])
    print(response.text)
 
except TimeoutError as e:
    print(f"タイムアウト: {e}")
    # 必要に応じてファイルを削除
    genai.delete_file(uploaded.name)
 
except RuntimeError as e:
    print(f"処理失敗: {e}")

import contextlib
 
@contextlib.contextmanager
def managed_file(file_path: str, mime_type: str):
    """アップロードしたファイルを自動的にクリーンアップするコンテキストマネージャー"""
    uploaded = genai.upload_file(file_path, mime_type=mime_type)
    try:
        active = wait_for_file_active(uploaded.name)
        yield active
    finally:
        # 正常終了・例外発生どちらの場合でも削除
        try:
            genai.delete_file(uploaded.name)
            print(f"🗑️ ファイル削除: {uploaded.name}")
        except Exception:
            pass  # 削除失敗は無視
 
# 使い方
with managed_file("video.mp4", "video/mp4") as file:
    response = model.generate_content(["動画の内容を説明してください。", file])
    print(response.text)
# withブロックを抜けた時点で自動削除

import mimetypes
 
def get_mime_type(file_path: str) -> str:
    """ファイルパスからMIMEタイプを判定する"""
    mime_type, _ = mimetypes.guess_type(file_path)
    if mime_type is None:
        raise ValueError(f"MIMEタイプを判定できませんでした: {file_path}")
    return mime_type
 
# 使い方
path = "sample_video.mp4"
mime = get_mime_type(path)  # → "video/mp4"
uploaded = genai.upload_file(path, mime_type=mime)

import asyncio
import google.generativeai as genai
 
async def async_wait_for_active(file_name: str, max_wait: int = 300) -> genai.protos.File:
    """非同期版のファイル待機関数"""
    start = asyncio.get_event_loop().time()
    interval = 5.0
    
    while True:
        elapsed = asyncio.get_event_loop().time() - start
        if elapsed > max_wait:
            raise TimeoutError(f"{file_name} がタイムアウト")
        
        # genai.get_file は同期関数なのでexecutorで実行
        loop = asyncio.get_event_loop()
        file = await loop.run_in_executor(None, genai.get_file, file_name)
        
        if file.state == genai.protos.File.State.ACTIVE:
            return file
        if file.state == genai.protos.File.State.FAILED:
            raise RuntimeError(f"{file_name} の処理が失敗")
        
        await asyncio.sleep(min(interval, 30.0))
        interval *= 1.5
 
async def batch_upload_and_wait(file_paths: list[str]) -> list[genai.protos.File]:
    """複数ファイルを並列アップロードし、全て ACTIVE になるまで待機"""
    loop = asyncio.get_event_loop()
    
    # 順番にアップロード（upload_fileは同期）
    uploaded_files = []
    for path in file_paths:
        mime = get_mime_type(path)
        f = await loop.run_in_executor(None, lambda p=path, m=mime: genai.upload_file(p, mime_type=m))
        uploaded_files.append(f)
        print(f"アップロード: {f.name}")
    
    # 全ファイルの完了を並列待機
    active_files = await asyncio.gather(
        *[async_wait_for_active(f.name) for f in uploaded_files]
    )
    return list(active_files)
 
# 実行例
# files = asyncio.run(batch_upload_and_wait(["img1.jpg", "img2.jpg", "img3.jpg"]))