feat: pipeline orchestrator with split GPU allocation

Adds pipeline_runner.py which orchestrates all 9 Mazinger dubbing stages.
GPU-heavy work is split across two @spaces.GPU(duration=120) functions
(_gpu_transcribe, _gpu_synthesize) to respect ZeroGPU's 120s per-request
hard cap. LLM calls use _call_with_retry with 5/15/45 s exponential
backoff on 429 errors. Audio duration is gated at 300 s via soundfile.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

pipeline_runner.py

@@ -0,0 +1,611 @@
+"""
+pipeline_runner.py — Mazinger Dubber pipeline orchestrator for HF ZeroGPU Spaces.
+
+Orchestrates all 9 dubbing stages. GPU-heavy stages are split into two separate
+@spaces.GPU-decorated functions to stay within the 120s-per-request hard cap.
+"""
+
+from __future__ import annotations
+
+import os
+import shutil
+import tempfile
+import time
+from typing import Any, Generator
+
+import spaces  # provided by HF ZeroGPU runtime
+
+# ---------------------------------------------------------------------------
+# Constants
+# ---------------------------------------------------------------------------
+
+HF_TOKEN: str = os.environ.get("HF_TOKEN", "")
+LLM_BASE_URL: str = "https://router.huggingface.co/v1"
+LLM_MODEL_TEXT: str = "Qwen/Qwen2.5-72B-Instruct"
+LLM_MODEL_VISION: str = "Qwen/Qwen2.5-VL-7B-Instruct"
+BASE_DIR: str = "/tmp/mazinger_output"
+
+STAGE_NAMES: list[str] = [
+    "Download",       # 1
+    "Transcribe",     # 2
+    "Thumbnails",     # 3
+    "Describe",       # 4
+    "Translate",      # 5
+    "Resegment",      # 6
+    "Synthesize",     # 7
+    "Assemble",       # 8
+    "Subtitle",       # 9
+]
+
+# ---------------------------------------------------------------------------
+# Mazinger imports (only available on HF Spaces where the package is installed)
+# ---------------------------------------------------------------------------
+
+from mazinger import ProjectPaths, LLMUsageTracker                              # noqa: E402
+from mazinger.llm import build_client                                           # noqa: E402
+from mazinger import (                                                          # noqa: E402
+    download,
+    transcribe,
+    thumbnails,
+    describe,
+    translate,
+    resegment,
+    tts,
+    assemble,
+    subtitle,
+)
+from mazinger.subtitle import SubtitleStyle, download_google_font               # noqa: E402
+from mazinger.srt import parse as parse_srt                                     # noqa: E402
+
+# ---------------------------------------------------------------------------
+# Helper: build LLM client
+# ---------------------------------------------------------------------------
+
+
+def _make_client(model: str = LLM_MODEL_TEXT):
+    """Return an OpenAI-compatible client pointed at HF Inference Router."""
+    return build_client(api_key=HF_TOKEN, base_url=LLM_BASE_URL)
+
+
+# ---------------------------------------------------------------------------
+# Helper: call with exponential backoff on 429
+# ---------------------------------------------------------------------------
+
+
+def _call_with_retry(fn, *args, max_retries: int = 3, **kwargs):
+    """
+    Call *fn* with *args*/*kwargs*, retrying up to *max_retries* times on
+    HTTP 429 (rate-limit) errors with exponential backoff (5 s / 15 s / 45 s).
+    """
+    delays = [5, 15, 45]
+    last_exc: Exception | None = None
+
+    for attempt in range(max_retries + 1):
+        try:
+            return fn(*args, **kwargs)
+        except Exception as exc:
+            # Detect rate-limit errors by status code attribute or message text
+            is_rate_limit = (
+                getattr(exc, "status_code", None) == 429
+                or "429" in str(exc)
+                or "rate limit" in str(exc).lower()
+                or "too many requests" in str(exc).lower()
+            )
+            if is_rate_limit and attempt < max_retries:
+                wait = delays[attempt]
+                print(
+                    f"[retry] 429 rate-limit hit — waiting {wait}s "
+                    f"(attempt {attempt + 1}/{max_retries})"
+                )
+                time.sleep(wait)
+                last_exc = exc
+                continue
+            raise
+
+    # Should never reach here, but satisfy the type checker
+    raise last_exc  # type: ignore[misc]
+
+
+# ---------------------------------------------------------------------------
+# Helper: audio duration guard
+# ---------------------------------------------------------------------------
+
+
+def _check_audio_duration(audio_path: str, max_seconds: float = 300.0) -> float:
+    """
+    Return audio duration in seconds.
+    Raises ValueError if the file exceeds *max_seconds*.
+    Requires the `soundfile` package (included in mazinger[all-qwen]).
+    """
+    import soundfile as sf  # lazy import — only needed here
+
+    info = sf.info(audio_path)
+    duration: float = info.duration
+    if duration > max_seconds:
+        raise ValueError(
+            f"Audio is {duration:.1f}s — exceeds the {max_seconds:.0f}s limit. "
+            "Please trim your video before uploading."
+        )
+    return duration
+
+
+# ---------------------------------------------------------------------------
+# GPU Stage 1 — Transcription  (120 s allocation)
+# ---------------------------------------------------------------------------
+
+
+@spaces.GPU(duration=120)
+def _gpu_transcribe(
+    audio_path: str,
+    output_path: str,
+    method: str = "whisperx",
+    model: str | None = None,
+    language: str | None = None,
+) -> str:
+    """
+    Run WhisperX (or the requested STT method) on *audio_path* with CUDA.
+
+    Returns the path to the written SRT file (*output_path*).
+    First invocation may be slow due to model weight downloads.
+    """
+    transcribe.transcribe(
+        audio_path=audio_path,
+        output_path=output_path,
+        method=method,
+        model=model,
+        language=language,
+        device="cuda",
+    )
+    return output_path
+
+
+# ---------------------------------------------------------------------------
+# GPU Stage 2 — TTS Synthesis  (120 s allocation)
+# ---------------------------------------------------------------------------
+
+
+@spaces.GPU(duration=120)
+def _gpu_synthesize(
+    tts_model_name: str,
+    voice_sample: str | None,
+    voice_script: str | None,
+    voice_theme: str | None,
+    clone_profile: str | None,
+    srt_entries: list[dict],
+    output_dir: str,
+    target_language: str,
+) -> str:
+    """
+    Load the TTS model, resolve a voice prompt via one of the supported modes,
+    then synthesize all SRT segments into *output_dir*.
+
+    Voice-prompt priority:
+      1. clone_profile  → load a pre-built voice profile
+      2. voice_sample + voice_script → create a voice prompt from recorded audio
+      3. voice_theme    → load a named built-in theme
+      4. fallback       → built-in "narrator-m" voice
+
+    Returns the output directory path containing rendered audio segments.
+    """
+    # Load TTS model onto GPU
+    tts_model = tts.load_model(tts_model_name)
+
+    # Resolve voice prompt
+    if clone_profile:
+        voice_prompt = tts.load_profile(clone_profile)
+    elif voice_sample and voice_script:
+        voice_prompt = tts.create_voice_prompt(
+            audio_path=voice_sample,
+            transcript=voice_script,
+        )
+    elif voice_theme:
+        voice_prompt = tts.load_theme(voice_theme)
+    else:
+        voice_prompt = tts.load_theme("narrator-m")
+
+    # Synthesize all segments
+    tts.synthesize_segments(
+        model=tts_model,
+        voice_prompt=voice_prompt,
+        srt_entries=srt_entries,
+        output_dir=output_dir,
+        language=target_language,
+    )
+
+    return output_dir
+
+
+# ---------------------------------------------------------------------------
+# Main pipeline generator
+# ---------------------------------------------------------------------------
+
+
+def run_pipeline(
+    source: str,
+    target_language: str,
+    voice_mode: str,
+    voice_theme: str | None = None,
+    voice_profile: str | None = None,
+    voice_sample: str | None = None,
+    voice_script: str | None = None,
+    output_type: str = "video",
+    embed_subtitles: bool = True,
+    subtitle_font: str = "Cairo",
+    subtitle_font_size: int = 28,
+) -> Generator[tuple[int, str, dict[str, Any]], None, None]:
+    """
+    Orchestrate all 9 dubbing stages for *source* (URL or local path).
+
+    Yields ``(stage_index, log_message, result_dict)`` tuples at each stage.
+    Stage indices are 1-based; a final yield with ``stage_index=9`` (all done)
+    is emitted after stage 9 completes.
+
+    Args:
+        source:            YouTube/local video URL or absolute file path.
+        target_language:   BCP-47 language code for the dubbed output (e.g. "ar").
+        voice_mode:        One of "theme", "profile", "clone", or "default".
+        voice_theme:       Named built-in voice theme (used when voice_mode="theme").
+        voice_profile:     Pre-built clone profile name (used when voice_mode="profile").
+        voice_sample:      Path to speaker audio sample (used when voice_mode="clone").
+        voice_script:      Transcript of *voice_sample* (used when voice_mode="clone").
+        output_type:       "video" or "audio_only".
+        embed_subtitles:   Burn translated subtitles into output video.
+        subtitle_font:     Google Font name for burned subtitles.
+        subtitle_font_size: Point size for burned subtitles.
+    """
+    os.makedirs(BASE_DIR, exist_ok=True)
+    project_dir = tempfile.mkdtemp(dir=BASE_DIR, prefix="run_")
+    paths = ProjectPaths(project_dir)
+    tracker = LLMUsageTracker()
+
+    # Temporary directory that survives across ZeroGPU ephemeral boundaries
+    result_tmp = tempfile.mkdtemp(prefix="mazinger_result_")
+
+    # -----------------------------------------------------------------------
+    # Stage 1 — Download
+    # -----------------------------------------------------------------------
+    stage = 1
+    yield stage, f"[{STAGE_NAMES[stage - 1]}] Starting download: {source}", {}
+
+    try:
+        if source.startswith("http://") or source.startswith("https://"):
+            dl_result = download.download_url(source, output_dir=paths.media_dir)
+        else:
+            dl_result = download.ingest_local(source, output_dir=paths.media_dir)
+    except Exception as exc:
+        yield stage, f"[{STAGE_NAMES[stage - 1]}] FAILED: {exc}", {"error": str(exc)}
+        return
+
+    audio_path: str = dl_result.get("audio_path", "")
+    video_path: str | None = dl_result.get("video_path")
+
+    # Duration guard
+    try:
+        duration = _check_audio_duration(audio_path, max_seconds=300.0)
+    except ValueError as exc:
+        yield stage, f"[{STAGE_NAMES[stage - 1]}] REJECTED: {exc}", {"error": str(exc)}
+        return
+
+    yield (
+        stage,
+        f"[{STAGE_NAMES[stage - 1]}] Done — {duration:.1f}s of audio.",
+        {"audio_path": audio_path, "video_path": video_path, "duration": duration},
+    )
+
+    # -----------------------------------------------------------------------
+    # Stage 2 — Transcribe (GPU)
+    # -----------------------------------------------------------------------
+    stage = 2
+    srt_path: str = paths.srt_raw
+
+    yield (
+        stage,
+        f"[{STAGE_NAMES[stage - 1]}] Transcribing with WhisperX "
+        "(first run may take a minute while model weights download)…",
+        {},
+    )
+
+    try:
+        _gpu_transcribe(
+            audio_path=audio_path,
+            output_path=srt_path,
+            method="whisperx",
+        )
+    except TimeoutError as exc:
+        yield (
+            stage,
+            f"[{STAGE_NAMES[stage - 1]}] GPU timeout — try a shorter clip. {exc}",
+            {"error": str(exc)},
+        )
+        return
+    except Exception as exc:
+        yield stage, f"[{STAGE_NAMES[stage - 1]}] FAILED: {exc}", {"error": str(exc)}
+        return
+
+    yield (
+        stage,
+        f"[{STAGE_NAMES[stage - 1]}] Done — SRT written to {srt_path}.",
+        {"srt_path": srt_path},
+    )
+
+    # -----------------------------------------------------------------------
+    # Stage 3 — Thumbnails
+    # -----------------------------------------------------------------------
+    stage = 3
+    thumb_paths: list[str] = []
+
+    yield stage, f"[{STAGE_NAMES[stage - 1]}] Extracting keyframes…", {}
+
+    if video_path and os.path.exists(video_path):
+        try:
+            timestamps = thumbnails.select_timestamps(video_path)
+            thumb_paths = thumbnails.extract_frames(
+                video_path=video_path,
+                timestamps=timestamps,
+                output_dir=paths.thumbs_dir,
+            )
+        except Exception as exc:
+            # Non-fatal — describe stage will be skipped gracefully
+            yield (
+                stage,
+                f"[{STAGE_NAMES[stage - 1]}] WARNING: could not extract frames — {exc}",
+                {"warning": str(exc)},
+            )
+    else:
+        yield (
+            stage,
+            f"[{STAGE_NAMES[stage - 1]}] No video file available — skipping frame extraction.",
+            {},
+        )
+
+    yield (
+        stage,
+        f"[{STAGE_NAMES[stage - 1]}] Done — {len(thumb_paths)} keyframe(s) extracted.",
+        {"thumb_paths": thumb_paths},
+    )
+
+    # -----------------------------------------------------------------------
+    # Stage 4 — Describe (vision LLM)
+    # -----------------------------------------------------------------------
+    stage = 4
+    scene_description: str = ""
+
+    yield stage, f"[{STAGE_NAMES[stage - 1]}] Describing scene content…", {}
+
+    if thumb_paths:
+        vision_client = _make_client(LLM_MODEL_VISION)
+        try:
+            scene_description = _call_with_retry(
+                describe.describe_content,
+                client=vision_client,
+                model=LLM_MODEL_VISION,
+                image_paths=thumb_paths,
+                tracker=tracker,
+            )
+        except Exception as exc:
+            yield (
+                stage,
+                f"[{STAGE_NAMES[stage - 1]}] WARNING: description failed — {exc}",
+                {"warning": str(exc)},
+            )
+    else:
+        yield (
+            stage,
+            f"[{STAGE_NAMES[stage - 1]}] No thumbnails — skipping visual description.",
+            {},
+        )
+
+    yield (
+        stage,
+        f"[{STAGE_NAMES[stage - 1]}] Done.",
+        {"scene_description": scene_description},
+    )
+
+    # -----------------------------------------------------------------------
+    # Stage 5 — Translate
+    # -----------------------------------------------------------------------
+    stage = 5
+    srt_translated_path: str = paths.srt_translated
+
+    yield (
+        stage,
+        f"[{STAGE_NAMES[stage - 1]}] Translating subtitles to {target_language}…",
+        {},
+    )
+
+    text_client = _make_client(LLM_MODEL_TEXT)
+
+    try:
+        _call_with_retry(
+            translate.translate_srt,
+            client=text_client,
+            model=LLM_MODEL_TEXT,
+            srt_path=srt_path,
+            output_path=srt_translated_path,
+            target_language=target_language,
+            scene_description=scene_description,
+            tracker=tracker,
+        )
+    except Exception as exc:
+        yield stage, f"[{STAGE_NAMES[stage - 1]}] FAILED: {exc}", {"error": str(exc)}
+        return
+
+    yield (
+        stage,
+        f"[{STAGE_NAMES[stage - 1]}] Done — translated SRT at {srt_translated_path}.",
+        {"srt_translated_path": srt_translated_path},
+    )
+
+    # -----------------------------------------------------------------------
+    # Stage 6 — Resegment
+    # -----------------------------------------------------------------------
+    stage = 6
+    srt_resegmented_path: str = paths.srt_resegmented
+
+    yield stage, f"[{STAGE_NAMES[stage - 1]}] Resegmenting for natural TTS phrasing…", {}
+
+    try:
+        _call_with_retry(
+            resegment.resegment_srt,
+            client=text_client,
+            model=LLM_MODEL_TEXT,
+            srt_path=srt_translated_path,
+            output_path=srt_resegmented_path,
+            target_language=target_language,
+            tracker=tracker,
+        )
+    except Exception as exc:
+        yield stage, f"[{STAGE_NAMES[stage - 1]}] FAILED: {exc}", {"error": str(exc)}
+        return
+
+    yield (
+        stage,
+        f"[{STAGE_NAMES[stage - 1]}] Done — resegmented SRT at {srt_resegmented_path}.",
+        {"srt_resegmented_path": srt_resegmented_path},
+    )
+
+    # -----------------------------------------------------------------------
+    # Stage 7 — Synthesize (GPU)
+    # -----------------------------------------------------------------------
+    stage = 7
+    segments_dir: str = paths.segments_dir
+    os.makedirs(segments_dir, exist_ok=True)
+
+    yield (
+        stage,
+        f"[{STAGE_NAMES[stage - 1]}] Synthesizing dubbed audio "
+        "(first run may take a minute while TTS model downloads)…",
+        {},
+    )
+
+    # Parse SRT into entry dicts for the TTS stage
+    srt_entries = parse_srt(srt_resegmented_path)
+
+    # Resolve voice-mode arguments
+    _voice_theme = voice_theme if voice_mode == "theme" else None
+    _clone_profile = voice_profile if voice_mode == "profile" else None
+    _voice_sample = voice_sample if voice_mode == "clone" else None
+    _voice_script = voice_script if voice_mode == "clone" else None
+
+    try:
+        _gpu_synthesize(
+            tts_model_name="Qwen/Qwen3-TTS",
+            voice_sample=_voice_sample,
+            voice_script=_voice_script,
+            voice_theme=_voice_theme,
+            clone_profile=_clone_profile,
+            srt_entries=srt_entries,
+            output_dir=segments_dir,
+            target_language=target_language,
+        )
+    except TimeoutError as exc:
+        yield (
+            stage,
+            f"[{STAGE_NAMES[stage - 1]}] GPU timeout — try a shorter clip. {exc}",
+            {"error": str(exc)},
+        )
+        return
+    except Exception as exc:
+        yield stage, f"[{STAGE_NAMES[stage - 1]}] FAILED: {exc}", {"error": str(exc)}
+        return
+
+    yield (
+        stage,
+        f"[{STAGE_NAMES[stage - 1]}] Done — audio segments in {segments_dir}.",
+        {"segments_dir": segments_dir},
+    )
+
+    # -----------------------------------------------------------------------
+    # Stage 8 — Assemble
+    # -----------------------------------------------------------------------
+    stage = 8
+    assembled_path: str = paths.assembled_audio
+
+    yield stage, f"[{STAGE_NAMES[stage - 1]}] Assembling dubbed timeline…", {}
+
+    try:
+        assemble.assemble_timeline(
+            srt_entries=srt_entries,
+            segments_dir=segments_dir,
+            original_audio_path=audio_path,
+            output_path=assembled_path,
+        )
+        assembled_path = assemble.post_process(assembled_path)
+    except Exception as exc:
+        yield stage, f"[{STAGE_NAMES[stage - 1]}] FAILED: {exc}", {"error": str(exc)}
+        return
+
+    yield (
+        stage,
+        f"[{STAGE_NAMES[stage - 1]}] Done — assembled audio at {assembled_path}.",
+        {"assembled_path": assembled_path},
+    )
+
+    # -----------------------------------------------------------------------
+    # Stage 9 — Subtitle / Mux
+    # -----------------------------------------------------------------------
+    stage = 9
+    final_path: str
+
+    yield stage, f"[{STAGE_NAMES[stage - 1]}] Finalising output…", {}
+
+    try:
+        if embed_subtitles and video_path and os.path.exists(video_path):
+            # Download the requested Google Font for burned subtitles
+            font_path: str | None = None
+            try:
+                font_path = download_google_font(subtitle_font, output_dir=paths.fonts_dir)
+            except Exception:
+                font_path = None  # fall back to FFmpeg default font
+
+            style = SubtitleStyle(
+                font_path=font_path,
+                font_size=subtitle_font_size,
+            )
+            final_path = subtitle.burn_subtitles(
+                video_path=video_path,
+                audio_path=assembled_path,
+                srt_path=srt_resegmented_path,
+                output_path=paths.final_video,
+                style=style,
+            )
+        else:
+            final_path = subtitle.mux_video(
+                video_path=video_path,
+                audio_path=assembled_path,
+                output_path=paths.final_video,
+            )
+    except Exception as exc:
+        yield stage, f"[{STAGE_NAMES[stage - 1]}] FAILED: {exc}", {"error": str(exc)}
+        return
+
+    # Copy results to the persistent temp directory (ZeroGPU ephemeral-safe)
+    result_video = os.path.join(result_tmp, os.path.basename(final_path))
+    result_srt = os.path.join(result_tmp, os.path.basename(srt_resegmented_path))
+
+    shutil.copy2(final_path, result_video)
+    shutil.copy2(srt_resegmented_path, result_srt)
+
+    yield (
+        stage,
+        f"[{STAGE_NAMES[stage - 1]}] Done — final output at {result_video}.",
+        {
+            "final_path": result_video,
+            "srt_path": result_srt,
+            "usage": tracker.summary(),
+        },
+    )
+
+    # -----------------------------------------------------------------------
+    # Final sentinel yield — all 9 stages complete
+    # -----------------------------------------------------------------------
+    yield (
+        9,
+        "Pipeline complete.",
+        {
+            "final_path": result_video,
+            "srt_path": result_srt,
+            "usage": tracker.summary(),
+            "done": True,
+        },
+    )