from __future__ import annotations
import copy
import hashlib
import json
import math
import random
import re
from dataclasses import dataclass
from pathlib import Path
from typing import Any
from h3_audio_intent import plan_h3_audio
from h3_audio_prompt import normalized_h3_audio_prompt
WORKFLOW_DIR = Path(__file__).resolve().parent / "workflows"
MAX_VIDEO_DURATION_SECONDS = 15
CAPABILITY_BY_MODE = {"t2i": "image", "i2i": "image", "t2i_i2i": "image", "text": "video", "image": "video"}
VISIBLE_TEXT_CONFLICTS = {
"text", "words", "readable words", "readable text", "letters", "digits", "numbers",
"chinese text", "chinese characters", "random text", "garbled text", "gibberish text",
"pseudo-text", "pseudo-chinese glyphs", "malformed han characters", "mojibake",
"corrupted glyphs", "labels", "signage", "typography", "text overlay", "logo",
"brand logo", "brand name", "trademark", "printed words", "printed numbers",
}
H3_DIALOGUE_TAG = re.compile(r"\[(?P[^\]]+)](?P.*?)", re.IGNORECASE | re.DOTALL)
H3_SPEECH_PREFIX = r"(?P(?:口播(?:内容|文案)?|台词|对白|旁白|朗读|说(?:道|出)?|讲(?:道)?|\b(?:says?|said|speaks?|spoke|narrates?)\b)\s*[：:]?\s*)"
H3_QUOTED_DIALOGUE = tuple(
re.compile(
H3_SPEECH_PREFIX + re.escape(left) + r"(?P(?:\\.|[^" + re.escape(right) + r"\\]"
+ (r"|(?<=[A-Za-z])'(?=[A-Za-z])" if right == "'" else "") + r")*)" + re.escape(right),
re.IGNORECASE,
)
for left, right in (("“", "”"), ('"', '"'), ("‘", "’"), ("'", "'"), ("「", "」"), ("『", "』"))
)
H3_BARE_DIALOGUE = re.compile(r"(?:^|(?<=\t))(?P[ ]*(?:口播(?:内容|文案)?|台词|对白|旁白|dialogue|voiceover|narration)[ ]*[：:][ ]*)(?P[^\t\r\n]*)(?=[\t\r\n]|$)", re.IGNORECASE | re.MULTILINE)
H3_QUOTED_SPEECH_START = re.compile(H3_SPEECH_PREFIX + r"[\"“‘'「『]", re.IGNORECASE)
H3_COMPLETE_SILENCE = re.compile(
r"(?:全程|完全|绝对)[ \t]*(?:绝对)?[ \t]*静音"
r"|(?:不要|无需|不需要|禁止|关闭|移除|不生成|不输出)[ \t]*(?:任何|全部|所有)?[ \t]*(?:音频|声音|音轨)"
r"|\b(?:no audio|mute all audio|complete silence|silent video)\b",
re.IGNORECASE,
)
H3_AUDIO_OUTPUT_POLICY_VERSION = "h3-audio-plan-verified-v25"
H3_CAPTION_PIPELINE_VERSION = "native-full-attention-realism04-manju-v24"
H3_VERIFIED_TEXT_ENCODER = "qwen3vl_32b_minimax_h3_int8_convrot.safetensors"
H3_CLEAR_FRAME_RULE = (
"No text, no subtitles, no captions, no watermark anywhere in the frame."
)
# Visual mouth movements or chatting are not consent to invent spoken words.
H3_SPEECH_INTENT = re.compile(r"口播|台词|对白|旁白|朗读|配音|(?:输出|生成|需要|播放)[^，。！？；\n]{0,12}(?:语音|人声)|(?:说(?:道|出)?|讲(?:道)?)\s*[：:\"“‘'「『]|\b(?:narrates?|narrator|dialogue|narration|voice[- ]?over|spoken words|spoken audio)\b", re.IGNORECASE)
H3_SILENT_CLAUSE = re.compile(
r"(?:无|没有|不要|不许|禁止|无需|不需要|不)(?:(?:任何|额外的?|其他|背景|人物|角色|进行|出现|加入|添加|自动|输出|生成|播放|允许|需要|加|再)){0,3}(?:口播|台词|对白|旁白|说话|讲话|聊天|交谈|朗读|配音|人声|语音)"
r"|\b(?:no|not|without|never)\s+(?:(?:additional|other|background|characters?|one|any|human|audible)\s+){0,4}(?:speech|dialogue|narration|voiceover|talking|speaking|speaks?|talks?)(?:\s*(?:,|or|and)\s*(?:speech|dialogue|narration|voiceover|talking|speaking|voices?))*",
re.IGNORECASE,
)
H3_NO_SPEECH_RULE = (
"H3 audio compliance: Background-only audio. Preserve the requested environmental ambience, "
"physical sound effects and instrumental background music. Do not add music unless requested. "
"No character speaks audibly. No voiceover, narration, dialogue, lyrics, chanting, "
"background conversation or invented language is audible. Do not generate human words or syllables. "
"Keep the visual actions in the prompt unchanged."
)
H3_DIALOGUE_RULE = (
"H3 audio compliance: Keep the speaker identities, on-screen/off-screen positions and audio sources described above. "
"Render the tagged dialogue in its declared language and written order. "
"Do not translate, paraphrase, replace, repeat, or invent any syllable beyond the written dialogue. "
"Keep all non-speech audio requirements unchanged."
)
@dataclass(frozen=True)
class WorkflowSpec:
key: str
name: str
mode: str
file: str
raw: dict[str, Any]
revision: str
@property
def capability(self) -> str:
return CAPABILITY_BY_MODE[self.mode]
@property
def path(self) -> Path:
return WORKFLOW_DIR / self.file
def load_registry() -> dict[str, WorkflowSpec]:
manifest = json.loads((WORKFLOW_DIR / "manifest.json").read_text(encoding="utf-8"))
result: dict[str, WorkflowSpec] = {}
for row in manifest.get("workflows") or []:
if not row.get("enabled", True):
continue
mode = str(row.get("mode") or "")
if mode not in CAPABILITY_BY_MODE:
raise ValueError(f"unsupported workflow mode: {mode}")
path = WORKFLOW_DIR / str(row["file"])
payload = path.read_bytes()
revision_material = payload + b"\0" + str(row.get("revision_salt") or "").encode("utf-8")
if row.get("speech_prompt_policy") == "minimax_h3":
revision_material += b"\0" + H3_AUDIO_OUTPUT_POLICY_VERSION.encode("utf-8")
revision_material += b"\0" + H3_CAPTION_PIPELINE_VERSION.encode("utf-8")
key = str(row["key"])
result[key] = WorkflowSpec(
key=key,
name=str(row.get("name") or key),
mode=mode,
file=str(row["file"]),
raw=dict(row),
revision=hashlib.sha256(revision_material).hexdigest()[:16],
)
return result
def public_workflow(spec: WorkflowSpec) -> dict[str, Any]:
return {
"key": spec.key,
"name": spec.name,
"mode": spec.mode,
"capability": spec.capability,
"revision": spec.revision,
"supportsReferences": bool(spec.raw.get("load_image_node") or spec.raw.get("reference_node") or spec.raw.get("image_refs_node")),
}
def public_model(spec: WorkflowSpec) -> dict[str, Any]:
supports_references = bool(spec.raw.get("load_image_node") or spec.raw.get("reference_node") or spec.raw.get("image_refs_node"))
if spec.mode == "t2i":
operations = ["text_to_image"]
min_images, max_images = 0, 0
elif spec.mode == "i2i":
operations = ["image_to_image"]
min_images, max_images = 1, 1
elif spec.mode == "t2i_i2i":
operations = ["text_to_image", "image_to_image"]
min_images, max_images = 0, int(spec.raw.get("max_reference_images") or 6)
elif spec.mode == "text":
operations = ["text_to_video"]
if supports_references:
operations.append("image_to_video")
min_images, max_images = 0, 1 if supports_references else 0
else:
operations = ["image_to_video"]
if spec.raw.get("reference_node"):
operations.append("reference_to_video")
min_images, max_images = 1, 9
else:
min_images, max_images = 1, 1
aspect_ratios = ["1:1", "3:2", "2:3", "4:3", "3:4", "16:9", "9:16", "21:9"]
endpoint_types = ["comfyui-workflow", *operations]
item: dict[str, Any] = {
"id": spec.key,
"object": "model",
"owned_by": "comfyui-workflow",
"display_name": spec.name,
"model_type": spec.capability,
"supported_endpoint_types": endpoint_types,
"supports_images": supports_references,
"min_images": min_images,
"max_images": max_images,
"default_parameters": {"aspect_ratio": "1:1" if spec.capability == "image" else "16:9"},
"options": {
"aspect_ratio": [{"value": value, "label": value} for value in aspect_ratios],
},
}
if spec.capability == "video":
item["default_parameters"].update({"duration_seconds": "5", "resolution": "720p"})
item["options"].update({
"duration_seconds": [{"value": str(value), "label": f"{value} 秒"} for value in range(1, MAX_VIDEO_DURATION_SECONDS + 1)],
"resolution": [{"value": value, "label": value.upper()} for value in ("480p", "720p", "1080p")],
})
return item
def _patch_node(workflow: dict[str, Any], node_id: str | None, values: dict[str, Any]) -> None:
if not node_id:
return
if node_id not in workflow:
raise ValueError(f"workflow is missing configured node {node_id}")
workflow[node_id].setdefault("inputs", {}).update(values)
def _fit_aligned_resolution(
width: int,
height: int,
*,
max_megapixels: float | None,
align: int,
) -> tuple[int, int]:
"""Preserve aspect ratio while fitting the first pass to the workflow budget."""
align = max(1, int(align))
target_width = max(align, int(width))
target_height = max(align, int(height))
if max_megapixels is not None and float(max_megapixels) > 0:
max_pixels = float(max_megapixels) * 1024 * 1024
area = target_width * target_height
if area > max_pixels:
scale = math.sqrt(max_pixels / area)
target_width = max(align, round(target_width * scale))
target_height = max(align, round(target_height * scale))
target_width = max(align, round(target_width / align) * align)
target_height = max(align, round(target_height / align) * align)
if max_megapixels is not None and float(max_megapixels) > 0:
max_pixels = float(max_megapixels) * 1024 * 1024
while target_width * target_height > max_pixels:
if target_width >= target_height and target_width > align:
target_width -= align
elif target_height > align:
target_height -= align
else:
break
return target_width, target_height
def _negative_prompt(spec: WorkflowSpec, current: str, requested: str) -> str:
parts: list[str] = []
seen: set[str] = set()
for value in (current, requested):
for raw in value.split(","):
term = raw.strip(" .")
normalized = " ".join(term.casefold().split())
if not term or normalized in seen:
continue
if spec.raw.get("preserve_visible_text") and normalized in VISIBLE_TEXT_CONFLICTS:
continue
seen.add(normalized)
parts.append(term)
return ", ".join(parts)
def _spoken_language(value: str) -> str:
if re.search(r"[\u3040-\u30ff]", value):
return "Japanese"
if re.search(r"[\uac00-\ud7af]", value):
return "Korean"
if re.search(r"[\u3400-\u9fff]", value):
return "Chinese"
return "English"
def _h3_audio_prompt(prompt: str, duration: float) -> str:
return normalized_h3_audio_prompt(prompt, duration)
def _h3_without_native_caption_instructions(prompt: str) -> str:
instruction = re.compile(
r"(?:请|需要|要求|同时|并且|并|自动|不要|不需要|无需|禁止|关闭|不显示|不加|不添加|无|添加|显示|生成|加上|加入|配上|带|烧录)+"
r"(?:任何|中文|英文|双语|同步|口播|对白|底部|居中|的)*字幕"
r"(?:[：:][^，,。；;\n]*)?"
r"|\b(?:add|show|generate|include|burn(?: in)?|with|no|without|disable)\s+"
r"(?:(?:chinese|english|bilingual|synchronized|any)\s+)?(?:subtitles|captions)\b", re.IGNORECASE)
protected = re.compile(H3_DIALOGUE_TAG.pattern + r'|“[^”]*”|"[^"\n]*"|「[^」]*」|『[^』]*』', re.IGNORECASE | re.DOTALL)
parts, cursor = [], 0
for match in protected.finditer(prompt):
parts.extend([instruction.sub("", prompt[cursor:match.start()]), match.group(0)])
cursor = match.end()
parts.append(instruction.sub("", prompt[cursor:]))
return "".join(parts)
def _apply_h3_generation_stage(graph: dict[str, Any]) -> None:
stage_id = "h3_generation_stage"
if stage_id in graph:
raise ValueError("H3_SUBTITLE_GRAPH_UNSUPPORTED: 生成阶段节点标识冲突")
loaders = [
("unet", "127", "UNETLoader", {"unet_name", "weight_dtype"}, set()),
("clip", "128", "CLIPLoader", {"clip_name", "type"}, {"device"}),
("video_vae", "119", "VAELoader", {"vae_name"}, set()),
("audio_vae", "120", "VAELoader", {"vae_name"}, set()),
]
models = {}
ports = {}
for index, (name, node_id, kind, required, optional) in enumerate(loaders):
node = graph.get(node_id, {})
inputs = node.get("inputs")
if (node.get("class_type") != kind or not isinstance(inputs, dict)
or not required <= set(inputs) or not set(inputs) <= required | optional
or any(not isinstance(v, str) or not v for v in inputs.values())):
raise ValueError("H3_SUBTITLE_GRAPH_UNSUPPORTED: 生成模型入口与已验证的阶段协议不一致")
models[name] = dict(inputs)
ports[node_id] = index
def rewire(value: Any) -> Any:
if isinstance(value, list):
if len(value) == 2 and isinstance(value[0], str) and value[0] in ports:
if type(value[1]) is not int or value[1] != 0:
raise ValueError("H3_SUBTITLE_GRAPH_UNSUPPORTED: 生成模型输出端口无效")
return [stage_id, ports[value[0]]]
return [rewire(item) for item in value]
if isinstance(value, dict):
return {key: rewire(item) for key, item in value.items()}
return value
# The gate receives filenames, not already-loaded GPU objects. Loader
# expansion occurs only after the private vision service confirms sleep.
for node in graph.values():
if "inputs" in node:
node["inputs"] = rewire(node["inputs"])
for node_id in ports:
del graph[node_id]
graph[stage_id] = {"class_type": "H3GenerationModelStage",
"inputs": {"models_json": json.dumps(models, ensure_ascii=False)}}
def _h3_manju_background_layout(value: str, references: int) -> str | None:
# Keep the GPU-verified layout narrow: literal speech, explicit sound/music
# and multi-reference identity assignments retain the original path.
if not value.endswith(H3_NO_SPEECH_RULE) or H3_DIALOGUE_TAG.search(value) or references > 1:
return None
scene = value[:-len(H3_NO_SPEECH_RULE)].rstrip()
if re.search(r'音乐|配乐|音效|环境音|声音|静音|声响|\b(?:music|BGM|sound|audio|silent|silence)\b', scene, re.IGNORECASE):
return None
reference = (
'Picture 1 is the reference for the people, setting and product. Preserve their appearance, '
'including the existing printing on the product packaging.\n\n'
) if references == 1 else ''
return (reference + 'integrated_multimodal_description:\n' + scene
+ '\n\noverall_soundscape: Environmental ambience only. The people remain silent throughout.'
+ '\n\nnon_diegetic_music: N/A'
+ '\n\nNo added text, no subtitles, no captions, no watermark anywhere in the frame.')
def _apply_h3_frame_policy(spec: WorkflowSpec, graph: dict[str, Any], request: dict[str, Any]) -> None:
metadata = request.get("metadata") or {}
if not isinstance(metadata, dict):
raise ValueError("H3 metadata must be an object")
for key in ("h3_native_subtitles", "h3_suppress_subtitles", "h3_subtitles"):
if key in metadata and not isinstance(metadata[key], bool):
raise ValueError(f"{key} must be a boolean")
if metadata.get("h3_native_subtitles") is True or metadata.get("h3_suppress_subtitles") is False:
raise ValueError("原生字幕已禁用，当前仅保留生成阶段字幕抑制，字幕烧录已暂停")
encoder = graph.get("128", {})
if encoder.get("class_type") != "CLIPLoader" or encoder.get("inputs", {}).get("type") != "minimax":
raise ValueError("H3_SUBTITLE_GRAPH_UNSUPPORTED: 当前工作流缺少已验证的 H3 文本编码器入口")
# The identical-seed P2 comparison passed literal speech validation only
# with the reference-quality encoder. Keep sampler, LoRA and audio untouched.
encoder["inputs"]["clip_name"] = H3_VERIFIED_TEXT_ENCODER
save = graph[str(spec.raw.get("save_node"))]
if save.get("class_type") != "VHS_VideoCombine" or "images" not in save.get("inputs", {}):
raise ValueError("H3_SUBTITLE_GRAPH_UNSUPPORTED: 当前工作流缺少可校验的视频帧输出")
if any(n.get("class_type") in {"BurnChineseSubtitles", "SoftChineseSubtitles", "H3VerifiedSubtitleFrames"} for n in graph.values()):
raise ValueError("H3_SUBTITLE_LEGACY_NODE: 字幕烧录已暂停，工作流仍包含旧字幕节点")
if any(n.get("class_type") in {"H3CaptionPipeline", "H3CaptionFinalize", "H3STTNRepair"}
for n in graph.values()):
raise ValueError("H3_SUBTITLE_GRAPH_UNSUPPORTED: 字幕后处理已暂停，工作流仍包含后处理节点")
# Operator policy: even a stored h3_subtitles=True does not re-enable the
# expensive cleanup/burn-in chain. Leave the raw decoded frames and the
# existing opt-in audio connection intact; this is suppression, not erasure.
visited_prompt_fields = set()
for node_key, input_key in (("prompt_node", "prompt_input"), ("motion_node", "motion_input")):
node = spec.raw.get(node_key)
if node:
key = str(spec.raw.get(input_key) or "text")
field = (str(node), key)
if field in visited_prompt_fields:
continue
visited_prompt_fields.add(field)
value = str(graph[str(node)]["inputs"].get(key) or "")
value = _h3_without_native_caption_instructions(value).rstrip()
structured = None
if spec.raw.get('h3_background_prompt_layout') == 'manju-v1':
reference_count = sum(k.startswith('canvas_reference_') for k in graph)
structured = _h3_manju_background_layout(value, reference_count)
graph[str(node)]["inputs"][key] = structured or (value + "\n\n" + H3_CLEAR_FRAME_RULE)
_apply_h3_generation_stage(graph)
def validate_h3_references(spec: WorkflowSpec, count: int) -> None:
if spec.raw.get("speech_prompt_policy") != "minimax_h3":
return
catalog = public_model(spec)
if catalog["max_images"] > 0 and count > catalog["max_images"]:
raise ValueError(f"H3_AUDIO_CONDITIONING_INVALID: this workflow supports at most {catalog['max_images']} reference images")
if not catalog["min_images"] <= count <= catalog["max_images"]:
raise ValueError("H3_AUDIO_CONDITIONING_INVALID: 参考图片数量与工作流不匹配，请选择对应图生或文生视频工作流；未忽略参考图")
def compile_workflow(spec: WorkflowSpec, request: dict[str, Any], uploaded_images: list[str], job_id: str, *, preserve_requested_audio: bool = False) -> dict[str, Any]:
if spec.capability == "video":
duration = float(request.get("duration") if request.get("duration") is not None else 5)
if not math.isfinite(duration) or duration <= 0 or duration > MAX_VIDEO_DURATION_SECONDS:
raise ValueError("视频时长必须大于 0 且最多 15 秒")
validate_h3_references(spec, len(uploaded_images))
workflow = copy.deepcopy(json.loads(spec.path.read_text(encoding="utf-8")))
prompt = str(request.get("prompt") or "").strip()
if not prompt:
raise ValueError("prompt is required")
native_audio_control = bool(spec.raw.get("native_audio_control"))
generate_audio = request.get("generate_audio")
if native_audio_control and generate_audio is None:
generate_audio = bool(spec.raw.get("native_audio_default", True))
if native_audio_control and generate_audio and spec.raw.get("speech_prompt_policy") == "minimax_h3":
audio_plan = plan_h3_audio(prompt, float(request.get("duration") or 5))
prompt = _h3_audio_prompt(prompt, float(request.get("duration") or 5))
# Explicit effects/music use the native track only behind the new ASR
# output gate. Unrequested audio remains disconnected as in v24.
generate_audio = not audio_plan["silent"] if preserve_requested_audio else audio_plan["audio_mode"] in {"dialogue", "effects"}
negative_prompt = str(request.get("negative_prompt") or "").strip()
seed = request.get("seed")
if seed is None:
seed = random.SystemRandom().randint(0, 2**63 - 1)
_patch_node(workflow, spec.raw.get("prompt_node"), {str(spec.raw.get("prompt_input") or "text"): prompt})
_patch_node(workflow, spec.raw.get("motion_node"), {str(spec.raw.get("motion_input") or "text"): prompt})
_patch_node(workflow, spec.raw.get("seed_node"), {str(spec.raw.get("seed_input") or "noise_seed"): int(seed)})
if negative_prompt and spec.raw.get("negative_node"):
node_id = str(spec.raw["negative_node"])
input_key = str(spec.raw.get("negative_input") or "text")
current = str((workflow.get(node_id, {}).get("inputs") or {}).get(input_key) or "").strip()
_patch_node(workflow, node_id, {input_key: _negative_prompt(spec, current, negative_prompt)})
requested_width = request.get("width")
requested_height = request.get("height")
if spec.raw.get("h3_native_optimized"):
frame = workflow[str(spec.raw["width_node"])]["inputs"]
width = int(requested_width if requested_width is not None else frame[str(spec.raw["width_input"])])
height = int(requested_height if requested_height is not None else frame[str(spec.raw["height_input"])])
width, height = _fit_aligned_resolution(width, height, max_megapixels=None,
align=int(spec.raw.get("dimension_align") or 32))
if width * height > 4096 * 2160:
raise ValueError("H3 字幕处理支持的最大画面面积为 4096×2160，请调整分辨率后提交")
_patch_node(workflow, spec.raw["width_node"], {str(spec.raw["width_input"]): width})
_patch_node(workflow, spec.raw["height_node"], {str(spec.raw["height_input"]): height})
elif requested_width is not None and requested_height is not None and spec.raw.get("base_max_megapixels") is not None:
align = max(1, int(spec.raw.get("dimension_align") or 1))
base_width, base_height = _fit_aligned_resolution(
int(requested_width),
int(requested_height),
max_megapixels=float(spec.raw["base_max_megapixels"]),
align=align,
)
_patch_node(workflow, spec.raw.get("width_node"), {str(spec.raw.get("width_input") or "value"): base_width})
_patch_node(workflow, spec.raw.get("height_node"), {str(spec.raw.get("height_input") or "value"): base_height})
if spec.raw.get("upscale_node"):
final_width, final_height = _fit_aligned_resolution(
int(requested_width),
int(requested_height),
max_megapixels=None,
align=align,
)
upscale_values: dict[str, Any] = {
str(spec.raw.get("upscale_mode_input") or "mode"): str(spec.raw.get("upscale_mode") or "target dimensions"),
str(spec.raw.get("upscale_width_input") or "mode.width"): final_width,
str(spec.raw.get("upscale_height_input") or "mode.height"): final_height,
}
chunking_input = spec.raw.get("upscale_chunking_input")
threshold = spec.raw.get("upscale_chunking_threshold_sec")
if chunking_input and threshold is not None:
upscale_values[str(chunking_input)] = float(request.get("duration") or 0) > float(threshold)
_patch_node(workflow, str(spec.raw["upscale_node"]), upscale_values)
else:
if requested_width is not None:
_patch_node(workflow, spec.raw.get("width_node"), {str(spec.raw.get("width_input") or "value"): int(requested_width)})
if requested_height is not None:
_patch_node(workflow, spec.raw.get("height_node"), {str(spec.raw.get("height_input") or "value"): int(requested_height)})
if request.get("duration") is not None:
_patch_node(workflow, spec.raw.get("duration_node"), {"value": float(request["duration"])})
count = max(1, min(int(request.get("batch_size") or 1), 4))
if spec.raw.get("batch_node"):
_patch_node(workflow, str(spec.raw["batch_node"]), {str(spec.raw.get("batch_input") or "batch_size"): count})
elif count > 1 and spec.raw.get("batch_source_node") and spec.raw.get("batch_target_node"):
source_node = str(spec.raw["batch_source_node"])
target_node = str(spec.raw["batch_target_node"])
if source_node not in workflow or target_node not in workflow:
raise ValueError("workflow batch source or target node is missing")
batch_node = "canvas_repeat_latent_batch"
while batch_node in workflow:
batch_node = "_" + batch_node
workflow[batch_node] = {"class_type": "RepeatLatentBatch", "inputs": {"samples": [source_node, 0], "amount": count}, "_meta": {"title": "Canvas Batch Output"}}
_patch_node(workflow, target_node, {str(spec.raw.get("batch_target_input") or "latent_image"): [batch_node, 0]})
if spec.raw.get("enhance_node") is not None and spec.raw.get("enhance_default") is not None:
_patch_node(workflow, str(spec.raw["enhance_node"]), {str(spec.raw.get("enhance_input") or "value"): bool(spec.raw["enhance_default"])})
if spec.raw.get("denoise_node") and (request.get("denoise") is not None or spec.raw.get("denoise_default") is not None):
denoise = request.get("denoise") if request.get("denoise") is not None else spec.raw.get("denoise_default")
_patch_node(workflow, str(spec.raw["denoise_node"]), {str(spec.raw.get("denoise_input") or "denoise"): float(denoise)})
if spec.raw.get("max_dimension_node") and (request.get("width") is not None or request.get("height") is not None):
maximum = max(int(request.get("width") or 0), int(request.get("height") or 0), 64)
_patch_node(workflow, str(spec.raw["max_dimension_node"]), {str(spec.raw.get("max_dimension_input") or "largest_size"): maximum})
if spec.raw.get("image_refs_node"):
max_reference_images = int(spec.raw.get("max_reference_images") or 6)
if len(uploaded_images) > max_reference_images:
raise ValueError(f"this workflow supports at most {max_reference_images} reference images")
if uploaded_images:
encode_node = str(spec.raw["image_refs_node"])
if encode_node not in workflow:
raise ValueError("workflow image reference node is missing")
if requested_width is not None and requested_height is not None:
aligned_width, aligned_height = _fit_aligned_resolution(
int(requested_width), int(requested_height),
max_megapixels=float(spec.raw.get("base_max_megapixels") or 1), align=32,
)
reference_resolution = max(32, round(math.sqrt(aligned_width * aligned_height) / 32) * 32)
_patch_node(workflow, encode_node, {"resolution": reference_resolution})
prefix = str(spec.raw.get("image_refs_input_prefix") or "images.image_")
for index, image_name in enumerate(uploaded_images, start=1):
node_id = f"canvas_reference_{index}"
while node_id in workflow:
node_id = "_" + node_id
workflow[node_id] = {"class_type": "LoadImage", "inputs": {"image": image_name}, "_meta": {"title": f"Canvas reference {index}"}}
_patch_node(workflow, encode_node, {f"{prefix}{index}": [node_id, 0]})
vae_node = str(spec.raw["image_refs_vae_node"])
_patch_node(workflow, encode_node, {"vae": [vae_node, 0]})
sampler_node = str(spec.raw["image_refs_sampler_node"])
latent_source = [encode_node, 2]
if count > 1:
batch_node = "canvas_repeat_reference_latent"
workflow[batch_node] = {"class_type": "RepeatLatentBatch", "inputs": {"samples": latent_source, "amount": count}, "_meta": {"title": "Canvas reference batch"}}
latent_source = [batch_node, 0]
_patch_node(workflow, sampler_node, {"latent_image": latent_source})
elif spec.raw.get("reference_node"):
if not uploaded_images:
raise ValueError("this workflow requires at least one reference image")
max_reference_images = int(spec.raw.get("max_reference_images") or 9)
if len(uploaded_images) > max_reference_images:
raise ValueError(f"this workflow supports at most {max_reference_images} reference images")
reference_inputs: dict[str, Any] = {}
prefix = str(spec.raw.get("reference_input_prefix") or "ref_images.ref_image_")
for index, image_name in enumerate(uploaded_images):
node_id = f"canvas_reference_{index + 1}"
while node_id in workflow:
node_id = "_" + node_id
workflow[node_id] = {"class_type": "LoadImage", "inputs": {"image": image_name}, "_meta": {"title": f"Canvas reference {index + 1}"}}
reference_inputs[f"{prefix}{index}"] = [node_id, 0]
_patch_node(workflow, str(spec.raw["reference_node"]), reference_inputs)
elif spec.raw.get("load_image_node"):
if spec.mode in {"i2i", "image"} and not uploaded_images:
raise ValueError("this workflow requires an input image")
if uploaded_images:
load_node = str(spec.raw["load_image_node"])
if load_node not in workflow and spec.mode == "text":
workflow[load_node] = {"class_type": "LoadImage", "inputs": {"image": uploaded_images[0]}, "_meta": {"title": "Canvas Reference Image"}}
else:
_patch_node(workflow, load_node, {"image": uploaded_images[0]})
if spec.raw.get("image_resize_node"):
resize_node = str(spec.raw["image_resize_node"])
if resize_node in workflow:
_patch_node(workflow, resize_node, {"input": [load_node, 0]})
if spec.raw.get("t2v_switch_node"):
_patch_node(workflow, str(spec.raw["t2v_switch_node"]), {"value": False})
elif spec.raw.get("t2v_switch_node"):
_patch_node(workflow, str(spec.raw["t2v_switch_node"]), {"value": True})
save_node = spec.raw.get("save_node")
if save_node:
prefix = str(spec.raw.get("save_prefix") or spec.key)
_patch_node(workflow, str(save_node), {"filename_prefix": f"{prefix}_{job_id}"})
if native_audio_control and not generate_audio:
workflow[str(save_node)].setdefault("inputs", {}).pop("audio", None)
if spec.raw.get("speech_prompt_policy") == "minimax_h3":
_apply_h3_frame_policy(spec, workflow, request)
return workflow