| 1 | """Helpers for exposing direct API media models in Streamlit pipeline UIs.""" |
| 2 | |
| 3 | import os |
| 4 | from typing import Any |
| 5 | |
| 6 | import streamlit as st |
| 7 | from loguru import logger |
| 8 | |
| 9 | from web.i18n import get_language |
| 10 | |
| 11 | |
| 12 | def is_api_workflow(workflow_key: str | None) -> bool: |
| 13 | """Return True for direct provider workflow keys such as api/dashscope/xxx.""" |
| 14 | return bool(workflow_key and workflow_key.startswith("api/")) |
| 15 | |
| 16 | |
| 17 | def is_source_workflow(workflow: dict, source: str) -> bool: |
| 18 | """Return True when a workflow belongs to a concrete source namespace.""" |
| 19 | key = workflow.get("key") or workflow.get("path") or "" |
| 20 | return key.startswith(f"{source}/") or workflow.get("source") == source |
| 21 | |
| 22 | |
| 23 | def workflow_source_label(source: str) -> str: |
| 24 | """Human-facing label for workflow source selectors.""" |
| 25 | zh = get_language() == "zh_CN" |
| 26 | labels = { |
| 27 | "selfhost": "本地 ComfyUI" if zh else "Local ComfyUI", |
| 28 | "runninghub": "RunningHub", |
| 29 | "api": "API 模型" if zh else "API models", |
| 30 | } |
| 31 | return labels.get(source, source) |
| 32 | |
| 33 | |
| 34 | def workflow_source_help(subject: str | None = None) -> str: |
| 35 | """Common help text for workflow/model source selectors.""" |
| 36 | zh = get_language() == "zh_CN" |
| 37 | subject_text = subject or ("当前步骤" if zh else "this step") |
| 38 | if zh: |
| 39 | return ( |
| 40 | f"选择{subject_text}使用的模型服务来源：" |
| 41 | "RunningHub 使用云端工作流；本地 ComfyUI 使用 selfhost 工作流；" |
| 42 | "API 调用直接请求模型供应商。选择后，下方列表只显示该来源下可用的工作流或模型。" |
| 43 | ) |
| 44 | return ( |
| 45 | f"Choose the model service source for {subject_text}: " |
| 46 | "RunningHub uses cloud workflows; Local ComfyUI uses selfhost workflows; " |
| 47 | "API call directly requests model providers. The list below only shows workflows or models from the selected source." |
| 48 | ) |
| 49 | |
| 50 | |
| 51 | def workflow_select_help() -> str: |
| 52 | """Common help text for workflow/model select boxes.""" |
| 53 | if get_language() == "zh_CN": |
| 54 | return "这里只显示上方所选模型服务来源下可用的工作流或模型。" |
| 55 | return "Only workflows or models from the selected model service source are shown here." |
| 56 | |
| 57 | |
| 58 | def list_local_media_workflows( |
| 59 | pixelle_video: Any, |
| 60 | media_type: str, |
| 61 | source: str, |
| 62 | key_contains: str | None = None, |
| 63 | key_prefix: str | None = None, |
| 64 | ) -> list[dict]: |
| 65 | """List non-API media workflows by source without mixing provider models.""" |
| 66 | try: |
| 67 | workflows = [] |
| 68 | seen_keys = set() |
| 69 | for workflow in pixelle_video.media.list_workflows(): |
| 70 | key = workflow.get("key") or workflow.get("path") or "" |
| 71 | if not key or key.startswith("api/"): |
| 72 | continue |
| 73 | if not is_source_workflow(workflow, source): |
| 74 | continue |
| 75 | if key_contains and key_contains.lower() not in key.lower(): |
| 76 | continue |
| 77 | if key_prefix: |
| 78 | fname = os.path.basename(key) |
| 79 | if not fname.startswith(key_prefix): |
| 80 | continue |
| 81 | workflow_media_type = workflow.get("media_type") |
| 82 | if media_type == "video": |
| 83 | if workflow_media_type and workflow_media_type != "video": |
| 84 | continue |
| 85 | if not workflow_media_type and "video_" not in key.lower() and not (key_prefix and os.path.basename(key).startswith(key_prefix)): |
| 86 | continue |
| 87 | elif workflow_media_type and workflow_media_type != media_type: |
| 88 | continue |
| 89 | elif media_type == "image" and "video_" in key.lower(): |
| 90 | continue |
| 91 | seen_keys.add(key) |
| 92 | workflows.append({ |
| 93 | "key": key, |
| 94 | "display_name": workflow.get("display_name") or key, |
| 95 | **workflow, |
| 96 | }) |
| 97 | |
| 98 | if key_prefix: |
| 99 | try: |
| 100 | from pixelle_video.utils.os_util import get_resource_path, list_resource_files |
| 101 | |
| 102 | for filename in list_resource_files("workflows", source): |
| 103 | if not filename.startswith(key_prefix) or not filename.endswith(".json"): |
| 104 | continue |
| 105 | key = f"{source}/{filename}" |
| 106 | if key in seen_keys: |
| 107 | continue |
| 108 | seen_keys.add(key) |
| 109 | workflows.append({ |
| 110 | "key": key, |
| 111 | "name": filename, |
| 112 | "display_name": f"{filename} - {source.title()}", |
| 113 | "source": source, |
| 114 | "path": get_resource_path("workflows", source, filename), |
| 115 | "media_type": media_type, |
| 116 | }) |
| 117 | except Exception as exc: |
| 118 | logger.warning(f"Failed to list {source}/{key_prefix} workflows from files: {exc}") |
| 119 | return workflows |
| 120 | except Exception as exc: |
| 121 | logger.warning(f"Failed to list {source} {media_type} workflows: {exc}") |
| 122 | return [] |
| 123 | |
| 124 | |
| 125 | def list_api_media_workflows( |
| 126 | pixelle_video: Any, |
| 127 | media_type: str, |
| 128 | required_adapter_abilities: list[str] | tuple[str, ...] | set[str] | None = None, |
| 129 | verified_only: bool = False, |
| 130 | ) -> list[dict]: |
| 131 | """List API-backed media workflows in the same option shape used by UIs.""" |
| 132 | api_media = getattr(pixelle_video, "api_media", None) |
| 133 | if api_media is None: |
| 134 | return [] |
| 135 | |
| 136 | required = set(required_adapter_abilities or []) |
| 137 | |
| 138 | try: |
| 139 | workflows = [] |
| 140 | for workflow in api_media.list_workflows(): |
| 141 | if workflow.get("media_type") != media_type: |
| 142 | continue |
| 143 | |
| 144 | if verified_only and not workflow.get("api_contract_verified", True): |
| 145 | continue |
| 146 | |
| 147 | adapter_abilities = set(workflow.get("adapter_ability_types") or []) |
| 148 | if required and not required.intersection(adapter_abilities): |
| 149 | continue |
| 150 | |
| 151 | workflows.append({ |
| 152 | "key": workflow["key"], |
| 153 | "display_name": workflow.get("display_name") or workflow["key"], |
| 154 | **workflow, |
| 155 | }) |
| 156 | |
| 157 | return workflows |
| 158 | except Exception as exc: |
| 159 | logger.warning(f"Failed to list API {media_type} workflows: {exc}") |
| 160 | return [] |
| 161 | |
| 162 | |
| 163 | def render_api_video_controls( |
| 164 | workflow: dict | None, |
| 165 | key_prefix: str, |
| 166 | default_duration: int = 5, |
| 167 | allow_audio_driven: bool = False, |
| 168 | show_duration: bool = True, |
| 169 | default_ratio: str | None = None, |
| 170 | ) -> dict: |
| 171 | """Render common API video controls based on verified adapter capability metadata.""" |
| 172 | if not workflow or not is_api_workflow(workflow.get("key")): |
| 173 | return {} |
| 174 | |
| 175 | zh = get_language() == "zh_CN" |
| 176 | capabilities = workflow.get("capabilities") or {} |
| 177 | adapter_abilities = set(workflow.get("adapter_ability_types") or []) |
| 178 | params: dict[str, Any] = {} |
| 179 | |
| 180 | title = "API 视频模型参数" if zh else "API video model options" |
| 181 | with st.expander(title, expanded=False): |
| 182 | ability_text = ", ".join(sorted(adapter_abilities)) or ("未标注" if zh else "unknown") |
| 183 | st.caption(("已接入能力：" if zh else "Adapter abilities: ") + ability_text) |
| 184 | |
| 185 | if not workflow.get("api_contract_verified", False): |
| 186 | st.warning( |
| 187 | "这个模型的公开 API 数据契约尚未完全确认，只会传递最基础的图生视频参数。" |
| 188 | if zh |
| 189 | else "This model's public API contract is not fully verified; only basic image-to-video parameters will be passed." |
| 190 | ) |
| 191 | |
| 192 | duration_contract = capabilities.get("duration") or {} |
| 193 | min_duration = int(duration_contract.get("min", 3)) |
| 194 | max_duration = int(duration_contract.get("max", 15)) |
| 195 | if show_duration: |
| 196 | default_value = min(max(int(default_duration or min_duration), min_duration), max_duration) |
| 197 | params["duration"] = st.slider( |
| 198 | "视频时长（秒）" if zh else "Duration (seconds)", |
| 199 | min_value=min_duration, |
| 200 | max_value=max_duration, |
| 201 | value=default_value, |
| 202 | step=1, |
| 203 | key=f"{key_prefix}_api_duration", |
| 204 | ) |
| 205 | else: |
| 206 | st.caption( |
| 207 | f"视频时长将自动跟随每段旁白音频长度，并限制在模型支持范围 {min_duration}-{max_duration}s。" |
| 208 | if zh |
| 209 | else f"Duration follows each scene's narration audio and is clamped to the model range {min_duration}-{max_duration}s." |
| 210 | ) |
| 211 | |
| 212 | resolutions = capabilities.get("resolutions") or [] |
| 213 | if resolutions: |
| 214 | params["resolution"] = st.selectbox( |
| 215 | "分辨率" if zh else "Resolution", |
| 216 | resolutions, |
| 217 | index=0, |
| 218 | key=f"{key_prefix}_api_resolution", |
| 219 | ) |
| 220 | |
| 221 | ratios = capabilities.get("ratios") or [] |
| 222 | if ratios: |
| 223 | preferred_ratio = default_ratio or "9:16" |
| 224 | default_ratio_index = ratios.index(preferred_ratio) if preferred_ratio in ratios else 0 |
| 225 | params["video_ratio"] = st.selectbox( |
| 226 | "画幅比例" if zh else "Aspect ratio", |
| 227 | ratios, |
| 228 | index=default_ratio_index, |
| 229 | key=f"{key_prefix}_api_ratio", |
| 230 | ) |
| 231 | |
| 232 | negative_prompt = st.text_area( |
| 233 | "负向提示词（可选）" if zh else "Negative prompt (optional)", |
| 234 | value="", |
| 235 | height=70, |
| 236 | key=f"{key_prefix}_api_negative_prompt", |
| 237 | ) |
| 238 | if negative_prompt.strip(): |
| 239 | params["negative_prompt"] = negative_prompt.strip() |
| 240 | |
| 241 | if workflow.get("api_contract_verified", False): |
| 242 | params["watermark"] = st.checkbox( |
| 243 | "添加水印" if zh else "Add watermark", |
| 244 | value=False, |
| 245 | key=f"{key_prefix}_api_watermark", |
| 246 | ) |
| 247 | |
| 248 | if workflow.get("provider") == "seedance" and workflow.get("api_contract_verified", False): |
| 249 | params["generate_audio"] = st.checkbox( |
| 250 | "让模型生成原生音频" if zh else "Generate native audio", |
| 251 | value=False, |
| 252 | key=f"{key_prefix}_api_generate_audio", |
| 253 | ) |
| 254 | |
| 255 | if workflow.get("provider") == "kling" and workflow.get("api_contract_verified", False): |
| 256 | params["sound"] = "on" if st.checkbox( |
| 257 | "让模型生成原生音频" if zh else "Generate native audio", |
| 258 | value=False, |
| 259 | key=f"{key_prefix}_api_kling_sound", |
| 260 | ) else "off" |
| 261 | |
| 262 | if allow_audio_driven and "audio_driven_i2v" in adapter_abilities: |
| 263 | params["use_narration_audio_as_driving_audio"] = st.checkbox( |
| 264 | "使用本场景旁白音频驱动画面" if zh else "Use narration audio as driving audio", |
| 265 | value=False, |
| 266 | help=( |
| 267 | "仅对已验证支持 driving_audio 的 API 模型生效。" |
| 268 | if zh |
| 269 | else "Only applies to verified API models that support driving_audio." |
| 270 | ), |
| 271 | key=f"{key_prefix}_api_audio_driven", |
| 272 | ) |
| 273 | |
| 274 | return {key: value for key, value in params.items() if value not in (None, "")} |
| 275 |