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"(.*?)", 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 section if "" in response: response = response.split("")[-1].strip() else: import re response = re.sub(r'.*', '', 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 = """
Try our multi-feature image editing demo here
Try it now
""" with gr.Blocks(title="Omni Video Factory") as demo: gr.HTML( f"""

🎬 {getattr(APP_CONFIG, "APP_TITLE", "Omni Video Factory")}

{banner_subtitle}
""" ) 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( """
🎬 Omni-Video-Factory
Industrial-grade creator workflow on ZeroGPU
What this project is
Omni-Video-Factory is an OmniAILab experimental integration that combines OmniVideo, ComfyUI workflows, and Hugging Face ZeroGPU to deliver industrial-grade video creator tooling in a lightweight public demo.
About OmniVideo
OmniVideo was first released in December 2025 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.
🔓 Open Source & Transparency
Comfy runtime code is based on minimal modifications of selfitcamera/ComfyUIVideo.

Except for OmniVideo model weights, the end-to-end pipeline code is visible.

Our workflows 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.
📜 Access & Usage Policy
This demo does not enforce a custom generation count cap. Your available runs are determined by your Hugging Face account quota and ZeroGPU availability.

Content generated using this Space must comply with the official Hugging Face content policy (https://huggingface.co/content-policy).

This demo is for research and experimentation only. Generated production content should not be used for commercial purposes.
""" ) return demo