fn_registry/python/functions/ml/comfyui_validate_workflow.py

"""Valida un workflow ComfyUI contra el catalogo /object_info del servidor.

Cruza los class_type del workflow contra los nodos disponibles y los nombres de
modelos (checkpoints, loras, vae, controlnet, upscale) contra los combos
enumerados de cada nodo. Asi se detectan nodos o modelos faltantes ANTES de
encolar (POST /prompt), evitando un HTTP 400 del servidor.

Compone comfyui_object_info (no reimplementa la consulta HTTP).

Impura: red (HTTP GET via comfyui_object_info). Solo stdlib.
"""
import os
import sys

_THIS_DIR = os.path.dirname(os.path.abspath(__file__))
if _THIS_DIR not in sys.path:
    sys.path.insert(0, _THIS_DIR)

from comfyui_object_info import comfyui_object_info  # noqa: E402

# inputs cuyo valor es el nombre (string) de un asset/modelo en disco
_MODEL_INPUTS = {
    "ckpt_name",
    "lora_name",
    "vae_name",
    "control_net_name",
    "model_name",
    "unet_name",
    "clip_name",
    "style_model_name",
    "gligen_name",
}


def comfyui_validate_workflow(
    workflow: dict,
    server: str = "127.0.0.1:8188",
    timeout: float = 30.0,
) -> dict:
    """Valida un workflow API format contra un servidor ComfyUI vivo.

    Args:
        workflow: dict en API format ({node_id: {class_type, inputs}}).
        server: host:port del servidor ComfyUI (sin esquema).
        timeout: timeout de la consulta HTTP en segundos.

    Returns:
        dict {ok, valid, missing_nodes, missing_models, error}:
          - ok: la validacion se pudo ejecutar (el servidor respondio).
          - valid: el workflow no tiene nodos ni modelos faltantes.
          - missing_nodes: lista de class_type ausentes en el servidor.
          - missing_models: lista de {node, input, value} con valores de modelo
            no presentes en el combo enumerado correspondiente.
          - error: mensaje si no se pudo consultar el servidor (ok=False).
    """
    try:
        obj_info = comfyui_object_info(server=server, timeout=timeout)
    except RuntimeError as exc:
        return {
            "ok": False,
            "valid": False,
            "missing_nodes": [],
            "missing_models": [],
            "error": str(exc),
        }

    missing_nodes = []
    missing_models = []
    for nid, node in workflow.items():
        if not isinstance(node, dict):
            continue
        ctype = node.get("class_type")
        if ctype not in obj_info:
            missing_nodes.append(ctype)
            continue
        spec = obj_info[ctype].get("input", {})
        allowed = {}
        for section in ("required", "optional"):
            for name, decl in (spec.get(section) or {}).items():
                if isinstance(decl, list) and decl and isinstance(decl[0], list):
                    allowed[name] = set(decl[0])
        for in_name, val in node.get("inputs", {}).items():
            if in_name in _MODEL_INPUTS and isinstance(val, str):
                opts = allowed.get(in_name)
                if opts is not None and val not in opts:
                    missing_models.append(
                        {"node": nid, "input": in_name, "value": val}
                    )

    # dedup missing_nodes preservando orden
    seen = set()
    missing_nodes = [c for c in missing_nodes if not (c in seen or seen.add(c))]
    valid = not missing_nodes and not missing_models
    return {
        "ok": True,
        "valid": valid,
        "missing_nodes": missing_nodes,
        "missing_models": missing_models,
        "error": "",
    }


if __name__ == "__main__":
    import json

    sys.path.insert(0, _THIS_DIR)
    from comfyui_build_txt2img_workflow import comfyui_build_txt2img_workflow

    wf = comfyui_build_txt2img_workflow("IMG_dreamshaper_8.safetensors", "a cat")
    print(json.dumps(comfyui_validate_workflow(wf), indent=2))