malkuthro
1732030175
* fix(setups): resolve switch-selected run results * fix(setups): harden switch result reporting * fix(setups): reject stale execution map writers
1997 lines
73 KiB
Python
1997 lines
73 KiB
Python
"""Published workflow setup registry boundary.
|
|
|
|
This module owns the server-side shape for externally callable Koolook
|
|
setups. Callers use this boundary instead of reading sidebar workflow or
|
|
snapshot storage directly.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
from copy import deepcopy
|
|
from dataclasses import dataclass
|
|
from datetime import datetime, timezone
|
|
import json
|
|
import math
|
|
import os
|
|
from pathlib import Path
|
|
import re
|
|
import tempfile
|
|
from typing import Any
|
|
|
|
|
|
DEFAULT_SETUPS_SUBDIR = "koolook-published-setups"
|
|
DEFAULT_SETUPS_FILENAME = "setups.json"
|
|
SAMPLE_SETUPS_PATH = (
|
|
Path(__file__).resolve().parent / "web" / "published_setups_sample.json"
|
|
)
|
|
SUPPORTED_SCHEMA_VERSION = 1
|
|
SETUP_ID_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_.-]*$")
|
|
VALIDATION_STATUSES = {"valid", "draft", "invalid"}
|
|
PUBLISH_INPUT_CLASS = "Koolook_PublishInput"
|
|
PUBLISH_OUTPUT_CLASS = "Koolook_PublishOutput"
|
|
PUBLISH_RESULT_CLASS = "Koolook_PublishResult"
|
|
PUBLISH_ROUTER_CLASS = "Koolook_PublishRouter"
|
|
VISUAL_ONLY_API_NODE_TYPES = {"Label (rgthree)", "Note"}
|
|
SINK_ONLY_API_NODE_TYPES = {"SetNode"}
|
|
PUBLISH_INPUT_FIELDS = (
|
|
("sequence_folder", "Sequence folder", True),
|
|
("qt_file", "QT file", True),
|
|
("single_file", "Single file", True),
|
|
("prompt", "Prompt", False),
|
|
)
|
|
PUBLISH_INPUT_MODES = (
|
|
(0, "EXR", "sequence_folder"),
|
|
(1, "QT", "qt_file"),
|
|
(2, "Img", "single_file"),
|
|
(3, "Prompt", "prompt"),
|
|
)
|
|
PUBLISH_OUTPUT_FIELDS = (
|
|
("folder", "Output folder", True),
|
|
("name", "Output name", True),
|
|
("version", "Version", True),
|
|
)
|
|
PUBLISH_RESULT_FIELDS = (
|
|
("result", "Result", True),
|
|
)
|
|
WIDGET_ONLY_INPUTS_BY_CLASS = {
|
|
"Text Multiline": ("text",),
|
|
"Koolook_PublishInput": ("mode", "sequence_folder", "qt_file", "single_file", "prompt"),
|
|
"Koolook_PublishOutput": ("folder", "name", "version"),
|
|
"Koolook_PublishResult": ("result",),
|
|
"Koolook_PublishRouter": (),
|
|
"EasyAIPipeline": (
|
|
"shot_duration",
|
|
"seed_value",
|
|
"instruction",
|
|
"base_directory_path",
|
|
"extension",
|
|
"shot_name",
|
|
"ai_method",
|
|
"version",
|
|
"disable_versioning",
|
|
"enable_overwrite",
|
|
"no_subfolders",
|
|
),
|
|
}
|
|
WIDGET_ONLY_INPUT_DEFAULTS = {
|
|
"EasyAIPipeline": {
|
|
"no_subfolders": False,
|
|
},
|
|
}
|
|
|
|
REQUIRED_SETUP_KEYS = (
|
|
"schemaVersion",
|
|
"id",
|
|
"version",
|
|
"updatedAt",
|
|
"metadata",
|
|
"visualGraph",
|
|
"apiPrompt",
|
|
"inputContract",
|
|
"outputContract",
|
|
"source",
|
|
"validation",
|
|
)
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class ValidationResult:
|
|
valid: bool
|
|
setup: dict[str, Any]
|
|
diagnostics: list[str]
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class ApiPromptConversionResult:
|
|
api_prompt: dict[str, Any] | None
|
|
diagnostics: list[str]
|
|
|
|
|
|
class StaticSetupStorage:
|
|
"""In-memory setup storage adapter for tests and seeded fixtures."""
|
|
|
|
def __init__(self, records: list[dict[str, Any]]) -> None:
|
|
self._records = records
|
|
self.diagnostics: list[str] = []
|
|
|
|
def load_setups(self) -> list[dict[str, Any]]:
|
|
return list(self._records)
|
|
|
|
def save_setups(self, records: list[dict[str, Any]]) -> None:
|
|
self._records = list(records)
|
|
|
|
|
|
class FileSetupStorage:
|
|
"""JSON-file storage adapter for published setup records."""
|
|
|
|
def __init__(self, path: Path, fallback_path: Path | None = None) -> None:
|
|
self.path = path
|
|
self.fallback_path = fallback_path
|
|
self.diagnostics: list[str] = []
|
|
|
|
def load_setups(self) -> list[dict[str, Any]]:
|
|
self.diagnostics = []
|
|
path = self.path
|
|
if not path.is_file() and self.fallback_path is not None:
|
|
path = self.fallback_path
|
|
self.diagnostics.append(
|
|
f"{self.path}: missing; using bundled sample {self.fallback_path}"
|
|
)
|
|
if not path.is_file():
|
|
return []
|
|
try:
|
|
raw = json.loads(path.read_text(encoding="utf-8"))
|
|
except (OSError, ValueError) as exc:
|
|
self.diagnostics.append(f"{path}: could not read published setups: {exc}")
|
|
return []
|
|
if isinstance(raw, dict):
|
|
raw = raw.get("setups", [])
|
|
if not isinstance(raw, list):
|
|
self.diagnostics.append(
|
|
f"{path}: published setup storage must be a list or {{setups: [...]}}"
|
|
)
|
|
return []
|
|
return [item for item in raw if isinstance(item, dict)]
|
|
|
|
def load_primary_setups(self) -> list[dict[str, Any]]:
|
|
self.diagnostics = []
|
|
if not self.path.is_file():
|
|
return []
|
|
try:
|
|
raw = json.loads(self.path.read_text(encoding="utf-8"))
|
|
except (OSError, ValueError) as exc:
|
|
self.diagnostics.append(f"{self.path}: could not read published setups: {exc}")
|
|
return []
|
|
if isinstance(raw, dict):
|
|
raw = raw.get("setups", [])
|
|
if not isinstance(raw, list):
|
|
self.diagnostics.append(
|
|
f"{self.path}: published setup storage must be a list or {{setups: [...]}}"
|
|
)
|
|
return []
|
|
return [item for item in raw if isinstance(item, dict)]
|
|
|
|
def save_setups(self, records: list[dict[str, Any]]) -> None:
|
|
self.diagnostics = []
|
|
self.path.parent.mkdir(parents=True, exist_ok=True)
|
|
payload = {"setups": records}
|
|
tmp_fd, tmp_name = tempfile.mkstemp(
|
|
prefix=f"{self.path.stem}.",
|
|
suffix=".tmp",
|
|
dir=str(self.path.parent),
|
|
)
|
|
try:
|
|
with os.fdopen(tmp_fd, "w", encoding="utf-8") as f:
|
|
json.dump(payload, f, indent=2)
|
|
f.write("\n")
|
|
os.replace(tmp_name, self.path)
|
|
except OSError:
|
|
try:
|
|
if os.path.exists(tmp_name):
|
|
os.unlink(tmp_name)
|
|
except OSError:
|
|
pass
|
|
raise
|
|
|
|
|
|
class PublishedSetupRegistry:
|
|
"""Read-only registry facade for published setup catalog consumers."""
|
|
|
|
def __init__(self, storage: StaticSetupStorage | FileSetupStorage) -> None:
|
|
self._storage = storage
|
|
self.diagnostics: list[str] = []
|
|
|
|
def _valid_setups(self) -> list[dict[str, Any]]:
|
|
self.diagnostics = []
|
|
out: list[dict[str, Any]] = []
|
|
for raw in self._storage.load_setups():
|
|
result = validate_setup(raw)
|
|
if result.valid:
|
|
out.append(result.setup)
|
|
continue
|
|
setup_id = raw.get("id", "<unknown>") if isinstance(raw, dict) else "<unknown>"
|
|
for diagnostic in result.diagnostics:
|
|
self.diagnostics.append(f"{setup_id}: {diagnostic}")
|
|
self.diagnostics.extend(getattr(self._storage, "diagnostics", []))
|
|
out.sort(key=lambda setup: str(setup["metadata"]["title"]).lower())
|
|
return out
|
|
|
|
def listSetups(self) -> list[dict[str, Any]]:
|
|
return [_catalog_summary(setup) for setup in self._valid_setups()]
|
|
|
|
def getSetup(self, setup_id: str) -> dict[str, Any] | None:
|
|
for setup in self._valid_setups():
|
|
if setup["id"] == setup_id:
|
|
return setup
|
|
return None
|
|
|
|
def publishSetup(
|
|
self,
|
|
*,
|
|
visualGraph: Any,
|
|
metadata: Any,
|
|
inputContract: Any,
|
|
outputContract: Any,
|
|
source: Any,
|
|
apiPrompt: Any = None,
|
|
) -> ValidationResult:
|
|
diagnostics = []
|
|
for name, value in (
|
|
("visualGraph", visualGraph),
|
|
("metadata", metadata),
|
|
("inputContract", inputContract),
|
|
("outputContract", outputContract),
|
|
("source", source),
|
|
):
|
|
if not isinstance(value, dict):
|
|
diagnostics.append(f"{name} must be an object")
|
|
if diagnostics:
|
|
return ValidationResult(False, {}, diagnostics)
|
|
|
|
if apiPrompt is None:
|
|
conversion = _convert_visual_graph_to_api_prompt(visualGraph)
|
|
if conversion.diagnostics:
|
|
return ValidationResult(False, {}, conversion.diagnostics)
|
|
api_prompt = conversion.api_prompt
|
|
elif isinstance(apiPrompt, dict):
|
|
api_prompt = deepcopy(apiPrompt)
|
|
else:
|
|
return ValidationResult(False, {}, ["apiPrompt must be an object when present"])
|
|
surface_geometry_diagnostics = []
|
|
if _uses_group_authored_surface(inputContract, outputContract):
|
|
surface_geometry_diagnostics = _validate_setup_surface_geometry(visualGraph)
|
|
if surface_geometry_diagnostics:
|
|
return ValidationResult(False, {}, surface_geometry_diagnostics)
|
|
setup = _build_draft_setup(
|
|
visualGraph=visualGraph,
|
|
apiPrompt=api_prompt,
|
|
metadata=metadata,
|
|
inputContract=inputContract,
|
|
outputContract=outputContract,
|
|
source=source,
|
|
)
|
|
result = validate_setup(setup)
|
|
diagnostics = list(result.diagnostics)
|
|
if result.valid:
|
|
setup = result.setup
|
|
diagnostics.extend(_validate_contract_targets(visualGraph, inputContract))
|
|
diagnostics.extend(_validate_api_prompt_contract_targets(setup.get("apiPrompt"), inputContract))
|
|
diagnostics.extend(_validate_setup_surface(setup, inputContract, outputContract))
|
|
if diagnostics:
|
|
return ValidationResult(False, {}, diagnostics)
|
|
|
|
load_for_write = getattr(self._storage, "load_primary_setups", self._storage.load_setups)
|
|
records = [
|
|
existing
|
|
for existing in load_for_write()
|
|
if existing.get("id") != setup["id"]
|
|
]
|
|
records.append(setup)
|
|
self._storage.save_setups(records)
|
|
return ValidationResult(True, setup, [])
|
|
|
|
|
|
def validate_setup(value: Any) -> ValidationResult:
|
|
"""Validate one published setup object through the public schema boundary."""
|
|
diagnostics: list[str] = []
|
|
if not isinstance(value, dict):
|
|
return ValidationResult(False, {}, ["setup must be a JSON object"])
|
|
setup = deepcopy(value)
|
|
|
|
for key in REQUIRED_SETUP_KEYS:
|
|
if key not in value:
|
|
diagnostics.append(f"missing required field: {key}")
|
|
|
|
if diagnostics:
|
|
return ValidationResult(False, {}, diagnostics)
|
|
|
|
_validate_scalar(setup, "id", str, diagnostics)
|
|
if isinstance(setup.get("id"), str) and not SETUP_ID_RE.match(setup["id"]):
|
|
diagnostics.append("id must be stable URL-safe text")
|
|
_validate_scalar(setup, "schemaVersion", int, diagnostics)
|
|
if setup.get("schemaVersion") != SUPPORTED_SCHEMA_VERSION:
|
|
diagnostics.append(
|
|
f"unsupported schemaVersion: {setup.get('schemaVersion')}; "
|
|
f"expected {SUPPORTED_SCHEMA_VERSION}"
|
|
)
|
|
_validate_scalar(setup, "version", (int, str), diagnostics)
|
|
_validate_iso_timestamp(setup.get("updatedAt"), diagnostics)
|
|
_validate_metadata(setup.get("metadata"), diagnostics)
|
|
_validate_mapping(setup.get("visualGraph"), "visualGraph", diagnostics)
|
|
if setup.get("apiPrompt") is not None:
|
|
_validate_mapping(setup.get("apiPrompt"), "apiPrompt", diagnostics)
|
|
if isinstance(setup.get("visualGraph"), dict) and isinstance(setup.get("apiPrompt"), dict):
|
|
conversion = None
|
|
if _api_prompt_needs_visual_normalization(setup["apiPrompt"], setup["visualGraph"]):
|
|
conversion = _convert_visual_graph_to_api_prompt(setup["visualGraph"])
|
|
if conversion is not None and conversion.diagnostics:
|
|
diagnostics.extend(conversion.diagnostics)
|
|
elif conversion is not None and conversion.api_prompt is not None:
|
|
setup["apiPrompt"] = conversion.api_prompt
|
|
_validate_contract(setup.get("inputContract"), "inputContract", "inputs", diagnostics)
|
|
_validate_contract(setup.get("outputContract"), "outputContract", "outputs", diagnostics)
|
|
_validate_persisted_setup_surface(setup, diagnostics)
|
|
_validate_source(setup.get("source"), diagnostics)
|
|
_validate_validation(setup.get("validation"), diagnostics)
|
|
if (
|
|
isinstance(setup.get("validation"), dict)
|
|
and setup["validation"].get("status") == "valid"
|
|
and setup.get("apiPrompt") is None
|
|
):
|
|
diagnostics.append("validation.status valid requires apiPrompt")
|
|
|
|
return ValidationResult(
|
|
valid=not diagnostics,
|
|
setup=setup if not diagnostics else {},
|
|
diagnostics=diagnostics,
|
|
)
|
|
|
|
|
|
def _build_draft_setup(
|
|
*,
|
|
visualGraph: dict[str, Any],
|
|
apiPrompt: dict[str, Any] | None,
|
|
metadata: dict[str, Any],
|
|
inputContract: dict[str, Any],
|
|
outputContract: dict[str, Any],
|
|
source: dict[str, Any],
|
|
) -> dict[str, Any]:
|
|
title = metadata.get("title", "")
|
|
setup_id = metadata.get("id", "")
|
|
setup_surface = _infer_setup_surface(visualGraph)
|
|
setup = {
|
|
"schemaVersion": SUPPORTED_SCHEMA_VERSION,
|
|
"id": setup_id,
|
|
"version": 1,
|
|
"updatedAt": datetime.now(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z"),
|
|
"metadata": {
|
|
"title": title,
|
|
"description": metadata.get("description", ""),
|
|
"category": metadata.get("category", ""),
|
|
"tags": metadata.get("tags", []),
|
|
"previewImage": metadata.get("previewImage", ""),
|
|
},
|
|
"visualGraph": visualGraph,
|
|
"apiPrompt": apiPrompt,
|
|
"inputContract": inputContract,
|
|
"outputContract": outputContract,
|
|
"setupSurface": setup_surface,
|
|
"source": source,
|
|
"validation": {
|
|
"status": "valid" if apiPrompt is not None else "draft",
|
|
"diagnostics": [] if apiPrompt is not None else ["API prompt conversion pending."],
|
|
},
|
|
}
|
|
execution_map = _build_execution_map(setup)
|
|
if execution_map is not None:
|
|
setup["executionMap"] = execution_map
|
|
return setup
|
|
|
|
|
|
def _infer_setup_surface(visual_graph: dict[str, Any]) -> dict[str, Any]:
|
|
return {
|
|
"sourceInputs": _nodes_in_named_groups(visual_graph, "Koolook Input"),
|
|
"outputs": _nodes_in_named_groups(visual_graph, "Koolook Output"),
|
|
"controls": [],
|
|
"app": _infer_app_surface(visual_graph),
|
|
}
|
|
|
|
|
|
def _infer_app_surface(visual_graph: dict[str, Any]) -> dict[str, Any]:
|
|
input_nodes = _nodes_in_group(visual_graph, "Koolook Input")
|
|
output_nodes = _nodes_in_group(visual_graph, "Koolook Output")
|
|
input_node = _first_node_of_type(input_nodes, PUBLISH_INPUT_CLASS)
|
|
output_node = _first_node_of_type(output_nodes, PUBLISH_OUTPUT_CLASS)
|
|
result_node = _first_node_of_type(output_nodes, PUBLISH_RESULT_CLASS)
|
|
inputs = _publish_input_fields(input_node)
|
|
outputs = _publish_output_fields(output_node)
|
|
app: dict[str, Any] = {
|
|
"inputs": inputs,
|
|
"outputs": outputs,
|
|
"results": _publish_result_fields(result_node),
|
|
}
|
|
if input_node is not None:
|
|
app["switch"] = _publish_input_switch(input_node, inputs)
|
|
return app
|
|
|
|
|
|
def _first_node_of_type(
|
|
nodes: list[dict[str, Any]],
|
|
class_type: str,
|
|
) -> dict[str, Any] | None:
|
|
for node in nodes:
|
|
if node.get("type") == class_type:
|
|
return node
|
|
return None
|
|
|
|
|
|
def _publish_input_fields(node: dict[str, Any] | None) -> list[dict[str, Any]]:
|
|
if node is None:
|
|
return []
|
|
return [
|
|
{
|
|
"key": key,
|
|
"label": label,
|
|
"visible": visible,
|
|
"target": {
|
|
"node": str(node.get("id")),
|
|
"input": key,
|
|
},
|
|
"default": _app_widget_value(node, key),
|
|
}
|
|
for key, label, visible in PUBLISH_INPUT_FIELDS
|
|
]
|
|
|
|
|
|
def _publish_output_fields(node: dict[str, Any] | None) -> list[dict[str, Any]]:
|
|
if node is None:
|
|
return []
|
|
return [
|
|
{
|
|
"key": key,
|
|
"label": label,
|
|
"visible": visible,
|
|
"target": {
|
|
"node": str(node.get("id")),
|
|
"input": key,
|
|
},
|
|
"default": _app_widget_value(node, key),
|
|
}
|
|
for key, label, visible in PUBLISH_OUTPUT_FIELDS
|
|
]
|
|
|
|
|
|
def _publish_result_fields(node: dict[str, Any] | None) -> list[dict[str, Any]]:
|
|
if node is None:
|
|
return []
|
|
return [
|
|
{
|
|
"key": key,
|
|
"label": label,
|
|
"visible": visible,
|
|
"target": {
|
|
"node": str(node.get("id")),
|
|
"input": key,
|
|
},
|
|
"default": _app_widget_value(node, key),
|
|
}
|
|
for key, label, visible in PUBLISH_RESULT_FIELDS
|
|
]
|
|
|
|
|
|
def _publish_input_switch(
|
|
node: dict[str, Any],
|
|
inputs: list[dict[str, Any]],
|
|
) -> dict[str, Any]:
|
|
inputs_by_key = {item["key"]: item for item in inputs}
|
|
return {
|
|
"key": "switch",
|
|
"label": "Input type",
|
|
"visible": True,
|
|
"target": {"node": str(node.get("id")), "input": "mode"},
|
|
"default": _publish_input_mode_index(_app_widget_value(node, "mode")),
|
|
"options": [
|
|
{
|
|
"value": value,
|
|
"label": label,
|
|
"visible": bool(inputs_by_key[input_key].get("visible")),
|
|
"input": input_key,
|
|
}
|
|
for value, label, input_key in PUBLISH_INPUT_MODES
|
|
],
|
|
}
|
|
|
|
|
|
def _publish_input_mode_index(value: Any) -> int:
|
|
if isinstance(value, int):
|
|
return value
|
|
for mode_value, label, _input_key in PUBLISH_INPUT_MODES:
|
|
if str(value).strip().lower() == label.lower():
|
|
return mode_value
|
|
return 2
|
|
|
|
|
|
def _build_execution_map(setup: dict[str, Any]) -> dict[str, Any] | None:
|
|
api_prompt = setup.get("apiPrompt")
|
|
if not isinstance(api_prompt, dict):
|
|
return None
|
|
app = setup.get("setupSurface", {}).get("app", {})
|
|
switch = app.get("switch") if isinstance(app, dict) else None
|
|
switch_target = switch.get("target") if isinstance(switch, dict) else None
|
|
if not isinstance(switch_target, dict):
|
|
return None
|
|
output_surface_ids = _setup_output_surface_node_ids(setup)
|
|
if not output_surface_ids:
|
|
return None
|
|
routers: list[dict[str, Any]] = []
|
|
for node_id, node in api_prompt.items():
|
|
if not isinstance(node, dict) or node.get("class_type") != PUBLISH_ROUTER_CLASS:
|
|
continue
|
|
inputs = node.get("inputs")
|
|
if not isinstance(inputs, dict):
|
|
continue
|
|
selector_ref = inputs.get("selector")
|
|
payload_ref = inputs.get("payload")
|
|
if not _api_ref_matches_target_output(selector_ref, switch_target, api_prompt, "switch"):
|
|
continue
|
|
branches: dict[str, dict[str, Any]] = {}
|
|
for value, label, _input_key in PUBLISH_INPUT_MODES:
|
|
branches[str(value)] = {
|
|
"label": label,
|
|
"output": value,
|
|
"writerNodes": _writer_nodes_for_router_output(
|
|
api_prompt,
|
|
str(node_id),
|
|
value,
|
|
output_surface_ids,
|
|
),
|
|
}
|
|
router: dict[str, Any] = {
|
|
"node": str(node_id),
|
|
"switchKey": str(switch.get("key", "switch")),
|
|
"selector": {"node": str(selector_ref[0]), "output": int(selector_ref[1])},
|
|
"branches": branches,
|
|
}
|
|
if _is_api_ref(payload_ref):
|
|
router["payload"] = {"node": str(payload_ref[0]), "output": int(payload_ref[1])}
|
|
routers.append(router)
|
|
if not routers:
|
|
return None
|
|
routers.sort(key=lambda item: item["node"])
|
|
return {"version": 1, "routers": routers}
|
|
|
|
|
|
def _writer_nodes_for_router_output(
|
|
api_prompt: dict[str, Any],
|
|
router_node_id: str,
|
|
router_output: int,
|
|
output_surface_ids: set[str],
|
|
) -> list[str]:
|
|
children = _prompt_children_by_ref(api_prompt)
|
|
out: list[str] = []
|
|
seen: set[str] = set()
|
|
pending = list(children.get((router_node_id, router_output), []))
|
|
while pending:
|
|
node_id = pending.pop(0)
|
|
if node_id in seen:
|
|
continue
|
|
seen.add(node_id)
|
|
if node_id in output_surface_ids and node_id != router_node_id:
|
|
out.append(node_id)
|
|
for child_ids in _children_for_node(children, node_id):
|
|
pending.extend(child_ids)
|
|
return sorted(dict.fromkeys(out))
|
|
|
|
|
|
def _children_for_node(
|
|
children: dict[tuple[str, int], list[str]],
|
|
node_id: str,
|
|
) -> list[list[str]]:
|
|
return [
|
|
child_ids
|
|
for (source_node_id, _source_slot), child_ids in children.items()
|
|
if source_node_id == node_id
|
|
]
|
|
|
|
|
|
def _prompt_children_by_ref(api_prompt: dict[str, Any]) -> dict[tuple[str, int], list[str]]:
|
|
children: dict[tuple[str, int], list[str]] = {}
|
|
for node_id, node in api_prompt.items():
|
|
if not isinstance(node, dict):
|
|
continue
|
|
inputs = node.get("inputs")
|
|
if not isinstance(inputs, dict):
|
|
continue
|
|
for value in inputs.values():
|
|
for ref in _api_refs(value):
|
|
children.setdefault((str(ref[0]), int(ref[1])), []).append(str(node_id))
|
|
return children
|
|
|
|
|
|
def _setup_output_surface_node_ids(setup: dict[str, Any]) -> set[str]:
|
|
surface = setup.get("setupSurface")
|
|
if not isinstance(surface, dict) or not isinstance(surface.get("outputs"), list):
|
|
return set()
|
|
out: set[str] = set()
|
|
for group in surface["outputs"]:
|
|
if not isinstance(group, dict) or not isinstance(group.get("nodes"), list):
|
|
continue
|
|
for node in group["nodes"]:
|
|
if isinstance(node, dict) and node.get("id") is not None:
|
|
out.add(str(node["id"]))
|
|
return out
|
|
|
|
|
|
def _api_ref_matches_target_output(
|
|
value: Any,
|
|
target: dict[str, Any],
|
|
api_prompt: dict[str, Any],
|
|
output_name: str,
|
|
) -> bool:
|
|
if not _is_api_ref(value):
|
|
return False
|
|
node_id = str(value[0])
|
|
if node_id != str(target.get("node")):
|
|
return False
|
|
node = api_prompt.get(node_id)
|
|
if not isinstance(node, dict):
|
|
return False
|
|
return _publish_input_output_slots(node).get(output_name) == int(value[1])
|
|
|
|
|
|
def _publish_input_output_slots(api_node: dict[str, Any]) -> dict[str, int]:
|
|
if api_node.get("class_type") != PUBLISH_INPUT_CLASS:
|
|
return {}
|
|
return {
|
|
"sequence_folder": 0,
|
|
"qt_file": 1,
|
|
"single_file": 2,
|
|
"prompt": 3,
|
|
"switch": 4,
|
|
}
|
|
|
|
|
|
def _is_api_ref(value: Any) -> bool:
|
|
return (
|
|
isinstance(value, list)
|
|
and len(value) == 2
|
|
and isinstance(value[0], (str, int))
|
|
and not isinstance(value[1], bool)
|
|
and isinstance(value[1], int)
|
|
)
|
|
|
|
|
|
def _api_refs(value: Any) -> list[list[Any]]:
|
|
if _is_api_ref(value):
|
|
return [value]
|
|
if isinstance(value, dict):
|
|
refs: list[list[Any]] = []
|
|
for child in value.values():
|
|
refs.extend(_api_refs(child))
|
|
return refs
|
|
if isinstance(value, list):
|
|
refs = []
|
|
for child in value:
|
|
refs.extend(_api_refs(child))
|
|
return refs
|
|
return []
|
|
|
|
|
|
def _app_widget_value(node: dict[str, Any], key: str) -> Any:
|
|
widget_values = node.get("widgets_values")
|
|
if isinstance(widget_values, dict):
|
|
return widget_values.get(key)
|
|
if isinstance(widget_values, list):
|
|
widget_names = WIDGET_ONLY_INPUTS_BY_CLASS.get(str(node.get("type")), ())
|
|
for index, name in enumerate(widget_names):
|
|
if name == key and index < len(widget_values):
|
|
return widget_values[index]
|
|
return None
|
|
|
|
|
|
def _validate_setup_surface(
|
|
setup: dict[str, Any],
|
|
input_contract: dict[str, Any],
|
|
output_contract: dict[str, Any],
|
|
) -> list[str]:
|
|
if not _uses_group_authored_surface(input_contract, output_contract):
|
|
return []
|
|
surface = setup.get("setupSurface")
|
|
diagnostics: list[str] = []
|
|
if not isinstance(surface, dict) or not surface.get("sourceInputs"):
|
|
diagnostics.append("setupSurface.sourceInputs requires a non-empty Koolook Input group")
|
|
if not isinstance(surface, dict) or not surface.get("outputs"):
|
|
diagnostics.append("setupSurface.outputs requires a non-empty Koolook Output group")
|
|
return diagnostics
|
|
|
|
|
|
def _validate_persisted_setup_surface(
|
|
setup: dict[str, Any],
|
|
diagnostics: list[str],
|
|
) -> None:
|
|
surface = setup.get("setupSurface")
|
|
input_contract = setup.get("inputContract")
|
|
output_contract = setup.get("outputContract")
|
|
group_authored = (
|
|
isinstance(input_contract, dict)
|
|
and isinstance(output_contract, dict)
|
|
and _uses_group_authored_surface(input_contract, output_contract)
|
|
)
|
|
if surface is None:
|
|
if group_authored:
|
|
diagnostics.append("setupSurface must be a JSON object for group-authored setups")
|
|
return
|
|
if not isinstance(surface, dict):
|
|
diagnostics.append("setupSurface must be a JSON object")
|
|
return
|
|
for key in ("sourceInputs", "outputs", "controls"):
|
|
if key in surface and not isinstance(surface[key], list):
|
|
diagnostics.append(f"setupSurface.{key} must be a list")
|
|
if group_authored:
|
|
if not isinstance(surface.get("sourceInputs"), list) or not surface["sourceInputs"]:
|
|
diagnostics.append("setupSurface.sourceInputs requires a non-empty Koolook Input group")
|
|
if not isinstance(surface.get("outputs"), list) or not surface["outputs"]:
|
|
diagnostics.append("setupSurface.outputs requires a non-empty Koolook Output group")
|
|
for key in ("sourceInputs", "outputs", "controls"):
|
|
items = surface.get(key)
|
|
if isinstance(items, list):
|
|
_validate_surface_entries(items, f"setupSurface.{key}", diagnostics)
|
|
if "app" in surface:
|
|
_validate_surface_app(
|
|
surface["app"],
|
|
"setupSurface.app",
|
|
diagnostics,
|
|
setup.get("visualGraph"),
|
|
setup.get("apiPrompt"),
|
|
)
|
|
|
|
|
|
def _validate_surface_entries(
|
|
entries: list[Any],
|
|
path: str,
|
|
diagnostics: list[str],
|
|
) -> None:
|
|
for entry_index, entry in enumerate(entries):
|
|
entry_path = f"{path}[{entry_index}]"
|
|
if not isinstance(entry, dict):
|
|
diagnostics.append(f"{entry_path} must be a JSON object")
|
|
continue
|
|
if not isinstance(entry.get("group"), str) or not entry["group"].strip():
|
|
diagnostics.append(f"{entry_path}.group must be non-empty text")
|
|
nodes = entry.get("nodes")
|
|
if not isinstance(nodes, list):
|
|
diagnostics.append(f"{entry_path}.nodes must be a list")
|
|
continue
|
|
for node_index, node in enumerate(nodes):
|
|
node_path = f"{entry_path}.nodes[{node_index}]"
|
|
if not isinstance(node, dict):
|
|
diagnostics.append(f"{node_path} must be a JSON object")
|
|
continue
|
|
for field in ("id", "type", "title"):
|
|
if not isinstance(node.get(field), str) or not node[field].strip():
|
|
diagnostics.append(f"{node_path}.{field} must be non-empty text")
|
|
|
|
|
|
def _validate_surface_app(
|
|
app: Any,
|
|
path: str,
|
|
diagnostics: list[str],
|
|
visual_graph: Any,
|
|
api_prompt: Any,
|
|
) -> None:
|
|
if not isinstance(app, dict):
|
|
diagnostics.append(f"{path} must be a JSON object")
|
|
return
|
|
node_by_id = _visual_node_index(visual_graph)
|
|
api_prompt = api_prompt if isinstance(api_prompt, dict) else None
|
|
input_keys: set[str] = set()
|
|
for key in ("inputs", "outputs", "results"):
|
|
value = app.get(key)
|
|
if value is None:
|
|
continue
|
|
if not isinstance(value, list):
|
|
diagnostics.append(f"{path}.{key} must be a list")
|
|
continue
|
|
keys = _validate_surface_app_fields(
|
|
value,
|
|
f"{path}.{key}",
|
|
diagnostics,
|
|
node_by_id,
|
|
api_prompt,
|
|
)
|
|
if key == "inputs":
|
|
input_keys = keys
|
|
if "switch" in app:
|
|
_validate_surface_app_switch(
|
|
app["switch"],
|
|
f"{path}.switch",
|
|
diagnostics,
|
|
node_by_id,
|
|
api_prompt,
|
|
input_keys,
|
|
)
|
|
|
|
|
|
def _validate_surface_app_fields(
|
|
fields: list[Any],
|
|
path: str,
|
|
diagnostics: list[str],
|
|
node_by_id: dict[str, dict[str, Any]],
|
|
api_prompt: dict[str, Any] | None,
|
|
) -> set[str]:
|
|
keys: set[str] = set()
|
|
for field_index, field in enumerate(fields):
|
|
field_path = f"{path}[{field_index}]"
|
|
if not isinstance(field, dict):
|
|
diagnostics.append(f"{field_path} must be a JSON object")
|
|
continue
|
|
_validate_non_empty_text(field.get("key"), f"{field_path}.key", diagnostics)
|
|
if isinstance(field.get("key"), str) and field["key"].strip():
|
|
keys.add(field["key"])
|
|
_validate_non_empty_text(field.get("label"), f"{field_path}.label", diagnostics)
|
|
if "visible" in field and not isinstance(field["visible"], bool):
|
|
diagnostics.append(f"{field_path}.visible must be a boolean")
|
|
_validate_surface_app_target(
|
|
field.get("target"),
|
|
f"{field_path}.target",
|
|
diagnostics,
|
|
node_by_id,
|
|
api_prompt,
|
|
)
|
|
return keys
|
|
|
|
|
|
def _validate_surface_app_switch(
|
|
switch: Any,
|
|
path: str,
|
|
diagnostics: list[str],
|
|
node_by_id: dict[str, dict[str, Any]],
|
|
api_prompt: dict[str, Any] | None,
|
|
input_keys: set[str],
|
|
) -> None:
|
|
if not isinstance(switch, dict):
|
|
diagnostics.append(f"{path} must be a JSON object")
|
|
return
|
|
_validate_non_empty_text(switch.get("key"), f"{path}.key", diagnostics)
|
|
_validate_non_empty_text(switch.get("label"), f"{path}.label", diagnostics)
|
|
if "visible" in switch and not isinstance(switch["visible"], bool):
|
|
diagnostics.append(f"{path}.visible must be a boolean")
|
|
_validate_surface_app_target(
|
|
switch.get("target"),
|
|
f"{path}.target",
|
|
diagnostics,
|
|
node_by_id,
|
|
api_prompt,
|
|
)
|
|
default = switch.get("default")
|
|
if default is not None and (isinstance(default, bool) or not isinstance(default, int)):
|
|
diagnostics.append(f"{path}.default must be an integer")
|
|
options = switch.get("options")
|
|
if not isinstance(options, list):
|
|
diagnostics.append(f"{path}.options must be a list")
|
|
return
|
|
for option_index, option in enumerate(options):
|
|
option_path = f"{path}.options[{option_index}]"
|
|
if not isinstance(option, dict):
|
|
diagnostics.append(f"{option_path} must be a JSON object")
|
|
continue
|
|
if isinstance(option.get("value"), bool) or not isinstance(option.get("value"), int):
|
|
diagnostics.append(f"{option_path}.value must be an integer")
|
|
_validate_non_empty_text(option.get("label"), f"{option_path}.label", diagnostics)
|
|
_validate_non_empty_text(option.get("input"), f"{option_path}.input", diagnostics)
|
|
if (
|
|
isinstance(option.get("input"), str)
|
|
and option["input"].strip()
|
|
and input_keys
|
|
and option["input"] not in input_keys
|
|
):
|
|
diagnostics.append(f"{option_path}.input must match a setupSurface.app.inputs key")
|
|
if "visible" in option and not isinstance(option["visible"], bool):
|
|
diagnostics.append(f"{option_path}.visible must be a boolean")
|
|
|
|
|
|
def _validate_surface_app_target(
|
|
target: Any,
|
|
path: str,
|
|
diagnostics: list[str],
|
|
node_by_id: dict[str, dict[str, Any]],
|
|
api_prompt: dict[str, Any] | None,
|
|
) -> None:
|
|
if not isinstance(target, dict):
|
|
diagnostics.append(f"{path} must be a JSON object")
|
|
return
|
|
_validate_non_empty_text(target.get("node"), f"{path}.node", diagnostics)
|
|
_validate_non_empty_text(target.get("input"), f"{path}.input", diagnostics)
|
|
node_id = target.get("node")
|
|
input_name = target.get("input")
|
|
if not isinstance(node_id, str) or not node_id.strip():
|
|
return
|
|
if not isinstance(input_name, str) or not input_name.strip():
|
|
return
|
|
node = node_by_id.get(node_id)
|
|
if node is None:
|
|
diagnostics.append(f"{path}.node not found in visualGraph")
|
|
return
|
|
if not _visual_node_has_input(node, input_name):
|
|
diagnostics.append(f"{path}.input not found in visualGraph node")
|
|
if api_prompt is None:
|
|
return
|
|
api_node = api_prompt.get(node_id)
|
|
if not isinstance(api_node, dict):
|
|
diagnostics.append(f"{path}.node not found in apiPrompt")
|
|
return
|
|
api_inputs = api_node.get("inputs")
|
|
if not isinstance(api_inputs, dict) or input_name not in api_inputs:
|
|
diagnostics.append(f"{path}.input not found in apiPrompt")
|
|
|
|
|
|
def _validate_non_empty_text(value: Any, path: str, diagnostics: list[str]) -> None:
|
|
if not isinstance(value, str) or not value.strip():
|
|
diagnostics.append(f"{path} must be non-empty text")
|
|
|
|
|
|
def _visual_node_index(visual_graph: Any) -> dict[str, dict[str, Any]]:
|
|
nodes = visual_graph.get("nodes") if isinstance(visual_graph, dict) else None
|
|
if not isinstance(nodes, list):
|
|
return {}
|
|
out: dict[str, dict[str, Any]] = {}
|
|
for node in nodes:
|
|
if not isinstance(node, dict):
|
|
continue
|
|
node_id = node.get("id")
|
|
if node_id is not None:
|
|
out[str(node_id)] = node
|
|
return out
|
|
|
|
|
|
def _visual_node_has_input(node: dict[str, Any], input_name: str) -> bool:
|
|
inputs = node.get("inputs")
|
|
if isinstance(inputs, list) and any(
|
|
isinstance(port, dict) and str(port.get("name", "")) == input_name
|
|
for port in inputs
|
|
):
|
|
return True
|
|
class_type = node.get("type")
|
|
widget_names = WIDGET_ONLY_INPUTS_BY_CLASS.get(str(class_type), ())
|
|
return input_name in widget_names
|
|
|
|
|
|
def _uses_group_authored_surface(
|
|
input_contract: dict[str, Any],
|
|
output_contract: dict[str, Any],
|
|
) -> bool:
|
|
inputs = input_contract.get("inputs")
|
|
outputs = output_contract.get("outputs")
|
|
return inputs == [] and outputs == []
|
|
|
|
|
|
def _nodes_in_named_groups(visual_graph: dict[str, Any], group_title: str) -> list[dict[str, Any]]:
|
|
groups = _nodes_in_group_sections(visual_graph, group_title)
|
|
return [
|
|
{"group": group_title, "nodes": [_surface_node_summary(node) for node in group_nodes]}
|
|
for group_nodes in groups
|
|
if group_nodes
|
|
]
|
|
|
|
|
|
def _nodes_in_group(visual_graph: dict[str, Any], group_title: str) -> list[dict[str, Any]]:
|
|
groups = _nodes_in_group_sections(visual_graph, group_title)
|
|
return [node for group_nodes in groups for node in group_nodes]
|
|
|
|
|
|
def _nodes_in_group_sections(
|
|
visual_graph: dict[str, Any],
|
|
group_title: str,
|
|
) -> list[list[dict[str, Any]]]:
|
|
groups = visual_graph.get("groups")
|
|
nodes = visual_graph.get("nodes")
|
|
if not isinstance(groups, list) or not isinstance(nodes, list):
|
|
return []
|
|
|
|
node_rects = [
|
|
(node, _rect_from_node(node))
|
|
for node in nodes
|
|
if isinstance(node, dict)
|
|
]
|
|
out: list[list[dict[str, Any]]] = []
|
|
for group in groups:
|
|
if not isinstance(group, dict) or group.get("title") != group_title:
|
|
continue
|
|
group_rect = _rect_from_group(group)
|
|
group_nodes = [
|
|
node
|
|
for node, node_rect in node_rects
|
|
if _rects_overlap(group_rect, node_rect)
|
|
]
|
|
if group_nodes:
|
|
out.append(sorted(group_nodes, key=lambda node: (_rect_from_node(node)["y"], _rect_from_node(node)["x"])))
|
|
return out
|
|
|
|
|
|
def _validate_setup_surface_geometry(visual_graph: dict[str, Any]) -> list[str]:
|
|
diagnostics: list[str] = []
|
|
groups = visual_graph.get("groups")
|
|
if isinstance(groups, list):
|
|
for group_index, group in enumerate(groups):
|
|
if not isinstance(group, dict) or group.get("title") not in {"Koolook Input", "Koolook Output"}:
|
|
continue
|
|
bounding = group.get("bounding")
|
|
pos = group.get("pos")
|
|
size = group.get("size")
|
|
if bounding is not None:
|
|
if (
|
|
not isinstance(bounding, list)
|
|
or len(bounding) < 4
|
|
or not all(_is_number(value) for value in bounding[:4])
|
|
):
|
|
diagnostics.append(
|
|
f"visualGraph.groups[{group_index}].bounding must contain numeric "
|
|
"x, y, width, height"
|
|
)
|
|
continue
|
|
if pos is not None or size is not None:
|
|
if (
|
|
not isinstance(pos, list)
|
|
or len(pos) < 2
|
|
or not all(_is_number(value) for value in pos[:2])
|
|
or not isinstance(size, list)
|
|
or len(size) < 2
|
|
or not all(_is_number(value) for value in size[:2])
|
|
):
|
|
diagnostics.append(
|
|
f"visualGraph.groups[{group_index}].pos/size must contain numeric "
|
|
"x, y, width, height"
|
|
)
|
|
continue
|
|
diagnostics.append(
|
|
f"visualGraph.groups[{group_index}] must define bounding or pos/size "
|
|
"for setup surface inference"
|
|
)
|
|
nodes = visual_graph.get("nodes")
|
|
if isinstance(nodes, list):
|
|
for node_index, node in enumerate(nodes):
|
|
if not isinstance(node, dict):
|
|
continue
|
|
pos = node.get("pos")
|
|
if pos is not None and (
|
|
not isinstance(pos, list)
|
|
or len(pos) < 2
|
|
or not all(_is_number(value) for value in pos[:2])
|
|
):
|
|
diagnostics.append(
|
|
f"visualGraph.nodes[{node_index}].pos must contain numeric x and y "
|
|
"for setup surface inference"
|
|
)
|
|
size = node.get("size")
|
|
if size is not None and (
|
|
not isinstance(size, list)
|
|
or len(size) < 2
|
|
or not all(_is_number(value) for value in size[:2])
|
|
):
|
|
diagnostics.append(
|
|
f"visualGraph.nodes[{node_index}].size must contain numeric width and height "
|
|
"for setup surface inference"
|
|
)
|
|
return diagnostics
|
|
|
|
|
|
def _surface_node_summary(node: dict[str, Any]) -> dict[str, str]:
|
|
node_type = node.get("type")
|
|
node_title = node.get("title")
|
|
return {
|
|
"id": str(node.get("id")),
|
|
"type": node_type if isinstance(node_type, str) else "",
|
|
"title": node_title if isinstance(node_title, str) and node_title else (
|
|
node_type if isinstance(node_type, str) else str(node.get("id"))
|
|
),
|
|
}
|
|
|
|
|
|
def _rect_from_group(group: dict[str, Any]) -> dict[str, float] | None:
|
|
bounding = group.get("bounding")
|
|
if isinstance(bounding, list) and len(bounding) >= 4:
|
|
return {
|
|
"x": _number_or_default(bounding[0], 0),
|
|
"y": _number_or_default(bounding[1], 0),
|
|
"w": _number_or_default(bounding[2], 0),
|
|
"h": _number_or_default(bounding[3], 0),
|
|
}
|
|
pos = group.get("pos")
|
|
size = group.get("size")
|
|
if isinstance(pos, list) and isinstance(size, list) and len(pos) >= 2 and len(size) >= 2:
|
|
return {
|
|
"x": _number_or_default(pos[0], 0),
|
|
"y": _number_or_default(pos[1], 0),
|
|
"w": _number_or_default(size[0], 0),
|
|
"h": _number_or_default(size[1], 0),
|
|
}
|
|
return None
|
|
|
|
|
|
def _rect_from_node(node: dict[str, Any]) -> dict[str, float] | None:
|
|
pos = node.get("pos")
|
|
if not isinstance(pos, list) or len(pos) < 2:
|
|
return None
|
|
size = node.get("size")
|
|
if not isinstance(size, list) or len(size) < 2:
|
|
size = [200, 100]
|
|
return {
|
|
"x": _number_or_default(pos[0], 0),
|
|
"y": _number_or_default(pos[1], 0),
|
|
"w": _number_or_default(size[0], 200),
|
|
"h": _number_or_default(size[1], 100),
|
|
}
|
|
|
|
|
|
def _rects_overlap(a: dict[str, float] | None, b: dict[str, float] | None) -> bool:
|
|
if not a or not b or a["w"] <= 0 or a["h"] <= 0 or b["w"] <= 0 or b["h"] <= 0:
|
|
return False
|
|
return (
|
|
a["x"] < b["x"] + b["w"]
|
|
and a["x"] + a["w"] > b["x"]
|
|
and a["y"] < b["y"] + b["h"]
|
|
and a["y"] + a["h"] > b["y"]
|
|
)
|
|
|
|
|
|
def _number_or_default(value: Any, default: float) -> float:
|
|
return value if isinstance(value, int | float) else default
|
|
|
|
|
|
def _is_number(value: Any) -> bool:
|
|
return not isinstance(value, bool) and isinstance(value, int | float) and math.isfinite(value)
|
|
|
|
|
|
def _convert_visual_graph_to_api_prompt(visual_graph: dict[str, Any]) -> ApiPromptConversionResult:
|
|
nodes = visual_graph.get("nodes")
|
|
if not isinstance(nodes, list):
|
|
return ApiPromptConversionResult(None, ["visualGraph.nodes must be a list"])
|
|
links = _visual_link_index(visual_graph.get("links"))
|
|
if links.diagnostics:
|
|
return ApiPromptConversionResult(None, links.diagnostics)
|
|
|
|
node_indexes_by_id: dict[str, int] = {}
|
|
node_ids = set[str]()
|
|
for node_index, node in enumerate(nodes):
|
|
if not isinstance(node, dict):
|
|
return ApiPromptConversionResult(None, [f"visualGraph.nodes[{node_index}] must be an object"])
|
|
node_id = node.get("id")
|
|
class_type = node.get("type")
|
|
if node_id is None or not isinstance(class_type, str) or not class_type.strip():
|
|
return ApiPromptConversionResult(
|
|
None,
|
|
[f"visualGraph.nodes[{node_index}].type must be non-empty text"],
|
|
)
|
|
node_id_key = str(node_id)
|
|
previous_node_index = node_indexes_by_id.get(node_id_key)
|
|
if previous_node_index is not None:
|
|
return ApiPromptConversionResult(
|
|
None,
|
|
[
|
|
f"visualGraph.nodes[{node_index}].id duplicates "
|
|
f"visualGraph.nodes[{previous_node_index}].id"
|
|
],
|
|
)
|
|
node_indexes_by_id[node_id_key] = node_index
|
|
node_ids.add(node_id_key)
|
|
|
|
node_by_id = {str(node["id"]): node for node in nodes}
|
|
subgraphs_by_id = _subgraph_definitions_by_id(visual_graph)
|
|
api_prompt: dict[str, Any] = {}
|
|
subgraph_output_refs: dict[str, dict[int, Any]] = {}
|
|
skipped_node_ids: set[str] = set()
|
|
expanded_node_ids: set[str] = set()
|
|
diagnostics: list[str] = []
|
|
|
|
def resolve_outer_link(link_id: Any, seen: set[str] | None = None) -> Any:
|
|
link_id_key = str(link_id)
|
|
seen = set() if seen is None else seen
|
|
if link_id_key in seen:
|
|
diagnostics.append(f"visualGraph.links[{link_id_key}] has a reroute cycle")
|
|
return None
|
|
seen.add(link_id_key)
|
|
link = links.links.get(link_id_key)
|
|
if link is None:
|
|
diagnostics.append(f"visualGraph.links[{link_id_key}] is missing")
|
|
return None
|
|
origin_id = str(link["origin_id"])
|
|
if _is_module_sentinel_id(link["origin_id"]):
|
|
diagnostics.append(f"visualGraph.links[{link_id_key}] uses unsupported module graph sentinel node")
|
|
return None
|
|
if not _is_non_negative_int(link["origin_slot"]):
|
|
diagnostics.append(
|
|
f"visualGraph.links[{link_id_key}].origin_slot must be a non-negative integer"
|
|
)
|
|
return None
|
|
origin_slot = int(link["origin_slot"])
|
|
origin_node = node_by_id.get(origin_id)
|
|
if origin_node is None:
|
|
diagnostics.append(f"visualGraph.links[{link_id_key}].origin_id not found in visualGraph")
|
|
return None
|
|
if origin_node.get("type") == "Reroute":
|
|
reroute_input = _first_linked_input(origin_node)
|
|
if reroute_input is None:
|
|
diagnostics.append(f"visualGraph.nodes[{node_indexes_by_id[origin_id]}] Reroute is unconnected")
|
|
return None
|
|
return resolve_outer_link(reroute_input, seen)
|
|
if origin_id in subgraph_output_refs:
|
|
ref = subgraph_output_refs[origin_id].get(origin_slot)
|
|
if ref is None:
|
|
diagnostics.append(
|
|
f"visualGraph.links[{link_id_key}] references missing subgraph output {origin_slot}"
|
|
)
|
|
return ref
|
|
if origin_id in skipped_node_ids or origin_id in expanded_node_ids:
|
|
diagnostics.append(
|
|
f"visualGraph.links[{link_id_key}] references non-executable node {origin_id}"
|
|
)
|
|
return None
|
|
return [origin_id, origin_slot]
|
|
|
|
def add_node_api(
|
|
*,
|
|
node: dict[str, Any],
|
|
node_index: int,
|
|
node_id_key: str,
|
|
class_type: str,
|
|
link_resolver,
|
|
) -> None:
|
|
node_inputs = node.get("inputs")
|
|
if node_inputs is None:
|
|
node_inputs = []
|
|
elif not isinstance(node_inputs, list):
|
|
diagnostics.append(f"visualGraph.nodes[{node_index}].inputs must be a list when present")
|
|
return
|
|
|
|
raw_widget_values = node.get("widgets_values")
|
|
widget_values_by_name = raw_widget_values if isinstance(raw_widget_values, dict) else {}
|
|
widget_values = raw_widget_values if isinstance(raw_widget_values, list) else []
|
|
if not widget_values and not widget_values_by_name:
|
|
widget_values = _proxy_widget_values_from_subgraph(visual_graph, node)
|
|
widget_index = 0
|
|
api_inputs: dict[str, Any] = {}
|
|
for input_index, input_port in enumerate(node_inputs):
|
|
if not isinstance(input_port, dict):
|
|
diagnostics.append(f"visualGraph.nodes[{node_index}].inputs must contain objects")
|
|
return
|
|
name = input_port.get("name")
|
|
if not isinstance(name, str) or not name.strip():
|
|
if class_type != "Reroute" or not isinstance(name, str):
|
|
diagnostics.append(f"visualGraph.nodes[{node_index}].inputs must have named ports")
|
|
return
|
|
link_id = input_port.get("link")
|
|
if link_id is not None:
|
|
link = links.links.get(str(link_id))
|
|
if link is not None and link["target_id"] is not None and str(link["target_id"]) != node_id_key:
|
|
diagnostics.append(
|
|
f"visualGraph.links[{link_id}].target does not match "
|
|
f"visualGraph.nodes[{node_index}].inputs[{input_index}]"
|
|
)
|
|
return
|
|
if (
|
|
link is not None
|
|
and link["target_slot"] is not None
|
|
and (
|
|
not _is_non_negative_int(link["target_slot"])
|
|
or int(link["target_slot"]) != input_index
|
|
)
|
|
):
|
|
diagnostics.append(
|
|
f"visualGraph.links[{link_id}].target does not match "
|
|
f"visualGraph.nodes[{node_index}].inputs[{input_index}]"
|
|
)
|
|
return
|
|
resolved = link_resolver(link_id)
|
|
if resolved is None:
|
|
return
|
|
api_inputs[name] = resolved
|
|
continue
|
|
widget = input_port.get("widget")
|
|
if isinstance(widget, dict):
|
|
if name in widget_values_by_name:
|
|
api_inputs[name] = widget_values_by_name[name]
|
|
continue
|
|
if widget_index >= len(widget_values):
|
|
diagnostics.append(
|
|
f"visualGraph.nodes[{node_index}].widgets_values is missing a value "
|
|
f"for input {name}"
|
|
)
|
|
return
|
|
api_inputs[name] = widget_values[widget_index]
|
|
widget_index += 1
|
|
|
|
if not node_inputs:
|
|
widget_only_diagnostics = _apply_widget_only_inputs(
|
|
class_type,
|
|
widget_values,
|
|
api_inputs,
|
|
node_index,
|
|
)
|
|
diagnostics.extend(widget_only_diagnostics)
|
|
if widget_only_diagnostics:
|
|
return
|
|
api_prompt[node_id_key] = {"class_type": class_type, "inputs": api_inputs}
|
|
|
|
def expand_subgraph_node(node: dict[str, Any], node_index: int) -> None:
|
|
node_id_key = str(node["id"])
|
|
subgraph = subgraphs_by_id.get(str(node["type"]))
|
|
if subgraph is None:
|
|
return
|
|
subgraph_links = _visual_link_index(subgraph.get("links"))
|
|
if subgraph_links.diagnostics:
|
|
diagnostics.extend(subgraph_links.diagnostics)
|
|
return
|
|
subgraph_nodes = subgraph.get("nodes")
|
|
if not isinstance(subgraph_nodes, list):
|
|
diagnostics.append(f"visualGraph.definitions.subgraphs[{node['type']}].nodes must be a list")
|
|
return
|
|
subgraph_inputs = subgraph.get("inputs")
|
|
if not isinstance(subgraph_inputs, list):
|
|
subgraph_inputs = []
|
|
subgraph_nodes_by_id = {
|
|
str(subgraph_node.get("id")): subgraph_node
|
|
for subgraph_node in subgraph_nodes
|
|
if isinstance(subgraph_node, dict) and subgraph_node.get("id") is not None
|
|
}
|
|
wrapper_inputs = _api_inputs_for_wrapper_node(
|
|
visual_graph,
|
|
node,
|
|
node_index,
|
|
links,
|
|
resolve_outer_link,
|
|
diagnostics,
|
|
)
|
|
skipped_internal_ids: set[str] = set()
|
|
for internal_node in subgraph_nodes:
|
|
if not isinstance(internal_node, dict):
|
|
continue
|
|
internal_id = str(internal_node.get("id"))
|
|
if _should_omit_api_node(internal_node):
|
|
skipped_internal_ids.add(internal_id)
|
|
|
|
def resolve_subgraph_link(link_id: Any, seen: set[str] | None = None) -> Any:
|
|
link_id_key = str(link_id)
|
|
seen = set() if seen is None else seen
|
|
if link_id_key in seen:
|
|
diagnostics.append(f"visualGraph.definitions.subgraphs[{node['type']}].links[{link_id_key}] has a reroute cycle")
|
|
return None
|
|
seen.add(link_id_key)
|
|
link = subgraph_links.links.get(link_id_key)
|
|
if link is None:
|
|
diagnostics.append(f"visualGraph.definitions.subgraphs[{node['type']}].links[{link_id_key}] is missing")
|
|
return None
|
|
origin_id = link["origin_id"]
|
|
if not _is_non_negative_int(link["origin_slot"]):
|
|
diagnostics.append(
|
|
f"visualGraph.definitions.subgraphs[{node['type']}].links[{link_id_key}].origin_slot "
|
|
"must be a non-negative integer"
|
|
)
|
|
return None
|
|
origin_slot = int(link["origin_slot"])
|
|
if _is_module_sentinel_id(origin_id):
|
|
if int(origin_id) != -10:
|
|
diagnostics.append(
|
|
f"visualGraph.definitions.subgraphs[{node['type']}].links[{link_id_key}] "
|
|
"uses unsupported module graph sentinel node"
|
|
)
|
|
return None
|
|
if origin_slot >= len(subgraph_inputs):
|
|
diagnostics.append(
|
|
f"visualGraph.definitions.subgraphs[{node['type']}].links[{link_id_key}] "
|
|
"references missing subgraph input"
|
|
)
|
|
return None
|
|
input_name = subgraph_inputs[origin_slot].get("name")
|
|
if not isinstance(input_name, str) or input_name not in wrapper_inputs:
|
|
return None
|
|
return wrapper_inputs[input_name]
|
|
origin_id_key = str(origin_id)
|
|
origin_node = subgraph_nodes_by_id.get(origin_id_key)
|
|
if origin_node is None:
|
|
diagnostics.append(
|
|
f"visualGraph.definitions.subgraphs[{node['type']}].links[{link_id_key}].origin_id not found"
|
|
)
|
|
return None
|
|
if origin_node.get("type") == "Reroute":
|
|
reroute_input = _first_linked_input(origin_node)
|
|
if reroute_input is None:
|
|
diagnostics.append(
|
|
f"visualGraph.definitions.subgraphs[{node['type']}].nodes[{origin_id_key}] Reroute is unconnected"
|
|
)
|
|
return None
|
|
return resolve_subgraph_link(reroute_input, seen)
|
|
if origin_id_key in skipped_internal_ids:
|
|
diagnostics.append(
|
|
f"visualGraph.definitions.subgraphs[{node['type']}].links[{link_id_key}] "
|
|
f"references non-executable node {origin_id_key}"
|
|
)
|
|
return None
|
|
return [f"{node_id_key}:{origin_id_key}", origin_slot]
|
|
|
|
for internal_index, internal_node in enumerate(subgraph_nodes):
|
|
if not isinstance(internal_node, dict):
|
|
diagnostics.append(
|
|
f"visualGraph.definitions.subgraphs[{node['type']}].nodes[{internal_index}] must be an object"
|
|
)
|
|
return
|
|
internal_class_type = internal_node.get("type")
|
|
internal_id = internal_node.get("id")
|
|
if internal_id is None or not isinstance(internal_class_type, str) or not internal_class_type.strip():
|
|
diagnostics.append(
|
|
f"visualGraph.definitions.subgraphs[{node['type']}].nodes[{internal_index}].type "
|
|
"must be non-empty text"
|
|
)
|
|
return
|
|
if str(internal_class_type) in subgraphs_by_id:
|
|
diagnostics.append(
|
|
f"visualGraph.definitions.subgraphs[{node['type']}].nodes[{internal_index}] "
|
|
"nested subgraphs are not supported"
|
|
)
|
|
return
|
|
if str(internal_id) in skipped_internal_ids:
|
|
continue
|
|
add_node_api(
|
|
node=internal_node,
|
|
node_index=node_index,
|
|
node_id_key=f"{node_id_key}:{internal_id}",
|
|
class_type=internal_class_type,
|
|
link_resolver=resolve_subgraph_link,
|
|
)
|
|
if diagnostics:
|
|
return
|
|
|
|
output_refs: dict[int, Any] = {}
|
|
for link_id, link in subgraph_links.links.items():
|
|
if not _is_module_sentinel_id(link["target_id"]) or int(link["target_id"]) != -20:
|
|
continue
|
|
target_slot = link.get("target_slot")
|
|
if not _is_non_negative_int(target_slot):
|
|
continue
|
|
ref = resolve_subgraph_link(link_id)
|
|
if ref is not None:
|
|
output_refs[int(target_slot)] = ref
|
|
subgraph_output_refs[node_id_key] = output_refs
|
|
expanded_node_ids.add(node_id_key)
|
|
|
|
for node_index, node in enumerate(nodes):
|
|
node_id_key = str(node["id"])
|
|
class_type = node["type"]
|
|
if _should_omit_api_node(node):
|
|
skipped_node_ids.add(node_id_key)
|
|
continue
|
|
if class_type == "Reroute":
|
|
skipped_node_ids.add(node_id_key)
|
|
continue
|
|
if class_type in subgraphs_by_id:
|
|
expand_subgraph_node(node, node_index)
|
|
if diagnostics:
|
|
return ApiPromptConversionResult(None, diagnostics)
|
|
continue
|
|
add_node_api(
|
|
node=node,
|
|
node_index=node_index,
|
|
node_id_key=node_id_key,
|
|
class_type=class_type,
|
|
link_resolver=resolve_outer_link,
|
|
)
|
|
if diagnostics:
|
|
return ApiPromptConversionResult(None, diagnostics)
|
|
|
|
return ApiPromptConversionResult(api_prompt, diagnostics)
|
|
|
|
|
|
def _api_inputs_for_wrapper_node(
|
|
visual_graph: dict[str, Any],
|
|
node: dict[str, Any],
|
|
node_index: int,
|
|
links: VisualLinkIndex,
|
|
link_resolver,
|
|
diagnostics: list[str],
|
|
) -> dict[str, Any]:
|
|
node_inputs = node.get("inputs")
|
|
if node_inputs is None:
|
|
return {}
|
|
if not isinstance(node_inputs, list):
|
|
diagnostics.append(f"visualGraph.nodes[{node_index}].inputs must be a list when present")
|
|
return {}
|
|
raw_widget_values = node.get("widgets_values")
|
|
widget_values_by_name = raw_widget_values if isinstance(raw_widget_values, dict) else {}
|
|
widget_values = raw_widget_values if isinstance(raw_widget_values, list) else []
|
|
widget_index = 0
|
|
values: dict[str, Any] = {}
|
|
for input_index, input_port in enumerate(node_inputs):
|
|
if not isinstance(input_port, dict):
|
|
diagnostics.append(f"visualGraph.nodes[{node_index}].inputs must contain objects")
|
|
return values
|
|
name = input_port.get("name")
|
|
if not isinstance(name, str) or not name.strip():
|
|
diagnostics.append(f"visualGraph.nodes[{node_index}].inputs must have named ports")
|
|
return values
|
|
link_id = input_port.get("link")
|
|
if link_id is not None:
|
|
link = links.links.get(str(link_id))
|
|
if link is not None and link["target_id"] is not None and str(link["target_id"]) != str(node.get("id")):
|
|
diagnostics.append(
|
|
f"visualGraph.links[{link_id}].target does not match "
|
|
f"visualGraph.nodes[{node_index}].inputs[{input_index}]"
|
|
)
|
|
return values
|
|
resolved = link_resolver(link_id)
|
|
if resolved is not None:
|
|
values[name] = resolved
|
|
continue
|
|
widget = input_port.get("widget")
|
|
if not isinstance(widget, dict):
|
|
continue
|
|
if name in widget_values_by_name:
|
|
values[name] = widget_values_by_name[name]
|
|
continue
|
|
if widget_index < len(widget_values):
|
|
values[name] = widget_values[widget_index]
|
|
widget_index += 1
|
|
proxy_values = _proxy_widget_values_from_subgraph(visual_graph, node)
|
|
proxy_index = 0
|
|
for input_port in node_inputs:
|
|
if proxy_index >= len(proxy_values) or not isinstance(input_port, dict):
|
|
continue
|
|
name = input_port.get("name")
|
|
if isinstance(name, str) and name not in values and input_port.get("widget"):
|
|
values[name] = proxy_values[proxy_index]
|
|
if input_port.get("widget"):
|
|
proxy_index += 1
|
|
return values
|
|
|
|
|
|
def _api_prompt_needs_visual_normalization(
|
|
api_prompt: dict[str, Any],
|
|
visual_graph: dict[str, Any],
|
|
) -> bool:
|
|
subgraph_ids = set(_subgraph_definitions_by_id(visual_graph))
|
|
for node in api_prompt.values():
|
|
if not isinstance(node, dict):
|
|
continue
|
|
class_type = node.get("class_type")
|
|
if not isinstance(class_type, str):
|
|
continue
|
|
if class_type == "Reroute":
|
|
return True
|
|
if class_type in VISUAL_ONLY_API_NODE_TYPES:
|
|
return True
|
|
if class_type in SINK_ONLY_API_NODE_TYPES:
|
|
return True
|
|
if class_type in subgraph_ids:
|
|
return True
|
|
return False
|
|
|
|
|
|
def _subgraph_definitions_by_id(visual_graph: dict[str, Any]) -> dict[str, dict[str, Any]]:
|
|
definitions = visual_graph.get("definitions")
|
|
if not isinstance(definitions, dict):
|
|
return {}
|
|
subgraphs = definitions.get("subgraphs")
|
|
if not isinstance(subgraphs, list):
|
|
return {}
|
|
return {
|
|
str(subgraph.get("id")): subgraph
|
|
for subgraph in subgraphs
|
|
if isinstance(subgraph, dict) and subgraph.get("id") is not None
|
|
}
|
|
|
|
|
|
def _should_omit_api_node(node: dict[str, Any]) -> bool:
|
|
class_type = str(node.get("type", ""))
|
|
if class_type in VISUAL_ONLY_API_NODE_TYPES:
|
|
return True
|
|
if class_type in SINK_ONLY_API_NODE_TYPES:
|
|
return not _node_has_linked_outputs(node)
|
|
return False
|
|
|
|
|
|
def _node_has_linked_outputs(node: dict[str, Any]) -> bool:
|
|
outputs = node.get("outputs")
|
|
if not isinstance(outputs, list):
|
|
return False
|
|
for output in outputs:
|
|
if not isinstance(output, dict):
|
|
continue
|
|
links = output.get("links")
|
|
if isinstance(links, list) and links:
|
|
return True
|
|
return False
|
|
|
|
|
|
def _first_linked_input(node: dict[str, Any]) -> Any | None:
|
|
node_inputs = node.get("inputs")
|
|
if not isinstance(node_inputs, list):
|
|
return None
|
|
for input_port in node_inputs:
|
|
if not isinstance(input_port, dict):
|
|
continue
|
|
link_id = input_port.get("link")
|
|
if link_id is not None:
|
|
return link_id
|
|
return None
|
|
|
|
|
|
def _apply_widget_only_inputs(
|
|
class_type: str,
|
|
widget_values: list[Any],
|
|
api_inputs: dict[str, Any],
|
|
node_index: int,
|
|
) -> list[str]:
|
|
if api_inputs:
|
|
return []
|
|
widget_names = WIDGET_ONLY_INPUTS_BY_CLASS.get(class_type)
|
|
if not widget_names:
|
|
return []
|
|
defaults = WIDGET_ONLY_INPUT_DEFAULTS.get(class_type, {})
|
|
diagnostics: list[str] = []
|
|
for index, name in enumerate(widget_names):
|
|
if index < len(widget_values):
|
|
value = widget_values[index]
|
|
if name == "no_subfolders" and not isinstance(value, bool):
|
|
value = defaults.get(name, False)
|
|
api_inputs[name] = value
|
|
elif name in defaults:
|
|
api_inputs[name] = defaults[name]
|
|
else:
|
|
diagnostics.append(
|
|
f"visualGraph.nodes[{node_index}].widgets_values is missing a value "
|
|
f"for widget-only input {name}"
|
|
)
|
|
return diagnostics
|
|
|
|
|
|
def _proxy_widget_values_from_subgraph(
|
|
visual_graph: dict[str, Any],
|
|
node: dict[str, Any],
|
|
) -> list[Any]:
|
|
properties = node.get("properties")
|
|
if not isinstance(properties, dict):
|
|
return []
|
|
proxy_widgets = properties.get("proxyWidgets")
|
|
if not isinstance(proxy_widgets, list):
|
|
return []
|
|
subgraph = _subgraph_definition(visual_graph, str(node.get("type", "")))
|
|
if subgraph is None:
|
|
return []
|
|
subgraph_nodes = subgraph.get("nodes")
|
|
if not isinstance(subgraph_nodes, list):
|
|
return []
|
|
|
|
nodes_by_id = {
|
|
str(subgraph_node.get("id")): subgraph_node
|
|
for subgraph_node in subgraph_nodes
|
|
if isinstance(subgraph_node, dict) and subgraph_node.get("id") is not None
|
|
}
|
|
values: list[Any] = []
|
|
for proxy_widget in proxy_widgets:
|
|
if not isinstance(proxy_widget, list) or len(proxy_widget) < 2:
|
|
continue
|
|
source_node = nodes_by_id.get(str(proxy_widget[0]))
|
|
widget_name = proxy_widget[1]
|
|
if not isinstance(source_node, dict) or not isinstance(widget_name, str):
|
|
continue
|
|
found, value = _node_widget_value(source_node, widget_name)
|
|
if found:
|
|
values.append(value)
|
|
return values
|
|
|
|
|
|
def _subgraph_definition(
|
|
visual_graph: dict[str, Any],
|
|
subgraph_id: str,
|
|
) -> dict[str, Any] | None:
|
|
definitions = visual_graph.get("definitions")
|
|
if not isinstance(definitions, dict):
|
|
return None
|
|
subgraphs = definitions.get("subgraphs")
|
|
if not isinstance(subgraphs, list):
|
|
return None
|
|
for subgraph in subgraphs:
|
|
if isinstance(subgraph, dict) and subgraph.get("id") == subgraph_id:
|
|
return subgraph
|
|
return None
|
|
|
|
|
|
def _node_widget_value(
|
|
node: dict[str, Any],
|
|
widget_name: str,
|
|
) -> tuple[bool, Any]:
|
|
inputs = node.get("inputs")
|
|
widget_values = node.get("widgets_values")
|
|
if not isinstance(inputs, list) or not isinstance(widget_values, list):
|
|
return False, None
|
|
|
|
widget_index = 0
|
|
for input_port in inputs:
|
|
if not isinstance(input_port, dict):
|
|
continue
|
|
widget = input_port.get("widget")
|
|
if not isinstance(widget, dict):
|
|
continue
|
|
if widget.get("name") == widget_name:
|
|
if widget_index >= len(widget_values):
|
|
return False, None
|
|
return True, widget_values[widget_index]
|
|
widget_index += 1
|
|
return False, None
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class VisualLinkIndex:
|
|
links: dict[str, dict[str, Any]]
|
|
diagnostics: list[str]
|
|
|
|
|
|
def _visual_link_index(value: Any) -> VisualLinkIndex:
|
|
links: dict[str, dict[str, Any]] = {}
|
|
if not isinstance(value, list):
|
|
return VisualLinkIndex(links, [])
|
|
link_indexes_by_id: dict[str, int] = {}
|
|
for link_index, item in enumerate(value):
|
|
if isinstance(item, dict):
|
|
link_id = item.get("id")
|
|
origin_id = item.get("origin_id")
|
|
origin_slot = item.get("origin_slot")
|
|
target_id = item.get("target_id")
|
|
target_slot = item.get("target_slot")
|
|
elif isinstance(item, list) and len(item) >= 5:
|
|
link_id = item[0]
|
|
origin_id = item[1]
|
|
origin_slot = item[2]
|
|
target_id = item[3]
|
|
target_slot = item[4]
|
|
elif isinstance(item, list) and len(item) >= 3:
|
|
link_id = item[0]
|
|
origin_id = item[1]
|
|
origin_slot = item[2]
|
|
target_id = None
|
|
target_slot = None
|
|
else:
|
|
continue
|
|
if link_id is None or origin_id is None or origin_slot is None:
|
|
continue
|
|
link_id_key = str(link_id)
|
|
previous_link_index = link_indexes_by_id.get(link_id_key)
|
|
if previous_link_index is not None:
|
|
return VisualLinkIndex(
|
|
links,
|
|
[
|
|
f"visualGraph.links[{link_index}].id duplicates "
|
|
f"visualGraph.links[{previous_link_index}].id"
|
|
],
|
|
)
|
|
link_indexes_by_id[link_id_key] = link_index
|
|
links[link_id_key] = {
|
|
"origin_id": origin_id,
|
|
"origin_slot": origin_slot,
|
|
"target_id": target_id,
|
|
"target_slot": target_slot,
|
|
}
|
|
return VisualLinkIndex(links, [])
|
|
|
|
|
|
def _is_module_sentinel_id(value: Any) -> bool:
|
|
try:
|
|
return int(value) < 0
|
|
except (TypeError, ValueError):
|
|
return False
|
|
|
|
|
|
def _is_non_negative_int(value: Any) -> bool:
|
|
return isinstance(value, int) and value >= 0
|
|
|
|
|
|
def _validate_contract_targets(
|
|
visual_graph: dict[str, Any],
|
|
input_contract: dict[str, Any],
|
|
) -> list[str]:
|
|
diagnostics: list[str] = []
|
|
nodes = visual_graph.get("nodes") if isinstance(visual_graph, dict) else None
|
|
node_by_id: dict[str, dict[str, Any]] = {}
|
|
if isinstance(nodes, list):
|
|
for node in nodes:
|
|
if not isinstance(node, dict):
|
|
continue
|
|
node_id = node.get("id")
|
|
if node_id is not None:
|
|
node_by_id[str(node_id)] = node
|
|
inputs = input_contract.get("inputs") if isinstance(input_contract, dict) else None
|
|
if not isinstance(inputs, list):
|
|
return diagnostics
|
|
for index, item in enumerate(inputs):
|
|
if not isinstance(item, dict):
|
|
continue
|
|
target = item.get("target")
|
|
if not isinstance(target, dict):
|
|
continue
|
|
node_id = str(target.get("node", ""))
|
|
input_name = str(target.get("input", ""))
|
|
node = node_by_id.get(node_id)
|
|
if node is None:
|
|
diagnostics.append(
|
|
f"inputContract.inputs[{index}].target.node not found in visualGraph"
|
|
)
|
|
continue
|
|
node_inputs = node.get("inputs")
|
|
if not isinstance(node_inputs, list):
|
|
continue
|
|
if not any(
|
|
isinstance(port, dict) and str(port.get("name", "")) == input_name
|
|
for port in node_inputs
|
|
):
|
|
diagnostics.append(
|
|
f"inputContract.inputs[{index}].target.input not found in visualGraph node"
|
|
)
|
|
return diagnostics
|
|
|
|
|
|
def _validate_api_prompt_contract_targets(
|
|
api_prompt: dict[str, Any] | None,
|
|
input_contract: dict[str, Any],
|
|
) -> list[str]:
|
|
diagnostics: list[str] = []
|
|
if api_prompt is None:
|
|
return diagnostics
|
|
inputs = input_contract.get("inputs") if isinstance(input_contract, dict) else None
|
|
if not isinstance(inputs, list):
|
|
return diagnostics
|
|
for index, item in enumerate(inputs):
|
|
if not isinstance(item, dict):
|
|
continue
|
|
target = item.get("target")
|
|
if not isinstance(target, dict):
|
|
continue
|
|
node_id = str(target.get("node", ""))
|
|
input_name = str(target.get("input", ""))
|
|
api_node = api_prompt.get(node_id)
|
|
if not isinstance(api_node, dict):
|
|
diagnostics.append(
|
|
f"inputContract.inputs[{index}].target.node not found in generated apiPrompt"
|
|
)
|
|
continue
|
|
api_inputs = api_node.get("inputs")
|
|
if not isinstance(api_inputs, dict) or input_name not in api_inputs:
|
|
diagnostics.append(
|
|
f"inputContract.inputs[{index}].target.input not found in generated apiPrompt"
|
|
)
|
|
return diagnostics
|
|
|
|
|
|
def default_storage_path() -> Path:
|
|
try:
|
|
import folder_paths
|
|
|
|
user_dir = folder_paths.get_user_directory()
|
|
return Path(user_dir) / DEFAULT_SETUPS_SUBDIR / DEFAULT_SETUPS_FILENAME
|
|
except Exception: # pragma: no cover - fallback for non-Comfy contexts
|
|
return Path.cwd() / "user" / "default" / DEFAULT_SETUPS_SUBDIR / DEFAULT_SETUPS_FILENAME
|
|
|
|
|
|
def default_registry() -> PublishedSetupRegistry:
|
|
return PublishedSetupRegistry(
|
|
FileSetupStorage(default_storage_path(), fallback_path=SAMPLE_SETUPS_PATH)
|
|
)
|
|
|
|
|
|
def _validate_scalar(
|
|
mapping: dict[str, Any],
|
|
key: str,
|
|
expected_type: type | tuple[type, ...],
|
|
diagnostics: list[str],
|
|
) -> None:
|
|
if not isinstance(mapping.get(key), expected_type):
|
|
diagnostics.append(f"{key} has invalid type")
|
|
|
|
|
|
def _validate_mapping(value: Any, path: str, diagnostics: list[str]) -> None:
|
|
if not isinstance(value, dict):
|
|
diagnostics.append(f"{path} must be a JSON object")
|
|
|
|
|
|
def _validate_iso_timestamp(value: Any, diagnostics: list[str]) -> None:
|
|
if not isinstance(value, str):
|
|
diagnostics.append("updatedAt must be an ISO timestamp")
|
|
return
|
|
normalized = value[:-1] + "+00:00" if value.endswith("Z") else value
|
|
try:
|
|
datetime.fromisoformat(normalized)
|
|
except ValueError:
|
|
diagnostics.append("updatedAt must be an ISO timestamp")
|
|
|
|
|
|
def _validate_metadata(value: Any, diagnostics: list[str]) -> None:
|
|
if not isinstance(value, dict):
|
|
diagnostics.append("metadata must be a JSON object")
|
|
return
|
|
if not isinstance(value.get("title"), str) or not value["title"].strip():
|
|
diagnostics.append("metadata.title must be non-empty text")
|
|
if "description" in value and not isinstance(value["description"], str):
|
|
diagnostics.append("metadata.description must be text")
|
|
if "category" in value and not isinstance(value["category"], str):
|
|
diagnostics.append("metadata.category must be text")
|
|
if "tags" in value and not _is_string_list(value["tags"]):
|
|
diagnostics.append("metadata.tags must be a list of text")
|
|
if "previewImage" in value and not isinstance(value["previewImage"], str):
|
|
diagnostics.append("metadata.previewImage must be text")
|
|
|
|
|
|
def _validate_contract(value: Any, path: str, items_key: str, diagnostics: list[str]) -> None:
|
|
if not isinstance(value, dict):
|
|
diagnostics.append(f"{path} must be a JSON object")
|
|
return
|
|
items = value.get(items_key)
|
|
if not isinstance(items, list):
|
|
diagnostics.append(f"{path}.{items_key} must be a list")
|
|
return
|
|
for index, item in enumerate(items):
|
|
item_path = f"{path}.{items_key}[{index}]"
|
|
if not isinstance(item, dict):
|
|
diagnostics.append(f"{item_path} must be a JSON object")
|
|
continue
|
|
for key in ("key", "type"):
|
|
if not isinstance(item.get(key), str) or not item[key].strip():
|
|
diagnostics.append(f"{item_path}.{key} must be non-empty text")
|
|
if "label" in item and not isinstance(item["label"], str):
|
|
diagnostics.append(f"{item_path}.label must be text")
|
|
if "required" in item and not isinstance(item["required"], bool):
|
|
diagnostics.append(f"{item_path}.required must be true or false")
|
|
if path == "inputContract":
|
|
target = item.get("target")
|
|
if not isinstance(target, dict):
|
|
diagnostics.append(f"{item_path}.target must be a JSON object")
|
|
else:
|
|
for key in ("node", "input"):
|
|
if not isinstance(target.get(key), str) or not target[key].strip():
|
|
diagnostics.append(f"{item_path}.target.{key} must be non-empty text")
|
|
|
|
|
|
def _validate_source(value: Any, diagnostics: list[str]) -> None:
|
|
if not isinstance(value, dict):
|
|
diagnostics.append("source must be a JSON object")
|
|
return
|
|
for key in ("kind", "path"):
|
|
if not isinstance(value.get(key), str) or not value[key].strip():
|
|
diagnostics.append(f"source.{key} must be non-empty text")
|
|
|
|
|
|
def _validate_validation(value: Any, diagnostics: list[str]) -> None:
|
|
if not isinstance(value, dict):
|
|
diagnostics.append("validation must be a JSON object")
|
|
return
|
|
if value.get("status") not in VALIDATION_STATUSES:
|
|
diagnostics.append("validation.status must be valid, draft, or invalid")
|
|
if "diagnostics" in value and not _is_string_list(value["diagnostics"]):
|
|
diagnostics.append("validation.diagnostics must be a list of text")
|
|
|
|
|
|
def _is_string_list(value: Any) -> bool:
|
|
return isinstance(value, list) and all(isinstance(item, str) for item in value)
|
|
|
|
|
|
def _catalog_summary(setup: dict[str, Any]) -> dict[str, Any]:
|
|
return {
|
|
"id": setup["id"],
|
|
"version": setup["version"],
|
|
"updatedAt": setup["updatedAt"],
|
|
"metadata": setup["metadata"],
|
|
"validation": setup["validation"],
|
|
"inputSummary": [
|
|
{
|
|
"key": item["key"],
|
|
"label": item.get("label", item["key"]),
|
|
"type": item["type"],
|
|
"required": bool(item.get("required", False)),
|
|
}
|
|
for item in setup["inputContract"].get("inputs", [])
|
|
],
|
|
"outputSummary": [
|
|
{
|
|
"key": item["key"],
|
|
"label": item.get("label", item["key"]),
|
|
"type": item["type"],
|
|
}
|
|
for item in setup["outputContract"].get("outputs", [])
|
|
],
|
|
}
|