停止するモデルで作った画像は、二度と同じものが作れない — 再生成可能性を台帳で管理する

# build_ledger.py — 既存アセットから provenance 台帳の初期行を作る
import hashlib
import json
import sqlite3
from pathlib import Path
 
from google import genai
 
client = genai.Client()  # GEMINI_API_KEY を環境変数から読む
ASSETS = Path("assets/wallpapers")
DB = Path("provenance.sqlite3")
 
SCHEMA = """
CREATE TABLE IF NOT EXISTS asset (
    output_sha256 TEXT PRIMARY KEY,
    path          TEXT NOT NULL,
    model_id      TEXT,
    prompt        TEXT,
    params_json   TEXT,
    embedding     BLOB,
    generated_at  TEXT,
    verdict       TEXT DEFAULT 'unjudged'
);
"""
 
 
def sha256_of(path: Path) -> str:
    h = hashlib.sha256()
    with path.open("rb") as f:
        for chunk in iter(lambda: f.read(1 << 20), b""):
            h.update(chunk)
    return h.hexdigest()
 
 
def embed_image(path: Path) -> list[float]:
    """gemini-embedding-2 で画像を埋め込む。画像とテキストが同一空間に載る。"""
    uploaded = client.files.upload(file=str(path))
    res = client.models.embed_content(
        model="gemini-embedding-2",
        contents=uploaded,
        config={"task_type": "RETRIEVAL_DOCUMENT"},
    )
    return res.embeddings[0].values
 
 
def main() -> None:
    conn = sqlite3.connect(DB)
    conn.executescript(SCHEMA)
 
    # 当時のスクリプトからサルベージしたプロンプト表（無いものは None のまま）
    salvaged = json.loads(Path("salvaged_prompts.json").read_text(encoding="utf-8"))
 
    for path in sorted(ASSETS.glob("*.png")):
        digest = sha256_of(path)
        row = conn.execute(
            "SELECT 1 FROM asset WHERE output_sha256 = ?", (digest,)
        ).fetchone()
        if row:
            continue  # 冪等: 同じ画像は二度入れない
 
        meta = salvaged.get(path.name, {})
        vec = embed_image(path)
        conn.execute(
            "INSERT INTO asset (output_sha256, path, model_id, prompt, params_json,"
            " embedding, generated_at) VALUES (?, ?, ?, ?, ?, ?, ?)",
            (
                digest,
                str(path),
                meta.get("model_id"),
                meta.get("prompt"),
                json.dumps(meta.get("params", {}), ensure_ascii=False),
                json.dumps(vec).encode("utf-8"),
                meta.get("generated_at"),
            ),
        )
        conn.commit()
        print(f"ledger += {path.name} ({digest[:12]})")
 
    conn.close()
 
 
if __name__ == "__main__":
    main()

# judge_regenerability.py — 新モデル出力と旧アセットの距離で判定する
import json
import sqlite3
 
import numpy as np
from google import genai
 
client = genai.Client()
NEW_MODEL = "gemini-3-image"  # 移行先の GA モデル
THRESHOLD = 0.86
 
 
def cosine(a: np.ndarray, b: np.ndarray) -> float:
    return float(a @ b / (np.linalg.norm(a) * np.linalg.norm(b)))
 
 
def regenerate(prompt: str, params: dict) -> bytes:
    res = client.models.generate_images(
        model=NEW_MODEL,
        prompt=prompt,
        config={
            "number_of_images": 1,
            "aspect_ratio": params.get("aspect_ratio", "9:16"),
        },
    )
    return res.generated_images[0].image.image_bytes
 
 
def judge(conn: sqlite3.Connection) -> None:
    rows = conn.execute(
        "SELECT output_sha256, prompt, params_json, embedding FROM asset"
        " WHERE prompt IS NOT NULL AND verdict = 'unjudged'"
    ).fetchall()
 
    for digest, prompt, params_json, emb_blob in rows:
        old_vec = np.array(json.loads(emb_blob.decode("utf-8")))
        image_bytes = regenerate(prompt, json.loads(params_json))
 
        new_vec = np.array(embed_bytes(image_bytes))
        score = cosine(old_vec, new_vec)
        verdict = "regenerable" if score >= THRESHOLD else "frozen"
 
        conn.execute(
            "UPDATE asset SET verdict = ? WHERE output_sha256 = ?", (verdict, digest)
        )
        conn.commit()
        print(f"{digest[:12]} score={score:.3f} -> {verdict}")
 
    # プロンプト不明の行は、入力が無い時点で再生成不能として確定させる
    conn.execute(
        "UPDATE asset SET verdict = 'frozen' WHERE prompt IS NULL AND verdict = 'unjudged'"
    )
    conn.commit()

def build_variant(digest: str, target_size: tuple[int, int], conn) -> Path:
    (verdict, path, prompt, params_json) = conn.execute(
        "SELECT verdict, path, prompt, params_json FROM asset WHERE output_sha256 = ?",
        (digest,),
    ).fetchone()
 
    if verdict == "regenerable":
        # 生成層: 新モデルで解像度に合わせて作り直す
        params = json.loads(params_json)
        params["aspect_ratio"] = to_aspect_ratio(target_size)
        return save(regenerate(prompt, params), digest, target_size)
 
    # 派生層: モデルを通さず、既存ファイルから画像処理だけで作る
    return resize_without_model(Path(path), target_size, digest)