import base64 import io import logging import os import shutil import subprocess import sys import tempfile import time import uuid from pathlib import Path logging.basicConfig( level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s", datefmt="%H:%M:%S", ) log = logging.getLogger(__name__) # --------------------------------------------------------------------------- # Install chatterbox-tts with --no-deps BEFORE any other import. # --------------------------------------------------------------------------- try: import chatterbox # noqa: F401 except ImportError: log.info("[boot] Installing chatterbox-tts --no-deps …") subprocess.run( [sys.executable, "-m", "pip", "install", "--no-deps", "--quiet", "chatterbox-tts"], check=True, ) log.info("[boot] chatterbox-tts installed.") from gradio import Server from fastapi import Request, BackgroundTasks from fastapi.responses import HTMLResponse, JSONResponse, FileResponse from commentary import STYLES, VIBE_TO_STYLE, generate_commentary from events import events_to_frames, pick_key_events from faces import annotate_event_frame, frame_at, scan_video, write_names_to_jsonl from ov_models import download_models from tts import generate_script_audio from video import get_duration, normalize_video MAX_SECONDS = 60.0 HERE = Path(__file__).parent SESSIONS: dict[str, dict] = {} TIERS = ["HEADLINER", "RISING", "DIVA", "WILDCARD", "BREAKOUT", "CAMEO"] SUPPORTED_VIDEO = {".mp4", ".mov", ".webm", ".mkv", ".avi", ".m4v"} log.info("[boot] Downloading OpenVINO models if needed…") try: download_models() log.info("[boot] OpenVINO models ready.") except Exception as e: log.warning(f"[boot] OV model download failed: {e}") def _crop_b64(pil, size=160): if pil is None: return None img = pil.copy(); img.thumbnail((size, size)) buf = io.BytesIO(); img.save(buf, format="PNG") return "data:image/png;base64," + base64.b64encode(buf.getvalue()).decode() def _build_ui() -> str: doc = (HERE / "Marquee.html").read_text() css = (HERE / "marquee.css").read_text() js = (HERE / "marquee.js").read_text() doc = doc.replace('', f"") doc = doc.replace('', f"") return doc app = Server() _UI_HTML = _build_ui() log.info("[boot] UI built and ready.") @app.get("/", response_class=HTMLResponse) async def root(): return HTMLResponse(_UI_HTML) @app.post("/api/scan") async def api_scan(request: Request): t0 = time.time() log.info("[scan] Request received") try: form = await request.form() upload = form.get("file") if upload is None: return JSONResponse({"ok": False, "error": "no file"}, status_code=400) suffix = Path(upload.filename).suffix.lower() log.info(f"[scan] File: {upload.filename} ({suffix})") if suffix not in SUPPORTED_VIDEO: return JSONResponse({"ok": False, "error": f"Unsupported format '{suffix}'. Supported: {', '.join(sorted(SUPPORTED_VIDEO))}"}) tmp = tempfile.NamedTemporaryFile(delete=False, suffix=suffix) data = await upload.read() tmp.write(data); tmp.flush(); tmp.close() log.info(f"[scan] Upload saved: {len(data)//1024}KB → {tmp.name}") dur = get_duration(tmp.name) log.info(f"[scan] Duration: {dur:.1f}s") if dur > MAX_SECONDS + 1: return JSONResponse({"ok": False, "error": "Clip longer than 60 seconds"}) log.info("[scan] Normalising video (ffmpeg)…") t1 = time.time() norm_path, w, h = normalize_video(tmp.name) log.info(f"[scan] Normalised in {time.time()-t1:.1f}s → {w}x{h}") log.info("[scan] Running person tracking scan…") tracks, motion, meta, jsonl_path = scan_video(norm_path, MAX_SECONDS) log.info(f"[scan] Tracking done — {len(tracks)} person(s) found") sid = uuid.uuid4().hex SESSIONS[sid] = { "norm": norm_path, "tracks": tracks, "jsonl_path": jsonl_path, "motion": motion, "meta": meta, } log.info(f"[scan] Session created: {sid}") stars = [ {"idx": trk["id"], "tier": TIERS[i % len(TIERS)], "crop": _crop_b64(trk["crop"])} for i, trk in enumerate(tracks) ] log.info(f"[scan] Completed in {time.time()-t0:.1f}s total") return JSONResponse({"ok": True, "session_id": sid, "video_url": f"/video/{sid}", "duration": get_duration(norm_path), "w": w, "h": h, "portrait": h > w, "stars": stars}) except Exception as e: import traceback; traceback.print_exc() log.error(f"[scan] FAILED: {e}") return JSONResponse({"ok": False, "error": str(e)}, status_code=500) @app.get("/video/{sid}") async def serve_video(sid: str): log.info(f"[video] Serving video for session {sid}") sess = SESSIONS.get(sid) if not sess: log.warning(f"[video] Session not found: {sid}") return JSONResponse({"error": "session not found"}, status_code=404) return FileResponse(sess["norm"], media_type="video/mp4", headers={"Accept-Ranges": "bytes"}) @app.post("/api/tts") async def api_tts(request: Request): t0 = time.time() log.info("[tts] Request received") try: body = await request.json() sid = body.get("session_id", "") vibe = body.get("vibe", "football") script = body.get("script", []) voice_model = body.get("voice_model", "chatterbox") log.info(f"[tts] sid={sid} vibe={vibe} voice={voice_model} lines={len(script)}") if not script: return JSONResponse({"ok": False, "error": "empty script"}, status_code=400) if sid and sid not in SESSIONS: return JSONResponse({"ok": False, "error": "session not found"}, status_code=404) log.info(f"[tts] Generating {len(script)} lines with {voice_model}…") lines = generate_script_audio(script, vibe, voice_model=voice_model) audio_count = sum(1 for l in lines if l.get("audio_b64")) log.info(f"[tts] Done in {time.time()-t0:.1f}s — " f"{audio_count}/{len(lines)} lines have audio") return JSONResponse({"ok": True, "lines": lines}) except Exception as e: import traceback; traceback.print_exc() log.error(f"[tts] FAILED: {e}") return JSONResponse({"ok": False, "error": str(e)}, status_code=500) @app.post("/api/export") async def api_export(request: Request, background_tasks: BackgroundTasks): """Burn video + TTS audio + soft subtitle captions into one MP4. FIX: tmpdir is cleaned up via BackgroundTasks AFTER the response is fully streamed — prevents the race condition where FileResponse tries to read a file that Python GC already deleted. """ import wave as _wave t0 = time.time() log.info("[export] Request received") tmpdir = None try: body = await request.json() sid = body.get("session_id", "") vibe = body.get("vibe", "") lines = body.get("lines", []) log.info(f"[export] sid={sid} vibe={vibe} lines={len(lines)}") sess = SESSIONS.get(sid) if not sess: return JSONResponse({"ok": False, "error": "session not found"}, status_code=404) if not lines: return JSONResponse({"ok": False, "error": "no lines"}, status_code=400) video_path = sess["norm"] tmpdir = tempfile.mkdtemp(prefix="mq_export_") log.info(f"[export] Working dir: {tmpdir}") # 1. Write TTS WAV files log.info("[export] Writing TTS WAV files…") audio_clips = [] for i, line in enumerate(lines): b64 = line.get("audio_b64") if not b64: continue wav_bytes = base64.b64decode(b64) wav_path = os.path.join(tmpdir, f"tts_{i:03d}.wav") with open(wav_path, "wb") as f: f.write(wav_bytes) with _wave.open(io.BytesIO(wav_bytes)) as wf: dur = wf.getnframes() / wf.getframerate() audio_clips.append({"path": wav_path, "t": float(line["t"]), "duration": dur}) log.info(f"[export] {len(audio_clips)} audio clips written") # 2. Build SRT subtitle file log.info("[export] Building subtitle file…") total_dur = sess["meta"]["n_frames"] / sess["meta"]["fps"] vibe_label = { "football": "COCO", "diva": "COCO", "wildlife": "COCO", "boxing": "COCO", "masterchef": "COCO", }.get(vibe, "COCO") def _srt_time(s): h=int(s//3600); m=int((s%3600)//60); sec=s%60 ms=int((sec%1)*1000); sec=int(sec) return f"{h:02d}:{m:02d}:{sec:02d},{ms:03d}" srt_rows = [] for i, line in enumerate(lines): t_start = float(line["t"]) t_end = float(lines[i+1]["t"]) if i+1 {_srt_time(t_end)}", f"{vibe_label} {line['text']}", "", ] srt_path = os.path.join(tmpdir, "subs.srt") with open(srt_path, "w", encoding="utf-8") as f: f.write("\n".join(srt_rows)) log.info(f"[export] SRT written: {len(lines)} cues") # 3. Build ffmpeg command out_path = os.path.join(tmpdir, "broadcast.mp4") n = len(audio_clips) cmd = ["ffmpeg", "-y", "-i", video_path, "-i", srt_path] for clip in audio_clips: cmd += ["-i", clip["path"]] if n > 0: parts, delayed = [], [] for idx, clip in enumerate(audio_clips): delay_ms = int(clip["t"] * 1000) lbl = f"d{idx}" cmd_idx = idx + 2 # 0=video, 1=srt, 2..=wavs parts.append(f"[{cmd_idx}:a]adelay={delay_ms}|{delay_ms},apad[{lbl}]") delayed.append(f"[{lbl}]") parts.append(f"{''.join(delayed)}amix=inputs={n}:normalize=0[commentary]") parts.append("[0:a]volume=0.25[orig];[orig][commentary]amix=inputs=2:normalize=0[mixed]") cmd += [ "-filter_complex", ";".join(parts), "-map", "0:v", "-map", "[mixed]", "-map", "1:s", "-c:v", "copy", # stream-copy video — no re-encode "-c:a", "aac", "-b:a", "128k", "-c:s", "mov_text", "-movflags", "+faststart", out_path, ] else: cmd += [ "-map", "0:v", "-map", "0:a", "-map", "1:s", "-c:v", "copy", "-c:a", "copy", "-c:s", "mov_text", "-movflags", "+faststart", out_path, ] log.info(f"[export] Running ffmpeg (n_audio={n})…") log.info(f"[export] CMD: {' '.join(cmd)}") t1 = time.time() res = subprocess.run(cmd, capture_output=True) log.info(f"[export] ffmpeg finished in {time.time()-t1:.1f}s, " f"returncode={res.returncode}") if res.returncode != 0: err = res.stderr.decode(errors="replace")[-1200:] log.error(f"[export] ffmpeg STDERR:\n{err}") return JSONResponse({"ok": False, "error": f"ffmpeg: {err}"}, status_code=500) out_size_mb = os.path.getsize(out_path) / 1024 / 1024 log.info(f"[export] Output: {out_path} ({out_size_mb:.1f}MB)") log.info(f"[export] Total time: {time.time()-t0:.1f}s") vibe_slug = vibe or "broadcast" fname = f"marquee_{vibe_slug}.mp4" # KEY FIX: schedule tmpdir cleanup as a background task so it runs # AFTER the file has been fully streamed to the client. # Previously tmpdir could be cleaned up mid-stream causing download to hang. def _cleanup(d: str): try: shutil.rmtree(d, ignore_errors=True) log.info(f"[export] Cleaned up tmpdir: {d}") except Exception as ex: log.warning(f"[export] Cleanup failed: {ex}") background_tasks.add_task(_cleanup, tmpdir) return FileResponse( out_path, media_type="video/mp4", filename=fname, headers={"Content-Disposition": f'attachment; filename="{fname}"'}, ) except Exception as e: import traceback; traceback.print_exc() log.error(f"[export] FAILED: {e}") if tmpdir: shutil.rmtree(tmpdir, ignore_errors=True) return JSONResponse({"ok": False, "error": str(e)}, status_code=500) @app.api(name="generate") def generate(session_id: str, names: list, vibe: str) -> dict: t0 = time.time() log.info(f"[generate] session={session_id} vibe={vibe} names={names}") sess = SESSIONS.get(session_id) if not sess: log.error(f"[generate] Session not found: {session_id}") return {"ok": False, "error": "Session expired — re-cast"} tracks = sess.get("tracks", []) id_to_name = {} for i, trk in enumerate(tracks): raw = str(names[i]).strip() if i < len(names) else "" if raw: id_to_name[trk["id"]] = raw log.info(f"[generate] Named tracks: {id_to_name}") if id_to_name: write_names_to_jsonl(sess["jsonl_path"], id_to_name) style = VIBE_TO_STYLE.get(vibe, next(iter(STYLES))) fps = sess["meta"]["fps"] event_times = pick_key_events(sess["motion"], fps) event_idxs = events_to_frames(event_times, fps) log.info(f"[generate] Key events: {event_times}") log.info("[generate] Annotating keyframes…") frames = [] for ts, fidx in zip(event_times, event_idxs): bgr = frame_at(sess["norm"], fidx) if bgr is None: log.warning(f"[generate] frame_at({fidx}) returned None, skipping") continue frames.append((ts, annotate_event_frame(bgr, sess["jsonl_path"], ts))) log.info(f"[generate] {len(frames)} annotated frames ready for VLM") log.info(f"[generate] Calling Qwen VLM (style={style})…") t1 = time.time() roster = list(id_to_name.values()) script = generate_commentary(frames, roster, style, vibe=vibe) log.info(f"[generate] VLM done in {time.time()-t1:.1f}s — {len(script)} lines") log.info(f"[generate] Total: {time.time()-t0:.1f}s") return {"ok": True, "script": [{"t": round(float(l["time"]),2), "text": l["text"]} for l in script]} app.launch(ssr_mode=False, show_error=True)