Spaces:

tddandroid
/

alloutmv

Sleeping

alloutmv / server.py

Fix ZeroGPU whisper CUDA fallback + GPU duration + error surfacing

c4c11ab verified 9 days ago

21.2 kB

"""AllOutMV backend v1.0 — finished, hardened.
Works out of the box:
  /api/health      GPU / ffmpeg / model status
  /api/transcribe  faster-whisper (word-timed segments)
  /api/plan        Director shot-plan from segments/lyrics (rule-based, or LLM via unified gateway if set)
  /api/render      ffmpeg join clips + audio, burn timed lyrics
Needs your model install (returns a clear 'not installed' error, never fake output):
  /api/lipsync     set LIPSYNC_CMD  e.g. Wav2Lip / MuseTalk / SadTalker
  /api/generate    set WAN_CMD      e.g. Wan2.2 ti2v-5B (from your screenshots: Apache-2.0,
                                    720p@24fps on a consumer 4090, MoE cinematic model)
Unified AI gateway (from your '251 AI Providers / one endpoint' screenshot):
  Set OPENROUTER_API_KEY (or LLM_BASE_URL + LLM_API_KEY) and /api/plan will use any
  provider/model through one OpenAI-compatible endpoint for lyric/director intelligence.
  Without a key it falls back to a deterministic local planner — no lock-in, no cost.
Security — adapted from the pentest/OSINT skill you supplied (defensive side):
  * ALLOUT_API_KEY: if set, every /api/* call (except /api/health) needs header X-API-Key
  * ALLOUT_ORIGINS: CORS allow-list, never leave '*' in production
  * Upload allow-list + size cap (ALLOUT_MAX_MB, default 500)
  * Filenames are server-generated; user names never touch the filesystem path
  * One GPU job at a time (queue), job errors truncated, no shell=True anywhere
  * Run a secrets scan before publishing this repo: trufflehog filesystem . / gitleaks detect
"""
import os, re, uuid, json, shlex, subprocess, threading, time, traceback
from pathlib import Path
from concurrent.futures import ThreadPoolExecutor
from fastapi import FastAPI, UploadFile, File, Form, Header, HTTPException, Request
from fastapi.middleware.cors import CORSMiddleware
from fastapi.staticfiles import StaticFiles
from fastapi.responses import JSONResponse, FileResponse

ROOT = Path(os.environ.get("ALLOUT_DATA", "./data")); FILES = ROOT / "files"; WORK = ROOT / "work"
FILES.mkdir(parents=True, exist_ok=True); WORK.mkdir(parents=True, exist_ok=True)
ORIGINS = [o.strip() for o in os.environ.get("ALLOUT_ORIGINS", "*").split(",") if o.strip()]
API_KEY = os.environ.get("ALLOUT_API_KEY", "")
MAX_BYTES = int(os.environ.get("ALLOUT_MAX_MB", "500")) * 1024 * 1024
LIPSYNC_CMD = os.environ.get("LIPSYNC_CMD", "")
WAN_CMD = os.environ.get("WAN_CMD", "")
# Auto-detect a local install made by install-wan.sh: models/Wan2.2 (+ its own venv + weights).
# Verified against Wan2.2's own README (single GPU, >=24GB VRAM): generate.py --task ti2v-5B
# --size 1280*704 --ckpt_dir ./Wan2.2-TI2V-5B --offload_model True --convert_model_dtype --t5_cpu
WAN_DIR = Path(__file__).parent / "models" / "Wan2.2"
WAN_CKPT = WAN_DIR / "Wan2.2-TI2V-5B"
WAN_PY = WAN_DIR / ".venv" / ("Scripts/python.exe" if os.name == "nt" else "bin/python")
def wan_available():
    return bool(WAN_CMD) or ((WAN_DIR / "generate.py").exists() and WAN_CKPT.exists() and WAN_PY.exists())
LLM_KEY = os.environ.get("OPENROUTER_API_KEY") or os.environ.get("LLM_API_KEY", "")
# Cloud video provider (fal.ai) — hosted Wan2.2, no GPU needed. Key from env FAL_KEY,
# or per-request from the Studio (X-Cloud-Key header) so a phone user never edits files.
# Verified 2026-11-05 against fal's published spec for fal-ai/wan/v2.2-5b/image-to-video:
# queue submit -> poll status_url -> GET response_url -> {video:{url}}; $0.15 per 720p clip.
FAL_KEY = os.environ.get("FAL_KEY") or os.environ.get("FAL_API_KEY", "")
FAL_QUEUE = os.environ.get("FAL_QUEUE_URL", "https://queue.fal.run").rstrip("/")
FAL_REST = os.environ.get("FAL_REST_URL", "https://rest.alpha.fal.ai").rstrip("/")
FAL_I2V = os.environ.get("FAL_I2V_MODEL", "fal-ai/wan/v2.2-5b/image-to-video")
FAL_T2V = os.environ.get("FAL_T2V_MODEL", "fal-ai/wan/v2.2-a14b/text-to-video")
LLM_BASE = os.environ.get("LLM_BASE_URL", "https://openrouter.ai/api/v1")
LLM_MODEL = os.environ.get("LLM_MODEL", "meta-llama/llama-3.3-70b-instruct:free")

ALLOWED = {".mp3", ".wav", ".m4a", ".flac", ".ogg", ".mp4", ".mov", ".webm", ".jpg", ".jpeg", ".png", ".webp"}

app = FastAPI(title="AllOutMV", version="1.0.0")
app.add_middleware(CORSMiddleware, allow_origins=ORIGINS, allow_methods=["*"], allow_headers=["*"])

@app.middleware("http")
async def guard(request: Request, call_next):
    if API_KEY and request.url.path.startswith("/api/") and request.url.path != "/api/health":
        if request.headers.get("x-api-key") != API_KEY:
            return JSONResponse({"detail": "Invalid or missing X-API-Key"}, status_code=401)
    resp = await call_next(request)
    resp.headers.update({"X-Content-Type-Options": "nosniff", "Referrer-Policy": "no-referrer"})
    return resp

app.mount("/files", StaticFiles(directory=FILES), name="files")
FRONTEND = Path(__file__).parent / "frontend"
if FRONTEND.exists():
    app.mount("/studio", StaticFiles(directory=FRONTEND, html=True), name="studio")

@app.get("/")
def home():
    idx = FRONTEND / "index.html"
    return FileResponse(idx) if idx.exists() else {"ok": True, "app": "AllOutMV", "docs": "/docs"}
pool = ThreadPoolExecutor(max_workers=1)
jobs, lock, _models = {}, threading.Lock(), {}

def save(up: UploadFile, jid: str, name: str):
    if up is None: return None
    ext = Path(up.filename or "").suffix[:8].lower()
    if ext and ext not in ALLOWED: raise HTTPException(400, f"File type {ext} not allowed")
    p = WORK / f"{jid}_{name}{ext}"; n = 0
    with open(p, "wb") as f:
        while chunk := up.file.read(1 << 20):
            n += len(chunk)
            if n > MAX_BYTES: f.close(); p.unlink(missing_ok=True); raise HTTPException(413, "File too large")
            f.write(chunk)
    return p

def submit(fn, *a):
    jid = uuid.uuid4().hex[:12]; jobs[jid] = {"status": "queued", "progress": 0, "created": time.time()}
    def run():
        jobs[jid]["status"] = "running"
        try:
            res = fn(jid, *a) or {}; jobs[jid].update(res); jobs[jid].update(status="completed", progress=100)
        except Exception as e:
            traceback.print_exc(); jobs[jid].update(status="failed", error=str(e)[:600])
    pool.submit(run); return jid

def prog(jid, p): jobs[jid]["progress"] = int(p)

def run_cmd(tpl, **kw):
    cmd = [part.format(**{k: str(v) for k, v in kw.items()}) for part in shlex.split(tpl)]
    r = subprocess.run(cmd, capture_output=True, text=True)  # never shell=True
    if r.returncode: raise RuntimeError((r.stderr or r.stdout)[-500:])

@app.get("/api/health")
def health():
    gpu = "unknown"
    try:
        import torch; gpu = torch.cuda.get_device_name(0) if torch.cuda.is_available() else "cpu only"
    except Exception: pass
    return {"ok": True, "version": "1.0.0", "gpu": gpu, "lipsync": bool(LIPSYNC_CMD), "wan": wan_available(),
            "photoclip": True, "cloud_wan": bool(FAL_KEY), "llm_director": bool(LLM_KEY), "llm_model": LLM_MODEL if LLM_KEY else None,
            "auth_required": bool(API_KEY),
            "ffmpeg": subprocess.run(["ffmpeg", "-version"], capture_output=True).returncode == 0}

@app.get("/api/providers")
def providers():
    """One-endpoint provider view (mirrors your 251-providers screenshot, honestly scoped)."""
    return {"gateway": LLM_BASE if LLM_KEY else None, "director_model": LLM_MODEL if LLM_KEY else "local-rules",
            "video": {"model": "Wan2.2 ti2v-5B / t2v", "installed": wan_available(), "note": "Apache-2.0, 720p@24fps, runs on 4090-class GPU"},
            "lipsync": {"installed": bool(LIPSYNC_CMD), "options": ["Wav2Lip", "MuseTalk", "SadTalker"]},
            "cloud": {"provider": "fal.ai", "configured_on_server": bool(FAL_KEY), "i2v_model": FAL_I2V, "price_note": "$0.15 per 720p clip (fal list price, checked 2026-11-05); a per-request key from the Studio also works"},
            "transcription": {"model": "faster-whisper", "installed": True}}

@app.get("/api/jobs/{jid}")
def job(jid: str): return jobs.get(jid) or {"status": "failed", "error": "unknown job"}

# ---------- transcription ----------
MODELS = {"faster-whisper": None, "whisper-large-v3": "large-v3", "tiny": "tiny", "base": "base", "small": "small"}
def _load_whisper(name):
    """Load a Whisper model, preferring CUDA but FALLING BACK to CPU when CUDA
    construction fails. Detection alone can lie: on HF ZeroGPU the driver
    reports a GPU that is only attached inside @spaces.GPU calls, so
    torch.cuda.is_available() is True while construction dies with 'no
    CUDA-capable device is detected'. Construction is the only real test.
    ALLOUT_WHISPER_DEVICE=cpu|cuda forces a device."""
    from faster_whisper import WhisperModel
    forced = os.environ.get("ALLOUT_WHISPER_DEVICE", "").lower()
    if forced == "cpu":
        tries = [("cpu", "int8")]
    elif forced == "cuda":
        tries = [("cuda", "float16")]
    else:
        try:
            import torch; cuda = torch.cuda.is_available()
        except Exception: cuda = False
        tries = [("cuda", "float16"), ("cpu", "int8")] if cuda else [("cpu", "int8")]
    last = None
    for dev, ct in tries:
        for attempt in range(2):  # first run downloads weights — retry once on a network drop
            try:
                return WhisperModel(name, device=dev, compute_type=ct)
            except Exception as e:
                last = e
    raise last

def do_transcribe(jid, audio, model):
    try:
        import torch; cuda = torch.cuda.is_available()
    except Exception: cuda = False
    name = MODELS.get(model) or ("large-v3" if cuda else "small")
    if name not in _models:
        _models[name] = _load_whisper(name)
    segs, info = _models[name].transcribe(str(audio), vad_filter=True)
    out, total = [], max(info.duration, 1)
    for s in segs:
        out.append({"start": round(s.start, 2), "end": round(s.end, 2), "text": s.text.strip()})
        prog(jid, 100 * s.end / total)
    return {"segments": out, "language": info.language, "duration": total}

@app.post("/api/transcribe")
def transcribe(audio: UploadFile = File(...), model: str = Form("faster-whisper")):
    p = save(audio, uuid.uuid4().hex[:8], "audio")
    return {"job_id": submit(lambda jid: do_transcribe(jid, p, model))}

# ---------- director plan (NEW in v1.0 — finishes the Studio's shot-plan in the backend) ----------
MOODS = {"Cinematic": "wide atmospheric frames, dramatic light, deliberate camera movement",
         "Dark": "deep shadows, high contrast", "Energetic": "dynamic framing, vivid accents, cuts on the beat",
         "Romantic": "soft light, intimate close-ups", "Dreamlike": "surreal imagery, haze, floating motifs",
         "Hopeful": "warm light, expansive frames", "Aggressive": "hard contrast, bold angles, punchy cuts"}
def local_plan(segments, mood, style, title):
    shots, cur, start = [], [], 0.0
    for s in segments or []:
        if s["start"] - start >= 15 and cur:
            shots.append(cur); cur, start = [], s["start"]
        cur.append(s)
    if cur: shots.append(cur)
    if not shots: shots = [[{"start": i * 15, "end": i * 15 + 15, "text": ""}] for i in range(4)]
    return [{"index": i + 1, "start": round(g[0]["start"], 2), "end": round(g[-1]["end"], 2),
             "caption": " ".join(x["text"] for x in g).strip(),
             "prompt": f"{style}. {MOODS.get(mood, MOODS['Cinematic'])}. Music video for '{title}'. "
                       f"Performer sings: \"{' '.join(x['text'] for x in g).strip()[:140]}\" "
                       f"Keep performer identity, wardrobe and colour grade consistent. Cinematic-level aesthetics: "
                       f"controlled lighting, composition and colour tone (Wan2.2). Complex natural motion."}
            for i, g in enumerate(shots)]

def do_plan(jid, segments, mood, style, title):
    plan = local_plan(segments, mood, style, title)
    if LLM_KEY:  # optional polish via unified gateway — failure never blocks a plan
        try:
            import urllib.request
            body = json.dumps({"model": LLM_MODEL, "messages": [
                {"role": "user", "content": f"Improve these Wan2.2 video prompts, return JSON list of strings only: {[p['prompt'] for p in plan]}"}]}).encode()
            req = urllib.request.Request(f"{LLM_BASE}/chat/completions", body, {"Authorization": f"Bearer {LLM_KEY}", "Content-Type": "application/json"})
            txt = json.loads(urllib.request.urlopen(req, timeout=60).read())["choices"][0]["message"]["content"]
            prompts = json.loads(txt[txt.find("["):txt.rfind("]") + 1])
            for p, t in zip(plan, prompts): p["prompt"] = str(t)
        except Exception: pass
    return {"shots": plan}

@app.post("/api/plan")
def plan(segments: str = Form("[]"), mood: str = Form("Cinematic"), style: str = Form("Modern cinematic"), title: str = Form("My Music Video")):
    return {"job_id": submit(lambda jid: do_plan(jid, json.loads(segments or "[]"), mood, style, title))}

# ---------- lip-sync ----------
def do_lipsync(jid, image, audio):
    if not LIPSYNC_CMD: raise RuntimeError("Lip-sync is not installed on this server. Set LIPSYNC_CMD (see README).")
    out = FILES / f"lip_{jid}.mp4"; run_cmd(LIPSYNC_CMD, image=image, audio=audio, out=out)
    return {"video_url": f"/files/{out.name}"}

@app.post("/api/lipsync")
def lipsync(audio: UploadFile = File(...), image: UploadFile = File(...), model: str = Form(""), lyrics: str = Form(""), instructions: str = Form("")):
    h = uuid.uuid4().hex[:8]; a, i = save(audio, h, "audio"), save(image, h, "image")
    return {"job_id": submit(lambda jid: do_lipsync(jid, i, a))}

# ---------- Wan2.2 clip ----------
def _http_json(url, key=None, data=None, method=None, timeout=120):
    import urllib.request
    body = json.dumps(data).encode() if data is not None else None
    req = urllib.request.Request(url, body, method=method or ("POST" if body else "GET"))
    if key: req.add_header("Authorization", f"Key {key}")
    if body: req.add_header("Content-Type", "application/json")
    with urllib.request.urlopen(req, timeout=timeout) as r:
        return json.loads(r.read() or b"{}")

def fal_upload(image, key):
    """Two-step fal storage upload (initiate -> PUT) -> public file_url for image_url."""
    import urllib.request, mimetypes
    image = Path(image)
    ctype = mimetypes.guess_type(image.name)[0] or "image/png"
    init = _http_json(f"{FAL_REST}/storage/upload/initiate", key, {"file_name": image.name, "content_type": ctype})
    req = urllib.request.Request(init["upload_url"], image.read_bytes(), method="PUT")
    req.add_header("Content-Type", ctype)
    urllib.request.urlopen(req, timeout=120).read()
    return init["file_url"]

def do_wan_cloud(jid, image, prompt, key, out):
    payload = {"prompt": str(prompt), "num_frames": 121, "frames_per_second": 24, "resolution": "720p"}
    model = FAL_T2V
    if image:
        payload["image_url"] = fal_upload(image, key); model = FAL_I2V
    sub = _http_json(f"{FAL_QUEUE}/{model}", key, payload)
    status_url, response_url = sub.get("status_url"), sub.get("response_url")
    if not status_url:
        result = sub
    else:
        result = None
        for i in range(120):  # up to ~10 min
            time.sleep(5); prog(jid, 30 + min(60, i))
            st = _http_json(status_url, key)
            if st.get("status") == "COMPLETED":
                result = _http_json(response_url, key); break
            if st.get("status") in ("FAILED", "ERROR"):
                raise RuntimeError(f"Cloud generation failed on fal: {str(st)[:300]}")
        if result is None: raise RuntimeError("Cloud generation timed out after ~10 minutes on fal.")
    url = (result.get("video") or {}).get("url")
    if not url: raise RuntimeError(f"fal returned no video URL: {str(result)[:300]}")
    import urllib.request
    urllib.request.urlretrieve(url, out)

def do_wan(jid, image, prompt, cloud_key=""):
    out = (FILES / f"clip_{jid}.mp4").resolve()  # absolute: the auto-detected run executes inside WAN_DIR
    if WAN_CMD:
        run_cmd(WAN_CMD, image=image or "", prompt=prompt, out=out)
    elif wan_available():
        cmd = [str(WAN_PY), "generate.py", "--task", "ti2v-5B", "--size", "1280*704",
               "--ckpt_dir", "./Wan2.2-TI2V-5B", "--offload_model", "True",
               "--convert_model_dtype", "--t5_cpu", "--prompt", str(prompt), "--save_file", str(out)]
        if image: cmd += ["--image", str(Path(image).resolve())]
        r = subprocess.run(cmd, cwd=WAN_DIR, capture_output=True, text=True)
        if r.returncode: raise RuntimeError((r.stderr or r.stdout)[-500:])
    elif cloud_key or FAL_KEY:
        do_wan_cloud(jid, image, prompt, cloud_key or FAL_KEY, out)
    else:
        raise RuntimeError("No video generator available: run ./install-wan.sh on a GPU machine, or add a fal.ai cloud key in the Studio (no PC needed).")
    if not out.exists(): raise RuntimeError("Generation finished but produced no file.")
    return {"video_url": f"/files/{out.name}"}

@app.post("/api/generate")
def generate(prompt: str = Form(...), image: UploadFile = File(None), audio: UploadFile = File(None),
             model: str = Form(""), scene: str = Form(""), has_image: str = Form(""),
             x_cloud_key: str = Header("")):
    i = save(image, uuid.uuid4().hex[:8], "image") if image else None
    return {"job_id": submit(lambda jid: do_wan(jid, i, prompt, x_cloud_key))}

# ---------- photo clip (CPU fallback — portrait -> slow-zoom clip, no GPU needed) ----------
def do_photoclip(jid, image, duration, variant=0):
    dur = max(1.0, min(float(duration or 5), 40.0))
    out = FILES / f"photo_{jid}.mp4"; frames = int(dur * 24)
    # Camera-move variants so consecutive fallback shots don't look identical:
    # 0 slow zoom-in, 1 slow zoom-out, 2 pan right, 3 pan left.
    moves = [
        ("min(zoom+0.0008,1.25)", "iw/2-(iw/zoom/2)", "ih/2-(ih/zoom/2)"),
        ("if(eq(on,1),1.25,max(zoom-0.0008,1.0))", "iw/2-(iw/zoom/2)", "ih/2-(ih/zoom/2)"),
        ("1.15", f"(iw-iw/zoom)*on/{frames}", "ih/2-(ih/zoom/2)"),
        ("1.15", f"(iw-iw/zoom)*(1-on/{frames})", "ih/2-(ih/zoom/2)"),
    ]
    z, x, y = moves[int(variant) % 4]
    vf = (f"scale=2560:1440:force_original_aspect_ratio=increase,crop=2560:1440,"
          f"zoompan=z='{z}':d={frames}:x='{x}':y='{y}':s=1280x720:fps=24")
    r = subprocess.run(["ffmpeg", "-y", "-loop", "1", "-i", str(image), "-vf", vf, "-t", str(dur),
                        "-c:v", "libx264", "-pix_fmt", "yuv420p", str(out)], capture_output=True, text=True)
    if r.returncode: raise RuntimeError(r.stderr[-500:])
    return {"video_url": f"/files/{out.name}"}

@app.post("/api/photoclip")
def photoclip(image: UploadFile = File(...), duration: str = Form("5")):
    i = save(image, uuid.uuid4().hex[:8], "image")
    return {"job_id": submit(lambda jid: do_photoclip(jid, i, duration))}

# ---------- final render ----------
def lyrics_to_srt(text, path):
    rows = []
    for l in (text or "").splitlines():
        m = re.match(r"^(\d+):(\d\d)\s+(.*)$", l.strip())
        if m: rows.append((int(m[1]) * 60 + int(m[2]), m[3]))
    ts = lambda s: f"{int(s//3600):02}:{int(s//60%60):02}:{int(s%60):02},000"
    with open(path, "w") as f:
        for i, (t, x) in enumerate(rows):
            end = rows[i + 1][0] if i + 1 < len(rows) else t + 4
            f.write(f"{i+1}\n{ts(t)} --> {ts(end)}\n{x}\n\n")
    return bool(rows)

def segments_to_srt(segments, path):
    ts = lambda s: f"{int(s//3600):02}:{int(s//60%60):02}:{int(s%60):02},{int((s%1)*1000):03}"
    with open(path, "w") as f:
        for i, s in enumerate(segments):
            f.write(f"{i+1}\n{ts(s['start'])} --> {ts(s['end'])}\n{s['text']}\n\n")
    return bool(segments)

def do_render(jid, audio, clips, lyrics, segments):
    files = [FILES / Path(str(c)).name for c in clips if c]  # basename only — no path traversal
    files = [f for f in files if f.exists()]
    if not files: raise RuntimeError("No generated clips found on the server to join.")
    lst = WORK / f"{jid}_list.txt"; lst.write_text("".join(f"file '{f.resolve()}'\n" for f in files))
    out = FILES / f"final_{jid}.mp4"; srt = WORK / f"{jid}.srt"
    has_subs = segments_to_srt(segments, srt) if segments else (lyrics_to_srt(lyrics, srt) if lyrics else False)
    cmd = ["ffmpeg", "-y", "-f", "concat", "-safe", "0", "-i", str(lst)]
    if audio: cmd += ["-i", str(audio), "-map", "0:v", "-map", "1:a", "-shortest"]
    if has_subs: cmd += ["-vf", f"subtitles={srt.as_posix()}"]
    cmd += ["-c:v", "libx264", "-pix_fmt", "yuv420p", "-c:a", "aac", str(out)]
    prog(jid, 20); r = subprocess.run(cmd, capture_output=True, text=True)
    if r.returncode: raise RuntimeError(r.stderr[-500:])
    return {"video_url": f"/files/{out.name}"}

@app.post("/api/render")
def render(audio: UploadFile = File(None), clips: str = Form("[]"), lipsync: str = Form(""),
           lyrics: str = Form(""), title: str = Form(""), segments: str = Form("[]")):
    a = save(audio, uuid.uuid4().hex[:8], "audio") if audio else None
    return {"job_id": submit(lambda jid: do_render(jid, a, json.loads(clips or "[]"), lyrics, json.loads(segments or "[]")))}

	"""AllOutMV backend v1.0 — finished, hardened.

	Works out of the box:
	/api/health GPU / ffmpeg / model status
	/api/transcribe faster-whisper (word-timed segments)
	/api/plan Director shot-plan from segments/lyrics (rule-based, or LLM via unified gateway if set)
	/api/render ffmpeg join clips + audio, burn timed lyrics

	Needs your model install (returns a clear 'not installed' error, never fake output):
	/api/lipsync set LIPSYNC_CMD e.g. Wav2Lip / MuseTalk / SadTalker
	/api/generate set WAN_CMD e.g. Wan2.2 ti2v-5B (from your screenshots: Apache-2.0,
	720p@24fps on a consumer 4090, MoE cinematic model)

	Unified AI gateway (from your '251 AI Providers / one endpoint' screenshot):
	Set OPENROUTER_API_KEY (or LLM_BASE_URL + LLM_API_KEY) and /api/plan will use any
	provider/model through one OpenAI-compatible endpoint for lyric/director intelligence.
	Without a key it falls back to a deterministic local planner — no lock-in, no cost.

	Security — adapted from the pentest/OSINT skill you supplied (defensive side):
	* ALLOUT_API_KEY: if set, every /api/* call (except /api/health) needs header X-API-Key
	* ALLOUT_ORIGINS: CORS allow-list, never leave '*' in production
	* Upload allow-list + size cap (ALLOUT_MAX_MB, default 500)
	* Filenames are server-generated; user names never touch the filesystem path
	* One GPU job at a time (queue), job errors truncated, no shell=True anywhere
	* Run a secrets scan before publishing this repo: trufflehog filesystem . / gitleaks detect
	"""
	import os, re, uuid, json, shlex, subprocess, threading, time, traceback
	from pathlib import Path
	from concurrent.futures import ThreadPoolExecutor
	from fastapi import FastAPI, UploadFile, File, Form, Header, HTTPException, Request
	from fastapi.middleware.cors import CORSMiddleware
	from fastapi.staticfiles import StaticFiles
	from fastapi.responses import JSONResponse, FileResponse

	ROOT = Path(os.environ.get("ALLOUT_DATA", "./data")); FILES = ROOT / "files"; WORK = ROOT / "work"
	FILES.mkdir(parents=True, exist_ok=True); WORK.mkdir(parents=True, exist_ok=True)
	ORIGINS = [o.strip() for o in os.environ.get("ALLOUT_ORIGINS", "*").split(",") if o.strip()]
	API_KEY = os.environ.get("ALLOUT_API_KEY", "")
	MAX_BYTES = int(os.environ.get("ALLOUT_MAX_MB", "500")) * 1024 * 1024
	LIPSYNC_CMD = os.environ.get("LIPSYNC_CMD", "")
	WAN_CMD = os.environ.get("WAN_CMD", "")
	# Auto-detect a local install made by install-wan.sh: models/Wan2.2 (+ its own venv + weights).
	# Verified against Wan2.2's own README (single GPU, >=24GB VRAM): generate.py --task ti2v-5B
	# --size 1280*704 --ckpt_dir ./Wan2.2-TI2V-5B --offload_model True --convert_model_dtype --t5_cpu
	WAN_DIR = Path(__file__).parent / "models" / "Wan2.2"
	WAN_CKPT = WAN_DIR / "Wan2.2-TI2V-5B"
	WAN_PY = WAN_DIR / ".venv" / ("Scripts/python.exe" if os.name == "nt" else "bin/python")
	def wan_available():
	return bool(WAN_CMD) or ((WAN_DIR / "generate.py").exists() and WAN_CKPT.exists() and WAN_PY.exists())
	LLM_KEY = os.environ.get("OPENROUTER_API_KEY") or os.environ.get("LLM_API_KEY", "")
	# Cloud video provider (fal.ai) — hosted Wan2.2, no GPU needed. Key from env FAL_KEY,
	# or per-request from the Studio (X-Cloud-Key header) so a phone user never edits files.
	# Verified 2026-11-05 against fal's published spec for fal-ai/wan/v2.2-5b/image-to-video:
	# queue submit -> poll status_url -> GET response_url -> {video:{url}}; $0.15 per 720p clip.
	FAL_KEY = os.environ.get("FAL_KEY") or os.environ.get("FAL_API_KEY", "")
	FAL_QUEUE = os.environ.get("FAL_QUEUE_URL", "https://queue.fal.run").rstrip("/")
	FAL_REST = os.environ.get("FAL_REST_URL", "https://rest.alpha.fal.ai").rstrip("/")
	FAL_I2V = os.environ.get("FAL_I2V_MODEL", "fal-ai/wan/v2.2-5b/image-to-video")
	FAL_T2V = os.environ.get("FAL_T2V_MODEL", "fal-ai/wan/v2.2-a14b/text-to-video")
	LLM_BASE = os.environ.get("LLM_BASE_URL", "https://openrouter.ai/api/v1")
	LLM_MODEL = os.environ.get("LLM_MODEL", "meta-llama/llama-3.3-70b-instruct:free")

	ALLOWED = {".mp3", ".wav", ".m4a", ".flac", ".ogg", ".mp4", ".mov", ".webm", ".jpg", ".jpeg", ".png", ".webp"}

	app = FastAPI(title="AllOutMV", version="1.0.0")
	app.add_middleware(CORSMiddleware, allow_origins=ORIGINS, allow_methods=[""], allow_headers=[""])

	@app.middleware("http")
	async def guard(request: Request, call_next):
	if API_KEY and request.url.path.startswith("/api/") and request.url.path != "/api/health":
	if request.headers.get("x-api-key") != API_KEY:
	return JSONResponse({"detail": "Invalid or missing X-API-Key"}, status_code=401)
	resp = await call_next(request)
	resp.headers.update({"X-Content-Type-Options": "nosniff", "Referrer-Policy": "no-referrer"})
	return resp

	app.mount("/files", StaticFiles(directory=FILES), name="files")
	FRONTEND = Path(__file__).parent / "frontend"
	if FRONTEND.exists():
	app.mount("/studio", StaticFiles(directory=FRONTEND, html=True), name="studio")

	@app.get("/")
	def home():
	idx = FRONTEND / "index.html"
	return FileResponse(idx) if idx.exists() else {"ok": True, "app": "AllOutMV", "docs": "/docs"}
	pool = ThreadPoolExecutor(max_workers=1)
	jobs, lock, _models = {}, threading.Lock(), {}

	def save(up: UploadFile, jid: str, name: str):
	if up is None: return None
	ext = Path(up.filename or "").suffix[:8].lower()
	if ext and ext not in ALLOWED: raise HTTPException(400, f"File type {ext} not allowed")
	p = WORK / f"{jid}_{name}{ext}"; n = 0
	with open(p, "wb") as f:
	while chunk := up.file.read(1 << 20):
	n += len(chunk)
	if n > MAX_BYTES: f.close(); p.unlink(missing_ok=True); raise HTTPException(413, "File too large")
	f.write(chunk)
	return p

	def submit(fn, *a):
	jid = uuid.uuid4().hex[:12]; jobs[jid] = {"status": "queued", "progress": 0, "created": time.time()}
	def run():
	jobs[jid]["status"] = "running"
	try:
	res = fn(jid, *a) or {}; jobs[jid].update(res); jobs[jid].update(status="completed", progress=100)
	except Exception as e:
	traceback.print_exc(); jobs[jid].update(status="failed", error=str(e)[:600])
	pool.submit(run); return jid

	def prog(jid, p): jobs[jid]["progress"] = int(p)

	def run_cmd(tpl, **kw):
	cmd = [part.format(**{k: str(v) for k, v in kw.items()}) for part in shlex.split(tpl)]
	r = subprocess.run(cmd, capture_output=True, text=True) # never shell=True
	if r.returncode: raise RuntimeError((r.stderr or r.stdout)[-500:])

	@app.get("/api/health")
	def health():
	gpu = "unknown"
	try:
	import torch; gpu = torch.cuda.get_device_name(0) if torch.cuda.is_available() else "cpu only"
	except Exception: pass
	return {"ok": True, "version": "1.0.0", "gpu": gpu, "lipsync": bool(LIPSYNC_CMD), "wan": wan_available(),
	"photoclip": True, "cloud_wan": bool(FAL_KEY), "llm_director": bool(LLM_KEY), "llm_model": LLM_MODEL if LLM_KEY else None,
	"auth_required": bool(API_KEY),
	"ffmpeg": subprocess.run(["ffmpeg", "-version"], capture_output=True).returncode == 0}

	@app.get("/api/providers")
	def providers():
	"""One-endpoint provider view (mirrors your 251-providers screenshot, honestly scoped)."""
	return {"gateway": LLM_BASE if LLM_KEY else None, "director_model": LLM_MODEL if LLM_KEY else "local-rules",
	"video": {"model": "Wan2.2 ti2v-5B / t2v", "installed": wan_available(), "note": "Apache-2.0, 720p@24fps, runs on 4090-class GPU"},
	"lipsync": {"installed": bool(LIPSYNC_CMD), "options": ["Wav2Lip", "MuseTalk", "SadTalker"]},
	"cloud": {"provider": "fal.ai", "configured_on_server": bool(FAL_KEY), "i2v_model": FAL_I2V, "price_note": "$0.15 per 720p clip (fal list price, checked 2026-11-05); a per-request key from the Studio also works"},
	"transcription": {"model": "faster-whisper", "installed": True}}

	@app.get("/api/jobs/{jid}")
	def job(jid: str): return jobs.get(jid) or {"status": "failed", "error": "unknown job"}

	# ---------- transcription ----------
	MODELS = {"faster-whisper": None, "whisper-large-v3": "large-v3", "tiny": "tiny", "base": "base", "small": "small"}
	def _load_whisper(name):
	"""Load a Whisper model, preferring CUDA but FALLING BACK to CPU when CUDA
	construction fails. Detection alone can lie: on HF ZeroGPU the driver
	reports a GPU that is only attached inside @spaces.GPU calls, so
	torch.cuda.is_available() is True while construction dies with 'no
	CUDA-capable device is detected'. Construction is the only real test.
	ALLOUT_WHISPER_DEVICE=cpu\|cuda forces a device."""
	from faster_whisper import WhisperModel
	forced = os.environ.get("ALLOUT_WHISPER_DEVICE", "").lower()
	if forced == "cpu":
	tries = [("cpu", "int8")]
	elif forced == "cuda":
	tries = [("cuda", "float16")]
	else:
	try:
	import torch; cuda = torch.cuda.is_available()
	except Exception: cuda = False
	tries = [("cuda", "float16"), ("cpu", "int8")] if cuda else [("cpu", "int8")]
	last = None
	for dev, ct in tries:
	for attempt in range(2): # first run downloads weights — retry once on a network drop
	try:
	return WhisperModel(name, device=dev, compute_type=ct)
	except Exception as e:
	last = e
	raise last

	def do_transcribe(jid, audio, model):
	try:
	import torch; cuda = torch.cuda.is_available()
	except Exception: cuda = False
	name = MODELS.get(model) or ("large-v3" if cuda else "small")
	if name not in _models:
	_models[name] = _load_whisper(name)
	segs, info = _models[name].transcribe(str(audio), vad_filter=True)
	out, total = [], max(info.duration, 1)
	for s in segs:
	out.append({"start": round(s.start, 2), "end": round(s.end, 2), "text": s.text.strip()})
	prog(jid, 100 * s.end / total)
	return {"segments": out, "language": info.language, "duration": total}

	@app.post("/api/transcribe")
	def transcribe(audio: UploadFile = File(...), model: str = Form("faster-whisper")):
	p = save(audio, uuid.uuid4().hex[:8], "audio")
	return {"job_id": submit(lambda jid: do_transcribe(jid, p, model))}

	# ---------- director plan (NEW in v1.0 — finishes the Studio's shot-plan in the backend) ----------
	MOODS = {"Cinematic": "wide atmospheric frames, dramatic light, deliberate camera movement",
	"Dark": "deep shadows, high contrast", "Energetic": "dynamic framing, vivid accents, cuts on the beat",
	"Romantic": "soft light, intimate close-ups", "Dreamlike": "surreal imagery, haze, floating motifs",
	"Hopeful": "warm light, expansive frames", "Aggressive": "hard contrast, bold angles, punchy cuts"}
	def local_plan(segments, mood, style, title):
	shots, cur, start = [], [], 0.0
	for s in segments or []:
	if s["start"] - start >= 15 and cur:
	shots.append(cur); cur, start = [], s["start"]
	cur.append(s)
	if cur: shots.append(cur)
	if not shots: shots = [[{"start": i * 15, "end": i * 15 + 15, "text": ""}] for i in range(4)]
	return [{"index": i + 1, "start": round(g[0]["start"], 2), "end": round(g[-1]["end"], 2),
	"caption": " ".join(x["text"] for x in g).strip(),
	"prompt": f"{style}. {MOODS.get(mood, MOODS['Cinematic'])}. Music video for '{title}'. "
	f"Performer sings: \"{' '.join(x['text'] for x in g).strip()[:140]}\" "
	f"Keep performer identity, wardrobe and colour grade consistent. Cinematic-level aesthetics: "
	f"controlled lighting, composition and colour tone (Wan2.2). Complex natural motion."}
	for i, g in enumerate(shots)]

	def do_plan(jid, segments, mood, style, title):
	plan = local_plan(segments, mood, style, title)
	if LLM_KEY: # optional polish via unified gateway — failure never blocks a plan
	try:
	import urllib.request
	body = json.dumps({"model": LLM_MODEL, "messages": [
	{"role": "user", "content": f"Improve these Wan2.2 video prompts, return JSON list of strings only: {[p['prompt'] for p in plan]}"}]}).encode()
	req = urllib.request.Request(f"{LLM_BASE}/chat/completions", body, {"Authorization": f"Bearer {LLM_KEY}", "Content-Type": "application/json"})
	txt = json.loads(urllib.request.urlopen(req, timeout=60).read())["choices"][0]["message"]["content"]
	prompts = json.loads(txt[txt.find("["):txt.rfind("]") + 1])
	for p, t in zip(plan, prompts): p["prompt"] = str(t)
	except Exception: pass
	return {"shots": plan}

	@app.post("/api/plan")
	def plan(segments: str = Form("[]"), mood: str = Form("Cinematic"), style: str = Form("Modern cinematic"), title: str = Form("My Music Video")):
	return {"job_id": submit(lambda jid: do_plan(jid, json.loads(segments or "[]"), mood, style, title))}

	# ---------- lip-sync ----------
	def do_lipsync(jid, image, audio):
	if not LIPSYNC_CMD: raise RuntimeError("Lip-sync is not installed on this server. Set LIPSYNC_CMD (see README).")
	out = FILES / f"lip_{jid}.mp4"; run_cmd(LIPSYNC_CMD, image=image, audio=audio, out=out)
	return {"video_url": f"/files/{out.name}"}

	@app.post("/api/lipsync")
	def lipsync(audio: UploadFile = File(...), image: UploadFile = File(...), model: str = Form(""), lyrics: str = Form(""), instructions: str = Form("")):
	h = uuid.uuid4().hex[:8]; a, i = save(audio, h, "audio"), save(image, h, "image")
	return {"job_id": submit(lambda jid: do_lipsync(jid, i, a))}

	# ---------- Wan2.2 clip ----------
	def _http_json(url, key=None, data=None, method=None, timeout=120):
	import urllib.request
	body = json.dumps(data).encode() if data is not None else None
	req = urllib.request.Request(url, body, method=method or ("POST" if body else "GET"))
	if key: req.add_header("Authorization", f"Key {key}")
	if body: req.add_header("Content-Type", "application/json")
	with urllib.request.urlopen(req, timeout=timeout) as r:
	return json.loads(r.read() or b"{}")

	def fal_upload(image, key):
	"""Two-step fal storage upload (initiate -> PUT) -> public file_url for image_url."""
	import urllib.request, mimetypes
	image = Path(image)
	ctype = mimetypes.guess_type(image.name)[0] or "image/png"
	init = _http_json(f"{FAL_REST}/storage/upload/initiate", key, {"file_name": image.name, "content_type": ctype})
	req = urllib.request.Request(init["upload_url"], image.read_bytes(), method="PUT")
	req.add_header("Content-Type", ctype)
	urllib.request.urlopen(req, timeout=120).read()
	return init["file_url"]

	def do_wan_cloud(jid, image, prompt, key, out):
	payload = {"prompt": str(prompt), "num_frames": 121, "frames_per_second": 24, "resolution": "720p"}
	model = FAL_T2V
	if image:
	payload["image_url"] = fal_upload(image, key); model = FAL_I2V
	sub = _http_json(f"{FAL_QUEUE}/{model}", key, payload)
	status_url, response_url = sub.get("status_url"), sub.get("response_url")
	if not status_url:
	result = sub
	else:
	result = None
	for i in range(120): # up to ~10 min
	time.sleep(5); prog(jid, 30 + min(60, i))
	st = _http_json(status_url, key)
	if st.get("status") == "COMPLETED":
	result = _http_json(response_url, key); break
	if st.get("status") in ("FAILED", "ERROR"):
	raise RuntimeError(f"Cloud generation failed on fal: {str(st)[:300]}")
	if result is None: raise RuntimeError("Cloud generation timed out after ~10 minutes on fal.")
	url = (result.get("video") or {}).get("url")
	if not url: raise RuntimeError(f"fal returned no video URL: {str(result)[:300]}")
	import urllib.request
	urllib.request.urlretrieve(url, out)

	def do_wan(jid, image, prompt, cloud_key=""):
	out = (FILES / f"clip_{jid}.mp4").resolve() # absolute: the auto-detected run executes inside WAN_DIR
	if WAN_CMD:
	run_cmd(WAN_CMD, image=image or "", prompt=prompt, out=out)
	elif wan_available():
	cmd = [str(WAN_PY), "generate.py", "--task", "ti2v-5B", "--size", "1280*704",
	"--ckpt_dir", "./Wan2.2-TI2V-5B", "--offload_model", "True",
	"--convert_model_dtype", "--t5_cpu", "--prompt", str(prompt), "--save_file", str(out)]
	if image: cmd += ["--image", str(Path(image).resolve())]
	r = subprocess.run(cmd, cwd=WAN_DIR, capture_output=True, text=True)
	if r.returncode: raise RuntimeError((r.stderr or r.stdout)[-500:])
	elif cloud_key or FAL_KEY:
	do_wan_cloud(jid, image, prompt, cloud_key or FAL_KEY, out)
	else:
	raise RuntimeError("No video generator available: run ./install-wan.sh on a GPU machine, or add a fal.ai cloud key in the Studio (no PC needed).")
	if not out.exists(): raise RuntimeError("Generation finished but produced no file.")
	return {"video_url": f"/files/{out.name}"}

	@app.post("/api/generate")
	def generate(prompt: str = Form(...), image: UploadFile = File(None), audio: UploadFile = File(None),
	model: str = Form(""), scene: str = Form(""), has_image: str = Form(""),
	x_cloud_key: str = Header("")):
	i = save(image, uuid.uuid4().hex[:8], "image") if image else None
	return {"job_id": submit(lambda jid: do_wan(jid, i, prompt, x_cloud_key))}

	# ---------- photo clip (CPU fallback — portrait -> slow-zoom clip, no GPU needed) ----------
	def do_photoclip(jid, image, duration, variant=0):
	dur = max(1.0, min(float(duration or 5), 40.0))
	out = FILES / f"photo_{jid}.mp4"; frames = int(dur * 24)
	# Camera-move variants so consecutive fallback shots don't look identical:
	# 0 slow zoom-in, 1 slow zoom-out, 2 pan right, 3 pan left.
	moves = [
	("min(zoom+0.0008,1.25)", "iw/2-(iw/zoom/2)", "ih/2-(ih/zoom/2)"),
	("if(eq(on,1),1.25,max(zoom-0.0008,1.0))", "iw/2-(iw/zoom/2)", "ih/2-(ih/zoom/2)"),
	("1.15", f"(iw-iw/zoom)*on/{frames}", "ih/2-(ih/zoom/2)"),
	("1.15", f"(iw-iw/zoom)*(1-on/{frames})", "ih/2-(ih/zoom/2)"),
	]
	z, x, y = moves[int(variant) % 4]
	vf = (f"scale=2560:1440:force_original_aspect_ratio=increase,crop=2560:1440,"
	f"zoompan=z='{z}':d={frames}:x='{x}':y='{y}':s=1280x720:fps=24")
	r = subprocess.run(["ffmpeg", "-y", "-loop", "1", "-i", str(image), "-vf", vf, "-t", str(dur),
	"-c:v", "libx264", "-pix_fmt", "yuv420p", str(out)], capture_output=True, text=True)
	if r.returncode: raise RuntimeError(r.stderr[-500:])
	return {"video_url": f"/files/{out.name}"}

	@app.post("/api/photoclip")
	def photoclip(image: UploadFile = File(...), duration: str = Form("5")):
	i = save(image, uuid.uuid4().hex[:8], "image")
	return {"job_id": submit(lambda jid: do_photoclip(jid, i, duration))}

	# ---------- final render ----------
	def lyrics_to_srt(text, path):
	rows = []
	for l in (text or "").splitlines():
	m = re.match(r"^(\d+):(\d\d)\s+(.*)$", l.strip())
	if m: rows.append((int(m[1]) * 60 + int(m[2]), m[3]))
	ts = lambda s: f"{int(s//3600):02}:{int(s//60%60):02}:{int(s%60):02},000"
	with open(path, "w") as f:
	for i, (t, x) in enumerate(rows):
	end = rows[i + 1][0] if i + 1 < len(rows) else t + 4
	f.write(f"{i+1}\n{ts(t)} --> {ts(end)}\n{x}\n\n")
	return bool(rows)

	def segments_to_srt(segments, path):
	ts = lambda s: f"{int(s//3600):02}:{int(s//60%60):02}:{int(s%60):02},{int((s%1)*1000):03}"
	with open(path, "w") as f:
	for i, s in enumerate(segments):
	f.write(f"{i+1}\n{ts(s['start'])} --> {ts(s['end'])}\n{s['text']}\n\n")
	return bool(segments)

	def do_render(jid, audio, clips, lyrics, segments):
	files = [FILES / Path(str(c)).name for c in clips if c] # basename only — no path traversal
	files = [f for f in files if f.exists()]
	if not files: raise RuntimeError("No generated clips found on the server to join.")
	lst = WORK / f"{jid}_list.txt"; lst.write_text("".join(f"file '{f.resolve()}'\n" for f in files))
	out = FILES / f"final_{jid}.mp4"; srt = WORK / f"{jid}.srt"
	has_subs = segments_to_srt(segments, srt) if segments else (lyrics_to_srt(lyrics, srt) if lyrics else False)
	cmd = ["ffmpeg", "-y", "-f", "concat", "-safe", "0", "-i", str(lst)]
	if audio: cmd += ["-i", str(audio), "-map", "0:v", "-map", "1:a", "-shortest"]
	if has_subs: cmd += ["-vf", f"subtitles={srt.as_posix()}"]
	cmd += ["-c:v", "libx264", "-pix_fmt", "yuv420p", "-c:a", "aac", str(out)]
	prog(jid, 20); r = subprocess.run(cmd, capture_output=True, text=True)
	if r.returncode: raise RuntimeError(r.stderr[-500:])
	return {"video_url": f"/files/{out.name}"}

	@app.post("/api/render")
	def render(audio: UploadFile = File(None), clips: str = Form("[]"), lipsync: str = Form(""),
	lyrics: str = Form(""), title: str = Form(""), segments: str = Form("[]")):
	a = save(audio, uuid.uuid4().hex[:8], "audio") if audio else None
	return {"job_id": submit(lambda jid: do_render(jid, a, json.loads(clips or "[]"), lyrics, json.loads(segments or "[]")))}