ki-fwi/tools/assets/comfyui_client.py

#!/usr/bin/env python3
"""
comfyui_client.py — génération des assets de FILIÈRE B (décor) via l'API HTTP de ComfyUI.

FILIÈRE B UNIQUEMENT. Ce script ne produit jamais d'image servant à identifier une
espèce : fonds, sprites du mode arcade, badges, écrans de titre, iconographie. La
séparation est vérifiée par `npm run validate:data`, qui refuse qu'un média de filière B
porte un usage « épreuve » ou « indice » (ADR 0004).

Exécution HORS LIGNE, AU BUILD. Aucun appel réseau ne subsiste dans le jeu livré : les
sorties sont des fichiers, et ComfyUI n'est pas une dépendance du site (playbook P2).

Corrections apportées au script du pack d'origine
-------------------------------------------------
1. `hash()` de Python est randomisé à chaque exécution (PYTHONHASHSEED). Le pack
   annonçait « même seed = même style » alors que ses seeds changeaient à chaque
   lancement. On utilise `zlib.crc32`, stable entre exécutions, machines et versions.
2. `json.loads(json.dumps(x))` devient `copy.deepcopy(x)` : même effet, sans passer
   par une sérialisation qui perd les types non JSON.
3. Chaque sortie porte un manifeste avec sa clé, sa seed et son empreinte, pour que la
   reproductibilité soit vérifiable et non pas affirmée.

Usage :
    python3 comfyui_client.py --workflow workflows/decor.json --plan plan-assets.json
    python3 comfyui_client.py --verifier   # rejoue les seeds et compare les empreintes
"""

from __future__ import annotations

import argparse
import copy
import hashlib
import json
import time
import urllib.error
import urllib.parse
import urllib.request
import zlib
from pathlib import Path

RACINE = Path(__file__).resolve().parent
SORTIE = RACINE / "out"
COMFY_DEFAUT = "http://127.0.0.1:8188"

# Réutilisable tel quel : c'est le meilleur apport du pack d'origine.
PROMPT_NEGATIF = (
    "text, watermark, signature, logo, blurry, low quality, jpeg artifacts, "
    "deformed, extra limbs, bad anatomy, photorealistic human faces, "
    "cartoon glare, oversaturated, plastic look, frame, border"
)


def seed_deterministe(cle: str) -> int:
    """
    Seed stable pour une clé donnée.

    `hash()` ne convient pas : il est randomisé par processus depuis Python 3.3, ce qui
    rend les seeds du pack d'origine non reproductibles malgré son argumentaire.
    CRC32 est stable partout et suffit largement — on cherche de la reproductibilité,
    pas de la résistance cryptographique.
    """
    return zlib.crc32(cle.encode("utf-8")) & 0xFFFFFFFF


def empreinte(chemin: Path) -> str:
    """SHA-256 d'un fichier — sert à prouver que deux exécutions produisent la même image."""
    h = hashlib.sha256()
    with chemin.open("rb") as f:
        for bloc in iter(lambda: f.read(65536), b""):
            h.update(bloc)
    return h.hexdigest()


def poster(serveur: str, workflow: dict, client_id: str) -> str:
    """Envoie un workflow à ComfyUI et renvoie l'identifiant de la tâche."""
    corps = json.dumps({"prompt": workflow, "client_id": client_id}).encode("utf-8")
    requete = urllib.request.Request(
        urllib.parse.urljoin(serveur, "/prompt"),
        data=corps,
        headers={"Content-Type": "application/json"},
    )
    with urllib.request.urlopen(requete, timeout=30) as reponse:
        return json.load(reponse)["prompt_id"]


def attendre(serveur: str, prompt_id: str, delai_max: int = 600) -> dict:
    """Interroge l'historique jusqu'à ce que la tâche soit terminée."""
    url = urllib.parse.urljoin(serveur, f"/history/{prompt_id}")
    debut = time.time()
    while time.time() - debut < delai_max:
        try:
            with urllib.request.urlopen(url, timeout=15) as reponse:
                historique = json.load(reponse)
            if prompt_id in historique:
                return historique[prompt_id]
        except urllib.error.URLError:
            pass
        time.sleep(2)
    raise TimeoutError(f"ComfyUI n'a pas rendu {prompt_id} en {delai_max}s")


def preparer(workflow_base: dict, noeud_seed: str, seed: int, noeud_prompt: str, prompt: str) -> dict:
    """
    Copie profonde du workflow avant modification.

    `copy.deepcopy` plutôt que `json.loads(json.dumps(...))` : même résultat, sans
    aller-retour par une sérialisation qui écraserait tout type non JSON.
    """
    workflow = copy.deepcopy(workflow_base)
    workflow[noeud_seed]["inputs"]["seed"] = seed
    workflow[noeud_prompt]["inputs"]["text"] = prompt
    return workflow


def generer(args: argparse.Namespace) -> int:
    workflow_base = json.loads(Path(args.workflow).read_text(encoding="utf-8"))
    plan = json.loads(Path(args.plan).read_text(encoding="utf-8"))
    SORTIE.mkdir(parents=True, exist_ok=True)

    manifeste = []
    client_id = "ki-fwi-assets"

    for asset in plan["assets"]:
        cle = f"{asset['id']}:{asset['type']}"
        seed = seed_deterministe(cle)
        prompt = f"{asset['prompt']}. Negative: {PROMPT_NEGATIF}"

        print(f"→ {cle} (seed {seed})", flush=True)

        if args.sec:
            manifeste.append({"cle": cle, "seed": seed, "fichier": None, "sha256": None})
            continue

        workflow = preparer(workflow_base, plan["noeud_seed"], seed, plan["noeud_prompt"], prompt)
        prompt_id = poster(args.serveur, workflow, client_id)
        resultat = attendre(args.serveur, prompt_id)

        fichiers = [
            img["filename"]
            for sortie in resultat.get("outputs", {}).values()
            for img in sortie.get("images", [])
        ]
        for nom in fichiers:
            chemin = SORTIE / nom
            manifeste.append(
                {
                    "cle": cle,
                    "seed": seed,
                    "fichier": nom,
                    "sha256": empreinte(chemin) if chemin.exists() else None,
                    # Tout asset généré entre dans le dépôt non validé, sans exception.
                    "filiere": "B",
                    "type_media": "image_generee",
                    "valide": False,
                }
            )

    chemin_manifeste = SORTIE / "manifeste.json"
    chemin_manifeste.write_text(json.dumps(manifeste, indent=2, ensure_ascii=False), encoding="utf-8")
    print(f"\n✔ {len(manifeste)} entrées → {chemin_manifeste}")
    print("  Tous les assets sortent en valide: false (ADR 0004).")
    return 0


def verifier(_args: argparse.Namespace) -> int:
    """
    Contrôle que les seeds sont bien déterministes.

    C'est le test du défaut n°1 du pack d'origine : deux appels doivent donner la même
    valeur, y compris dans deux processus différents.
    """
    echantillon = ["kowosol:fond", "zikak:sprite", "mansiniye:badge", "lakou:titre"]
    attendu = {cle: seed_deterministe(cle) for cle in echantillon}
    rejoue = {cle: seed_deterministe(cle) for cle in echantillon}

    for cle in echantillon:
        etat = "✔" if attendu[cle] == rejoue[cle] else "✘"
        print(f"{etat} {cle:24s} {attendu[cle]}")

    stable = attendu == rejoue
    print("\n✔ Seeds déterministes." if stable else "\n✘ Seeds instables.")
    print("  Rejouer ce script dans un autre processus doit donner exactement ces valeurs.")
    return 0 if stable else 1


def main() -> int:
    parseur = argparse.ArgumentParser(description="Génération des assets de filière B (décor)")
    parseur.add_argument("--workflow", help="workflow ComfyUI au format API")
    parseur.add_argument("--plan", help="plan des assets à générer")
    parseur.add_argument("--serveur", default=COMFY_DEFAUT)
    parseur.add_argument("--sec", action="store_true", help="calcule les seeds sans appeler ComfyUI")
    parseur.add_argument("--verifier", action="store_true", help="teste le déterminisme des seeds")
    args = parseur.parse_args()

    if args.verifier:
        return verifier(args)
    if not args.workflow or not args.plan:
        parseur.error("--workflow et --plan sont requis (ou --verifier)")
    return generer(args)


if __name__ == "__main__":
    raise SystemExit(main())