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 vider_cache(serveur: str) -> None:
    """
    Vide le cache d'exécution de ComfyUI.

    INDISPENSABLE avant tout test de reproductibilité. ComfyUI met en cache le résultat
    d'un graphe identique : relancer le même workflow ne recalcule rien et renvoie la
    sortie précédente. Comparer deux « exécutions » sans vider le cache ne prouve donc
    rien du tout — seulement que le cache fonctionne.

    Constaté en session 3 : la deuxième exécution n'avait écrit aucun fichier dans le
    dossier de sortie de ComfyUI. Le hasard a voulu que la conclusion soit juste ; la
    méthode, elle, était fausse.
    """
    corps = json.dumps({"unload_models": True, "free_memory": True}).encode("utf-8")
    requete = urllib.request.Request(
        urllib.parse.urljoin(serveur, "/free"),
        data=corps,
        headers={"Content-Type": "application/json"},
    )
    with urllib.request.urlopen(requete, timeout=30):
        pass


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, plan: dict, seed: int, 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.

    Le prompt négatif va dans SON PROPRE nœud. Le pack d'origine le concaténait au
    prompt positif — sur SDXL, cela revient à demander au modèle de produire ce qu'on
    voulait exclure.
    """
    workflow = copy.deepcopy(workflow_base)
    workflow[plan["noeud_seed"]]["inputs"]["seed"] = seed
    workflow[plan["noeud_prompt"]]["inputs"]["text"] = prompt
    if plan.get("noeud_prompt_negatif"):
        workflow[plan["noeud_prompt_negatif"]]["inputs"]["text"] = PROMPT_NEGATIF
    return workflow


def recuperer(serveur: str, image: dict, destination: Path) -> None:
    """
    Rapatrie une image depuis ComfyUI.

    Sans cela, les fichiers restent dans le dossier de sortie de ComfyUI et le manifeste
    référence des chemins qui n'existent pas côté projet.
    """
    parametres = urllib.parse.urlencode(
        {
            "filename": image["filename"],
            "subfolder": image.get("subfolder", ""),
            "type": image.get("type", "output"),
        }
    )
    url = urllib.parse.urljoin(serveur, f"/view?{parametres}")
    with urllib.request.urlopen(url, timeout=60) as reponse:
        destination.write_bytes(reponse.read())


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 = asset["prompt"]

        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, seed, prompt)
        prompt_id = poster(args.serveur, workflow, client_id)
        resultat = attendre(args.serveur, prompt_id)

        images = [
            img
            for sortie in resultat.get("outputs", {}).values()
            for img in sortie.get("images", [])
            if img.get("type") == "output"
        ]
        for image in images:
            nom = f"{asset['id']}-{asset['type']}-{seed}.png"
            chemin = SORTIE / nom
            recuperer(args.serveur, image, chemin)
            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 reproductibilite(args: argparse.Namespace) -> int:
    """
    Test de reproductibilité RÉEL : génère, vide le cache, régénère, compare.

    Sans le vidage intermédiaire, ce test ne mesure que le cache de ComfyUI.
    """
    args_gen = argparse.Namespace(**vars(args))
    args_gen.sec = False

    print("— première génération")
    generer(args_gen)
    premier = json.loads((SORTIE / "manifeste.json").read_text(encoding="utf-8"))

    print("\n— vidage du cache d'exécution")
    vider_cache(args.serveur)

    print("\n— seconde génération, cache vidé")
    generer(args_gen)
    second = json.loads((SORTIE / "manifeste.json").read_text(encoding="utf-8"))

    identiques = True
    print()
    for a, b in zip(premier, second):
        egal = a["sha256"] == b["sha256"]
        identiques &= egal
        print(f"{'✔' if egal else '✘'} {a['cle']} — seed {a['seed']} — {a['sha256'][:24]}")

    print(
        "\n✔ Sorties bit-identiques après vidage du cache : la reproductibilité est réelle."
        if identiques
        else "\n✘ Sorties divergentes : la reproductibilité n'est PAS acquise."
    )
    return 0 if identiques else 1


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")
    parseur.add_argument(
        "--reproductibilite",
        action="store_true",
        help="génère deux fois en vidant le cache entre les deux, et compare les empreintes",
    )
    args = parseur.parse_args()

    if args.verifier:
        return verifier(args)
    if args.reproductibilite:
        if not args.workflow or not args.plan:
            parseur.error("--workflow et --plan sont requis pour --reproductibilite")
        return reproductibilite(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())