556 lines
31 KiB
Python
556 lines
31 KiB
Python
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"<d>\[(?P<language>[^\]]+)](?P<text>.*?)</d>", re.IGNORECASE | re.DOTALL)
|
||
H3_SPEECH_PREFIX = r"(?P<prefix>(?:口播(?:内容|文案)?|台词|对白|旁白|朗读|说(?:道|出)?|讲(?:道)?|\b(?:says?|said|speaks?|spoke|narrates?)\b)\s*[：:]?\s*)"
|
||
H3_QUOTED_DIALOGUE = tuple(
|
||
re.compile(
|
||
H3_SPEECH_PREFIX + re.escape(left) + r"(?P<text>(?:\\.|[^" + 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<prefix>[ ]*(?:口播(?:内容|文案)?|台词|对白|旁白|dialogue|voiceover|narration)[ ]*[：:][ ]*)(?P<text>[^\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
|