File Search のストアが本番で静かに陳腐化する — カタログ同期とドリフト検知を実装した運用メモ

# diff_manifest.py
import json
from pathlib import Path
 
 
def load(path: str) -> dict[str, str]:
    p = Path(path)
    return json.loads(p.read_text()) if p.exists() else {}
 
 
def diff(prev: dict[str, str], curr: dict[str, str]):
    prev_ids, curr_ids = set(prev), set(curr)
    added = curr_ids - prev_ids
    removed = prev_ids - curr_ids
    # 両方に存在し、指紋が変わったものが「更新」
    updated = {i for i in (prev_ids & curr_ids) if prev[i] != curr[i]}
    return added, updated, removed
 
 
if __name__ == "__main__":
    prev = load("manifest.synced.json")   # 前回ストアに反映済みの状態
    curr = load("manifest.current.json")  # 今の正本
    added, updated, removed = diff(prev, curr)
    print(f"added={len(added)} updated={len(updated)} removed={len(removed)}")
    # 期待する出力例: added=42 updated=18 removed=7

# incremental_import.py
from google import genai
 
client = genai.Client()
STORE_NAME = "wallpaper-catalog-active"  # アクティブなストア名（後述のポインタ）
 
 
def import_asset(asset: dict) -> None:
    """1アセットをストアへ取り込む。File Search 側のメソッド名・引数は
    SDK のバージョンで動くため、ここはプレビュー仕様を確認してから固定する。
    本質は『この ID の内容をこの指紋でストアに載せた』という事実をこちらが台帳化すること。"""
    client.file_search_stores.upload_to_file_search_store(
        file_search_store_name=STORE_NAME,
        file=asset["image_path"],
        config={
            "display_name": asset["id"],
            "custom_metadata": [
                {"key": "title", "string_value": asset["title"]},
                {"key": "category", "string_value": asset["category"]},
            ],
        },
    )
 
 
def apply_incremental(catalog_by_id: dict, added: set, updated: set) -> None:
    # 更新は「いったん同じIDの古い文書を消してから入れ直す」を基本にする。
    # 消し漏れがあると同じ壁紙が二重に引用されるため、消えたことを必ず確認する。
    for asset_id in sorted(added | updated):
        import_asset(catalog_by_id[asset_id])
    print(f"imported {len(added | updated)} documents")

# bluegreen_rebuild.py
import json
import time
from pathlib import Path
from google import genai
 
client = genai.Client()
POINTER = Path("active_store.json")  # {"store": "wallpaper-catalog-active-20260623"}
 
 
def active_store() -> str:
    return json.loads(POINTER.read_text())["store"]
 
 
def rebuild(catalog_by_id: dict) -> str:
    # 1) 日付つきの新ストアを作る（緑）。古い方（青）はまだ生かしておく。
    new_store = f"wallpaper-catalog-active-{time.strftime('%Y%m%d-%H%M')}"
    client.file_search_stores.create(config={"display_name": new_store})
 
    # 2) 配信中の正本だけを全件取り込む
    for asset in catalog_by_id.values():
        client.file_search_stores.upload_to_file_search_store(
            file_search_store_name=new_store,
            file=asset["image_path"],
            config={"display_name": asset["id"]},
        )
 
    # 3) 検証クエリ（削除済みアセットの名前で引いて、ヒットしないことを確認）
    if not passes_smoke_test(new_store):
        raise RuntimeError("smoke test failed — ポインタは切り替えない")
 
    # 4) ポインタを切り替え（ここで初めて本番が新ストアを向く）
    POINTER.write_text(json.dumps({"store": new_store}))
    print(f"switched active store -> {new_store}")
    return new_store

# drift_audit.py
import json
from pathlib import Path
 
 
def audit() -> int:
    current = json.loads(Path("manifest.current.json").read_text())
    synced = json.loads(Path("manifest.synced.json").read_text())
 
    added, updated, removed = (
        set(current) - set(synced),
        {i for i in set(current) & set(synced) if current[i] != synced[i]},
        set(synced) - set(current),
    )
    drift = len(added) + len(updated) + len(removed)
    total = max(len(current), 1)
    rate = drift / total * 100
 
    print(f"drift: {drift}/{total} ({rate:.1f}%) "
          f"add={len(added)} upd={len(updated)} del={len(removed)}")
    # しきい値を超えたら通知。私は 1.0% を超えたら手を止めて見にいくと決めている。
    return drift
 
 
if __name__ == "__main__":
    if audit() > 0:
        # ここで Slack 通知や exit code で CI を赤くする
        print("⚠️ ストアと正本にズレがあります。同期ジョブのログを確認してください。")