Antigravity × Gemini File API で長時間メディアを AI に読ませる実装ガイド — 動画・音声・PDFの要約とハイライト抽出パイプライン

# .venv を作成して Gen AI SDK を入れる
python -m venv .venv
source .venv/bin/activate  # Windows なら .venv\Scripts\activate
pip install "google-genai>=1.0" "pydantic>=2.5" "python-dotenv>=1.0" "rich>=13.0"

# .env
GEMINI_API_KEY=YOUR_GEMINI_API_KEY

# file_ingest.py — 冪等なアップロード
from __future__ import annotations
import hashlib
import json
import os
import time
from pathlib import Path
from google import genai
from google.genai import types
from dotenv import load_dotenv
 
load_dotenv()
client = genai.Client(api_key=os.environ["GEMINI_API_KEY"])
CACHE_DIR = Path(".cache/files")
CACHE_DIR.mkdir(parents=True, exist_ok=True)
 
def _sha1(path: Path) -> str:
    """ファイルの SHA-1 を返す。キャッシュキーに使う"""
    h = hashlib.sha1()
    with path.open("rb") as f:
        for chunk in iter(lambda: f.read(1024 * 1024), b""):
            h.update(chunk)
    return h.hexdigest()
 
def upload_or_reuse(local_path: str, display_name: str | None = None) -> dict:
    """同一ファイルは再アップロードせず、キャッシュに保存した File オブジェクト情報を返す"""
    path = Path(local_path)
    key = _sha1(path)
    cache_path = CACHE_DIR / f"{key}.json"
 
    if cache_path.exists():
        cached = json.loads(cache_path.read_text())
        # Gemini File API は 48 時間で削除される
        if time.time() - cached["uploaded_at"] < 47 * 3600:
            print(f"♻️  reuse cache: {cached['name']}")
            return cached
 
    print(f"⬆️  uploading {path.name} ({path.stat().st_size / 1024 / 1024:.1f} MB)...")
    file = client.files.upload(
        file=path,
        config=types.UploadFileConfig(display_name=display_name or path.name),
    )
    # ACTIVE になるまで待機
    while file.state.name == "PROCESSING":
        time.sleep(2)
        file = client.files.get(name=file.name)
 
    if file.state.name == "FAILED":
        raise RuntimeError(f"upload failed: {file.error}")
 
    info = {
        "name": file.name,
        "uri": file.uri,
        "mime_type": file.mime_type,
        "display_name": file.display_name,
        "uploaded_at": time.time(),
    }
    cache_path.write_text(json.dumps(info, ensure_ascii=False, indent=2))
    print(f"✅ ready: {file.name}")
    return info

# summary.py — 全体要約ステージ
from google.genai import types
from file_ingest import client
 
SUMMARY_MODEL = "gemini-2.5-pro"  # 長時間メディアは Pro 系が安定
 
def summarize(file_info: dict, language: str = "ja") -> str:
    """アップロード済みファイルの全体要約を返す"""
    system = (
        "あなたはメディア要約の専門家です。"
        "与えられた動画・音声・PDF の内容を、時系列に沿って 400〜600 文字で要約してください。"
        "推測や一般論は書かず、メディア内で実際に述べられている事実のみを書きます。"
    )
    contents = [
        types.Part.from_uri(file_uri=file_info["uri"], mime_type=file_info["mime_type"]),
        types.Part.from_text(text=f"上記のメディアを {language} で要約してください。"),
    ]
    try:
        resp = client.models.generate_content(
            model=SUMMARY_MODEL,
            contents=contents,
            config=types.GenerateContentConfig(
                system_instruction=system,
                temperature=0.2,  # 要約は再現性重視で低め
            ),
        )
    except Exception as e:
        # レート制限や一時的な失敗は呼び出し側でリトライする
        raise RuntimeError(f"summary failed: {e}") from e
    return resp.text.strip()

# extract.py — タイムスタンプ付きハイライト抽出
from __future__ import annotations
from pydantic import BaseModel, Field
from google.genai import types
from file_ingest import client
 
EXTRACT_MODEL = "gemini-2.5-pro"
 
class Chapter(BaseModel):
    start: str = Field(..., description="章の開始タイムスタンプ HH:MM:SS")
    end: str = Field(..., description="章の終了タイムスタンプ HH:MM:SS")
    title: str = Field(..., description="章タイトル 30 文字以内")
    summary: str = Field(..., description="章の要約 120〜200 文字")
 
class Highlight(BaseModel):
    timestamp: str = Field(..., description="ハイライト開始 HH:MM:SS")
    title: str
    reason: str = Field(..., description="なぜハイライトか 60〜100 文字")
 
class MediaDigest(BaseModel):
    overall_summary: str
    chapters: list[Chapter]
    highlights: list[Highlight]
 
def extract_digest(file_info: dict, overall_summary: str, language: str = "ja") -> MediaDigest:
    system = (
        "あなたはメディアの構造抽出エンジンです。"
        "渡されたメディア本体と事前要約を元に、章立てとハイライトを抽出します。"
        "タイムスタンプは必ず HH:MM:SS 形式（必要なら MM:SS でも可）で、"
        "メディア内で実際に到達する時刻を記述すること。架空のタイムスタンプは禁止。"
    )
    contents = [
        types.Part.from_uri(file_uri=file_info["uri"], mime_type=file_info["mime_type"]),
        types.Part.from_text(
            text=(
                f"# 事前要約\n{overall_summary}\n\n"
                f"# 指示\n上記メディアを {language} で、章立てとハイライトを抽出してください。"
                "章は5〜10個、ハイライトは3〜8個が目安です。"
            )
        ),
    ]
    try:
        resp = client.models.generate_content(
            model=EXTRACT_MODEL,
            contents=contents,
            config=types.GenerateContentConfig(
                system_instruction=system,
                response_mime_type="application/json",
                response_schema=MediaDigest,  # Pydantic をそのまま渡せる
                temperature=0.3,
            ),
        )
    except Exception as e:
        raise RuntimeError(f"extract failed: {e}") from e
    # SDK が Pydantic モデルへパース済みの結果を返すのでそのまま返却
    if resp.parsed is None:
        raise ValueError("Gemini response could not be parsed into MediaDigest")
    return resp.parsed  # type: ignore[return-value]

# 後処理: タイムスタンプの正規化
def normalize_ts(ts: str) -> str:
    parts = ts.split(":")
    if len(parts) == 2:
        return "00:" + ts
    return ts

# cost.py — 使用量とコスト計算
# 料金は 2026 年 4 月時点の公式レート例。最新は公式ドキュメントで要確認
PRICE_PER_1M_INPUT_USD = 1.25   # Gemini 2.5 Pro テキスト換算
PRICE_PER_1M_OUTPUT_USD = 10.0
USD_JPY = 150  # レートは環境に合わせて上書き
 
def estimate_cost(usage) -> dict:
    """response.usage_metadata を受け取り、推定コストを返す"""
    in_tok = getattr(usage, "prompt_token_count", 0) or 0
    out_tok = getattr(usage, "candidates_token_count", 0) or 0
    usd = (
        in_tok / 1_000_000 * PRICE_PER_1M_INPUT_USD
        + out_tok / 1_000_000 * PRICE_PER_1M_OUTPUT_USD
    )
    return {
        "input_tokens": in_tok,
        "output_tokens": out_tok,
        "usd": round(usd, 4),
        "jpy": round(usd * USD_JPY, 1),
    }

# main.py
import argparse
import json
from pathlib import Path
from rich.console import Console
from file_ingest import upload_or_reuse
from summary import summarize
from extract import extract_digest, normalize_ts
from cost import estimate_cost
 
console = Console()
 
def run(input_path: str, out_dir: str = "out", language: str = "ja") -> None:
    Path(out_dir).mkdir(parents=True, exist_ok=True)
    file_info = upload_or_reuse(input_path)
 
    console.print("[bold cyan]Stage 2: summary[/bold cyan]")
    overall = summarize(file_info, language=language)
 
    console.print("[bold cyan]Stage 3: extract[/bold cyan]")
    digest = extract_digest(file_info, overall, language=language)
 
    # タイムスタンプ正規化
    for ch in digest.chapters:
        ch.start = normalize_ts(ch.start)
        ch.end = normalize_ts(ch.end)
    for hl in digest.highlights:
        hl.timestamp = normalize_ts(hl.timestamp)
 
    out = {
        "overall_summary": overall,
        "digest": digest.model_dump(),
    }
    out_path = Path(out_dir) / (Path(input_path).stem + ".json")
    out_path.write_text(json.dumps(out, ensure_ascii=False, indent=2))
    console.print(f"💾 saved: {out_path}")
 
if __name__ == "__main__":
    ap = argparse.ArgumentParser()
    ap.add_argument("input")
    ap.add_argument("--out-dir", default="out")
    ap.add_argument("--lang", default="ja")
    args = ap.parse_args()
    run(args.input, args.out_dir, args.lang)

$ python main.py ./samples/meeting_20260410.m4a --lang ja
⬆️  uploading meeting_20260410.m4a (183.5 MB)...
✅ ready: files/abc123
Stage 2: summary
Stage 3: extract
💾 saved: out/meeting_20260410.json

{
  "overall_summary": "本会議は 2026-04-10 に行われた、...",
  "digest": {
    "chapters": [
      {"start": "00:00:00", "end": "00:12:30", "title": "議題確認とKPIレビュー", "summary": "..."},
      ...
    ],
    "highlights": [
      {"timestamp": "00:47:15", "title": "新プランの価格合意", "reason": "..."}
    ]
  }
}