wehub-resource-sync
275 lines
11 KiB
Python
275 lines
11 KiB
Python
"""Helpers for exposing direct API media models in Streamlit pipeline UIs."""
|
|
|
|
import os
|
|
from typing import Any
|
|
|
|
import streamlit as st
|
|
from loguru import logger
|
|
|
|
from web.i18n import get_language
|
|
|
|
|
|
def is_api_workflow(workflow_key: str | None) -> bool:
|
|
"""Return True for direct provider workflow keys such as api/dashscope/xxx."""
|
|
return bool(workflow_key and workflow_key.startswith("api/"))
|
|
|
|
|
|
def is_source_workflow(workflow: dict, source: str) -> bool:
|
|
"""Return True when a workflow belongs to a concrete source namespace."""
|
|
key = workflow.get("key") or workflow.get("path") or ""
|
|
return key.startswith(f"{source}/") or workflow.get("source") == source
|
|
|
|
|
|
def workflow_source_label(source: str) -> str:
|
|
"""Human-facing label for workflow source selectors."""
|
|
zh = get_language() == "zh_CN"
|
|
labels = {
|
|
"selfhost": "本地 ComfyUI" if zh else "Local ComfyUI",
|
|
"runninghub": "RunningHub",
|
|
"api": "API 模型" if zh else "API models",
|
|
}
|
|
return labels.get(source, source)
|
|
|
|
|
|
def workflow_source_help(subject: str | None = None) -> str:
|
|
"""Common help text for workflow/model source selectors."""
|
|
zh = get_language() == "zh_CN"
|
|
subject_text = subject or ("当前步骤" if zh else "this step")
|
|
if zh:
|
|
return (
|
|
f"选择{subject_text}使用的模型服务来源："
|
|
"RunningHub 使用云端工作流；本地 ComfyUI 使用 selfhost 工作流；"
|
|
"API 调用直接请求模型供应商。选择后，下方列表只显示该来源下可用的工作流或模型。"
|
|
)
|
|
return (
|
|
f"Choose the model service source for {subject_text}: "
|
|
"RunningHub uses cloud workflows; Local ComfyUI uses selfhost workflows; "
|
|
"API call directly requests model providers. The list below only shows workflows or models from the selected source."
|
|
)
|
|
|
|
|
|
def workflow_select_help() -> str:
|
|
"""Common help text for workflow/model select boxes."""
|
|
if get_language() == "zh_CN":
|
|
return "这里只显示上方所选模型服务来源下可用的工作流或模型。"
|
|
return "Only workflows or models from the selected model service source are shown here."
|
|
|
|
|
|
def list_local_media_workflows(
|
|
pixelle_video: Any,
|
|
media_type: str,
|
|
source: str,
|
|
key_contains: str | None = None,
|
|
key_prefix: str | None = None,
|
|
) -> list[dict]:
|
|
"""List non-API media workflows by source without mixing provider models."""
|
|
try:
|
|
workflows = []
|
|
seen_keys = set()
|
|
for workflow in pixelle_video.media.list_workflows():
|
|
key = workflow.get("key") or workflow.get("path") or ""
|
|
if not key or key.startswith("api/"):
|
|
continue
|
|
if not is_source_workflow(workflow, source):
|
|
continue
|
|
if key_contains and key_contains.lower() not in key.lower():
|
|
continue
|
|
if key_prefix:
|
|
fname = os.path.basename(key)
|
|
if not fname.startswith(key_prefix):
|
|
continue
|
|
workflow_media_type = workflow.get("media_type")
|
|
if media_type == "video":
|
|
if workflow_media_type and workflow_media_type != "video":
|
|
continue
|
|
if not workflow_media_type and "video_" not in key.lower() and not (key_prefix and os.path.basename(key).startswith(key_prefix)):
|
|
continue
|
|
elif workflow_media_type and workflow_media_type != media_type:
|
|
continue
|
|
elif media_type == "image" and "video_" in key.lower():
|
|
continue
|
|
seen_keys.add(key)
|
|
workflows.append({
|
|
"key": key,
|
|
"display_name": workflow.get("display_name") or key,
|
|
**workflow,
|
|
})
|
|
|
|
if key_prefix:
|
|
try:
|
|
from pixelle_video.utils.os_util import get_resource_path, list_resource_files
|
|
|
|
for filename in list_resource_files("workflows", source):
|
|
if not filename.startswith(key_prefix) or not filename.endswith(".json"):
|
|
continue
|
|
key = f"{source}/{filename}"
|
|
if key in seen_keys:
|
|
continue
|
|
seen_keys.add(key)
|
|
workflows.append({
|
|
"key": key,
|
|
"name": filename,
|
|
"display_name": f"{filename} - {source.title()}",
|
|
"source": source,
|
|
"path": get_resource_path("workflows", source, filename),
|
|
"media_type": media_type,
|
|
})
|
|
except Exception as exc:
|
|
logger.warning(f"Failed to list {source}/{key_prefix} workflows from files: {exc}")
|
|
return workflows
|
|
except Exception as exc:
|
|
logger.warning(f"Failed to list {source} {media_type} workflows: {exc}")
|
|
return []
|
|
|
|
|
|
def list_api_media_workflows(
|
|
pixelle_video: Any,
|
|
media_type: str,
|
|
required_adapter_abilities: list[str] | tuple[str, ...] | set[str] | None = None,
|
|
verified_only: bool = False,
|
|
) -> list[dict]:
|
|
"""List API-backed media workflows in the same option shape used by UIs."""
|
|
api_media = getattr(pixelle_video, "api_media", None)
|
|
if api_media is None:
|
|
return []
|
|
|
|
required = set(required_adapter_abilities or [])
|
|
|
|
try:
|
|
workflows = []
|
|
for workflow in api_media.list_workflows():
|
|
if workflow.get("media_type") != media_type:
|
|
continue
|
|
|
|
if verified_only and not workflow.get("api_contract_verified", True):
|
|
continue
|
|
|
|
adapter_abilities = set(workflow.get("adapter_ability_types") or [])
|
|
if required and not required.intersection(adapter_abilities):
|
|
continue
|
|
|
|
workflows.append({
|
|
"key": workflow["key"],
|
|
"display_name": workflow.get("display_name") or workflow["key"],
|
|
**workflow,
|
|
})
|
|
|
|
return workflows
|
|
except Exception as exc:
|
|
logger.warning(f"Failed to list API {media_type} workflows: {exc}")
|
|
return []
|
|
|
|
|
|
def render_api_video_controls(
|
|
workflow: dict | None,
|
|
key_prefix: str,
|
|
default_duration: int = 5,
|
|
allow_audio_driven: bool = False,
|
|
show_duration: bool = True,
|
|
default_ratio: str | None = None,
|
|
) -> dict:
|
|
"""Render common API video controls based on verified adapter capability metadata."""
|
|
if not workflow or not is_api_workflow(workflow.get("key")):
|
|
return {}
|
|
|
|
zh = get_language() == "zh_CN"
|
|
capabilities = workflow.get("capabilities") or {}
|
|
adapter_abilities = set(workflow.get("adapter_ability_types") or [])
|
|
params: dict[str, Any] = {}
|
|
|
|
title = "API 视频模型参数" if zh else "API video model options"
|
|
with st.expander(title, expanded=False):
|
|
ability_text = ", ".join(sorted(adapter_abilities)) or ("未标注" if zh else "unknown")
|
|
st.caption(("已接入能力：" if zh else "Adapter abilities: ") + ability_text)
|
|
|
|
if not workflow.get("api_contract_verified", False):
|
|
st.warning(
|
|
"这个模型的公开 API 数据契约尚未完全确认，只会传递最基础的图生视频参数。"
|
|
if zh
|
|
else "This model's public API contract is not fully verified; only basic image-to-video parameters will be passed."
|
|
)
|
|
|
|
duration_contract = capabilities.get("duration") or {}
|
|
min_duration = int(duration_contract.get("min", 3))
|
|
max_duration = int(duration_contract.get("max", 15))
|
|
if show_duration:
|
|
default_value = min(max(int(default_duration or min_duration), min_duration), max_duration)
|
|
params["duration"] = st.slider(
|
|
"视频时长（秒）" if zh else "Duration (seconds)",
|
|
min_value=min_duration,
|
|
max_value=max_duration,
|
|
value=default_value,
|
|
step=1,
|
|
key=f"{key_prefix}_api_duration",
|
|
)
|
|
else:
|
|
st.caption(
|
|
f"视频时长将自动跟随每段旁白音频长度，并限制在模型支持范围 {min_duration}-{max_duration}s。"
|
|
if zh
|
|
else f"Duration follows each scene's narration audio and is clamped to the model range {min_duration}-{max_duration}s."
|
|
)
|
|
|
|
resolutions = capabilities.get("resolutions") or []
|
|
if resolutions:
|
|
params["resolution"] = st.selectbox(
|
|
"分辨率" if zh else "Resolution",
|
|
resolutions,
|
|
index=0,
|
|
key=f"{key_prefix}_api_resolution",
|
|
)
|
|
|
|
ratios = capabilities.get("ratios") or []
|
|
if ratios:
|
|
preferred_ratio = default_ratio or "9:16"
|
|
default_ratio_index = ratios.index(preferred_ratio) if preferred_ratio in ratios else 0
|
|
params["video_ratio"] = st.selectbox(
|
|
"画幅比例" if zh else "Aspect ratio",
|
|
ratios,
|
|
index=default_ratio_index,
|
|
key=f"{key_prefix}_api_ratio",
|
|
)
|
|
|
|
negative_prompt = st.text_area(
|
|
"负向提示词（可选）" if zh else "Negative prompt (optional)",
|
|
value="",
|
|
height=70,
|
|
key=f"{key_prefix}_api_negative_prompt",
|
|
)
|
|
if negative_prompt.strip():
|
|
params["negative_prompt"] = negative_prompt.strip()
|
|
|
|
if workflow.get("api_contract_verified", False):
|
|
params["watermark"] = st.checkbox(
|
|
"添加水印" if zh else "Add watermark",
|
|
value=False,
|
|
key=f"{key_prefix}_api_watermark",
|
|
)
|
|
|
|
if workflow.get("provider") == "seedance" and workflow.get("api_contract_verified", False):
|
|
params["generate_audio"] = st.checkbox(
|
|
"让模型生成原生音频" if zh else "Generate native audio",
|
|
value=False,
|
|
key=f"{key_prefix}_api_generate_audio",
|
|
)
|
|
|
|
if workflow.get("provider") == "kling" and workflow.get("api_contract_verified", False):
|
|
params["sound"] = "on" if st.checkbox(
|
|
"让模型生成原生音频" if zh else "Generate native audio",
|
|
value=False,
|
|
key=f"{key_prefix}_api_kling_sound",
|
|
) else "off"
|
|
|
|
if allow_audio_driven and "audio_driven_i2v" in adapter_abilities:
|
|
params["use_narration_audio_as_driving_audio"] = st.checkbox(
|
|
"使用本场景旁白音频驱动画面" if zh else "Use narration audio as driving audio",
|
|
value=False,
|
|
help=(
|
|
"仅对已验证支持 driving_audio 的 API 模型生效。"
|
|
if zh
|
|
else "Only applies to verified API models that support driving_audio."
|
|
),
|
|
key=f"{key_prefix}_api_audio_driven",
|
|
)
|
|
|
|
return {key: value for key, value in params.items() if value not in (None, "")}
|