#!/usr/bin/env python3 # -*- coding: utf-8 -*- """验证 comfyui-AICG3D 自带工作流「11.AICG3D_测试样板」能否直接使用。 用法： python3 aicg3d_run.py original # 原样提交（预期被校验拒绝） python3 aicg3d_run.py fixed # 换成本机资源后提交（预期能出片） """ import json import sys import time import urllib.error import urllib.request BASE = "http://127.0.0.1:8189" OUR_IMAGE = "h3_rv2v_char_grok.jpg" # 本机 input/ 里已有的参考图 PROMPT = ( "角色与参考图完全一致，微微转头看向镜头并自然眨眼，" "镜头缓慢推近，头发随动作轻微摆动，背景保持静止。" ) LORA_SLOTS = ("lora_1", "lora_2", "lora_3", "lora_4", "lora_5", "lora_6", "lora_7", "lora_8", "lora_9") def loader_inputs(fl2va, ref2va, lora1, non_h3="自动回落"): d = { "fl2va_model": fl2va, "ref2va_model": ref2va, "text_encoder": "qwen3vl_32b_minimax_h3_nvfp4_awq.safetensors", "video_vae": "minimax_h3_video_vae_int8_convrot.safetensors", "audio_vae": "minimax_h3_audio_vae_fp32.safetensors", } for i, name in enumerate(LORA_SLOTS): d[name] = lora1 if i == 0 else "none" d[name + "_strength"] = 1.0 d["non_h3_policy"] = non_h3 return d def build(mode="fixed"): if mode == "original": # ---- 完全照抄工作流 JSON 里的值 ---- loader = loader_inputs( "MiniMax/Minimax-h3_Singularity_ref2va_v1.3_int8.safetensors", "MiniMax/minimax_h3_fused_refdelta_r1024_turbo8_mystic07_int8_convrot.safetensors", "minimax\\taomate_h3_3step_comfy.safetensors", # 注意：Windows 反斜杠 ) img = "oMAEICyBGQF62YfOAlLAOI7ttFehVAJIIegFBW~noop.jpeg" h3 = { "mode": "图生或首尾帧", "resolution": "720P", "aspect_ratio": "9:16", "width": 1344, "height": 768, "seconds": 5, "advanced": False, "fps": 24, "keyframe_role": "首帧优先", "ref_image_size": "1K 像素（1024×1024 等比）", "reference_mention_mode": "按序号", "prompt_optimizer_settings": False, "prompt_optimizer_scene_guide": "通用", } steps, seed = 3, 921547326318013 prefix = "video/AICG3D_original" elif mode == "author": # ---- 用作者写在 13 号工作流里的模型/LoRA 原值（含反斜杠 LoRA），枚举用前端翻译后的英文 ---- loader = loader_inputs( "MiniMax/Minimax-h3_Singularity_ref2va_v1.3_int8.safetensors", "MiniMax/minimax_h3_fused_refdelta_r1024_turbo8_mystic07_int8_convrot.safetensors", "minimax\\taomate_h3_3step_comfy.safetensors", ) img = OUR_IMAGE h3 = { "mode": "image", "resolution": "768P", "aspect_ratio": "16:9", "width": 1344, "height": 768, "seconds": 5, "advanced": False, "fps": 24, "keyframe_role": "first", "ref_image_size": "1k", "reference_mention_mode": "index", "prompt_optimizer_settings": False, "prompt_optimizer_scene_guide": "none", } steps, seed = 3, 20260928 # TaoMate 3 步 LoRA → 3 步 prefix = "video/AICG3D_author_assets" else: # ---- 换成本机真实存在的资源 ---- loader = loader_inputs( "minimax_h3_fl2va_pruned_int8_convrot.safetensors", "minimax_h3_ref2va_pruned_int8_convrot.safetensors", "minimax_h3_fl2v_turbo_4step_v1.0_768p_comfyui_bf16.safetensors", ) img = OUR_IMAGE h3 = { "mode": "image", "resolution": "768P", "aspect_ratio": "16:9", "width": 1344, "height": 768, "seconds": 5, "advanced": False, "fps": 24, "keyframe_role": "first", "ref_image_size": "match", "reference_mention_mode": "index", "prompt_optimizer_settings": False, "prompt_optimizer_scene_guide": "none", } steps, seed = 4, 20260927 prefix = "video/AICG3D_test_fixed" h3.update({ "h3_bundle": ["1", 0], "media": ["2", 0], "prompt": PROMPT, }) media = {"media_state": json.dumps( {"images": [{"filename": img}], "audios": [], "videos": []}, ensure_ascii=False)} render = { "h3_context": ["3", 1], "noise_seed": seed, "sampler_name": "res_multistep", "scheduler": "simple", "steps": steps, "denoise": 1.0, "cleanup_after_run": "卸载模型", } save = {"video": ["4", 0], "filename_prefix": prefix, "format": "auto", "codec": "auto"} return { "1": {"class_type": "AICG3D_H3Loader", "inputs": loader}, "2": {"class_type": "AICG3D_H3MediaLoader", "inputs": media}, "3": {"class_type": "AICG3D_H3", "inputs": h3}, "4": {"class_type": "AICG3D_H3RenderAdvanced", "inputs": render}, "5": {"class_type": "SaveVideo", "inputs": save}, } def post(path, payload): req = urllib.request.Request( BASE + path, data=json.dumps(payload).encode("utf-8"), headers={"Content-Type": "application/json"}) try: with urllib.request.urlopen(req, timeout=60) as r: return r.status, json.loads(r.read()) except urllib.error.HTTPError as e: return e.code, e.read().decode("utf-8", "replace") def main(): mode = sys.argv[1] if len(sys.argv) > 1 else "fixed" prompt = build(mode) open(f"/tmp/aicg3d_prompt_{mode}.json", "w", encoding="utf-8").write( json.dumps(prompt, ensure_ascii=False, indent=1)) print(f"=== 提交模式：{mode} ===", flush=True) code, resp = post("/prompt", {"prompt": prompt, "client_id": "aicg3d-probe"}) if code != 200: print(f"HTTP {code} —— 被拒绝。原始返回：", flush=True) print(resp if isinstance(resp, str) else json.dumps(resp, ensure_ascii=False), flush=True) return 1 pid = resp.get("prompt_id") print(f"HTTP 200 —— 已进入队列，prompt_id={pid}", flush=True) if mode == "original": return 0 t0 = time.time() while True: time.sleep(10) with urllib.request.urlopen(BASE + f"/history/{pid}", timeout=30) as r: hist = json.loads(r.read()) if pid in hist: entry = hist[pid] st = entry.get("status", {}) print(f"[{int(time.time()-t0)}s] status={st.get('status_str')} completed={st.get('completed')}", flush=True) if st.get("completed") or st.get("status_str") == "error": for m in st.get("messages", [])[-6:]: print(" 消息:", json.dumps(m, ensure_ascii=False)[:300], flush=True) for nid, out in (entry.get("outputs") or {}).items(): print(f" 输出节点 {nid}: {json.dumps(out, ensure_ascii=False)[:400]}", flush=True) return 0 if st.get("completed") else 2 else: with urllib.request.urlopen(BASE + "/queue", timeout=30) as r: q = json.loads(r.read()) run = [x for x in q.get("queue_running", []) if x[1] == pid] pend = [x for x in q.get("queue_pending", []) if x[1] == pid] print(f"[{int(time.time()-t0)}s] 运行中={bool(run)} 排队中={bool(pend)}", flush=True) if not run and not pend: print("不在队列也不在历史里 —— 可能已失败", flush=True) return 3 if __name__ == "__main__": sys.exit(main())