Update main.py

main.py

@@ -41,12 +41,12 @@ for d in [AUDIO_DIR, OUTPUT_DIR, JOBS_DIR, CLIENT_DIR]:
 
 # Rate limiting
 RATE_LIMITS: Dict[str, list] = {}
-GENERATE_LIMIT   = 3     # per hour per client
-STATUS_LIMIT     = 1000  # per hour per job
+GENERATE_LIMIT = 3      # per hour per client
+STATUS_LIMIT   = 1000   # per hour per job
 
 # Job timeout
-JOBS: Dict[str, Dict]      = {}
-JOB_TIMEOUT      = 900    # 15 min hard cap
+JOBS: Dict[str, Dict] = {}
+JOB_TIMEOUT = 900       # 15 min hard cap
 
 # Video listing cache
 VIDEO_CACHE: Dict[str, tuple] = {}
@@ -55,12 +55,12 @@ CACHE_TIMEOUT = 300
 # =============================================================================
 # QUEUE SYSTEM
 # =============================================================================
-MAX_CONCURRENT = 1
-AVG_JOB_DURATION_S = 360   # 6 minutes per job
+MAX_CONCURRENT      = 1
+AVG_JOB_DURATION_S  = 360   # 6 minutes per job
 
-_JOB_QUEUE: deque         = deque()   # ordered list of waiting job_ids
-_RUNNING_JOBS: set         = set()    # job_ids currently encoding
-_QUEUE_LOCK: asyncio.Lock  = None     # created at startup
+_JOB_QUEUE:    deque        = deque()
+_RUNNING_JOBS: set          = set()
+_QUEUE_LOCK:   asyncio.Lock = None   # created at startup
 
 # =============================================================================
 # CLIENT ID SYSTEM
@@ -112,7 +112,7 @@ def _create_client(fingerprint: str) -> Dict:
         "jobs":        [],
     }
     _save_client(rec)
-    logger.info(f"🆕 New client {rec['token'][:8]}…")
+    logger.info(f"New client {rec['token'][:8]}...")
     return rec
 
 def _fingerprint(req: Request) -> str:
@@ -142,7 +142,7 @@ def resolve_client(req: Request) -> Dict:
             "jobs":        [],
         }
         _save_client(rec)
-        logger.info(f"♻️  Restored client {token[:8]}… from header")
+        logger.info(f"Restored client {token[:8]}... from header")
         return rec
 
     fp = _fingerprint(req)
@@ -168,7 +168,7 @@ logger = logging.getLogger(__name__)
 # =============================================================================
 app = FastAPI(
     title="ArchNemix Shorts Generator API",
-    version="5.0.0",
+    version="5.1.0",
     docs_url="/docs",
     redoc_url="/redoc",
 )
@@ -195,8 +195,8 @@ app.add_middleware(
 # MODELS
 # =============================================================================
 class GenerateRequest(BaseModel):
-    audio_base64:  str            = Field(..., min_length=10)
-    subtitles_ass: str            = Field(..., min_length=10)
+    audio_base64:  str             = Field(..., min_length=10)
+    subtitles_ass: str             = Field(..., min_length=10)
     background:    str
     duration:      Optional[float] = Field(None, ge=1, le=180)
     request_id:    Optional[str]   = None
@@ -323,7 +323,7 @@ def _make_job(job_id: str, client_token: str) -> Dict:
         "status":           "queued",
         "queue_position":   None,
         "progress":         0,
-        "message":          "Waiting in queue…",
+        "message":          "Waiting in queue...",
         "created_at":       datetime.now(),
         "updated_at":       datetime.now(),
         "error":            None,
@@ -376,10 +376,10 @@ def cleanup_old_jobs():
             except ValueError:
                 pass
             for p in [
-                JOBS_DIR  / f"{job_id}.json",
-                AUDIO_DIR / f"{job_id}.mp3",
-                AUDIO_DIR / f"{job_id}.ass",
-                AUDIO_DIR / f"{job_id}_trimmed.mp4",
+                JOBS_DIR   / f"{job_id}.json",
+                AUDIO_DIR  / f"{job_id}.mp3",
+                AUDIO_DIR  / f"{job_id}.ass",
+                AUDIO_DIR  / f"{job_id}_trimmed.mp4",
                 OUTPUT_DIR / f"{job_id}.mp4",
             ]:
                 try:
@@ -400,10 +400,10 @@ def queue_position(job_id: str) -> Optional[int]:
 def queue_eta(position: int, job_duration: Optional[float] = None) -> Dict:
     running_remaining = AVG_JOB_DURATION_S
     for jid in _RUNNING_JOBS:
-        j = JOBS.get(jid, {})
+        j     = JOBS.get(jid, {})
         total = j.get("total_duration", 0) or AVG_JOB_DURATION_S
         out   = j.get("ffmpeg_out_time", 0) or 0
-        remaining = max(0, total - out)
+        remaining         = max(0, total - out)
         running_remaining = min(running_remaining, remaining)
 
     queued_ahead = max(0, position - 1)
@@ -421,13 +421,13 @@ def queue_eta(position: int, job_duration: Optional[float] = None) -> Dict:
         return f"{m}m {sec:02d}s" if sec else f"{m}m"
 
     return {
-        "queue_position":     position,
-        "jobs_ahead":         position,
-        "wait_low_s":         low_s,
-        "wait_high_s":        high_s,
-        "wait_low_human":     _fmt(low_s),
-        "wait_high_human":    _fmt(high_s),
-        "total_with_own_s":   int(total_s),
+        "queue_position":   position,
+        "jobs_ahead":       position,
+        "wait_low_s":       low_s,
+        "wait_high_s":      high_s,
+        "wait_low_human":   _fmt(low_s),
+        "wait_high_human":  _fmt(high_s),
+        "total_with_own_s": int(total_s),
         "message": (
             f"Position {position} in queue — "
             f"roughly {_fmt(low_s)}–{_fmt(high_s)} wait, "
@@ -441,12 +441,11 @@ async def _queue_worker():
 
         if len(_RUNNING_JOBS) >= MAX_CONCURRENT:
             continue
-
         if not _JOB_QUEUE:
             continue
 
         job_id = _JOB_QUEUE[0]
-        job = JOBS.get(job_id)
+        job    = JOBS.get(job_id)
         if not job or job["status"] in ("failed", "completed"):
             _JOB_QUEUE.popleft()
             continue
@@ -474,9 +473,9 @@ async def _run_and_release(job_id, audio_path, subs_path, background, duration):
 def _refresh_queue_positions():
     for i, jid in enumerate(_JOB_QUEUE):
         if jid in JOBS:
-            pos = i + 1
+            pos              = i + 1
             JOBS[jid]["queue_position"] = pos
-            JOBS[jid]["message"] = queue_eta(pos)["message"]
+            JOBS[jid]["message"]        = queue_eta(pos)["message"]
             _persist_job(jid)
 
 # =============================================================================
@@ -489,7 +488,7 @@ def _kill_ffmpeg_for_job(job_id: str):
     if proc and proc.returncode is None:
         try:
             proc.kill()
-            logger.info(f"🔪 Killed FFmpeg for job {job_id}")
+            logger.info(f"Killed FFmpeg for job {job_id}")
         except Exception:
             pass
 
@@ -498,27 +497,27 @@ def _parse_ffmpeg_kv(line: str) -> Dict:
     return {parts[0].strip(): parts[1].strip()} if len(parts) == 2 else {}
 
 async def _stream_ffmpeg_progress(
-    proc: asyncio.subprocess.Process,
-    job_id: str,
+    proc:           asyncio.subprocess.Process,
+    job_id:         str,
     total_duration: float,
     progress_start: int = 50,
     progress_end:   int = 95,
 ):
     frame = fps = speed = 0
-    out_time_s = 0.0
+    out_time_s       = 0.0
     last_update_time = time.time()
 
     async for raw_line in proc.stderr:
         line = raw_line.decode(errors="replace").strip()
         if not line:
             continue
-            
+
         kv = _parse_ffmpeg_kv(line)
         if not kv:
             continue
         key, val = next(iter(kv.items()))
 
-        if   key == "frame":        
+        if   key == "frame":
             frame = int(val) if val.isdigit() else frame
         elif key == "fps":
             try: fps = float(val)
@@ -528,13 +527,13 @@ async def _stream_ffmpeg_progress(
             except ValueError: pass
         elif key == "out_time":
             try:
-                parts = val.split(':')
+                parts = val.split(":")
                 if len(parts) == 3:
                     h, m, s = parts
                     out_time_s = int(h) * 3600 + int(m) * 60 + float(s)
             except ValueError: pass
         elif key == "speed":
-            try: 
+            try:
                 speed = float(val.replace("x", ""))
             except ValueError: pass
         elif key == "progress":
@@ -553,8 +552,8 @@ async def _stream_ffmpeg_progress(
                     total_duration=total_duration,
                     message=(
                         f"Encoding {out_time_s:.1f}s / {total_duration:.1f}s  "
-                        f"({speed:.1f}× speed)"
-                        if total_duration > 0 and speed > 0 else "Encoding…"
+                        f"({speed:.1f}x speed)"
+                        if total_duration > 0 and speed > 0 else "Encoding..."
                     ),
                 )
             if val == "end":
@@ -574,9 +573,9 @@ async def process_video_task(
     duration:   Optional[float],
 ):
     try:
-        logger.info(f"🚀 Job {job_id}: starting single-pass encode")
+        logger.info(f"Job {job_id}: starting encode")
         update_job(job_id, status="processing", progress=10,
-                   message="Downloading background video…", queue_position=None)
+                   message="Downloading background video...", queue_position=None)
 
         try:
             bg_path = await download_video_from_dataset(background, "minecraft")
@@ -584,33 +583,33 @@ async def process_video_task(
             update_job(job_id, status="failed", error=str(e.detail))
             return
 
-        update_job(job_id, progress=20, message="Analysing media…")
+        update_job(job_id, progress=20, message="Analysing media...")
 
         audio_dur = duration or get_media_duration(audio_path)
         video_dur = get_media_duration(Path(bg_path))
 
         if not (1 <= audio_dur <= 180):
-            update_job(job_id, status="failed", error="Audio duration out of range (1–180 s)")
+            update_job(job_id, status="failed", error="Audio duration out of range (1-180 s)")
             return
 
         logger.info(f"Job {job_id}: audio={audio_dur:.2f}s  bg={video_dur:.2f}s")
 
         if video_dur > audio_dur:
-            max_start = video_dur - audio_dur
-            start     = random.uniform(0, max_start)
+            max_start     = video_dur - audio_dur
+            start         = random.uniform(0, max_start)
             bg_input_args = [
                 "-ss", f"{start:.3f}",
                 "-t",  f"{audio_dur:.3f}",
                 "-i",  str(bg_path),
             ]
-            logger.info(f"Job {job_id}: trim start={start:.2f}s (input-side seek)")
+            logger.info(f"Job {job_id}: trim start={start:.2f}s")
         else:
-            loop_n = int(audio_dur / video_dur) + 2
+            loop_n        = int(audio_dur / video_dur) + 2
             bg_input_args = [
                 "-stream_loop", str(loop_n),
                 "-i",           str(bg_path),
             ]
-            logger.info(f"Job {job_id}: loop ×{loop_n}")
+            logger.info(f"Job {job_id}: loop x{loop_n}")
 
         subs_str = str(subs_path).replace("\\", "/").replace(":", "\\:")
         vf = (
@@ -633,17 +632,17 @@ async def process_video_task(
             "-filter_complex", vf,
             "-map", "[v]",
             "-map", "1:a",
-            "-c:v",      "libx264",
-            "-preset",   "medium",
-            "-crf",      "16",
-            "-minrate",  "8M",
-            "-maxrate",  "16M",
-            "-bufsize",  "16M",
+            "-c:v",       "libx264",
+            "-preset",    "medium",
+            "-crf",       "16",
+            "-minrate",   "8M",
+            "-maxrate",   "16M",
+            "-bufsize",   "16M",
             "-profile:v", "high",
             "-level:v",   "4.2",
             "-pix_fmt",   "yuv420p",
-            "-c:a",  "aac",
-            "-b:a",  "192k",
+            "-c:a",       "aac",
+            "-b:a",       "192k",
             "-shortest",
             "-movflags", "+faststart",
             "-threads",  "0",
@@ -652,11 +651,7 @@ async def process_video_task(
             str(output_path),
         ]
 
-        update_job(job_id, progress=40, message="Encoding… (single-pass)", total_duration=audio_dur)
-        logger.info(
-            f"Job {job_id}: CRF=16 preset=medium "
-            f"minrate=8M maxrate=16M bufsize=16M threads=auto"
-        )
+        update_job(job_id, progress=40, message="Encoding... (single-pass)", total_duration=audio_dur)
 
         timeout = max(480, int(audio_dur * 5) + 120)
 
@@ -695,7 +690,7 @@ async def process_video_task(
             return
 
         size_mb = output_path.stat().st_size / 1024 / 1024
-        logger.info(f"✅ Job {job_id}: {size_mb:.2f} MB")
+        logger.info(f"Job {job_id}: done {size_mb:.2f} MB")
 
         for p in [audio_path, subs_path]:
             try:
@@ -707,7 +702,7 @@ async def process_video_task(
             job_id,
             status="completed",
             progress=100,
-            message=f"Done! {size_mb:.1f} MB • CRF 16 • medium • 8–16 Mbps",
+            message=f"Done! {size_mb:.1f} MB",
             output_path=str(output_path),
         )
 
@@ -715,6 +710,24 @@ async def process_video_task(
         logger.exception(f"process_video_task unhandled for {job_id}")
         update_job(job_id, status="failed", error="Unexpected server error")
 
+# =============================================================================
+# FILE CLEANUP HELPER
+# =============================================================================
+def _delete_output_file(job_id: str, output_path: Path):
+    """
+    Delete the video file after it has been sent to the user.
+    Called as a background task from /download so /tmp stays clean.
+    """
+    try:
+        output_path.unlink(missing_ok=True)
+        if job_id in JOBS:
+            JOBS[job_id]["output_path"] = None
+            JOBS[job_id]["message"]     = "Downloaded and cleaned up"
+            _persist_job(job_id)
+        logger.info(f"Cleaned up {job_id} after download")
+    except Exception as e:
+        logger.warning(f"Cleanup failed for {job_id}: {e}")
+
 # =============================================================================
 # API ENDPOINTS
 # =============================================================================
@@ -723,18 +736,19 @@ async def process_video_task(
 async def root():
     return {
         "name":    "ArchNemix Shorts Generator API",
-        "version": "5.0.0",
+        "version": "5.1.0",
         "queue": {
-            "max_concurrent": MAX_CONCURRENT,
+            "max_concurrent":    MAX_CONCURRENT,
             "avg_job_duration_s": AVG_JOB_DURATION_S,
         },
         "endpoints": {
-            "register":   "GET  /register         — get or create your client token",
-            "generate":   "POST /generate          — submit a job",
-            "cancel":     "DELETE /job/{id}        — cancel job (called by controller)",
-            "job_status": "GET  /job/{id}           — poll status + queue position",
-            "queue":      "GET  /queue             — overall queue snapshot",
-            "download":   "GET  /download/{id}",
+            "register":   "GET  /register",
+            "generate":   "POST /generate",
+            "cancel":     "DELETE /job/{id}",
+            "job_status": "GET  /job/{id}",
+            "queue":      "GET  /queue",
+            "stream":     "GET  /stream/{id}   — inline video playback",
+            "download":   "GET  /download/{id} — file download (deletes after)",
             "delete":     "DELETE /video/{id}",
             "videos":     "GET  /videos/{category}",
             "health":     "GET  /health",
@@ -777,12 +791,12 @@ async def queue_status():
     running = len(_RUNNING_JOBS)
     waiting = len(_JOB_QUEUE)
     return {
-        "running":           running,
-        "waiting":           waiting,
-        "total_active":      running + waiting,
-        "avg_job_s":         AVG_JOB_DURATION_S,
-        "estimated_wait_s":  waiting * AVG_JOB_DURATION_S if running == 0 else
-                             (waiting + 1) * AVG_JOB_DURATION_S,
+        "running":          running,
+        "waiting":          waiting,
+        "total_active":     running + waiting,
+        "avg_job_s":        AVG_JOB_DURATION_S,
+        "estimated_wait_s": waiting * AVG_JOB_DURATION_S if running == 0 else
+                            (waiting + 1) * AVG_JOB_DURATION_S,
         "message": (
             "No jobs running — submit yours now!" if (running + waiting) == 0 else
             f"{running} running, {waiting} in queue"
@@ -799,9 +813,9 @@ async def list_videos(category: str = "minecraft"):
 
 @app.post("/generate")
 async def generate_video(
-    req: GenerateRequest,
+    req:             GenerateRequest,
     background_tasks: BackgroundTasks,
-    http_req: Request,
+    http_req:        Request,
 ):
     if not validate_app_key(http_req):
         raise HTTPException(403, "Invalid API key")
@@ -836,7 +850,7 @@ async def generate_video(
     create_job(job_id, token)
 
     try:
-        audio_size = await save_base64_audio(req.audio_base64, audio_path)
+        await save_base64_audio(req.audio_base64, audio_path)
         await save_subtitles(req.subtitles_ass, subs_path)
     except ValueError as e:
         update_job(job_id, status="failed", error=str(e))
@@ -856,28 +870,23 @@ async def generate_video(
         client["jobs"] = client["jobs"][-50:]
     _save_client(client)
 
-    position = len(_JOB_QUEUE) + 1
-    if _RUNNING_JOBS:
-        pass
-    else:
-        position = 0
-
+    position = len(_JOB_QUEUE) + 1 if _RUNNING_JOBS else 0
     _JOB_QUEUE.append(job_id)
     _refresh_queue_positions()
 
     if position == 0:
-        eta = {"message": "Starting very soon…", "wait_low_s": 0, "wait_high_s": 30}
+        eta = {"message": "Starting very soon...", "wait_low_s": 0, "wait_high_s": 30}
     else:
         eta = queue_eta(position, req.duration)
 
     return JSONResponse(
         {
-            "job_id":           job_id,
-            "client_token":     token,
-            "status":           "queued",
-            "queue_position":   position if position > 0 else 1,
-            "queue":            eta,
-            "check_status":     f"/job/{job_id}",
+            "job_id":         job_id,
+            "client_token":   token,
+            "status":         "queued",
+            "queue_position": position if position > 0 else 1,
+            "queue":          eta,
+            "check_status":   f"/job/{job_id}",
         },
         headers={"X-Client-ID": token},
     )
@@ -922,30 +931,66 @@ async def job_status(job_id: str, http_req: Request):
         },
     }
 
+    # ── Return both URLs when completed — controller stores these directly ───
     if job["status"] == "completed" and job.get("output_path"):
         resp["download_url"] = f"/download/{job_id}"
+        resp["stream_url"]   = f"/stream/{job_id}"
+
     if job.get("error"):
         resp["error"] = job["error"]
 
     return JSONResponse(resp)
 
+
+@app.get("/stream/{job_id}")
+async def stream_video(job_id: str):
+    """
+    Serve the video for inline browser playback.
+    Frontend uses this as <video src="..."> so user can watch immediately.
+    File is NOT deleted after streaming — user may rewatch or seek.
+    """
+    output_path = OUTPUT_DIR / f"{job_id}.mp4"
+    if not output_path.exists():
+        raise HTTPException(404, "Video not found or not ready")
+    if job_id in JOBS and JOBS[job_id]["status"] != "completed":
+        raise HTTPException(400, "Video not ready yet")
+
+    return FileResponse(
+        output_path,
+        media_type="video/mp4",
+        headers={
+            "Cache-Control": "no-cache",
+            "Accept-Ranges": "bytes",   # enables seeking in the browser player
+        },
+    )
+
+
 @app.get("/download/{job_id}")
-async def download_video(job_id: str):
+async def download_video(job_id: str, background_tasks: BackgroundTasks):
+    """
+    Serve the video as a file download.
+    Deletes the file from /tmp after sending so disk stays clean.
+    """
     output_path = OUTPUT_DIR / f"{job_id}.mp4"
     if not output_path.exists():
         raise HTTPException(404, "Video not found or not ready")
     if job_id in JOBS and JOBS[job_id]["status"] != "completed":
         raise HTTPException(400, "Video not ready yet")
+
+    # Delete file from /tmp after response is sent
+    background_tasks.add_task(_delete_output_file, job_id, output_path)
+
     return FileResponse(
         output_path,
         media_type="video/mp4",
         filename=f"archnemix-short-{job_id[:8]}.mp4",
         headers={
             "Content-Disposition": f'attachment; filename="archnemix-short-{job_id[:8]}.mp4"',
-            "Cache-Control": "no-cache",
+            "Cache-Control":       "no-cache",
         },
     )
 
+
 @app.delete("/video/{job_id}")
 async def delete_video(job_id: str, http_req: Request):
     if not validate_app_key(http_req):
@@ -957,7 +1002,7 @@ async def delete_video(job_id: str, http_req: Request):
         try:
             output_path.unlink()
             removed = True
-            logger.info(f"🗑  Deleted {job_id}")
+            logger.info(f"Deleted {job_id}")
         except Exception as e:
             raise HTTPException(500, f"Delete failed: {e}")
 
@@ -968,6 +1013,7 @@ async def delete_video(job_id: str, http_req: Request):
 
     return {"job_id": job_id, "deleted": removed}
 
+
 @app.delete("/job/{job_id}")
 async def cancel_job(job_id: str, http_req: Request):
     """Called by controller when frontend goes silent."""
@@ -988,12 +1034,17 @@ async def cancel_job(job_id: str, http_req: Request):
     except ValueError:
         pass
 
+    # Clean up output file if it exists
+    output_path = OUTPUT_DIR / f"{job_id}.mp4"
+    output_path.unlink(missing_ok=True)
+
     update_job(job_id, status="failed", error="Cancelled by controller — client disconnected")
     _refresh_queue_positions()
 
-    logger.info(f"🛑 Job {job_id} cancelled via controller request")
+    logger.info(f"Job {job_id} cancelled via controller request")
     return {"job_id": job_id, "cancelled": True}
 
+
 @app.get("/debug/dataset")
 async def debug_dataset():
     try:
@@ -1013,24 +1064,25 @@ async def startup():
     _QUEUE_LOCK = asyncio.Lock()
 
     logger.info("=" * 60)
-    logger.info("ArchNemix Shorts Generator API v5.0.0")
+    logger.info("ArchNemix Shorts Generator API v5.1.0")
     logger.info(f"Dataset    : {HF_DATASET}")
     logger.info(f"Concurrent : {MAX_CONCURRENT}")
-    logger.info(f"Avg job    : {AVG_JOB_DURATION_S}s ({AVG_JOB_DURATION_S//60}m)")
-    logger.info("Quality    : CRF 16 · medium · 8–16 Mbps · AAC 192k")
-    logger.info("Speed opts : single-pass trim · ass-first filter · threads=auto")
+    logger.info(f"Avg job    : {AVG_JOB_DURATION_S}s ({AVG_JOB_DURATION_S // 60}m)")
+    logger.info("Quality    : CRF 16 · medium · 8-16 Mbps · AAC 192k")
+    logger.info("Video path : stream from /tmp, delete on download")
     logger.info("=" * 60)
 
     asyncio.create_task(_queue_worker())
 
     try:
         vids = await list_videos_from_dataset("minecraft")
-        logger.info(f"✅ {len(vids)} videos cached")
+        logger.info(f"{len(vids)} videos cached")
     except Exception as e:
-        logger.warning(f"⚠️  Video cache warm-up: {e}")
+        logger.warning(f"Video cache warm-up: {e}")
 
     cleanup_old_jobs()
-    logger.info("🚀 Ready")
+    logger.info("Ready")
+
 
 @app.on_event("shutdown")
 async def shutdown():