src/app_lib.py

import os
import json
import re
import importlib.machinery
import importlib.util
import shutil
import subprocess
import sys
import time
import threading
import types
import zlib
import urllib.error
import urllib.request
from pathlib import Path
from typing import Dict, List, Optional, Tuple, Any

import gradio as gr
import spaces
import torch
from transformers import AutoModelForCausalLM, AutoTokenizer

try:
    import redis
except Exception:
    redis = None

redis = None

os.environ.setdefault("HF_HUB_DISABLE_PROGRESS_BARS", "1")

PLACEHOLDER_VAE = "__OMNI_VAE__"
PLACEHOLDER_TEXT_ENCODER = "__OMNI_TEXT_ENCODER__"
PLACEHOLDER_UNET_Q6KH = "__OMNI_UNET_Q6KH__"
PLACEHOLDER_UNET_Q6KL = "__OMNI_UNET_Q6KL__"

PLACEHOLDER_TO_KEY = {
    PLACEHOLDER_VAE: "vae",
    PLACEHOLDER_TEXT_ENCODER: "text_encoder",
    PLACEHOLDER_UNET_Q6KH: "unet_q6kh",
    PLACEHOLDER_UNET_Q6KL: "unet_q6kl",
}

MODEL_KEY_TO_SUBDIR = {
    "vae": "vae",
    "text_encoder": "text_encoders",
    "unet_q6kh": "diffusion_models",
    "unet_q6kl": "diffusion_models",
}
WORKFLOW_FILES = ("video_t2v_api.json", "video_i2v_api.json", "video_v2v_api.json")
WORKFLOW_PACK_MARKER = b"OMNIWF1"
WORKFLOW_PACK_SUFFIX = ".pack"

DEFAULT_OMNI_VIDEOS = (
    "wan_2.1_vae.safetensors@Comfy-Org/Wan_2.1_ComfyUI_repackaged@split_files/vae/wan_2.1_vae.safetensors"
    "#nsfw_wan_umt5-xxl_fp8_scaled.safetensors@geceff/Wan2.2-Custom-Models-GGUF@text_encoders/nsfw_wan_umt5-xxl_fp8_scaled.safetensors"
    "#wan22EnhancedNSFWCameraPrompt_nsfwFASTMOVEFP8High.safetensors@jorgmikel76/wan22EnhancedNSFWCameraPrompt_nsfwFASTMOVEFP8High@wan22EnhancedNSFWCameraPrompt_nsfwFASTMOVEFP8High.safetensors"
    "#wan22EnhancedNSFWCameraPrompt_nsfwFASTMOVEFP8Low.safetensors@jorgmikel76/wan22EnhancedNSFWCameraPrompt_nsfwFASTMOVEFP8High@wan22EnhancedNSFWCameraPrompt_nsfwFASTMOVEFP8Low.safetensors"
)

_COMFY_CLIENT = None
_MODEL_PREP_LOCK = threading.Lock()
_MODEL_PREP_RUNNING = False
_MODEL_PREP_STATUS = "model preparation not started"
_RUNTIME_PREP_LOCK = threading.Lock()
_RUNTIME_PREP_RUNNING = False
_RUNTIME_PREP_STATUS = "runtime preparation not started"
_PREP_IO_LOCK = threading.Lock()

_QWEN_MODEL = None
_QWEN_TOKENIZER = None
_QWEN_LOCK = threading.Lock()

def _load_qwen_model():
    global _QWEN_MODEL, _QWEN_TOKENIZER
    with _QWEN_LOCK:
        if _QWEN_MODEL is None:
            model_id = "Qwen/Qwen3-0.6B"
            _QWEN_TOKENIZER = AutoTokenizer.from_pretrained(model_id)
            _QWEN_MODEL = AutoModelForCausalLM.from_pretrained(
                model_id,
                dtype=torch.float32,
                device_map="cpu"
            )
    return _QWEN_MODEL, _QWEN_TOKENIZER

def _split_parts(value: Optional[str] = None) -> Tuple[str, str, str, str]:
    raw = (value if value is not None else os.getenv("OMNI_VIDEOS") or DEFAULT_OMNI_VIDEOS).strip()
    parts = [p.strip() for p in raw.split("#") if p.strip()]
    if len(parts) != 4:
        raise ValueError("OMNI_VIDEOS must have 4 non-empty parts separated by '#'.")
    return parts[0], parts[1], parts[2], parts[3]

def _parse_entry(entry: str) -> Tuple[str, Optional[str], Optional[str]]:
    parts = (entry or "").strip().split("@", 2)
    filename = parts[0].strip() if parts and parts[0].strip() else ""
    repo_id = parts[1].strip() if len(parts) >= 2 and parts[1].strip() else None
    repo_relpath = parts[2].strip() if len(parts) >= 3 and parts[2].strip() else None
    return filename, repo_id, repo_relpath

def parse_model_names(value: Optional[str] = None) -> Dict[str, str]:
    a, b, c, d = _split_parts(value)
    vae, _, _ = _parse_entry(a)
    text_encoder, _, _ = _parse_entry(b)
    unet_q6kh, _, _ = _parse_entry(c)
    unet_q6kl, _, _ = _parse_entry(d)
    if not all([vae, text_encoder, unet_q6kh, unet_q6kl]):
        raise ValueError("Each OMNI_VIDEOS segment must include a filename.")
    return {
        "vae": vae,
        "text_encoder": text_encoder,
        "unet_q6kh": unet_q6kh,
        "unet_q6kl": unet_q6kl,
    }

def parse_model_entries(value: Optional[str] = None) -> Dict[str, Tuple[str, Optional[str], Optional[str]]]:
    a, b, c, d = _split_parts(value)
    return {
        "vae": _parse_entry(a),
        "text_encoder": _parse_entry(b),
        "unet_q6kh": _parse_entry(c),
        "unet_q6kl": _parse_entry(d),
    }

def _split_one_key(
    value: Optional[str] = None,
) -> Tuple[Optional[str], Optional[str], Optional[str], Optional[str], Optional[str]]:
    raw = (value if value is not None else os.getenv("ONE_KEY") or "").strip()
    if not raw:
        return None, None, None, None, None
    parts = [p.strip() for p in raw.split("#")]
    while len(parts) < 5:
        parts.append("")
    return (
        parts[0] or None,
        parts[1] or None,
        parts[2] or None,
        parts[3] or None,
        parts[4] or None,
    )

def _get_hf_token() -> Optional[str]:
    hf_from_one_key, _, _, _, _ = _split_one_key()
    if hf_from_one_key:
        return hf_from_one_key
    for key in ("HF_TOKEN", "HUGGINGFACEHUB_API_TOKEN", "HUGGINGFACE_TOKEN"):
        value = (os.getenv(key) or "").strip()
        if value:
            return value
    return None

def _get_worker_api_token() -> Optional[str]:
    _, _, _, worker_from_one_key, _ = _split_one_key()
    if worker_from_one_key:
        return worker_from_one_key
    for key in ("WORKER_API_TOKEN", "OMNI_WORKER_API_TOKEN"):
        value = (os.getenv(key) or "").strip()
        if value:
            return value
    return None

def _get_legacy_api_base() -> Optional[str]:
    _, _, legacy_api_base, _, _ = _split_one_key()
    return legacy_api_base

def _get_github_token() -> Optional[str]:
    _, _, _, _, github_from_one_key = _split_one_key()
    if github_from_one_key:
        return github_from_one_key
    for key in ("GITHUB_TOKEN", "OMNI_GITHUB_TOKEN"):
        value = (os.getenv(key) or "").strip()
        if value:
            return value
    return None

def _model_root_dir() -> Path:
    src_dir = Path(__file__).resolve().parent
    repo_dir = src_dir.parent
    return repo_dir / "ComfyUIVideo" / "models"

def _repo_dir() -> Path:
    src_dir = Path(__file__).resolve().parent
    return src_dir.parent

def _comfy_dir() -> Path:
    return _repo_dir() / "ComfyUIVideo"

def _load_app_config():
    path = _repo_dir() / "src" / "config.pyc"
    if not path.exists():
        raise RuntimeError(f"missing private config bytecode: {path}")

    raw = path.read_bytes()
    marker = b"OMNICFG1"

    if raw.startswith(marker):
        payload = raw[len(marker) :]
        try:
            decoded = zlib.decompress(payload)
            data = json.loads(decoded.decode("utf-8"))
        except Exception as exc:
            raise RuntimeError(f"failed to decode packed private config: {type(exc).__name__}: {exc}") from exc
        if not isinstance(data, dict) or not data:
            raise RuntimeError(f"packed private config is empty: {path}")
        print(f"[config] loaded packed private config: {path}")
        return types.SimpleNamespace(**data)

    loader = importlib.machinery.SourcelessFileLoader("omni_video_factory_private_config", str(path))
    spec = importlib.util.spec_from_loader("omni_video_factory_private_config", loader)
    module = importlib.util.module_from_spec(spec) if spec else None
    if not spec or not module or not spec.loader:
        raise RuntimeError(f"failed to load private config module spec from {path}")
    try:
        spec.loader.exec_module(module)
    except Exception as exc:
        raise RuntimeError(f"failed to load sourceless private config: {type(exc).__name__}: {exc}") from exc

    data = {}
    for key in dir(module):
        if key.isupper() and key != "APP_RUNTIME_CONFIG":
            data[key] = getattr(module, key)
    class_obj = getattr(module, "APP_RUNTIME_CONFIG", None)
    if class_obj is not None:
        for key in dir(class_obj):
            if key.isupper() and key not in data:
                data[key] = getattr(class_obj, key)
    if not data:
        raise RuntimeError(f"private config loaded but empty: {path}")
    print(f"[config] loaded sourceless private config: {path}")
    return types.SimpleNamespace(**data)

APP_CONFIG = _load_app_config()

def _runtime_git_repo() -> str:
    repo = (os.getenv("OMNI_RUNTIME_GIT_REPO") or "selfitcamera/ComfyUIVideo").strip()
    return repo.removesuffix(".git")

def _runtime_git_revision() -> str:
    return (os.getenv("OMNI_RUNTIME_GIT_REF") or "17b9fb3").strip()

def _is_commit_hash(ref: str) -> bool:
    ref = (ref or "").strip()
    if len(ref) < 7 or len(ref) > 40:
        return False
    return all(ch in "0123456789abcdefABCDEF" for ch in ref)

def _runtime_git_user() -> str:
    return (os.getenv("OMNI_RUNTIME_GIT_USER") or "selfitcamera").strip()

def _workflow_repo_id() -> str:
    env_space_id = (os.getenv("SPACE_ID") or "").strip()
    if env_space_id:
        return env_space_id
    return (os.getenv("OMNI_WORKFLOW_REPO") or "FrameAI4687/AI-Video-0213-02").strip()

def _runtime_git_clone_url() -> str:
    github_token = _get_github_token()
    repo = _runtime_git_repo()
    if github_token:
        return f"https://{_runtime_git_user()}:{github_token}@github.com/{repo}.git"
    return f"https://github.com/{repo}.git"

def _redact_sensitive(text: str) -> str:
    out = text
    for secret in (_get_hf_token(), _get_worker_api_token(), _get_github_token()):
        if secret:
            out = out.replace(secret, "***")
    return out

def _llm_api_base() -> str:
    return (os.getenv("OMNI_LLM_API_BASE") or "https://omnifilm.net").strip().rstrip("/")

def _llm_api_base_candidates() -> List[str]:
    candidates: List[str] = []
    env_base = (os.getenv("OMNI_LLM_API_BASE") or "").strip().rstrip("/")
    if env_base:
        candidates.append(env_base)
    candidates.append("https://omnifilm.net")
    legacy_base = (_get_legacy_api_base() or "").strip().rstrip("/")
    if legacy_base.startswith("http://") or legacy_base.startswith("https://"):
        parts = legacy_base.split("/", 3)
        if len(parts) >= 3:
            candidates.append(parts[0] + "//" + parts[2])
    seen = set()
    ordered = []
    for base in candidates:
        if base and base not in seen:
            ordered.append(base)
            seen.add(base)
    return ordered

def _http_json(
    method: str,
    url: str,
    payload: Optional[dict] = None,
    headers: Optional[Dict[str, str]] = None,
    timeout: int = 30,
) -> dict:
    def _decode_json(raw_text: str) -> dict:
        try:
            return json.loads(raw_text or "{}")
        except Exception as exc:
            raise RuntimeError(f"invalid json response: {type(exc).__name__}: {exc}") from exc

    def _http_json_via_curl() -> dict:
        cmd = ["curl", "-sS", "-X", method.upper(), url, "--max-time", str(int(timeout))]
        for k, v in req_headers.items():
            cmd.extend(["-H", f"{k}: {v}"])
        if payload is not None:
            cmd.extend(["--data", json.dumps(payload, ensure_ascii=False)])
        result = subprocess.run(cmd, capture_output=True, text=True, check=False)
        stdout = (result.stdout or "").strip()
        stderr = _redact_sensitive((result.stderr or "").strip())
        if result.returncode != 0:
            raise RuntimeError(f"curl failed (exit={result.returncode}): {stderr}")
        return _decode_json(stdout)

    data = None
    req_headers = dict(headers or {})
    req_headers.setdefault(
        "User-Agent",
        "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 "
        "(KHTML, like Gecko) Chrome/122.0.0.0 Safari/537.36",
    )
    req_headers.setdefault("Accept", "application/json, text/plain, */*")
    if payload is not None:
        data = json.dumps(payload, ensure_ascii=False).encode("utf-8")
        req_headers.setdefault("Content-Type", "application/json")
    req = urllib.request.Request(url, data=data, headers=req_headers, method=method.upper())
    try:
        with urllib.request.urlopen(req, timeout=timeout) as resp:
            raw = resp.read().decode("utf-8", errors="replace")
        return _decode_json(raw)
    except urllib.error.HTTPError as exc:
        detail = exc.read().decode("utf-8", errors="replace")
        if int(getattr(exc, "code", 0) or 0) == 403 and "1010" in detail:
            try:
                return _http_json_via_curl()
            except Exception:
                pass
        raise RuntimeError(f"HTTP {exc.code}: {detail}") from exc
    except Exception as exc:
        raise RuntimeError(f"{type(exc).__name__}: {exc}") from exc

def _extract_result_text(obj) -> str:
    def _strip_answer_wrapper(text: str) -> str:
        raw = (text or "").strip()
        if not raw:
            return ""
        matched = re.search(r"<answer>(.*?)</answer>", raw, flags=re.IGNORECASE | re.DOTALL)
        if matched:
            inner = (matched.group(1) or "").strip()
            if inner:
                return inner
        return raw

    if obj is None:
        return ""
    if isinstance(obj, str):
        return _strip_answer_wrapper(obj)
    if isinstance(obj, list):
        texts = [_extract_result_text(x) for x in obj]
        texts = [t for t in texts if t]
        if not texts:
            return ""
        with_sep = [t for t in texts if "¥" in t]
        return max(with_sep or texts, key=len)
    if isinstance(obj, dict):
        if "choices" in obj and isinstance(obj["choices"], list):
            for choice in obj["choices"]:
                if isinstance(choice, dict):
                    msg = choice.get("message")
                    if isinstance(msg, dict) and isinstance(msg.get("content"), str):
                        text = _strip_answer_wrapper(msg["content"])
                        if text:
                            return text
                    if isinstance(choice.get("text"), str):
                        text = _strip_answer_wrapper(choice["text"])
                        if text:
                            return text
        texts = [_extract_result_text(v) for v in obj.values()]
        texts = [t for t in texts if t]
        if not texts:
            return ""
        with_sep = [t for t in texts if "¥" in t]
        return max(with_sep or texts, key=len)
    return ""

_NSFW_PIPELINE = None
_NSFW_STATE_LOCK = threading.Lock()
_USAGE_STATE_LOCK = threading.Lock()
_COUNTRY_CACHE: Dict[str, str] = {}
_COUNTRY_CACHE_TS: Dict[str, float] = {}
_REDIS_STATE_LOCK = threading.Lock()
_REDIS_CLIENT = None
_REDIS_MEMORY_LAST_CHECK_TS = 0.0

def _redis_url() -> str:
    for key in ("REDIS_KEY", "REDIS_URL", "OMNI_REDIS_URL"):
        value = (os.getenv(key) or "").strip()
        if value:
            return value
    return ""

def _redis_enabled() -> bool:
    return bool(_redis_url())

def _runtime_boot_marker() -> str:
    host = (os.getenv("HOSTNAME") or "unknown-host").strip() or "unknown-host"
    proc_boot = ""
    try:
        raw = Path("/proc/1/stat").read_text(encoding="utf-8", errors="ignore").strip()
        parts = raw.split()
        if len(parts) >= 22:
            proc_boot = str(parts[21]).strip()
    except Exception:
        proc_boot = ""
    if proc_boot:
        return f"{host}:{proc_boot}"
    return host

def _redis_scan_delete(client, pattern: str) -> int:
    total = 0
    cursor = 0
    while True:
        cursor, keys = client.scan(cursor=cursor, match=pattern, count=256)
        if keys:
            total += int(client.delete(*keys) or 0)
        if cursor == 0:
            break
    return total

def _redis_prepare_usage_on_boot(client) -> None:
    marker_key = "ovf:meta:boot_marker"
    marker = _runtime_boot_marker()
    try:
        old = str(client.get(marker_key) or "")
    except Exception:
        old = ""
    if old == marker:
        return
    cleared_usage = 0
    try:
        cleared_usage += _redis_scan_delete(client, "ovf:usage:*")
        cleared_usage += _redis_scan_delete(client, "ovf:usage_window:*")
    except Exception as exc:
        pass
    try:
        client.set(marker_key, marker)
    except Exception:
        pass

def _redis_maybe_flush_all(client) -> None:
    global _REDIS_MEMORY_LAST_CHECK_TS
    now = time.time()
    interval_default = int(getattr(APP_CONFIG, "REDIS_MEMORY_CHECK_INTERVAL_SECONDS", 120))
    interval = max(15, int((os.getenv("REDIS_MEMORY_CHECK_INTERVAL_SECONDS") or str(interval_default)).strip() or str(interval_default)))
    with _REDIS_STATE_LOCK:
        if (now - float(_REDIS_MEMORY_LAST_CHECK_TS or 0.0)) < interval:
            return
        _REDIS_MEMORY_LAST_CHECK_TS = now
    try:
        info = client.info(section="memory")
        used = int(info.get("used_memory") or 0)
        max_memory = int(info.get("maxmemory") or 0)
        if max_memory <= 0:
            max_memory = int((os.getenv("REDIS_MEMORY_LIMIT_BYTES") or str(30 * 1024 * 1024)).strip() or str(30 * 1024 * 1024))
        if max_memory <= 0:
            return
        ratio = float(used) / float(max_memory)
        threshold_default = float(getattr(APP_CONFIG, "REDIS_FLUSH_ALL_RATIO", 0.95))
        threshold = float((os.getenv("REDIS_FLUSH_ALL_RATIO") or str(threshold_default)).strip() or str(threshold_default))
        threshold = min(0.999, max(0.6, threshold))
        if ratio >= threshold:
            client.flushdb()
    except Exception as exc:
        pass

def _redis_client():
    global _REDIS_CLIENT
    if _REDIS_CLIENT is not None:
        return _REDIS_CLIENT
    if redis is None:
        return None
    url = _redis_url()
    if not url:
        return None
    with _REDIS_STATE_LOCK:
        if _REDIS_CLIENT is not None:
            return _REDIS_CLIENT
        try:
            max_connections_default = int(getattr(APP_CONFIG, "REDIS_MAX_CONNECTIONS", 1))
            max_connections = int((os.getenv("REDIS_MAX_CONNECTIONS") or str(max_connections_default)).strip() or str(max_connections_default))
            max_connections = max(1, min(2, max_connections))
            client = redis.Redis.from_url(
                url,
                decode_responses=True,
                max_connections=max_connections,
                socket_connect_timeout=1.5,
                socket_timeout=1.5,
                health_check_interval=30,
                retry_on_timeout=True,
            )
            client.ping()
            _redis_prepare_usage_on_boot(client)
            _REDIS_CLIENT = client
        except Exception as exc:
            _REDIS_CLIENT = None
    return _REDIS_CLIENT

def _nsfw_enabled() -> bool:
    return False

def _nsfw_policy_applies(mode: str) -> bool:
    return False

def _normalize_ip(ip: str) -> str:
    ip = (ip or "").strip()
    if not ip:
        return ""
    if "," in ip:
        ip = ip.split(",", 1)[0].strip()
    if "." in ip and ip.count(":") == 1:
        host, _, maybe_port = ip.rpartition(":")
        if host and maybe_port.isdigit():
            ip = host
    if ip.startswith("[") and ip.endswith("]"):
        ip = ip[1:-1].strip()
    return ip if 0 < len(ip) <= 128 else ""

def _request_ip(request: Any = None) -> str:
    if request is None:
        return ""
    try:
        headers = getattr(request, "headers", {}) or {}
        lowered = {str(k).lower(): str(v) for k, v in dict(headers).items()}
        for key in ("cf-connecting-ip", "x-forwarded-for", "x-real-ip", "x-client-ip", "fly-client-ip"):
            val = lowered.get(key, "")
            got = _normalize_ip(val)
            if got:
                return got
    except Exception:
        pass
    try:
        client = getattr(request, "client", None)
        host = getattr(client, "host", "") if client is not None else ""
        got = _normalize_ip(host)
        if got:
            return got
    except Exception:
        pass
    return ""

def _usage_entry_key(ip: str) -> str:
    key = _normalize_ip(ip)
    return key if key else "__unknown__"

def _usage_key(ip: str) -> str:
    return f"ovf:usage:{_usage_entry_key(ip)}"

def _usage_country_stats_key(metric: str = "video") -> str:
    m = str(metric or "video").strip().lower()
    if m == "auto":
        return "ovf:usage:global:country_auto_prompt_total"
    if m == "nsfw":
        return "ovf:usage:global:country_nsfw_total"
    return "ovf:usage:global:country_video_total"

def _usage_global_video_total_key() -> str:
    return "ovf:usage:global:video_total"

def _usage_country_ip_set_key(country_code: str) -> str:
    return f"ovf:usage:global:country_ips:{country_code}"

def _stats_country(country_code: str) -> str:
    cc = (country_code or "").strip().upper()
    if len(cc) == 2 and cc.isalpha():
        return cc
    return "UNKNOWN"

def _usage_window_key(ip: str, window_seconds: int, now: float) -> str:
    bucket = int(float(now) // max(1, int(window_seconds)))
    return f"ovf:usage_window:auto:{_usage_entry_key(ip)}:{bucket}"

def _geo_key(ip: str) -> str:
    return f"ovf:geo:{_usage_entry_key(ip)}"

def _usage_ttl_seconds() -> int:
    return max(300, int((os.getenv("REDIS_USAGE_TTL_SECONDS") or str(7 * 24 * 3600)).strip() or str(7 * 24 * 3600)))

def _geo_ttl_seconds() -> int:
    return max(3600, int((os.getenv("REDIS_GEO_TTL_SECONDS") or str(180 * 24 * 3600)).strip() or str(180 * 24 * 3600)))

def _set_local_country_cache(ip: str, country: str, now_ts: Optional[float] = None) -> None:
    now_ts = float(now_ts if now_ts is not None else time.time())
    with _USAGE_STATE_LOCK:
        _COUNTRY_CACHE[ip] = country
        _COUNTRY_CACHE_TS[ip] = now_ts

def _request_country(request: Any = None, ip: str = "") -> str:
    return ""

def _snapshot_usage_counts(ip: str) -> Tuple[int, int, int]:
    client = _redis_client()
    if client is None:
        return (0, 0, 0)
    try:
        vals = client.hmget(_usage_key(ip), "auto_prompt_total", "video_total", "nsfw_total")
        return (
            int(vals[0] or 0),
            int(vals[1] or 0),
            int(vals[2] or 0),
        )
    except Exception as exc:
        return (0, 0, 0)

def _usage_country(ip: str, fallback_country_code: str = "") -> str:
    cc = (fallback_country_code or "").strip().upper()
    if cc:
        return cc
    client = _redis_client()
    if client is not None:
        try:
            cc = str(client.hget(_usage_key(ip), "country") or "").strip().upper()
            if cc:
                return cc
        except Exception:
            pass
    return ""

def _log_usage_snapshot(event: str, ip: str, country_code: str = "") -> None:
    return

def _log_global_country_video_stats_if_needed(client, global_total: int) -> None:
    return

def _allow_auto_prompt_and_record(ip: str, country_code: str = "") -> Tuple[bool, str]:
    return True, ""

def _record_video_generation(ip: str, country_code: str = "") -> None:
    return

def _record_nsfw_hit(ip: str, country_code: str = "") -> int:
    return 0

def _nsfw_counter_endpoint() -> Optional[str]:
    return None

def _nsfw_counter_key(ip: str) -> str:
    return "__unknown__"

def _nsfw_remote_inc_total(ip: str) -> Optional[int]:
    return None

def _nsfw_remote_register_nsfw(ip: str) -> Optional[int]:
    return None

def _nsfw_keyword_match(label: str) -> bool:
    return False

def _get_nsfw_pipeline():
    return None

def _nsfw_predict_label_from_pil(pil_image) -> str:
    return "unknown"

def _video_frames_for_nsfw(video_path: str):
    return []

def _nsfw_check_video(video_path: str) -> Tuple[bool, str]:
    return False, ""

def _nsfw_blur_preview_from_video(video_path: str) -> Optional[str]:
    return None

def _nsfw_warning_card_html() -> str:
    return ""

def _nsfw_blocked_card_html() -> str:
    return ""

def _nsfw_card_update_for_status(status_text: str):
    return gr.update(value="", visible=False)

def _nsfw_preview_update(path: Optional[str]):
    p = str(path or "").strip()
    if p and Path(p).exists():
        return gr.update(value=p, visible=True)
    return gr.update(value=None, visible=False)

def _public_generation_status(status_text: str) -> str:
    lines = [str(raw or "").strip() for raw in str(status_text or "").splitlines() if str(raw or "").strip()]
    if not lines:
        return "ZeroGPU elapsed: --"
    elapsed_line = "ZeroGPU elapsed: --"
    detail_lines: List[str] = []
    for line in lines:
        if line.startswith("ZeroGPU elapsed:"):
            elapsed_line = line
        else:
            detail_lines.append(line)
    if not detail_lines:
        return elapsed_line
    detail = detail_lines[0]
    if len(detail) > 320:
        detail = detail[:317] + "..."
    return f"{detail}\n{elapsed_line}"

def _should_show_like_tip(ip: str) -> bool:
    return False

def _download_upsell_card_html(show_like_tip: bool = False) -> str:
    return ""

def _download_upsell_update(video_path: Optional[str], status_text: str = "", show_like_tip: bool = False):
    return gr.update(value="", visible=False)

def _normalize_scene_parts(content: str, scene_count: int, fallback_prompt: str) -> List[str]:
    parts = [p.strip() for p in (content or "").split("¥") if p.strip()]
    if not parts:
        parts = [fallback_prompt.strip() or "cinematic scene"]
    normalized = parts[:scene_count]
    while len(normalized) < scene_count:
        normalized.append(normalized[-1])
    return normalized

def _llm_generate_scene_prompt_text(
    mode: str,
    prompt_text: str,
    scene_count: int,
    seconds_per_scene: int,
) -> str:
    if scene_count <= 1:
        text, _note = _ensure_scene_prompt_text(prompt_text or "", scene_count)
        return text

    model, tokenizer = _load_qwen_model()
    system_prompt = (
        "You are a cinematic video prompt generator. "
        f"You MUST output EXACTLY {scene_count} scenes. "
        "Separate each scene using the '¥' character. "
        "Example format: Scene 1 description ¥ Scene 2 description"
    )
    user_prompt = f"Expand this prompt into {scene_count} continuous cinematic scenes: {prompt_text}"
    messages = [
        {"role": "system", "content": system_prompt},
        {"role": "user", "content": user_prompt}
    ]
    text_input = tokenizer.apply_chat_template(
        messages,
        tokenize=False,
        add_generation_prompt=True
    )
    model_inputs = tokenizer([text_input], return_tensors="pt")
    
    # Increase max_new_tokens to allow the model to finish its reasoning process
    generated_ids = model.generate(
        **model_inputs,
        max_new_tokens=1536
    )
    generated_ids = [
        output_ids[len(input_ids):] for input_ids, output_ids in zip(model_inputs.input_ids, generated_ids)
    ]
    response = tokenizer.batch_decode(generated_ids, skip_special_tokens=True)[0]
    
    # Strictly remove the <think> section
    if "</think>" in response:
        response = response.split("</think>")[-1].strip()
    else:
        import re
        response = re.sub(r'<think>.*', '', response, flags=re.DOTALL).strip()
        
    # Fallback to the original prompt if the response is empty
    if not response.strip():
        response = " ¥ ".join([prompt_text] * scene_count)
    
    import re
    if '¥' not in response and scene_count > 1:
        parts = re.split(r'(?i)scene\s*\d+:', response)
        parts = [p.strip() for p in parts if p.strip()]
        if len(parts) >= scene_count:
            response = ' ¥ '.join(parts[:scene_count])
        else:
            sentences = [s.strip() for s in response.split('.') if s.strip()]
            if len(sentences) >= scene_count:
                chunk_size = max(1, len(sentences) // scene_count)
                chunks = [' '.join(sentences[i:i+chunk_size]) for i in range(0, len(sentences), chunk_size)]
                response = ' ¥ '.join(chunks[:scene_count])
            
    return response

def _auto_prompt_for_mode(mode: str, prompt_text: str, scene_count: int, seconds_per_scene: int) -> Tuple[str, str]:
    prompt_text = (prompt_text or "").strip()
    llm_text = _llm_generate_scene_prompt_text(mode, prompt_text, scene_count, seconds_per_scene)
    scenes = _normalize_scene_parts(llm_text, scene_count, prompt_text)
    return "¥".join(scenes), f"auto-prompt ok: {scene_count} scenes"

def _ensure_scene_prompt_text(prompt_text: str, scene_count: int) -> Tuple[str, str]:
    prompt_text = (prompt_text or "").strip()
    if scene_count <= 1:
        return prompt_text, "scene prompt: single"
    scenes = _normalize_scene_parts(prompt_text, scene_count, prompt_text)
    return "¥".join(scenes), "scene prompt: expanded from base prompt"

def _normalize_scene_inputs(scene_count: int, base_prompt: str, scenes: List[str]) -> List[str]:
    scene_count = max(1, min(int(scene_count), 4))
    base_prompt = (base_prompt or "").strip()
    normalized = [((s or "").strip()) for s in (scenes or [])]
    while len(normalized) < 4:
        normalized.append("")
    normalized = normalized[:4]
    if not any(normalized[:scene_count]):
        return [""] * scene_count
    for i in range(4):
        if not normalized[i]:
            normalized[i] = normalized[i - 1] if i > 0 and normalized[i - 1] else base_prompt
    return normalized[:scene_count]

def _scene_values_for_ui(scene_count: int, scene_array: List[str]) -> Tuple[str, str, str, str]:
    vals = list(scene_array or [])
    while len(vals) < 4:
        vals.append("")
    vals = vals[:4]
    count = max(1, min(int(scene_count), 4))
    if count < 4:
        vals[count] = vals[count] if count < len(vals) else ""
    return vals[0], vals[1], vals[2], vals[3]

def _ui_generate_scenes(mode: str, prompt_text: str, scene_count: int, seconds_per_scene: int, request: Any = None):
    count = max(1, min(int(scene_count or 1), 4))
    try:
        joined, note = _auto_prompt_for_mode(mode, prompt_text, count, int(seconds_per_scene or 3))
        scenes = _normalize_scene_parts(joined, count, prompt_text)
        s1, s2, s3, s4 = _scene_values_for_ui(count, scenes)
        return s1, s2, s3, s4, note
    except Exception as exc:
        fallback = _normalize_scene_inputs(count, "", ["", "", "", ""])
        s1, s2, s3, s4 = _scene_values_for_ui(count, fallback)
        return s1, s2, s3, s4, f"auto-prompt failed: {exc}"

def _workflow_pack_path(wf_dir: Path, workflow_name: str) -> Path:
    return wf_dir / f"{workflow_name}{WORKFLOW_PACK_SUFFIX}"

def _decode_packed_workflow(raw: bytes) -> str:
    if not raw.startswith(WORKFLOW_PACK_MARKER):
        raise RuntimeError("invalid workflow pack marker")
    payload = raw[len(WORKFLOW_PACK_MARKER) :]
    decoded = zlib.decompress(payload)
    return decoded.decode("utf-8")

def _materialize_workflow_from_pack(wf_dir: Path, workflow_name: str) -> Tuple[bool, str]:
    json_path = wf_dir / workflow_name
    if json_path.exists():
        return True, "workflow json already exists"
    packed_path = _workflow_pack_path(wf_dir, workflow_name)
    if not packed_path.exists():
        return False, f"packed workflow not found: {packed_path}"
    try:
        text = _decode_packed_workflow(packed_path.read_bytes())
        data = json.loads(text)
        json_path.parent.mkdir(parents=True, exist_ok=True)
        json_path.write_text(json.dumps(data, ensure_ascii=False), encoding="utf-8")
        return True, f"materialized from pack: {packed_path.name}"
    except Exception as exc:
        return False, f"failed to materialize {workflow_name}: {type(exc).__name__}: {exc}"

def _materialize_packed_workflows(wf_dir: Path) -> None:
    if not wf_dir.exists():
        return
    notes = []
    for name in WORKFLOW_FILES:
        ok, detail = _materialize_workflow_from_pack(wf_dir, name)
        if ok and detail.startswith("materialized"):
            notes.append(detail)

def _download_workflow_files_from_hf(dest_dir: Path) -> str:
    try:
        from huggingface_hub import hf_hub_download
    except Exception as exc:
        return f"workflow hf download unavailable: {type(exc).__name__}: {exc}"
    repo_id = _workflow_repo_id()
    token = _get_hf_token()
    dest_dir.mkdir(parents=True, exist_ok=True)
    got = []
    misses = []
    candidate_filenames = {
        name: (
            f"src/workflows/{name}",
            f"workflows/{name}",
            f"src/workflows/{name}{WORKFLOW_PACK_SUFFIX}",
            f"workflows/{name}{WORKFLOW_PACK_SUFFIX}",
        )
        for name in WORKFLOW_FILES
    }
    for name in WORKFLOW_FILES:
        resolved = False
        for candidate in candidate_filenames[name]:
            try:
                cached = Path(
                    hf_hub_download(
                        repo_id=repo_id,
                        repo_type="space",
                        filename=candidate,
                        token=token,
                    )
                )
                if candidate.endswith(WORKFLOW_PACK_SUFFIX):
                    packed_dst = dest_dir / f"{name}{WORKFLOW_PACK_SUFFIX}"
                    shutil.copy2(cached, packed_dst)
                    ok, detail = _materialize_workflow_from_pack(dest_dir, name)
                    if not ok:
                        raise RuntimeError(detail)
                else:
                    shutil.copy2(cached, dest_dir / name)
                got.append(name)
                resolved = True
                break
            except Exception:
                continue
        if not resolved:
            misses.append(name)
    if misses:
        return f"workflow hf download partial from space:{repo_id}; downloaded={got or 'none'} missing={misses}"
    return f"workflow hf download ok from space:{repo_id}; downloaded={got}"

def _expected_model_paths() -> Dict[str, Path]:
    entries = parse_model_entries()
    root = _model_root_dir()
    out = {}
    for key, (filename, _repo_id, _repo_relpath) in entries.items():
        out[key] = root / MODEL_KEY_TO_SUBDIR[key] / filename
    return out

def check_required_models_status() -> str:
    try:
        entries = parse_model_entries()
    except Exception as exc:
        return f"OMNI_VIDEOS invalid: {type(exc).__name__}: {exc}"
    root = _model_root_dir()
    lines = [f"model root: {root}"]
    for key, path in _expected_model_paths().items():
        filename, repo_id, repo_relpath = entries[key]
        exists = path.exists()
        source = f"{repo_id}@{repo_relpath}" if repo_id and repo_relpath else "no source"
        lines.append(f"- {key}: {'ok' if exists else 'missing'} -> {path.name} ({source})")
    return "\n".join(lines)

def download_missing_models() -> str:
    try:
        from huggingface_hub import hf_hub_download
    except Exception as exc:
        return f"huggingface_hub unavailable: {type(exc).__name__}: {exc}"
    try:
        entries = parse_model_entries()
    except Exception as exc:
        return f"OMNI_VIDEOS invalid: {type(exc).__name__}: {exc}"
    token = _get_hf_token()
    results = []
    for key, path in _expected_model_paths().items():
        filename, repo_id, repo_relpath = entries[key]
        if path.exists():
            results.append(f"- {key}: already exists")
            continue
        if not repo_id or not repo_relpath:
            results.append(f"- {key}: missing and no download source")
            continue
        path.parent.mkdir(parents=True, exist_ok=True)
        try:
            downloaded = Path(
                hf_hub_download(
                    repo_id=repo_id,
                    filename=repo_relpath,
                    token=token,
                    local_dir=str(path.parent),
                )
            )
            if downloaded.resolve() != path.resolve():
                shutil.copy2(downloaded, path)
                saved = path
            else:
                saved = downloaded
            results.append(f"- {key}: downloaded -> {saved.name}")
        except Exception as exc:
            results.append(f"- {key}: download failed ({type(exc).__name__}: {exc})")
    return "\n".join(results)

def ensure_models_ready_on_startup() -> str:
    try:
        entries = parse_model_entries()
    except Exception as exc:
        return f"skip auto-download: OMNI_VIDEOS invalid ({type(exc).__name__}: {exc})"
    missing_keys = []
    for key, path in _expected_model_paths().items():
        if not path.exists():
            missing_keys.append(key)
    if not missing_keys:
        return "all required models already present"
    lines = [f"missing models at startup: {', '.join(missing_keys)}"]
    lines.append(download_missing_models())
    final_missing = [key for key, path in _expected_model_paths().items() if not path.exists()]
    if final_missing:
        lines.append(f"still missing after auto-download: {', '.join(final_missing)}")
    else:
        lines.append("all required models are ready")
    return "\n".join(lines)

def _set_model_prep_state(*, running: Optional[bool] = None, status: Optional[str] = None) -> None:
    global _MODEL_PREP_RUNNING, _MODEL_PREP_STATUS
    with _MODEL_PREP_LOCK:
        if running is not None:
            _MODEL_PREP_RUNNING = running
        if status is not None:
            _MODEL_PREP_STATUS = status

def _get_model_prep_state() -> Tuple[bool, str]:
    with _MODEL_PREP_LOCK:
        return _MODEL_PREP_RUNNING, _MODEL_PREP_STATUS

def _run_model_prep_job() -> None:
    try:
        with _PREP_IO_LOCK:
            result = ensure_models_ready_on_startup()
    except Exception as exc:
        result = f"startup model preparation failed: {type(exc).__name__}: {exc}"
    _set_model_prep_state(running=False, status=result)

def kickoff_model_prepare_background() -> str:
    if _all_models_present():
        _set_model_prep_state(running=False, status="all required models already present")
        return "all required models already present"
    running, status = _get_model_prep_state()
    if running:
        return f"background model preparation already running\n{status}"
    _set_model_prep_state(running=True, status="background model preparation started")
    worker = threading.Thread(target=_run_model_prep_job, name="model-prep", daemon=True)
    worker.start()
    return "background model preparation started"

def ensure_models_ready_for_generation() -> Tuple[bool, str]:
    if _all_models_present():
        return True, "all required models are ready"
    running, status = _get_model_prep_state()
    if running:
        return False, "models are preparing in background\n" + status
    _set_model_prep_state(running=True, status="on-demand model preparation started")
    try:
        with _PREP_IO_LOCK:
            result = ensure_models_ready_on_startup()
    except Exception as exc:
        result = f"on-demand model preparation failed: {type(exc).__name__}: {exc}"
    _set_model_prep_state(running=False, status=result)
    if _all_models_present():
        return True, result
    return False, result

def _set_runtime_prep_state(*, running: Optional[bool] = None, status: Optional[str] = None) -> None:
    global _RUNTIME_PREP_RUNNING, _RUNTIME_PREP_STATUS
    with _RUNTIME_PREP_LOCK:
        if running is not None:
            _RUNTIME_PREP_RUNNING = running
        if status is not None:
            _RUNTIME_PREP_STATUS = status

def _get_runtime_prep_state() -> Tuple[bool, str]:
    with _RUNTIME_PREP_LOCK:
        return _RUNTIME_PREP_RUNNING, _RUNTIME_PREP_STATUS

def _has_comfy_runtime() -> bool:
    comfy_dir = _comfy_dir()
    required = (
        comfy_dir / "ComfyApi.py",
        comfy_dir / "comfy" / "options.py",
        comfy_dir / "comfy" / "cli_args.py",
        comfy_dir / "nodes.py",
        comfy_dir / "folder_paths.py",
        comfy_dir / "execution.py",
    )
    return all(path.exists() for path in required)

def _runtime_missing_paths() -> list[str]:
    comfy_dir = _comfy_dir()
    required = (
        comfy_dir / "ComfyApi.py",
        comfy_dir / "comfy" / "options.py",
        comfy_dir / "comfy" / "cli_args.py",
        comfy_dir / "nodes.py",
        comfy_dir / "folder_paths.py",
        comfy_dir / "execution.py",
    )
    return [str(path.relative_to(comfy_dir)) for path in required if not path.exists()]

def _run_git(cmd: list[str], cwd: Optional[Path] = None) -> Tuple[bool, str]:
    try:
        result = subprocess.run(
            cmd,
            cwd=str(cwd) if cwd is not None else None,
            capture_output=True,
            text=True,
            check=False,
        )
    except Exception as exc:
        return False, f"{type(exc).__name__}: {exc}"
    merged = "\n".join([result.stdout.strip(), result.stderr.strip()]).strip()
    merged = _redact_sensitive(merged)
    if result.returncode != 0:
        return False, merged or f"git command failed (exit={result.returncode})"
    return True, merged

def _runtime_git_head(comfy_dir: Path) -> str:
    ok, detail = _run_git(["git", "-C", str(comfy_dir), "rev-parse", "--short", "HEAD"])
    if not ok:
        return "unknown"
    detail = (detail or "").strip()
    if not detail:
        return "unknown"
    return detail.splitlines()[-1]

def _download_runtime_on_demand() -> str:
    repo = _runtime_git_repo()
    revision = _runtime_git_revision()
    comfy_dir = _comfy_dir()
    models_dir = comfy_dir / "models"
    models_backup = _repo_dir() / ".runtime_models_backup"
    clone_url = _runtime_git_clone_url()
    if models_backup.exists():
        shutil.rmtree(models_backup, ignore_errors=True)
    if comfy_dir.exists() and not (comfy_dir / ".git").exists():
        if models_dir.exists():
            try:
                shutil.move(str(models_dir), str(models_backup))
            except Exception as exc:
                return f"runtime clone failed while preserving models: {type(exc).__name__}: {exc}"
        shutil.rmtree(comfy_dir, ignore_errors=True)
    if not comfy_dir.exists():
        comfy_dir.parent.mkdir(parents=True, exist_ok=True)
        if _is_commit_hash(revision):
            ok, detail = _run_git(
                ["git", "clone", "--depth", "1", clone_url, str(comfy_dir)],
                cwd=_repo_dir(),
            )
            if not ok:
                return f"runtime clone failed: {detail}"
            ok, detail = _run_git(["git", "-C", str(comfy_dir), "fetch", "--depth", "1", "origin", revision])
            if not ok:
                ok, detail = _run_git(["git", "-C", str(comfy_dir), "fetch", "origin", revision])
                if not ok:
                    return f"runtime clone failed (fetch commit): {detail}"
            ok, detail = _run_git(["git", "-C", str(comfy_dir), "reset", "--hard", "FETCH_HEAD"])
            if not ok:
                return f"runtime clone failed (checkout commit): {detail}"
        else:
            ok, detail = _run_git(
                ["git", "clone", "--depth", "1", "--branch", revision, clone_url, str(comfy_dir)],
                cwd=_repo_dir(),
            )
            if not ok:
                return f"runtime clone failed: {detail}"
    else:
        ok, detail = _run_git(["git", "-C", str(comfy_dir), "remote", "set-url", "origin", clone_url])
        if not ok:
            return f"runtime sync failed (set-url): {detail}"
        ok, detail = _run_git(["git", "-C", str(comfy_dir), "fetch", "--depth", "1", "origin", revision])
        if not ok and _is_commit_hash(revision):
            ok, detail = _run_git(["git", "-C", str(comfy_dir), "fetch", "origin", revision])
        if not ok:
            return f"runtime sync failed (fetch): {detail}"
        ok, detail = _run_git(["git", "-C", str(comfy_dir), "reset", "--hard", "FETCH_HEAD"])
        if not ok:
            return f"runtime sync failed (reset): {detail}"
    if models_backup.exists():
        restored_models_dir = comfy_dir / "models"
        try:
            if restored_models_dir.exists():
                shutil.copytree(models_backup, restored_models_dir, dirs_exist_ok=True)
                shutil.rmtree(models_backup, ignore_errors=True)
            else:
                restored_models_dir.parent.mkdir(parents=True, exist_ok=True)
                shutil.move(str(models_backup), str(restored_models_dir))
        except Exception as exc:
            return f"runtime sync finished but restoring models failed: {type(exc).__name__}: {exc}"
    if _has_comfy_runtime():
        return f"runtime ready from github:{repo}@{revision}"
    missing = _runtime_missing_paths()
    return "runtime git sync finished but files are still missing: " + ", ".join(missing)

def _run_runtime_prep_job() -> None:
    try:
        with _PREP_IO_LOCK:
            result = _download_runtime_on_demand()
    except Exception as exc:
        result = f"startup runtime preparation failed: {type(exc).__name__}: {exc}"
    _set_runtime_prep_state(running=False, status=result)

def kickoff_runtime_prepare_background() -> str:
    if _has_comfy_runtime():
        _set_runtime_prep_state(running=False, status="runtime already present")
        return "runtime already present"
    running, status = _get_runtime_prep_state()
    if running:
        return f"background runtime preparation already running\n{status}"
    _set_runtime_prep_state(running=True, status="background runtime preparation started")
    worker = threading.Thread(target=_run_runtime_prep_job, name="runtime-prep", daemon=True)
    worker.start()
    return "background runtime preparation started"

def ensure_runtime_ready_for_generation() -> Tuple[bool, str]:
    if _has_comfy_runtime():
        return True, "runtime already present"
    running, status = _get_runtime_prep_state()
    if running:
        return False, "runtime is preparing in background\n" + status
    _set_runtime_prep_state(running=True, status="on-demand runtime preparation started")
    try:
        with _PREP_IO_LOCK:
            result = _download_runtime_on_demand()
    except Exception as exc:
        result = f"on-demand runtime preparation failed: {type(exc).__name__}: {exc}"
    _set_runtime_prep_state(running=False, status=result)
    if _has_comfy_runtime():
        return True, result
    return False, result

def resolve_placeholders(obj):
    names = parse_model_names()
    resolved = _deep_resolve_placeholders(obj, names)
    return _apply_model_name_aliases(resolved, names)

def _install_comfyapi_src_shims() -> None:
    if "src" not in sys.modules:
        pkg = types.ModuleType("src")
        pkg.__path__ = []
        sys.modules["src"] = pkg
    mod = sys.modules.get("src.model_names")
    if mod is None:
        mod = types.ModuleType("src.model_names")
        sys.modules["src.model_names"] = mod
    mod.resolve_placeholders = resolve_placeholders

def _drop_conflicting_modules(base_dir: Path) -> None:
    checks = {
        "comfy": base_dir / "comfy",
        "utils": base_dir / "utils",
        "folder_paths": base_dir,
        "nodes": base_dir,
        "execution": base_dir,
        "server": base_dir,
    }
    for root_name, expected_path in checks.items():
        mod = sys.modules.get(root_name)
        if mod is None:
            continue
        mod_file = os.path.abspath(getattr(mod, "__file__", "") or "")
        expected_prefix = os.path.abspath(str(expected_path)) + os.sep
        if mod_file.startswith(expected_prefix):
            continue
        for key in list(sys.modules.keys()):
            if key == root_name or key.startswith(root_name + "."):
                del sys.modules[key]

def _get_comfy_client():
    global _COMFY_CLIENT
    if _COMFY_CLIENT is not None:
        return _COMFY_CLIENT

    comfy_dir = _comfy_dir()
    if not _has_comfy_runtime():
        raise RuntimeError("ComfyUIVideo runtime is not ready. Please run generation again.")

    _install_comfyapi_src_shims()
    _drop_conflicting_modules(comfy_dir)

    try:
        config_path = comfy_dir / "custom_nodes" / "was-node-suite-comfyui" / "was_suite_config.json"
        if config_path.parent.exists():
            config_data = {}
            if config_path.exists():
                try:
                    config_data = json.loads(config_path.read_text(encoding="utf-8"))
                except Exception:
                    pass
            config_data["ffmpeg_bin_path"] = "/usr/bin/ffmpeg"
            config_path.write_text(json.dumps(config_data, indent=4), encoding="utf-8")
    except Exception:
        pass

    comfy_dir_str = str(comfy_dir)
    if comfy_dir_str not in sys.path:
        sys.path.insert(0, comfy_dir_str)

    try:
        from ComfyApi import ComfyApi
    except Exception as exc:
        raise RuntimeError(f"Failed to import ComfyApi: {type(exc).__name__}: {exc}") from exc

    workflows_dir = _workflow_dir()
    missing_workflows = [name for name in WORKFLOW_FILES if not (workflows_dir / name).exists()]
    if missing_workflows:
        repo_dir = _repo_dir()
        discovered = {}
        for name in missing_workflows:
            matches = sorted(repo_dir.rglob(name))
            discovered[name] = [str(path) for path in matches[:3]]
        raise RuntimeError(
            f"workflow files missing in {workflows_dir}: {', '.join(missing_workflows)}; discovered={discovered}"
        )
    _COMFY_CLIENT = ComfyApi(
        base_dir=comfy_dir_str,
        workflows_dir=str(workflows_dir),
        preload_models=False,
    )
    return _COMFY_CLIENT

def _resolve_model_name(value: str, names: Dict[str, str]) -> str:
    key = PLACEHOLDER_TO_KEY.get((value or "").strip())
    if not key:
        return value
    return names.get(key, value)

def _deep_resolve_placeholders(obj, names: Dict[str, str]):
    if isinstance(obj, dict):
        return {k: _deep_resolve_placeholders(v, names) for k, v in obj.items()}
    if isinstance(obj, list):
        return [_deep_resolve_placeholders(v, names) for v in obj]
    if isinstance(obj, str):
        return _resolve_model_name(obj, names)
    return obj

def _pick_unet_key(node: Dict, original_name: str) -> str:
    title = ((node.get("_meta") or {}).get("title") or "")
    text = f"{title} {original_name}".lower()
    if any(k in text for k in (" low", " q6kl", "q6kl", "fp8l", "_low", "low)")):
        return "unet_q6kl"
    return "unet_q6kh"

def _apply_model_name_aliases(obj, names: Dict[str, str]):
    if not isinstance(obj, dict):
        return obj
    for _, node in obj.items():
        if not isinstance(node, dict):
            continue
        class_type = str(node.get("class_type") or "")
        inputs = node.get("inputs")
        if not isinstance(inputs, dict):
            continue
        if "vae_name" in inputs and isinstance(inputs.get("vae_name"), str):
            inputs["vae_name"] = names["vae"]
        if "clip_name" in inputs and isinstance(inputs.get("clip_name"), str):
            inputs["clip_name"] = names["text_encoder"]
        if "unet_name" in inputs and isinstance(inputs.get("unet_name"), str):
            if class_type in ("UnetLoaderGGUF", "UnetLoaderGGUFAdvanced", "UNETLoader"):
                key = _pick_unet_key(node, inputs["unet_name"])
                inputs["unet_name"] = names[key]
    return obj

def _collect_placeholders(obj, out: set[str]) -> None:
    if isinstance(obj, dict):
        for v in obj.values():
            _collect_placeholders(v, out)
        return
    if isinstance(obj, list):
        for v in obj:
            _collect_placeholders(v, out)
        return
    if isinstance(obj, str) and obj in PLACEHOLDER_TO_KEY:
        out.add(obj)

def _workflow_dir() -> Path:
    repo_dir = _repo_dir()
    candidates = (
        repo_dir / "src" / "workflows",
    )
    for wf_dir in candidates:
        _materialize_packed_workflows(wf_dir)
        if all((wf_dir / name).exists() for name in WORKFLOW_FILES):
            return wf_dir
    primary = candidates[0]
    primary.mkdir(parents=True, exist_ok=True)
    found = {}
    search_roots = [repo_dir / "src", repo_dir / "workflows", repo_dir]
    for name in WORKFLOW_FILES:
        for root in search_roots:
            if not root.exists():
                continue
            matches = sorted(root.rglob(name))
            if matches:
                found[name] = matches[0]
                break
    if len(found) == len(WORKFLOW_FILES):
        parents = {}
        for path in found.values():
            parents[path.parent] = parents.get(path.parent, 0) + 1
        best_parent = max(parents.items(), key=lambda item: item[1])[0]
        for name, src_path in found.items():
            dst_path = primary / name
            if not dst_path.exists() and src_path.exists():
                shutil.copy2(src_path, dst_path)
        if all((primary / name).exists() for name in WORKFLOW_FILES):
            return primary
        if all((best_parent / name).exists() for name in WORKFLOW_FILES):
            return best_parent
    _download_workflow_files_from_hf(primary)
    return primary

def validate_workflow_placeholders() -> str:
    try:
        names = parse_model_names()
    except Exception as exc:
        return f"OMNI_VIDEOS invalid: {type(exc).__name__}: {exc}"
    wf_dir = _workflow_dir()
    missing = []
    unresolved_msgs = []
    for wf_name in WORKFLOW_FILES:
        wf_path = wf_dir / wf_name
        if not wf_path.exists():
            missing.append(wf_name)
            continue
        try:
            data = json.loads(wf_path.read_text(encoding="utf-8"))
        except Exception as exc:
            unresolved_msgs.append(f"{wf_name}: invalid json ({type(exc).__name__})")
            continue
        resolved = _deep_resolve_placeholders(data, names)
        unresolved = set()
        _collect_placeholders(resolved, unresolved)
        if unresolved:
            unresolved_msgs.append(f"{wf_name}: unresolved placeholders={sorted(unresolved)}")
        else:
            unresolved_msgs.append(f"{wf_name}: ok")
    lines = []
    if missing:
        lines.append(f"missing workflow files: {', '.join(missing)}")
    lines.extend(unresolved_msgs)
    return "\n".join(lines) if lines else "no workflow files found"

def _env_status_text() -> str:
    lines = []
    raw = (os.getenv("OMNI_VIDEOS") or "").strip()
    if not raw:
        lines.append("- OMNI_VIDEOS: missing")
    else:
        try:
            parsed = parse_model_names(raw)
            lines.append(f"- OMNI_VIDEOS: ok ({parsed['vae']}, {parsed['text_encoder']})")
        except Exception as exc:
            lines.append(f"- OMNI_VIDEOS: invalid ({type(exc).__name__})")
    one_key = (os.getenv("ONE_KEY") or "").strip()
    lines.append("- ONE_KEY: configured" if one_key else "- ONE_KEY: missing")
    lines.append("- WORKER_API_TOKEN: configured" if _get_worker_api_token() else "- WORKER_API_TOKEN: missing")
    lines.append("- GITHUB_TOKEN: configured" if _get_github_token() else "- GITHUB_TOKEN: missing")
    return "\n".join(lines)

@spaces.GPU(duration=20)
def healthcheck_gpu(name: str) -> str:
    who = (name or "").strip() or "world"
    return f"Omni-Video-Factory bootstrap is running. Hello, {who}!"

def _all_models_present() -> bool:
    try:
        return all(path.exists() for path in _expected_model_paths().values())
    except Exception:
        return False

def _maybe_file_path(value) -> str:
    if isinstance(value, str):
        return value
    if isinstance(value, dict):
        for key in ("name", "path"):
            got = value.get(key)
            if isinstance(got, str):
                return got
    return ""

def _output_dir() -> Path:
    out = _repo_dir() / "datas" / "outputs"
    out.mkdir(parents=True, exist_ok=True)
    return out

def _scene_count() -> int:
    default_scene = int(getattr(APP_CONFIG, "DEFAULT_SCENE_COUNT"))
    raw = (os.getenv("OMNI_SCENE_COUNT") or str(default_scene)).strip()
    try:
        value = int(raw)
    except (TypeError, ValueError):
        value = default_scene
    return max(1, min(value, 4))

def _vid_resolution_default() -> int:
    raw = (os.getenv("OMNI_VID_RES") or "384").strip()
    try:
        value = int(raw)
    except (TypeError, ValueError):
        value = 384
    if value not in (384, 512):
        return 384
    return value

def _aspect_ratio_default() -> str:
    raw = (os.getenv("OMNI_ASPECT_RATIO") or "3:4").strip()
    allowed = {"16:9", "4:3", "1:1", "3:4", "9:16"}
    return raw if raw in allowed else "3:4"

def _seconds_per_scene_default() -> int:
    raw = (os.getenv("OMNI_SECONDS_PER_SCENE") or "3").strip()
    try:
        value = int(raw)
    except (TypeError, ValueError):
        value = 3
    return 5 if value >= 5 else 3

def _allowed_scene_counts(resolution: int) -> List[int]:
    rules = {
        384: [1, 2, 3, 4],
        512: [1, 2],
    }
    return rules.get(int(resolution), [1])

def _normalize_generation_options(
    mode: str,
    scene_count: Optional[int],
    resolution: Optional[int],
    seconds_per_scene: Optional[int],
) -> Tuple[int, int, int, str]:
    res = int(resolution) if resolution else _vid_resolution_default()
    if res not in (384, 512):
        res = 384
    allowed_scenes = _allowed_scene_counts(res)
    if scene_count is None:
        scenes = _scene_count()
    else:
        try:
            scenes = int(scene_count)
        except (TypeError, ValueError):
            scenes = _scene_count()
    if scenes not in allowed_scenes:
        scenes = allowed_scenes[-1]
    secs = int(seconds_per_scene) if seconds_per_scene else _seconds_per_scene_default()
    secs = 5 if secs >= 5 else 3
    if mode == "v2v":
        scenes = 1
    notes = f"config: res={res}, scenes={scenes}, seconds_per_scene={secs}"
    return scenes, res, secs, notes

def _frames_for_seconds(seconds_per_scene: int) -> int:
    return 81 if int(seconds_per_scene) >= 5 else 49

def _round_to_multiple(value: int, multiple: int = 16) -> int:
    if multiple <= 1:
        return int(value)
    return int((int(value) + multiple - 1) // multiple * multiple)

def _compute_t2v_dims(aspect_ratio: str, resolution: int) -> Tuple[int, int]:
    res = int(resolution)
    aspect_map = {
        "16:9": (16, 9),
        "4:3": (4, 3),
        "1:1": (1, 1),
        "3:4": (3, 4),
        "9:16": (9, 16),
    }
    w_ratio, h_ratio = aspect_map.get(aspect_ratio, (3, 4))
    target_area = max(256, int(res) * int(res))
    width = int((target_area * (w_ratio / h_ratio)) ** 0.5)
    height = int((target_area * (h_ratio / w_ratio)) ** 0.5)
    return _round_to_multiple(width, 16), _round_to_multiple(height, 16)

def _gpu_duration_default() -> int:
    default_fixed = int(getattr(APP_CONFIG, "GPU_DEFAULT_FIXED_SECONDS"))
    raw = (os.getenv("OMNI_GPU_SECONDS") or str(default_fixed)).strip()
    try:
        value = int(raw)
    except (TypeError, ValueError):
        value = default_fixed
    min_secs = int(getattr(APP_CONFIG, "GPU_MIN_SECONDS"))
    max_secs = int(getattr(APP_CONFIG, "GPU_MAX_SECONDS"))
    return max(min_secs, min(value, max_secs))

def _estimated_runtime_seconds(
    mode: str,
    scene_count: Optional[int],
    resolution: Optional[int],
    seconds_per_scene: Optional[int],
) -> float:
    scenes, res, secs, _ = _normalize_generation_options(
        mode=mode,
        scene_count=scene_count,
        resolution=resolution,
        seconds_per_scene=seconds_per_scene,
    )
    raw_base_by_res = dict(getattr(APP_CONFIG, "GPU_BASE_SECONDS_BY_RES") or {})
    base_by_res: Dict[int, float] = {}
    for key, value in raw_base_by_res.items():
        try:
            base_by_res[int(key)] = float(value)
        except (TypeError, ValueError):
            continue
    base = float(base_by_res.get(int(res), 27.0))
    scene_extra = float(getattr(APP_CONFIG, "GPU_SCENE_EXTRA_FACTOR"))
    scene_multiplier = 1.0 + scene_extra * max(0, int(scenes) - 1)
    seconds_multiplier = 1.0 if int(secs) <= 3 else float(getattr(APP_CONFIG, "GPU_SECONDS_5_MULTIPLIER"))
    mode_multiplier_map = dict(getattr(APP_CONFIG, "GPU_MODE_MULTIPLIER"))
    mode_multiplier = float(mode_multiplier_map.get(str(mode or "").lower().strip(), 1.0))
    return max(10.0, base * scene_multiplier * seconds_multiplier * mode_multiplier)

def _dynamic_gpu_duration(
    mode: str,
    prompt: str,
    image_file=None,
    video_file=None,
    scene_count: Optional[int] = None,
    resolution: Optional[int] = None,
    seconds_per_scene: Optional[int] = None,
    auto_prompt: bool = False,
    aspect_ratio: str = "3:4",
    client_ip: Optional[str] = None,
    client_country: Optional[str] = None,
    target_fps: int = 16,
) -> int:
    raw_fixed = (os.getenv("OMNI_GPU_SECONDS") or "").strip()
    if raw_fixed:
        try:
            min_secs = int(getattr(APP_CONFIG, "GPU_MIN_SECONDS"))
            max_secs = int(getattr(APP_CONFIG, "GPU_MAX_SECONDS"))
            return max(min_secs, min(int(raw_fixed), max_secs))
        except (TypeError, ValueError):
            pass
    estimated = _estimated_runtime_seconds(mode, scene_count, resolution, seconds_per_scene)
    ratio_default = float(getattr(APP_CONFIG, "GPU_BUFFER_RATIO_DEFAULT"))
    extra_default = float(getattr(APP_CONFIG, "GPU_BUFFER_SECONDS_DEFAULT"))
    raw_ratio = (os.getenv("OMNI_GPU_BUFFER_RATIO") or str(ratio_default)).strip()
    raw_extra = (os.getenv("OMNI_GPU_BUFFER_SECONDS") or str(extra_default)).strip()
    try:
        ratio = float(raw_ratio)
    except (TypeError, ValueError):
        ratio = ratio_default
    try:
        extra = float(raw_extra)
    except (TypeError, ValueError):
        extra = extra_default
    requested = int(estimated * max(1.0, ratio) + max(0.0, extra) + 0.5)
    min_secs = int(getattr(APP_CONFIG, "GPU_MIN_SECONDS"))
    max_secs = int(getattr(APP_CONFIG, "GPU_MAX_SECONDS"))
    return max(min_secs, min(requested, max_secs))

def _patch_workflow_settings(target_fps: int) -> None:
    wf_dir = _workflow_dir()
    for wf_name in WORKFLOW_FILES:
        wf_path = wf_dir / wf_name
        if not wf_path.exists():
            continue
        try:
            data = json.loads(wf_path.read_text(encoding="utf-8"))
            changed = False
            for key, node in data.items():
                class_type = str(node.get("class_type", ""))
                if class_type == "VHS_VideoCombine":
                    if "inputs" in node:
                        if "frame_rate" in node["inputs"] and node["inputs"]["frame_rate"] != target_fps:
                            node["inputs"]["frame_rate"] = target_fps
                            changed = True
                        if "crf" in node["inputs"] and node["inputs"]["crf"] != 0:
                            node["inputs"]["crf"] = 0
                            changed = True
                if "VFI" in class_type or class_type == "RIFE VFI":
                    if "inputs" in node:
                        if "multiplier" in node["inputs"] and node["inputs"]["multiplier"] != 2:
                            node["inputs"]["multiplier"] = 2
                            changed = True
            if changed:
                wf_path.write_text(json.dumps(data, indent=4), encoding="utf-8")
        except Exception:
            pass

@spaces.GPU(duration=_dynamic_gpu_duration)
def run_generation_real(
    mode: str,
    prompt: str,
    image_file=None,
    video_file=None,
    scene_count: Optional[int] = None,
    resolution: Optional[int] = None,
    seconds_per_scene: Optional[int] = None,
    auto_prompt: bool = False,
    aspect_ratio: str = "3:4",
    client_ip: Optional[str] = None,
    client_country: Optional[str] = None,
    target_fps: int = 16,
):
    os.environ["OMNI_TARGET_FPS"] = str(target_fps)
    os.environ["OMNI_BASE_FPS"] = "16"
    os.environ["OMNI_FRAME_MULTIPLIER"] = "2"
    os.environ["OMNI_CRF"] = "0"
    
    _patch_workflow_settings(target_fps)
    
    started_at = time.perf_counter()
    warming_up_message = "warming up, please retry in 15-30 seconds"

    def _status_text(base: str) -> str:
        elapsed = max(0.0, time.perf_counter() - started_at)
        message = (base or "").strip()
        if message:
            return f"{message}\nZeroGPU elapsed: {elapsed:.1f}s"
        return f"ZeroGPU elapsed: {elapsed:.1f}s"

    prompt = (prompt or "").strip()
    if not prompt:
        return _status_text("prompt is empty"), None, None
    
    models_ok, prep_status = ensure_models_ready_for_generation()
    if not models_ok:
        return (
            _status_text(
                "models are not ready\n"
            + prep_status
            + "\n"
            + check_required_models_status()
            )
        ), None, None

    image_path = _maybe_file_path(image_file)
    video_path = _maybe_file_path(video_file)

    if mode == "i2v" and not image_path:
        return _status_text("i2v requires an image file"), None, None
    if mode == "v2v" and not video_path:
        return _status_text("v2v requires a video file"), None, None

    runtime_ok, runtime_status = ensure_runtime_ready_for_generation()
    if not runtime_ok:
        return _status_text(f"{warming_up_message}\n{runtime_status}"), None, None

    try:
        comfy = _get_comfy_client()
    except Exception as exc:
        exc_text = f"{type(exc).__name__}: {exc}"
        warmup_keywords = ("missing workflow", "runtime is not ready", "Failed to import ComfyApi")
        if any(key.lower() in exc_text.lower() for key in warmup_keywords):
            return _status_text(f"{warming_up_message}\n{exc_text}"), None, None
        return _status_text(f"runtime unavailable: {exc_text}"), None, None

    stamp = time.strftime("%Y%m%d_%H%M%S")
    output_path = _output_dir() / f"{mode}_{stamp}_{int(time.time())}.mp4"
    scene_count, vid_res, secs_per_scene, _option_note = _normalize_generation_options(
        mode=mode,
        scene_count=scene_count,
        resolution=resolution,
        seconds_per_scene=seconds_per_scene,
    )
    num_frames = _frames_for_seconds(secs_per_scene)
    t2v_width, t2v_height = _compute_t2v_dims(aspect_ratio, vid_res)
    if mode in {"t2v", "i2v"}:
        if auto_prompt:
            try:
                prompt, _ = _auto_prompt_for_mode(mode, prompt, scene_count, secs_per_scene)
            except Exception as exc:
                prompt, _ = _ensure_scene_prompt_text(prompt, scene_count)
        else:
            prompt, _ = _ensure_scene_prompt_text(prompt, scene_count)

    try:
        if mode == "t2v":
            saved = comfy.video_t2v(
                prompt=prompt,
                vid_resolution=vid_res,
                num_frames=num_frames,
                num_scene=scene_count,
                width=t2v_width,
                height=t2v_height,
                output_path=str(output_path),
            )
        elif mode == "i2v":
            saved = comfy.video_i2v(
                prompt=prompt,
                img_path=image_path,
                vid_resolution=vid_res,
                num_frames=num_frames,
                num_scene=scene_count,
                output_path=str(output_path),
            )
        elif mode == "v2v":
            saved = comfy.video_v2v(
                prompt=prompt,
                video_path=video_path,
                num_frames=num_frames,
                vid_resolution=vid_res,
                output_path=str(output_path),
            )
        else:
            return _status_text(f"unknown mode: {mode}"), None, None
    except Exception as exc:
        return _status_text(f"{mode} failed: {type(exc).__name__}: {exc}"), None, None

    final_path = str(saved or output_path)
    if not Path(final_path).exists():
        return _status_text(f"{mode} finished but output missing: {final_path}"), None, None

    status_lines: List[str] = []
    return _status_text("\n".join(status_lines)), final_path, None

def build_demo() -> gr.Blocks:
    banner_subtitle = "Please give us a ❤️ if you find it helpful. No usage limits beyond your ZeroGPU daily quota."
    manual_prompt_min_chars = 10
    image_editor_upsell_html = """
    <div style="margin-top:14px;">
      <a href="https://huggingface.co/spaces/selfit-camera/Omni-Image-Editor" target="_blank" style="display:block; text-decoration:none;">
        <div style="display:flex; align-items:center; justify-content:center; gap:18px; padding:16px 20px; border-radius:22px; background:linear-gradient(90deg,#e38cf4 0%,#f55978 100%); box-shadow:0 10px 24px rgba(245,89,120,0.18);">
          <div style="color:#fff; font-size:1.3em; font-weight:800; line-height:1.2; text-align:center;">Try our multi-feature image editing demo here</div>
          <div style="flex:0 0 auto; background:#fff; color:#f25575; border-radius:18px; padding:10px 24px; font-size:1.25em; font-weight:800; line-height:1;">Try it now</div>
        </div>
      </a>
    </div>
    """
    with gr.Blocks(title="Omni Video Factory") as demo:
        gr.HTML(
            f"""
            <div style="text-align: center; margin: 20px auto 10px auto; max-width: 800px;">
                <h1 style="color: #2c3e50; margin: 0; font-size: 3.5em; font-weight: 800; letter-spacing: 3px; text-shadow: 2px 2px 4px rgba(0,0,0,0.1);">
                    🎬 {getattr(APP_CONFIG, "APP_TITLE", "Omni Video Factory")}
                </h1>
            </div>
            <style>
            @keyframes breathe {{
                0%, 100% {{ transform: scale(1); }}
                50% {{ transform: scale(1.02); }}
            }}
            .breathing-banner {{ animation: breathe 3s ease-in-out infinite; }}
            </style>
            <div class="breathing-banner" style="
                background: linear-gradient(135deg, #667eea 0%, #764ba2 100%);
                margin: 0 auto 20px auto; padding: 12px 60px; border-radius: 24px;
                min-width: 460px; max-width: 900px; box-shadow: 0 2px 8px rgba(102, 126, 234, 0.3);
                text-align: center;
            ">
                <span style="color: white; font-weight: 600; font-size: 1.0em;">
                    {banner_subtitle}
                </span>
            </div>
            """
        )
        default_resolution = _vid_resolution_default()
        allowed_default_scenes = _allowed_scene_counts(default_resolution)
        default_scene = _scene_count()
        if default_scene not in allowed_default_scenes:
            default_scene = allowed_default_scenes[-1]
        default_ratio = _aspect_ratio_default()
        d_w, d_h = _compute_t2v_dims(default_ratio, default_resolution)
        resolution_notes = {
            384: "384 is for learning the workflow only, not recommended for daily use.",
            512: "Balanced speed and quality, but face ID consistency is weaker than 800.",
        }

        def _on_dims_change(ratio_value, res_value):
            try:
                w, h = _compute_t2v_dims(str(ratio_value), int(res_value))
            except Exception:
                w, h = _compute_t2v_dims("3:4", 384)
            return f"T2V dims: **{w}x{h}**"

        def _resolution_note_text(res_value):
            try:
                res = int(res_value)
            except Exception:
                res = 384
            return str(resolution_notes.get(res, ""))

        def _scene_box_visibility(scene_value):
            try:
                count = max(1, min(int(scene_value), 4))
            except Exception:
                count = 1
            return (
                gr.update(visible=(count >= 1)),
                gr.update(visible=(count >= 2)),
                gr.update(visible=(count >= 3)),
                gr.update(visible=(count >= 4)),
            )

        def _on_scene_count_change(scene_value):
            updates = _scene_box_visibility(scene_value)
            return updates + updates

        def _on_resolution_change_with_scene_boxes(res_value, scene_value):
            try:
                allowed = _allowed_scene_counts(int(res_value))
            except Exception:
                allowed = [1]
            try:
                current = int(scene_value)
            except Exception:
                current = allowed[0]
            if current not in allowed:
                current = allowed[-1]
            scene_updates = _on_scene_count_change(current)
            return (gr.update(choices=allowed, value=current),) + scene_updates

        def _manual_scene_validation_error(scene_count, s1, s2, s3, s4):
            try:
                count = max(1, min(int(scene_count), 4))
            except Exception:
                count = 1
            scenes = [s1, s2, s3, s4]
            for idx, text in enumerate(scenes[:count], start=1):
                if len((text or "").strip()) < manual_prompt_min_chars:
                    return (
                        f"Manual prompt requires at least {manual_prompt_min_chars} characters "
                        f"for each visible scene (Scene {idx})."
                    )
            return ""

        def _manual_submit_rejected(message: str):
            return (
                message,
                None,
                gr.update(value="", visible=False),
                gr.update(value=None, visible=False),
                gr.update(value="", visible=False),
            )

        def _submit_t2v(
            scene_count,
            seconds_per_scene,
            resolution,
            aspect_ratio,
            target_fps,
            base_prompt,
            s1,
            s2,
            s3,
            s4,
            request: gr.Request,
        ):
            scenes = _normalize_scene_inputs(int(scene_count), base_prompt, [s1, s2, s3, s4])
            prompt = "¥".join(scenes)
            ip = _request_ip(request)
            cc = _request_country(request, ip=ip)
            status, video, preview = run_generation_real(
                "t2v",
                prompt,
                None,
                None,
                int(scene_count),
                int(resolution),
                int(seconds_per_scene),
                False,
                str(aspect_ratio),
                ip,
                cc,
                int(target_fps),
            )
            show_like_tip = _should_show_like_tip(ip)
            return (
                _public_generation_status(status),
                video,
                _nsfw_card_update_for_status(status),
                _nsfw_preview_update(preview),
                _download_upsell_update(video, status, show_like_tip=show_like_tip),
            )

        def _submit_i2v(
            scene_count,
            seconds_per_scene,
            resolution,
            target_fps,
            image_file,
            base_prompt,
            s1,
            s2,
            s3,
            s4,
            request: gr.Request,
        ):
            scenes = _normalize_scene_inputs(int(scene_count), base_prompt, [s1, s2, s3, s4])
            prompt = "¥".join(scenes)
            ip = _request_ip(request)
            cc = _request_country(request, ip=ip)
            status, video, preview = run_generation_real(
                "i2v",
                prompt,
                image_file,
                None,
                int(scene_count),
                int(resolution),
                int(seconds_per_scene),
                False,
                "3:4",
                ip,
                cc,
                int(target_fps),
            )
            show_like_tip = _should_show_like_tip(ip)
            return (
                _public_generation_status(status),
                video,
                _nsfw_card_update_for_status(status),
                _nsfw_preview_update(preview),
                _download_upsell_update(video, status, show_like_tip=show_like_tip),
            )

        def _submit_t2v_manual(
            scene_count,
            seconds_per_scene,
            resolution,
            aspect_ratio,
            target_fps,
            base_prompt,
            s1,
            s2,
            s3,
            s4,
            request: gr.Request,
        ):
            err = _manual_scene_validation_error(scene_count, s1, s2, s3, s4)
            if err:
                return _manual_submit_rejected(err)
            return _submit_t2v(
                scene_count,
                seconds_per_scene,
                resolution,
                aspect_ratio,
                target_fps,
                base_prompt,
                s1,
                s2,
                s3,
                s4,
                request,
            )

        def _submit_i2v_manual(
            scene_count,
            seconds_per_scene,
            resolution,
            target_fps,
            image_file,
            base_prompt,
            s1,
            s2,
            s3,
            s4,
            request: gr.Request,
        ):
            err = _manual_scene_validation_error(scene_count, s1, s2, s3, s4)
            if err:
                return _manual_submit_rejected(err)
            return _submit_i2v(
                scene_count,
                seconds_per_scene,
                resolution,
                target_fps,
                image_file,
                base_prompt,
                s1,
                s2,
                s3,
                s4,
                request,
            )

        def _submit_v2v(prompt, video_file, resolution, seconds_per_scene, aspect_ratio, target_fps, request: gr.Request):
            ip = _request_ip(request)
            cc = _request_country(request, ip=ip)
            status, video, preview = run_generation_real(
                "v2v",
                prompt,
                None,
                video_file,
                1,
                int(resolution),
                int(seconds_per_scene),
                False,
                str(aspect_ratio),
                ip,
                cc,
                int(target_fps),
            )
            show_like_tip = _should_show_like_tip(ip)
            return (
                _public_generation_status(status),
                video,
                _nsfw_card_update_for_status(status),
                _nsfw_preview_update(preview),
                _download_upsell_update(video, status, show_like_tip=show_like_tip),
            )

        def _generate_t2v_scenes(base_prompt, scene_count, seconds_per_scene, request: gr.Request):
            return _ui_generate_scenes("t2v", base_prompt, scene_count, seconds_per_scene, request)

        def _generate_i2v_scenes(base_prompt, scene_count, seconds_per_scene, request: gr.Request):
            return _ui_generate_scenes("i2v", base_prompt, scene_count, seconds_per_scene, request)

        with gr.Tab("🖼️ Image to Video"):
            with gr.Row():
                with gr.Column(scale=1):
                    scene_count_i2v = gr.Radio([1, 2, 3, 4], value=default_scene, label="Scene Count")
                    seconds_i2v = gr.Radio([3, 5], value=_seconds_per_scene_default(), label="Seconds per Scene")
                    resolution_i2v = gr.Radio([384, 512], value=default_resolution, label="Resolution")
                    target_fps_i2v = gr.Radio([16, 32, 64, 128], value=16, label="Target FPS (Frame Interpolation)")
                    resolution_note_i2v = gr.Markdown(_resolution_note_text(default_resolution))

                    image_i2v = gr.File(label="Image file", file_types=["image"])
                    gr.Markdown("I2V aspect ratio follows the uploaded image automatically.")
                    with gr.Tabs():
                        with gr.Tab("Manual Prompt"):
                            s1_i2v_manual = gr.Textbox(label="Scene Prompt 1", lines=2, visible=True)
                            s2_i2v_manual = gr.Textbox(label="Scene Prompt 2", lines=2, visible=(default_scene >= 2))
                            s3_i2v_manual = gr.Textbox(label="Scene Prompt 3", lines=2, visible=(default_scene >= 3))
                            s4_i2v_manual = gr.Textbox(label="Scene Prompt 4", lines=2, visible=(default_scene >= 4))
                            run_btn_i2v_manual = gr.Button("Submit Generate I2V", variant="primary")
                        with gr.Tab("Auto Prompt"):
                            base_prompt_i2v = gr.Textbox(label="Prompt", value="A drone shot over mountains", lines=3)
                            gen_scene_btn_i2v = gr.Button("Generate Scene Prompts (Auto)")
                            s1_i2v_auto = gr.Textbox(label="Scene Prompt 1", lines=2, visible=True)
                            s2_i2v_auto = gr.Textbox(label="Scene Prompt 2", lines=2, visible=(default_scene >= 2))
                            s3_i2v_auto = gr.Textbox(label="Scene Prompt 3", lines=2, visible=(default_scene >= 3))
                            s4_i2v_auto = gr.Textbox(label="Scene Prompt 4", lines=2, visible=(default_scene >= 4))
                            scene_note_i2v = gr.Textbox(label="Auto Prompt Status", interactive=False)
                            run_btn_i2v_auto = gr.Button("Submit Generate I2V", variant="primary")
                with gr.Column(scale=1):
                    out_video_i2v = gr.Video(label="Generated video", height=384)
                    nsfw_card_i2v = gr.HTML(value="", visible=False)
                    download_card_i2v = gr.HTML(value="", visible=False)
                    nsfw_preview_i2v = gr.Image(label="NSFW Preview (Blurred 80% Frame)", type="filepath", visible=False)
                    status_i2v = gr.Textbox(label="Generation status", lines=8)
                    gr.HTML(image_editor_upsell_html)

            resolution_i2v.change(
                _on_resolution_change_with_scene_boxes,
                [resolution_i2v, scene_count_i2v],
                [
                    scene_count_i2v,
                    s1_i2v_manual,
                    s2_i2v_manual,
                    s3_i2v_manual,
                    s4_i2v_manual,
                    s1_i2v_auto,
                    s2_i2v_auto,
                    s3_i2v_auto,
                    s4_i2v_auto,
                ],
            )
            resolution_i2v.change(_resolution_note_text, [resolution_i2v], [resolution_note_i2v])
            scene_count_i2v.change(
                _on_scene_count_change,
                [scene_count_i2v],
                [s1_i2v_manual, s2_i2v_manual, s3_i2v_manual, s4_i2v_manual, s1_i2v_auto, s2_i2v_auto, s3_i2v_auto, s4_i2v_auto],
            )
            gen_scene_btn_i2v.click(
                _generate_i2v_scenes,
                inputs=[base_prompt_i2v, scene_count_i2v, seconds_i2v],
                outputs=[s1_i2v_auto, s2_i2v_auto, s3_i2v_auto, s4_i2v_auto, scene_note_i2v],
            )
            run_btn_i2v_manual.click(
                _submit_i2v_manual,
                inputs=[scene_count_i2v, seconds_i2v, resolution_i2v, target_fps_i2v, image_i2v, base_prompt_i2v, s1_i2v_manual, s2_i2v_manual, s3_i2v_manual, s4_i2v_manual],
                outputs=[status_i2v, out_video_i2v, nsfw_card_i2v, nsfw_preview_i2v, download_card_i2v],
            )
            run_btn_i2v_auto.click(
                _submit_i2v,
                inputs=[scene_count_i2v, seconds_i2v, resolution_i2v, target_fps_i2v, image_i2v, base_prompt_i2v, s1_i2v_auto, s2_i2v_auto, s3_i2v_auto, s4_i2v_auto],
                outputs=[status_i2v, out_video_i2v, nsfw_card_i2v, nsfw_preview_i2v, download_card_i2v],
            )

        with gr.Tab("✨ Text to Video"):
            with gr.Row():
                with gr.Column(scale=1):
                    scene_count_t2v = gr.Radio([1, 2, 3, 4], value=default_scene, label="Scene Count")
                    seconds_t2v = gr.Radio([3, 5], value=_seconds_per_scene_default(), label="Seconds per Scene")
                    resolution_t2v = gr.Radio([384, 512], value=default_resolution, label="Resolution")
                    target_fps_t2v = gr.Radio([16, 32, 64, 128], value=16, label="Target FPS (Frame Interpolation)")
                    resolution_note_t2v = gr.Markdown(_resolution_note_text(default_resolution))
                    aspect_ratio_t2v = gr.Radio(["16:9", "4:3", "1:1", "3:4", "9:16"], value=default_ratio, label="Aspect Ratio")
                    dims_hint_t2v = gr.Markdown(f"T2V dims: **{d_w}x{d_h}**")

                    with gr.Tabs():
                        with gr.Tab("Manual Prompt"):
                            s1_t2v_manual = gr.Textbox(label="Scene Prompt 1", lines=2, visible=True)
                            s2_t2v_manual = gr.Textbox(label="Scene Prompt 2", lines=2, visible=(default_scene >= 2))
                            s3_t2v_manual = gr.Textbox(label="Scene Prompt 3", lines=2, visible=(default_scene >= 3))
                            s4_t2v_manual = gr.Textbox(label="Scene Prompt 4", lines=2, visible=(default_scene >= 4))
                            run_btn_t2v_manual = gr.Button("Submit Generate T2V", variant="primary")
                        with gr.Tab("Auto Prompt"):
                            base_prompt_t2v = gr.Textbox(label="Prompt", value="A cinematic ocean wave at sunset", lines=3)
                            gen_scene_btn_t2v = gr.Button("Generate Scene Prompts (Auto)")
                            s1_t2v_auto = gr.Textbox(label="Scene Prompt 1", lines=2, visible=True)
                            s2_t2v_auto = gr.Textbox(label="Scene Prompt 2", lines=2, visible=(default_scene >= 2))
                            s3_t2v_auto = gr.Textbox(label="Scene Prompt 3", lines=2, visible=(default_scene >= 3))
                            s4_t2v_auto = gr.Textbox(label="Scene Prompt 4", lines=2, visible=(default_scene >= 4))
                            scene_note_t2v = gr.Textbox(label="Auto Prompt Status", interactive=False)
                            run_btn_t2v_auto = gr.Button("Submit Generate T2V", variant="primary")
                with gr.Column(scale=1):
                    out_video_t2v = gr.Video(label="Generated video", height=384)
                    nsfw_card_t2v = gr.HTML(value="", visible=False)
                    download_card_t2v = gr.HTML(value="", visible=False)
                    nsfw_preview_t2v = gr.Image(label="NSFW Preview (Blurred 80% Frame)", type="filepath", visible=False)
                    status_t2v = gr.Textbox(label="Generation status", lines=8)
                    gr.HTML(image_editor_upsell_html)

            resolution_t2v.change(
                _on_resolution_change_with_scene_boxes,
                [resolution_t2v, scene_count_t2v],
                [
                    scene_count_t2v,
                    s1_t2v_manual,
                    s2_t2v_manual,
                    s3_t2v_manual,
                    s4_t2v_manual,
                    s1_t2v_auto,
                    s2_t2v_auto,
                    s3_t2v_auto,
                    s4_t2v_auto,
                ],
            )
            resolution_t2v.change(_resolution_note_text, [resolution_t2v], [resolution_note_t2v])
            scene_count_t2v.change(
                _on_scene_count_change,
                [scene_count_t2v],
                [s1_t2v_manual, s2_t2v_manual, s3_t2v_manual, s4_t2v_manual, s1_t2v_auto, s2_t2v_auto, s3_t2v_auto, s4_t2v_auto],
            )
            aspect_ratio_t2v.change(_on_dims_change, [aspect_ratio_t2v, resolution_t2v], [dims_hint_t2v])
            resolution_t2v.change(_on_dims_change, [aspect_ratio_t2v, resolution_t2v], [dims_hint_t2v])
            gen_scene_btn_t2v.click(
                _generate_t2v_scenes,
                inputs=[base_prompt_t2v, scene_count_t2v, seconds_t2v],
                outputs=[s1_t2v_auto, s2_t2v_auto, s3_t2v_auto, s4_t2v_auto, scene_note_t2v],
            )
            run_btn_t2v_manual.click(
                _submit_t2v_manual,
                inputs=[scene_count_t2v, seconds_t2v, resolution_t2v, aspect_ratio_t2v, target_fps_t2v, base_prompt_t2v, s1_t2v_manual, s2_t2v_manual, s3_t2v_manual, s4_t2v_manual],
                outputs=[status_t2v, out_video_t2v, nsfw_card_t2v, nsfw_preview_t2v, download_card_t2v],
            )
            run_btn_t2v_auto.click(
                _submit_t2v,
                inputs=[scene_count_t2v, seconds_t2v, resolution_t2v, aspect_ratio_t2v, target_fps_t2v, base_prompt_t2v, s1_t2v_auto, s2_t2v_auto, s3_t2v_auto, s4_t2v_auto],
                outputs=[status_t2v, out_video_t2v, nsfw_card_t2v, nsfw_preview_t2v, download_card_t2v],
            )

        with gr.Tab("🎞️ Video Extend"):
            with gr.Row():
                with gr.Column(scale=1):
                    seconds_v2v = gr.Radio([3, 5], value=_seconds_per_scene_default(), label="Additional Seconds")
                    resolution_v2v = gr.Radio([384, 512], value=default_resolution, label="Resolution")
                    aspect_ratio_v2v = gr.Radio(["16:9", "4:3", "1:1", "3:4", "9:16"], value=default_ratio, label="Aspect Ratio")
                    target_fps_v2v = gr.Radio([16, 32, 64, 128], value=16, label="Target FPS (Frame Interpolation)")
                    prompt_v2v = gr.Textbox(label="Prompt", value="Enhance this clip with cinematic lighting", lines=3)
                    video_v2v = gr.File(label="Video file", file_types=["video"])
                    run_btn_v2v = gr.Button("Submit Generate V2V", variant="primary")
                with gr.Column(scale=1):
                    out_video_v2v = gr.Video(label="Generated video", height=384)
                    nsfw_card_v2v = gr.HTML(value="", visible=False)
                    download_card_v2v = gr.HTML(value="", visible=False)
                    nsfw_preview_v2v = gr.Image(label="NSFW Preview (Blurred 80% Frame)", type="filepath", visible=False)
                    status_v2v = gr.Textbox(label="Generation status", lines=8)
                    gr.HTML(image_editor_upsell_html)

            run_btn_v2v.click(
                _submit_v2v,
                inputs=[prompt_v2v, video_v2v, resolution_v2v, seconds_v2v, aspect_ratio_v2v, target_fps_v2v],
                outputs=[status_v2v, out_video_v2v, nsfw_card_v2v, nsfw_preview_v2v, download_card_v2v],
            )
        gr.HTML(
            """
            <div style="width:100%; margin:42px 0 20px 0; padding:0 8px;">
              <div style="background:linear-gradient(135deg,#667eea 0%,#764ba2 100%); border-radius:24px; padding:34px 28px; box-shadow:0 14px 36px rgba(102,126,234,0.32);">
                <div style="text-align:center; margin-bottom:18px;">
                  <div style="font-size:2.7em; font-weight:900; line-height:1.15; color:#1f2544;">🎬 Omni-Video-Factory</div>
                  <div style="margin-top:8px; font-size:1.02em; color:#2d3363; font-weight:700; letter-spacing:0.2px;">Industrial-grade creator workflow on ZeroGPU</div>
                </div>

                <div style="display:grid; grid-template-columns:repeat(auto-fit,minmax(320px,1fr)); gap:14px;">
                  <div style="background:rgba(255,255,255,0.88); border:1px solid rgba(255,255,255,0.65); border-radius:16px; padding:18px 18px;">
                    <div style="font-size:1.06em; font-weight:800; color:#1f2544; margin-bottom:8px;">What this project is</div>
                    <div style="font-size:0.98em; line-height:1.78; color:#334155;">
                      Omni-Video-Factory is an OmniAILab experimental integration that combines <strong>OmniVideo</strong>, <strong>ComfyUI workflows</strong>, and <strong>Hugging Face ZeroGPU</strong> to deliver industrial-grade video creator tooling in a lightweight public demo.
                    </div>
                  </div>

                  <div style="background:rgba(255,255,255,0.88); border:1px solid rgba(255,255,255,0.65); border-radius:16px; padding:18px 18px;">
                    <div style="font-size:1.06em; font-weight:800; color:#1f2544; margin-bottom:8px;">About OmniVideo</div>
                    <div style="font-size:0.98em; line-height:1.78; color:#334155;">
                      OmniVideo was first released in <strong>December 2025</strong> as a multi-modal generation and editing model for video and image. It has strong compatibility with ComfyUI and is designed for high-fidelity, identity-consistent video generation.
                    </div>
                  </div>
                </div>
              </div>

              <div style="display:grid; grid-template-columns:repeat(auto-fit,minmax(300px,1fr)); gap:16px; margin-top:18px;">
                <div style="background:#f8fafc; border:1px solid #dbe3f1; border-radius:16px; padding:20px; box-shadow:0 6px 18px rgba(15,23,42,0.06);">
                  <div style="font-weight:800; color:#1e293b; font-size:1.15em; margin-bottom:8px;">🔓 Open Source & Transparency</div>
                  <div style="color:#334155; font-size:0.97em; line-height:1.76;">
                    Comfy runtime code is based on minimal modifications of
                    <a href="https://github.com/selfitcamera/ComfyUIVideo" target="_blank" style="color:#2563eb; text-decoration:none; font-weight:700;">selfitcamera/ComfyUIVideo</a>.
                    <br><br>
                    Except for OmniVideo model weights, the end-to-end pipeline code is visible.
                    <br><br>
                    Our <a href="https://github.com/selfitcamera/ComfyUIVideo/tree/main/workflows" target="_blank" style="color:#2563eb; text-decoration:none; font-weight:700;">workflows</a> are also highly compatible with LTX-Video, WanX 2.2, and WanX 2.1. You can use AI coding tools to adapt this stack to corresponding open-source models.
                  </div>
                </div>

                <div style="background:#fff7ed; border:1px solid #fed7aa; border-radius:16px; padding:20px; box-shadow:0 6px 18px rgba(249,115,22,0.12);">
                  <div style="font-weight:800; color:#9a3412; font-size:1.15em; margin-bottom:8px;">📜 Access & Usage Policy</div>
                  <div style="color:#7c2d12; font-size:0.97em; line-height:1.76;">
                    This demo does not enforce a custom generation count cap. Your available runs are determined by your Hugging Face account quota and ZeroGPU availability.
                    <br><br>
                    Content generated using this Space must comply with the official Hugging Face content policy
                    (<a href="https://huggingface.co/content-policy" target="_blank" style="color:#b45309;font-weight:700;text-decoration:none;">https://huggingface.co/content-policy</a>).
                    <br><br>
                    This demo is for research and experimentation only. Generated production content should <strong>not</strong> be used for commercial purposes.
                  </div>
                </div>
              </div>
            </div>
            """
        )
    return demo