Blame for app.py

"""AllOutMV — native Gradio app for a free Hugging Face ZeroGPU Space.

Engine: server.py (transcribe / director plan / photo-zoom clips / render)
plus providers.py, a free-tier provider chain (NovAI -> DeepInfra ->
Novita -> fal) that hops providers per shot when one runs dry.

Provider keys: read from the Space's private SECRETS (NOVAI_KEY,
DEEPINFRA_KEY, NOVITA_KEY, FAL_KEY) so Tyler never re-pastes them, with
the on-screen fields as per-session overrides. Keys are never in the code
— this Space is public, and a key printed in code would be scraped and
drained within days.

Long-form: the planner covers the WHOLE song (1–5 min) with 3–15s shots
cut on the lyric timings, instrumental gaps filled with their own shots,
and the render crossfades between clips (each clip is generated fade-long
so transitions overlap without drifting the lyric sync).
"""
import os
import subprocess
import time
import uuid
from pathlib import Path

import gradio as gr
import spaces

import ltx
import providers
import server
import talking
from server import FILES, do_photoclip, do_render, do_transcribe, local_plan, segments_to_srt

FADE = 0.4  # seconds of crossfade between clips
ENV_KEYS = {
    "novai": os.environ.get("NOVAI_KEY", ""),
    "deepinfra": os.environ.get("DEEPINFRA_KEY", ""),
    "novita": os.environ.get("NOVITA_KEY", ""),
    "fal": os.environ.get("FAL_KEY", "") or os.environ.get("FAL_API_KEY", ""),
}


def _job():
    jid = uuid.uuid4().hex[:12]
    server.jobs[jid] = {"status": "running", "progress": 0, "created": time.time()}
    return jid


def _fmt_ts(sec):
    return f"{int(sec // 60)}:{int(sec % 60):02d}"


def _keys(novai, deepinfra, novita, fal):
    fields = {"novai": novai, "deepinfra": deepinfra, "novita": novita, "fal": fal}
    return {k: (fields.get(k) or "").strip() or ENV_KEYS.get(k, "") for k in ENV_KEYS}


def _audio_duration(path):
    try:
        r = subprocess.run(["ffprobe", "-v", "error", "-show_entries", "format=duration",
                            "-of", "csv=p=0", str(path)], capture_output=True, text=True)
        return float(r.stdout.strip())
    except Exception:
        return 0.0


@spaces.GPU(duration=120)
def _transcribe_gpu(audio_path, model_choice):
    """Real GPU work (Whisper on CUDA) — the Space's @spaces.GPU function.
    ZeroGPU attaches the GPU only inside decorated calls and aborts decorated
    functions that never use it, so the GPU belongs on transcription, not on
    the HTTP provider chain."""
    return do_transcribe(_job(), Path(audio_path), model_choice or "small")


def transcribe(audio_path, model_choice):
    if not audio_path:
        return None, "", "Upload a song first."
    try:
        res = _transcribe_gpu(audio_path, model_choice)  # fast path: real GPU
    except Exception:
        try:
            res = do_transcribe(_job(), Path(audio_path), model_choice or "small")  # CPU fallback
        except Exception as e:
            return None, "", f"⚠️ Transcription failed — {type(e).__name__}: {e}"
    segs = res["segments"]
    text = "\n".join(f"{_fmt_ts(s['start'])}  {s['text']}" for s in segs)
    return segs, text, f"Transcribed {len(segs)} lines ({res['language']}, {res['duration']:.0f}s). Now build the director plan."


def _split_span(start, end, prompt, caption):
    """Split one span into <=15s shots (first keeps the caption)."""
    out, t = [], start
    while end - t > 15:
        out.append({"start": t, "end": t + 15.0, "prompt": prompt, "caption": caption if t == start else ""})
        t += 15.0
    out.append({"start": t, "end": end, "prompt": prompt, "caption": caption if t == start else ""})
    return out


def _normalize_spans(shots):
    """Guarantee every shot is 3–15s: oversize spans split into near-equal
    parts; a <3s sliver merges into its neighbour and the pair re-splits."""
    out = []
    for s in shots:
        span = s["end"] - s["start"]
        if span > 15:
            n = int(span // 15) + 1
            step = span / n
            for k in range(n):
                out.append({**s, "start": s["start"] + k * step, "end": s["start"] + (k + 1) * step,
                            "caption": s["caption"] if k == 0 else ""})
        elif span < 3 and out:
            prev = out.pop()
            combined = s["end"] - prev["start"]
            n = max(1, int(combined // 15) + (1 if combined % 15 > 0 else 0))
            step = combined / n
            for k in range(n):
                src = prev if k == 0 else s
                out.append({**src, "start": prev["start"] + k * step, "end": prev["start"] + (k + 1) * step,
                            "caption": prev["caption"] if k == 0 else (s["caption"] if k == n - 1 else "")})
        else:
            out.append(dict(s))
    return out


def long_plan(segs, mood, style, title, duration):
    """Full-song shot plan on a CONTINUOUS timeline: lyric shots from
    local_plan (each holds until the next begins), instrumental gaps get
    their own shots, every shot normalized to 3–15s, 0 -> duration covered."""
    base = local_plan(segs, mood or "Cinematic", style or "Modern cinematic", title or "My Music Video")
    if not duration:
        duration = max((s["end"] for s in base), default=60.0)
    inst_prompt = (f"{style or 'Modern cinematic'}. Instrumental break in a music video for "
                   f"'{title or 'My Music Video'}': sweeping atmospheric visuals, the performer's world, "
                   f"no singing. Keep wardrobe and colour grade consistent. Cinematic-level aesthetics (Wan2.2).")
    anchors = sorted(base, key=lambda x: x["start"])
    shots, cursor = [], 0.0
    for i, s in enumerate(anchors):
        start = float(s["start"])
        nxt = float(anchors[i + 1]["start"]) if i + 1 < len(anchors) else duration
        end = min(max(float(s["end"]), min(start + 3.0, nxt)), duration)
        if start - cursor >= 3:
            shots += _split_span(cursor, start, inst_prompt, "(instrumental)")
        elif start > cursor and shots:
            shots[-1]["end"] = start  # sub-3s hole: previous shot simply holds
        shots.append({"start": max(start, cursor), "end": end, "prompt": s["prompt"], "caption": s["caption"]})
        cursor = end
    if duration - cursor >= 1:
        shots += _split_span(cursor, duration, inst_prompt, "(instrumental)")
    shots = _normalize_spans([s for s in shots if s["end"] - s["start"] >= 0.5])
    for i, s in enumerate(shots):
        s["index"] = i + 1
    return shots


def build_plan(segs, mood, style, title, audio_path=None):
    if not segs:
        return None, "Transcribe your song first — the plan is built from the lyrics."
    dur = _audio_duration(audio_path) if audio_path else 0.0
    shots = long_plan(segs, mood, style, title, dur)
    md = "\n\n".join(
        f"**Shot {s['index']}** ({_fmt_ts(s['start'])}–{_fmt_ts(s['end'])})\n\n_{s['caption'] or '—'}_\n\n{s['prompt']}"
        for s in shots
    )
    return shots, md


def render_smooth(clips, windows, audio, segments, out_path):
    """Crossfade render: clip i was generated (window_i + FADE) long (except
    the last), so each xfade overlaps the extension and cuts land exactly on
    the lyric windows — total length stays sum(windows), sync preserved."""
    if len(clips) == 1:
        raise ValueError("single clip — plain render is fine")
    jid = uuid.uuid4().hex[:10]
    srt = server.WORK / f"{jid}.srt"
    has_subs = segments_to_srt(segments, srt) if segments else False
    cmd = ["ffmpeg", "-y"]
    for c in clips:
        cmd += ["-i", str(c)]
    cmd += ["-i", str(audio)]
    filt, prev, offset = [], "[0:v]", 0.0
    for i in range(len(clips) - 1):
        offset += windows[i]
        lbl = f"[x{i}]"
        filt.append(f"{prev}[{i + 1}:v]xfade=transition=fade:duration={FADE}:offset={offset:.3f}{lbl}")
        prev = lbl
    if has_subs:
        filt.append(f"{prev}subtitles={srt.as_posix()}[vout]")
        prev = "[vout]"
    graph = ";".join(filt)
    cmd += ["-filter_complex", graph, "-map", prev, "-map", f"{len(clips)}:a",
            "-shortest", "-c:v", "libx264", "-pix_fmt", "yuv420p", "-c:a", "aac", str(out_path)]
    r = subprocess.run(cmd, capture_output=True, text=True)
    if r.returncode or not Path(out_path).exists():
        raise RuntimeError(r.stderr[-400:])
    return Path(out_path)


def ai_clip(image_path, prompt, novai_key, deepinfra_key, novita_key, fal_key):
    """Generate ONE AI clip through the provider chain (first free tier that
    answers wins). Deliberately NOT @spaces.GPU: ZeroGPU aborts decorated
    functions that never touch the GPU, and this work is pure HTTP polling —
    it runs on CPU. The decorator lives on _transcribe_gpu instead."""
    keys = _keys(novai_key, deepinfra_key, novita_key, fal_key)
    if not any(keys.values()):
        raise gr.Error("No provider keys available — add one in the key fields above or save them as Space secrets.")
    if not (prompt or "").strip():
        raise gr.Error("Write a prompt for the clip.")
    providers.reset_chain()
    out = FILES / f"chain_{uuid.uuid4().hex[:10]}.mp4"
    try:
        path, who = providers.generate_with_chain(keys, image_path, prompt, 6.0, out)
    except providers.ProviderError as e:
        raise gr.Error(f"Every provider failed: {e}")
    return str(path)


def hero_clip(image_path, prompt):
    """Animate the performer's OWN portrait with LTX-Video on this Space's
    GPU (provider zero). Weights download on CPU first so the GPU window is
    spent generating, not downloading. Free ZeroGPU ration: a few clips/day.
    (image_path arrives as a plain path, same as ai_clip's.)"""
    if not image_path:
        raise gr.Error("Upload a performer portrait first (step 1).")
    if not (prompt or "").strip():
        prompt = ("The performer raps energetically to the camera, bouncing to the beat, "
                  "hands gesturing and thrown up on the rhythm, head nodding, mouth moving "
                  "as if performing the song, cinematic music-video lighting")
    try:
        ltx.ensure_weights()
    except Exception as e:
        raise gr.Error(f"LTX weights download failed: {str(e)[:400]}")
    out = FILES / f"hero_{uuid.uuid4().hex[:10]}.mp4"
    try:
        ltx.ltx_clip(image_path, prompt, 5.0, out)
    except Exception as e:
        raise gr.Error(f"LTX hero clip failed: {str(e)[:400]}")
    norm = FILES / f"hero_{uuid.uuid4().hex[:10]}_720.mp4"
    try:
        providers.normalize_clip(out, 5.0, norm)
        return str(norm)
    except Exception:
        return str(out)


def talking_clip(image_path, audio_path, start_sec, progress=gr.Progress()):
    """Talking head (SadTalker): the portrait is WARPED, never repainted —
    his face stays his, only the mouth and head move to the song. Takes a
    5-second slice of the song at start_sec. First use installs the model
    into an isolated vendor env on CPU, then renders on the Space's GPU."""
    if not image_path:
        raise gr.Error("Upload a performer portrait first (step 1).")
    if not audio_path:
        raise gr.Error("Upload your song first (step 1).")
    start = max(0.0, float(start_sec or 0))
    slice_wav = FILES / f"talk_slice_{uuid.uuid4().hex[:10]}.wav"
    r = subprocess.run(["ffmpeg", "-y", "-ss", f"{start}", "-t", "5", "-i", str(audio_path),
                        "-ar", "16000", "-ac", "1", str(slice_wav)],
                       capture_output=True, text=True)
    if r.returncode or not slice_wav.exists():
        raise gr.Error(f"Could not cut the audio slice: {r.stderr[-300:]}")
    progress(0.1, desc="Preparing SadTalker (first run installs the model — a few minutes)…")
    try:
        talking.ensure_talking()
    except Exception as e:
        raise gr.Error(f"SadTalker setup failed: {str(e)[:400]}")
    progress(0.55, desc="Animating his face on your words (GPU)…")
    raw = FILES / f"talk_raw_{uuid.uuid4().hex[:10]}.mp4"
    try:
        talking.sadtalker_gpu(image_path, str(slice_wav), raw)
    except Exception as e:
        raise gr.Error(f"Talking-head failed: {str(e)[:400]}")
    progress(0.9, desc="Polishing to 720p…")
    final = FILES / f"talk_{uuid.uuid4().hex[:10]}.mp4"
    vf = ("scale=1280:720:force_original_aspect_ratio=decrease,"
          "pad=1280:720:(ow-iw)/2:(oh-ih)/2,fps=24")
    r = subprocess.run(["ffmpeg", "-y", "-i", str(raw), "-vf", vf,
                        "-c:v", "libx264", "-pix_fmt", "yuv420p", "-c:a", "aac", str(final)],
                       capture_output=True, text=True)
    if r.returncode or not final.exists():
        return str(raw), "Talking-head clip ready (720p polish failed — showing raw render). His face, your words."
    return str(final), "Talking-head clip ready — his face, your words. (SadTalker beta)"


def resume_clip(task_id, novai_key):
    """Recover an already-finished NovAI clip by task ID — free-queue tasks
    outlive the run that submitted them, so a timed-out generation's video
    can still be collected without spending another one."""
    keys = _keys(novai_key, "", "", "")
    if not keys.get("novai"):
        raise gr.Error("No NovAI key available — paste one above or save NOVAI_KEY as a Space secret.")
    if not (task_id or "").strip():
        raise gr.Error("Paste a NovAI task ID first.")
    out = FILES / f"resume_{uuid.uuid4().hex[:10]}.mp4"
    raw = Path(str(out) + ".raw.mp4")
    try:
        got = providers.novai_resume(keys["novai"], task_id, raw)
        providers.normalize_clip(got, 6.0, out)
    except providers.ProviderError as e:
        raise gr.Error(f"Resume failed: {e}")
    return str(out)


def make_video(audio_path, image_path, segs, shots, use_ai,
               novai_key, deepinfra_key, novita_key, fal_key,
               progress=gr.Progress()):
    """One-click pipeline for a full 1–5 min song: transcribe/plan if needed,
    per-shot clips (3–15s on the lyric windows) via the free-tier chain with
    varied photo-zoom fallback, crossfade render with burned-in lyrics."""
    if not audio_path:
        yield None, "Upload a song first."
        return
    if not image_path:
        yield None, "Upload a portrait photo — clips are built from it."
        return
    keys = _keys(novai_key, deepinfra_key, novita_key, fal_key)
    chain_on = bool(use_ai and any(keys.values()))
    chain_note = ""
    if use_ai and not chain_on:
        chain_note = "\n\n⚠️ Provider chain was OFF — no provider keys found (key fields empty and no Space secrets visible), so every shot used photo-zoom."
    chain_error = None
    duration = _audio_duration(audio_path)
    if segs is None:
        progress(0.05, desc="Transcribing your song…")
        jid = _job()
        res = do_transcribe(jid, Path(audio_path), "small")
        segs = res["segments"]
        duration = duration or res["duration"]
    if not shots:
        shots = long_plan(segs, "Cinematic", "Modern cinematic", "My Music Video", duration)
    providers.reset_chain()
    providers.image_reference(image_path, keys, cache={})  # warm the portrait URL once
    clip_paths, windows, used = [], [], []
    total = len(shots)
    for i, shot in enumerate(shots):
        window = max(3.0, min(float(shot["end"]) - float(shot["start"]), 15.0))
        target = window + (FADE if i < total - 1 else 0.0)  # extension the crossfade will eat
        progress(0.1 + 0.7 * i / total, desc=f"Clip {i + 1} of {total}…")
        made = None
        if chain_on:
            out = FILES / f"shot_{uuid.uuid4().hex[:10]}.mp4"
            try:
                path, who = providers.generate_with_chain(keys, image_path, shot["prompt"], target, out)
                made = (path, who)
            except providers.ProviderError as e:
                made = None  # photo-zoom fills this shot
                if chain_error is None:
                    chain_error = str(e)
        if made is None:
            jid = _job()
            res = do_photoclip(jid, Path(image_path), target, variant=i % 4)
            made = (FILES / Path(res["video_url"]).name, "photo-zoom (free, built-in)")
        clip_paths.append(made[0])
        windows.append(window)
        used.append(f"Shot {i + 1} ({_fmt_ts(shot['start'])}): {made[1]}")
    progress(0.9, desc="Rendering your final video…")
    final = FILES / f"final_{uuid.uuid4().hex[:10]}.mp4"
    try:
        render_smooth(clip_paths, windows, Path(audio_path), segs, final)
        how = f"{total} shots, crossfaded, {_fmt_ts(sum(windows))} of video"
    except Exception:
        jid = _job()
        res = do_render(jid, Path(audio_path), [p.name for p in clip_paths], "", segs)
        final = FILES / Path(res["video_url"]).name
        how = f"{total} shots, hard-cut render (crossfade unavailable)"
    if chain_error:
        chain_note += f"\n\n⚠️ Provider chain report (first failure): {chain_error}"
    yield str(final), f"Done — your music video is ready ({how}).\n" + "\n".join(used) + chain_note


with gr.Blocks(title="AllOutMV") as demo:
    gr.Markdown(
        "# 🎬 AllOutMV — AI Music Video Studio\n"
        "Upload a song (1–5 min) + a portrait, get a full music video: shots are cut on your lyric timings (3–15s each), "
        "generated through a **chain of free tiers** that hops providers when one runs dry, and crossfaded together so it "
        "flows with the music. Transcription and rendering run free on this Space; the first run downloads the speech model."
    )
    segs_state = gr.State(None)
    shots_state = gr.State(None)
    with gr.Row():
        audio_in = gr.Audio(label="Your song (mp3/wav/m4a)", type="filepath")
        image_in = gr.Image(label="Performer portrait", type="filepath")
    with gr.Accordion("🔑 Free-tier provider keys — saved as Space secrets, or paste per session", open=True):
        gr.Markdown(
            "Keys saved in this Space's **Settings → Variables and secrets** (`NOVAI_KEY`, `DEEPINFRA_KEY`, `NOVITA_KEY`, `FAL_KEY`) "
            "are used automatically — the fields below are only per-session overrides, and nothing pasted here is stored. "
            "**NovAI** — aiapi-pro.com: CogVideoX-Flash is **$0 per call** (5 free videos/day) + $2 trial credit. "
            "**DeepInfra** — deepinfra.com: Wan image-to-video (needs a funded balance). "
            "**Novita** — novita.ai: Kling image-to-video 3–15s (needs credit). "
            "**fal.ai** — pay-as-you-go safety net, $0.15/clip. With no keys at all, videos still finish with photo-zoom clips."
        )
        with gr.Row():
            novai_in = gr.Textbox(label="NovAI API key override (free tier)", type="password")
            deepinfra_in = gr.Textbox(label="DeepInfra API key override", type="password")
        with gr.Row():
            novita_in = gr.Textbox(label="Novita API key override", type="password")
            fal_in = gr.Textbox(label="fal.ai API key override (optional)", type="password")
    with gr.Row():
        model_in = gr.Dropdown(["tiny", "base", "small"], value="small", label="Whisper model (small = best balance)")
        title_in = gr.Textbox(label="Video title", value="My Music Video")
        mood_in = gr.Dropdown(["Cinematic", "Dark", "Energetic", "Romantic", "Dreamlike", "Hopeful", "Aggressive"], value="Cinematic", label="Mood")
        style_in = gr.Textbox(label="Visual style", value="Modern cinematic")
    status_out = gr.Textbox(label="Status", interactive=False, lines=8)

    transcribe_btn = gr.Button("1️⃣ Transcribe my song")
    transcript_out = gr.Textbox(label="Lyrics (timed)", lines=8, interactive=False)
    transcribe_btn.click(transcribe, [audio_in, model_in], [segs_state, transcript_out, status_out], api_name="transcribe")

    plan_btn = gr.Button("2️⃣ Build director plan")
    plan_out = gr.Markdown()
    plan_btn.click(build_plan, [segs_state, mood_in, style_in, title_in, audio_in], [shots_state, plan_out], api_name="plan")

    use_ai_in = gr.Checkbox(label="Use the free-tier provider chain for AI clips (photo-zoom fills any gap)", value=True)
    video_btn = gr.Button("⚡ Make my music video", variant="primary")
    video_out = gr.Video(label="Your music video")
    video_btn.click(make_video,
                    [audio_in, image_in, segs_state, shots_state, use_ai_in, novai_in, deepinfra_in, novita_in, fal_in],
                    [video_out, status_out], api_name="make_video")

    gr.Markdown("---\n### Or generate a single AI clip through the chain")
    with gr.Row():
        clip_prompt_in = gr.Textbox(label="Clip prompt", placeholder="Performer singing under neon city lights, cinematic…")
        clip_btn = gr.Button("🎬 Generate one AI clip")
    clip_out = gr.Video(label="AI clip")
    clip_btn.click(ai_clip, [image_in, clip_prompt_in, novai_in, deepinfra_in, novita_in, fal_in], [clip_out], api_name="ai_clip")
    with gr.Row():
        resume_in = gr.Textbox(label="Resume a NovAI task ID (if a generation timed out, its video may still be waiting)", placeholder="Paste task ID…")
        resume_btn = gr.Button("↩️ Resume task")
    resume_btn.click(resume_clip, [resume_in, novai_in], [clip_out], api_name="resume_clip")
    gr.Markdown("### 🎬 Animate my performer — LTX on this Space's own GPU\n"
                "Animates **your uploaded portrait itself** (bouncing, gesturing, performing). "
                "Free GPU ration is only a few clips a day — save these for hero moments like the chorus.")
    with gr.Row():
        hero_prompt_in = gr.Textbox(label="Performance direction",
                                    value="The performer raps energetically to the camera, bouncing to the beat, hands gesturing and thrown up on the rhythm, mouth moving as if performing the song, cinematic music-video lighting")
        hero_btn = gr.Button("🧟 Animate my performer", variant="primary")
    hero_out = gr.Video(label="Hero clip (your performer, animated)")
    hero_btn.click(hero_clip, [image_in, hero_prompt_in], [hero_out], api_name="hero_clip")
    gr.Markdown("### 🗣️ Talking head (beta) — SadTalker on this Space's GPU\n"
                "The opposite trick from LTX: **his face is never repainted** — your photo is warped by 3D facial "
                "motion read off your song, so only the mouth and head move. Uses a 5-second slice of your song. "
                "First use installs the model (a few minutes); save it for the lines that matter most.")
    with gr.Row():
        talk_start_in = gr.Number(label="Slice start (seconds into the song)", value=78)
        talk_btn = gr.Button("🗣️ Make him perform it", variant="primary")
    talk_out = gr.Video(label="Talking-head clip (his face, your voice)")
    talk_btn.click(talking_clip, [image_in, audio_in, talk_start_in], [talk_out, status_out], api_name="talking_clip")

if __name__ == "__main__":
    demo.queue(max_size=16).launch()