Gemini Files API の孤児ファイルを棚卸しする — 多アプリ運用の照合と自動クリーンアップ設計

import sqlite3
import datetime
 
def init_db(conn):
    conn.execute("""
        CREATE TABLE IF NOT EXISTS uploads (
            file_name   TEXT PRIMARY KEY,   -- Files API の name (files/xxxx)
            app_id      TEXT NOT NULL,      -- どのアプリ向けか
            local_path  TEXT NOT NULL,
            uploaded_at TEXT NOT NULL,      -- ISO8601 (UTC)
            purpose     TEXT NOT NULL,      -- classify / variation など
            done        INTEGER DEFAULT 0   -- 推論まで完了したか
        )
    """)
    conn.commit()
 
def record_upload(conn, f, app_id: str, local_path: str, purpose: str):
    conn.execute(
        "INSERT OR REPLACE INTO uploads VALUES (?, ?, ?, ?, ?, 0)",
        (f.name, app_id, local_path,
         datetime.datetime.now(datetime.timezone.utc).isoformat(), purpose),
    )
    conn.commit()

def reconcile(conn):
    remote = {f.name: f for f in client.files.list()}
    rows = conn.execute(
        "SELECT file_name, app_id, uploaded_at FROM uploads WHERE done = 0"
    ).fetchall()
    local_pending = {r[0]: r for r in rows}
 
    only_remote = set(remote) - set(local_pending)   # 素性不明
    only_local  = set(local_pending) - set(remote)   # 失効済み
    both        = set(remote) & set(local_pending)   # 進行中
 
    # 失効していた未完了ファイルは、DB を畳んで再アップロード対象にする
    for name in only_local:
        conn.execute("UPDATE uploads SET done = -1 WHERE file_name = ?", (name,))
    conn.commit()
    return {"unknown": only_remote, "expired": only_local, "active": both}

# After: アップロード→記録→推論→削除→完了記録 を1サイクルで閉じる
def classify_wallpaper(conn, path: str, app_id: str):
    f = upload_and_wait(path)
    record_upload(conn, f, app_id, path, purpose="classify")
    try:
        resp = client.models.generate_content(
            model="gemini-2.5-flash",
            contents=[f, "この画像を30カテゴリのいずれかに分類してください"],
        )
        category = resp.text.strip()
    finally:
        # 成否にかかわらず、このサイクルで上げたファイルは責任を持って削除する
        try:
            client.files.delete(name=f.name)
        except Exception:
            pass  # 既に消えていても問題ない。次回の照合で吸収する
        conn.execute("UPDATE uploads SET done = 1 WHERE file_name = ?", (f.name,))
        conn.commit()
    return category

def sweep_unknown(unknown_names, min_age_hours: float = 1.0):
    now = datetime.datetime.now(datetime.timezone.utc)
    swept = 0
    for name in unknown_names:
        f = client.files.get(name=name)
        created = datetime.datetime.fromisoformat(
            f.create_time.isoformat() if hasattr(f.create_time, "isoformat")
            else str(f.create_time)
        )
        if (now - created).total_seconds() >= min_age_hours * 3600:
            client.files.delete(name=name)
            swept += 1
    return swept

def cleanup_job(db_path: str = "uploads.db"):
    conn = sqlite3.connect(db_path)
    init_db(conn)
    result = reconcile(conn)
 
    swept = sweep_unknown(result["unknown"], min_age_hours=1.0)
    metrics = {
        "active": len(result["active"]),
        "expired": len(result["expired"]),
        "unknown_found": len(result["unknown"]),
        "unknown_swept": swept,
        "at": datetime.datetime.now(datetime.timezone.utc).isoformat(),
    }
    # 失効として畳んだ未完了ファイルは、再アップロードのキューに戻す
    requeue = conn.execute(
        "SELECT local_path, app_id FROM uploads WHERE done = -1"
    ).fetchall()
    conn.close()
    print(metrics)
    return metrics, requeue