(workflow: dict[str, Any], *, prompt: str | None, seed: int | None)
| 499 | |
| 500 | |
| 501 | def _patch_comfy_workflow(workflow: dict[str, Any], *, prompt: str | None, seed: int | None) -> dict[str, Any]: |
| 502 | workflow = json.loads(json.dumps(workflow)) |
| 503 | nodes = workflow.get("prompt", workflow) |
| 504 | if not isinstance(nodes, dict): |
| 505 | raise ModlyCliError("ComfyUI workflow must be API format (top-level node-id object, or {'prompt': {...}})") |
| 506 | if "nodes" in workflow and "links" in workflow: |
| 507 | raise ModlyCliError("ComfyUI workflow is editor format; export it as API format first") |
| 508 | |
| 509 | if prompt is not None: |
| 510 | patched = False |
| 511 | for node in nodes.values(): |
| 512 | if not isinstance(node, dict): |
| 513 | continue |
| 514 | class_type = str(node.get("class_type", "")).lower() |
| 515 | inputs = node.get("inputs") if isinstance(node.get("inputs"), dict) else {} |
| 516 | text = str(inputs.get("text", "")).lower() |
| 517 | if "cliptextencode" in class_type and "negative" not in text: |
| 518 | inputs["text"] = prompt |
| 519 | patched = True |
| 520 | break |
| 521 | if not patched: |
| 522 | for node in nodes.values(): |
| 523 | if isinstance(node, dict) and isinstance(node.get("inputs"), dict): |
| 524 | inputs = node["inputs"] |
| 525 | for key in ("prompt", "positive", "text"): |
| 526 | if key in inputs and isinstance(inputs[key], str): |
| 527 | inputs[key] = prompt |
| 528 | patched = True |
| 529 | break |
| 530 | if patched: |
| 531 | break |
| 532 | if not patched: |
| 533 | raise ModlyCliError("Could not find a text/prompt input to patch in ComfyUI workflow") |
| 534 | |
| 535 | if seed is not None: |
| 536 | for node in nodes.values(): |
| 537 | if not isinstance(node, dict) or not isinstance(node.get("inputs"), dict): |
| 538 | continue |
| 539 | inputs = node["inputs"] |
| 540 | for key in ("seed", "noise_seed"): |
| 541 | if key in inputs and isinstance(inputs[key], int): |
| 542 | inputs[key] = seed |
| 543 | return nodes |
| 544 | |
| 545 | |
| 546 | def _run_comfy_workflow(args: argparse.Namespace) -> dict[str, Any]: |
no test coverage detected