"""End-to-end ComfyUI workflow submission test.""" import json, time, urllib.request, urllib.error, uuid COMFY = "http://127.0.0.1:8188" WF_PATH = r"D:\Comfy\App\ComfyUI_windows_portable\ComfyUI\custom_nodes\ComfyUI-MusicVideoDirector\workflow_music_video_director.json" def http_json(method, path, body=None): url = COMFY + path data = json.dumps(body).encode() if body is not None else None req = urllib.request.Request(url, data=data, method=method, headers={"Content-Type": "application/json"}) try: with urllib.request.urlopen(req, timeout=30) as r: return r.status, json.loads(r.read().decode()) except urllib.error.HTTPError as e: return e.code, json.loads(e.read().decode()) # === Load UI workflow === wf_ui = json.load(open(WF_PATH, encoding="utf-8")) # Load object_info for schema-driven conversion obj_info = json.load(open(r"D:\Comfy\object_info.json", encoding="utf-8")) # === Build node id → node map and links map === nodes_by_id = {n["id"]: n for n in wf_ui["nodes"]} # link format: [link_id, src_node, src_slot, dst_node, dst_slot, type] links_by_id = {l[0]: l for l in wf_ui["links"]} # === UI → API conversion === # Strategy: For each node, get its INPUT_TYPES schema (req + opt order). # - If input name is connected (in node["inputs"] with link), use [src_node_id, src_slot]. # - Otherwise pull from widgets_values in schema order. # Widget inputs come AFTER linked inputs in the schema iteration. def convert(wf_ui, mute_ids=set(), force_dummy=None): api = {} for n in wf_ui["nodes"]: nid = n["id"] if n.get("mode") == 2 or nid in mute_ids: continue ctype = n["type"] if ctype == "Note": continue if ctype not in obj_info: print(f" ! Unknown class_type: {ctype}") continue schema = obj_info[ctype]["input"] req = schema.get("required", {}) opt = schema.get("optional", {}) all_inputs = list(req.keys()) + list(opt.keys()) # Map of input name → linked source # Also track "widget-converted-to-input" names so we skip their stale widget slot link_inputs = {} widget_converted_names = set() for inp in n.get("inputs", []) or []: if inp.get("widget"): widget_converted_names.add(inp["name"]) if inp.get("link") is not None: lk = links_by_id.get(inp["link"]) if lk: link_inputs[inp["name"]] = [str(lk[1]), lk[2]] widgets = list(n.get("widgets_values", []) or []) widget_idx = 0 inputs = {} for name in all_inputs: if name in link_inputs: inputs[name] = link_inputs[name] # If this was a widget converted to input, advance past its stale widget slot if name in widget_converted_names and widget_idx < len(widgets): widget_idx += 1 else: # Determine if this input is a widget (not a socket-only input) spec = req.get(name) or opt.get(name) # spec is either [type_str, options_dict] or [list_of_choices] if widget_idx < len(widgets): inputs[name] = widgets[widget_idx] widget_idx += 1 else: # missing — try default from spec if isinstance(spec, list) and len(spec) > 1 and isinstance(spec[1], dict): if "default" in spec[1]: inputs[name] = spec[1]["default"] # Apply forced overrides if force_dummy and nid in force_dummy: inputs.update(force_dummy[nid]) api[str(nid)] = {"class_type": ctype, "inputs": inputs} return api # Strategy A: mute everything heavy, feed SegmentEditor with dummy segments_json # We want to actually exercise OllamaPromptGen (id=6). It needs: # - segments (from SegmentEditor id=4) # - extra_style_notes (from LLMChat id=20) — let's keep that too # SegmentEditor needs segments_in (link) OR can be standalone with segments_json widget. # So we MUTE: 1 (AudioLoader), 2 (Whisper), 3 (SmartSegmenter), 5 (Splitter), # and all image-gen nodes 8-17. # Keep: 4 (SegmentEditor with dummy json), 6 (OllamaPromptGen), 7 (CharBank), 20 (LLMChat) mute = {1, 2, 3, 5, 8, 9, 10, 11, 12, 13, 14, 16, 17} # keep 15 (PromptDisplay, OUTPUT_NODE) dummy_segments = json.dumps([ {"id": 0, "start": 0.0, "end": 8.0, "text": "Istanbul'da yagmur yagiyor"}, {"id": 1, "start": 8.0, "end": 16.0, "text": "Cati ustunde durup gunbatimini izliyorum"}, ]) force = { 4: {"segments_json": dummy_segments}, # SegmentEditor standalone # Workflow's stale model name (supergemma) is not installed — swap to a real one 6: { "ollama_model": "llama3.2:3b-instruct-q4_K_M", }, 20: { "model": "(otomatik)", }, } # Remove the segments_in link from node 4 (since upstream is muted) # Patch: clear that input from node 4 by removing it before conversion n4 = nodes_by_id[4] n4["inputs"] = [i for i in n4.get("inputs", []) if i.get("name") != "segments_in"] api_wf = convert(wf_ui, mute_ids=mute, force_dummy=force) print("=== Converted nodes ===") for k, v in api_wf.items(): print(f" [{k}] {v['class_type']} inputs={list(v['inputs'].keys())}") print("\n=== Submitting to /prompt ===") client_id = str(uuid.uuid4()) status, resp = http_json("POST", "/prompt", {"prompt": api_wf, "client_id": client_id}) print(f"HTTP {status}") print(json.dumps(resp, indent=2, ensure_ascii=False)[:3000]) if status == 200 and "prompt_id" in resp: pid = resp["prompt_id"] print(f"\n=== Polling /history/{pid} ===") for i in range(60): time.sleep(2) st, hist = http_json("GET", f"/history/{pid}") if hist and pid in hist: h = hist[pid] status_info = h.get("status", {}) print(f" [{i*2}s] status={status_info.get('status_str')} completed={status_info.get('completed')}") if status_info.get("completed") or status_info.get("status_str") in ("success", "error"): print("\n=== FINAL ===") print(json.dumps(status_info, indent=2, ensure_ascii=False)[:2000]) # Pull outputs outputs = h.get("outputs", {}) for nid, out in outputs.items(): print(f"\n Node {nid} ({api_wf.get(nid,{}).get('class_type','?')}) outputs:") print(" ", json.dumps(out, ensure_ascii=False)[:500]) break else: # Check queue _, q = http_json("GET", "/queue") print(f" [{i*2}s] queue running={len(q.get('queue_running',[]))} pending={len(q.get('queue_pending',[]))}")