"""AllOutMV free-tier provider chain. One legit free account per provider, strung together: for every shot the chain tries providers in order and hops to the next the moment one is out of credits, rate-limited, or errors — so a single music video can draw shots from several free tiers and still finish. Built from each provider's published API spec (checked 2026-11-05): 1. NovAI — CogVideoX-Flash is $0 per call (free tier, ~10 calls/day, 6s text-to-video clips). POST /v1/video/generations -> poll. https://aiapi-pro.com (base overridable via NOVAI_BASE) 2. DeepInfra — Wan image-to-video, synchronous POST /v1/inference/{model} -> {video_url}. Small signup credit. Model default Wan-AI/Wan2.6-I2V (DEEPINFRA_I2V_MODEL). 3. Novita — Kling image-to-video (3–15s), async POST /v3/async/kling-v3.0-pro-i2v -> poll /v3/async/task-result. Welcome voucher on signup. 4. fal.ai — Wan2.2 image-to-video via server.do_wan_cloud ($0.15/clip, pay-as-you-go; no free tier, kept last as the reliable paid net). Image input: providers that need the portrait get a public URL when one is available (fal storage upload when a fal key exists), otherwise a base64 data URI. If every provider fails, the caller falls back to the built-in photo-zoom clip — a run never dies because a free tier ran dry. Keys are the user's own, one account per provider. The chain exists so each provider's own free allowance is used in turn — not to evade quotas: a provider that reports quota/auth failure is parked for the rest of the run and the next provider takes over. """ import base64 import json import mimetypes import os import re import subprocess import time import urllib.error import urllib.request import uuid from pathlib import Path import server NOVAI_BASE = os.environ.get("NOVAI_BASE", "https://aiapi-pro.com").rstrip("/") DEEPINFRA_BASE = os.environ.get("DEEPINFRA_BASE", "https://api.deepinfra.com").rstrip("/") DEEPINFRA_I2V_MODEL = os.environ.get("DEEPINFRA_I2V_MODEL", "Wan-AI/Wan2.6-I2V") NOVITA_BASE = os.environ.get("NOVITA_BASE", "https://api.novita.ai").rstrip("/") NOVITA_I2V = os.environ.get("NOVITA_I2V_MODEL", "kling-v3.0-pro-i2v") _exhausted = set() # providers parked for this process after quota/auth failure class ProviderError(Exception): pass class QuotaError(ProviderError): """Out of credits / bad key / rate-limited — park this provider, hop on.""" pass def reset_chain(): _exhausted.clear() def _http(url, key=None, data=None, timeout=180, auth="Bearer"): body = json.dumps(data).encode() if data is not None else None req = urllib.request.Request(url, body, method="POST" if body is not None else "GET") if key: req.add_header("Authorization", f"{auth} {key}") if body is not None: req.add_header("Content-Type", "application/json") try: with urllib.request.urlopen(req, timeout=timeout) as r: return json.loads(r.read() or b"{}") except urllib.error.HTTPError as e: detail = e.read().decode("utf-8", "ignore")[:300] if e.code in (401, 402, 403, 429): raise QuotaError(f"HTTP {e.code}: {detail}") raise ProviderError(f"HTTP {e.code}: {detail}") except Exception as e: raise ProviderError(str(e)[:300]) def _download(url, out): try: urllib.request.urlretrieve(url, out) except Exception as e: raise ProviderError(f"download failed: {str(e)[:200]}") if not Path(out).exists() or Path(out).stat().st_size < 10000: raise ProviderError("downloaded clip is empty or too small") def _data_uri(path): path = Path(path) ctype = mimetypes.guess_type(path.name)[0] or "image/jpeg" return f"data:{ctype};base64," + base64.b64encode(path.read_bytes()).decode() def normalize_clip(src, seconds, out): """Make any provider clip concat-ready and exactly the shot's length: 1280x720, 24fps, H.264/yuv420p, no audio; freeze-pad or trim to seconds.""" seconds = max(1.0, float(seconds)) vf = ("scale=1280:720:force_original_aspect_ratio=decrease," "pad=1280:720:(ow-iw)/2:(oh-ih)/2,fps=24," f"tpad=stop_mode=clone:stop_duration={seconds}") r = subprocess.run(["ffmpeg", "-y", "-i", str(src), "-vf", vf, "-t", f"{seconds}", "-an", "-c:v", "libx264", "-pix_fmt", "yuv420p", str(out)], capture_output=True, text=True) if r.returncode or not Path(out).exists(): raise ProviderError(f"normalize failed: {r.stderr[-300:]}") return Path(out) # ---------------- providers (each returns a downloaded raw clip path) ---------------- def _novai(key, image_path, image_ref, prompt, seconds, raw): # The free CogVideoX upstream has a concurrency cap: it answers HTTP 429 # "Free video model is busy — retry in 30-60 seconds". That is NOT quota # exhaustion, so retry the submit with backoff instead of parking NovAI # for the whole run (the bug that zeroed the first live video's AI shots). sub, last = None, None for attempt in range(5): try: sub = _http(f"{NOVAI_BASE}/v1/video/generations", key, {"model": "cogvideox-flash", "prompt": str(prompt)[:500]}) break except QuotaError as e: last = e if attempt < 4: time.sleep(45) if sub is None: raise last or ProviderError("NovAI submit failed") tid = sub.get("id") or sub.get("task_id") if not tid: raise ProviderError(f"NovAI returned no task id: {str(sub)[:200]}") return _novai_poll(key, tid, raw) def _novai_extract_url(u): """Find the video URL in a NovAI task payload, whatever shape it takes: plain fields, an output object, or the OpenAI-chat shape the gateway actually returns — choices[0].message.content (URL or JSON string).""" url = u.get("video_url") or u.get("url") if not url and isinstance(u.get("output"), dict): url = u["output"].get("video_url") or u["output"].get("url") vr = u.get("video_result") # the shape Zhipu actually returns (found live) if not url and isinstance(vr, list) and vr and isinstance(vr[0], dict): url = vr[0].get("url") or vr[0].get("video_url") if not url and isinstance(vr, dict): url = vr.get("url") or vr.get("video_url") if not url: try: content = u["choices"][0]["message"]["content"] except Exception: content = None if isinstance(content, str) and content.strip(): try: parsed = json.loads(content) if isinstance(parsed, dict): url = parsed.get("video_url") or parsed.get("url") except Exception: pass if not url: m = re.search(r"https?://[^\s\"'\)\]]+", content) if m: url = m.group(0) return url def _novai_poll(key, tid, raw): """Poll a submitted NovAI task until it finishes. Free-queue generations can take a long while (~18 min budget). The gateway's real shape (found live 2026-11-06): status is `task_status` ("SUCCESS"), and the video URL rides in choices[0].message.content — the first adapter read `status` only, so a SUCCEEDED task polled to a blind timeout.""" def _unwrap(st): d = st.get("data") if isinstance(d, dict): return {**st, **d} if isinstance(d, list) and d and isinstance(d[0], dict): return {**st, **d[0]} return st last_seen = "" for _ in range(170): time.sleep(6) try: st = _http(f"{NOVAI_BASE}/v1/video/generations/{tid}", key) except QuotaError: time.sleep(30) # busy while polling — wait and keep polling continue last_seen = str(st)[:1200] u = _unwrap(st) status = str(u.get("task_status") or u.get("status") or st.get("task_status") or "").lower() if status in ("succeeded", "success", "completed", "succeed", "done"): url = _novai_extract_url(u) or _novai_extract_url(st) if not url: raise ProviderError(f"NovAI finished without a video URL: {last_seen}") _download(url, raw) # writes to raw; returns None by design return Path(raw) if status in ("failed", "error", "cancelled", "canceled", "fail"): raise ProviderError(f"NovAI task failed: {last_seen}") raise ProviderError(f"NovAI task timed out (~18 min). Last poll payload: {last_seen}") def novai_resume(key, tid, raw): """Fetch the result of an already-submitted NovAI task by ID — free-queue tasks outlive the client that submitted them, so a timed-out run's video can still be recovered without spending another generation.""" return _novai_poll(key, str(tid).strip(), raw) def _deepinfra(key, image_path, image_ref, prompt, seconds, raw): payload = {"img_url": image_ref, "prompt": str(prompt)[:800]} res = _http(f"{DEEPINFRA_BASE}/v1/inference/{DEEPINFRA_I2V_MODEL}", key, payload, timeout=600) url = res.get("video_url") if not url: raise ProviderError(f"DeepInfra returned no video_url: {str(res)[:200]}") if url.startswith("/"): url = DEEPINFRA_BASE + url return _download(url, raw) def _novita(key, image_path, image_ref, prompt, seconds, raw): dur = str(int(max(3, min(15, round(seconds))))) sub = _http(f"{NOVITA_BASE}/v3/async/{NOVITA_I2V}", key, {"image": image_ref, "prompt": str(prompt)[:800], "duration": dur}) tid = sub.get("task_id") or (sub.get("data") or {}).get("task_id") if not tid: raise ProviderError(f"Novita returned no task_id: {str(sub)[:200]}") for _ in range(100): # up to ~10 min time.sleep(6) st = _http(f"{NOVITA_BASE}/v3/async/task-result?task_id={tid}", key) status = str((st.get("task") or {}).get("status", "")) if status == "TASK_STATUS_SUCCEED": vids = st.get("videos") or [] url = vids[0].get("video_url") if vids else None if not url: raise ProviderError(f"Novita finished without a video URL: {str(st)[:200]}") return _download(url, raw) if status == "TASK_STATUS_FAILED": raise ProviderError(f"Novita task failed: {str(st)[:200]}") raise ProviderError("Novita task timed out (~10 min)") def _fal(key, image_path, image_ref, prompt, seconds, raw): jid = uuid.uuid4().hex[:12] server.jobs[jid] = {"status": "running", "progress": 0, "created": time.time()} server.do_wan_cloud(jid, Path(image_path) if image_path else None, prompt, key, Path(raw)) return Path(raw) PROVIDERS = [ {"id": "novai", "name": "NovAI CogVideoX-Flash (free)", "fn": _novai}, {"id": "deepinfra", "name": "DeepInfra Wan (signup credit)", "fn": _deepinfra}, {"id": "novita", "name": "Novita Kling (welcome voucher)", "fn": _novita}, {"id": "fal", "name": "fal.ai Wan2.2 ($0.15/clip)", "fn": _fal}, ] def image_reference(image_path, keys, cache={}): """Public URL for the portrait when possible (fal storage via the fal key), else a base64 data URI. Cached per run.""" if not image_path: return None if "url" in cache: return cache["url"] ref = None fal_key = (keys.get("fal") or "").strip() if fal_key: try: ref = server.fal_upload(Path(image_path), fal_key) except Exception: ref = None if not ref: ref = _data_uri(image_path) cache["url"] = ref return ref def generate_with_chain(keys, image_path, prompt, seconds, out_path, log=lambda m: None): """Try each configured provider in order; first success wins and the clip is normalized to exactly `seconds`. Returns (path, provider_name). Raises ProviderError listing every failure when the whole chain fails.""" keys = {k: (v or "").strip() for k, v in (keys or {}).items()} ref = image_reference(image_path, keys) if image_path else None raw = Path(str(out_path) + ".raw.mp4") errors = [] for p in PROVIDERS: key = keys.get(p["id"]) if not key: continue if p["id"] in _exhausted: errors.append(f"{p['name']}: parked (quota earlier this run)") continue try: log(f"trying {p['name']}…") p["fn"](key, image_path, ref, prompt, seconds, raw) normalize_clip(raw, seconds, out_path) raw.unlink(missing_ok=True) return Path(out_path), p["name"] except QuotaError as e: _exhausted.add(p["id"]) errors.append(f"{p['name']}: {e}") log(f"{p['name']} is out — hopping to the next provider") except ProviderError as e: errors.append(f"{p['name']}: {e}") log(f"{p['name']} failed ({e}) — hopping to the next provider") raise ProviderError(" | ".join(errors) if errors else "no provider keys supplied")