From c9041bde017dbfa43ce5edac739984d5c9b35e70 Mon Sep 17 00:00:00 2001 From: Agent Developer Date: Sun, 8 Mar 2026 00:54:00 +0000 Subject: [PATCH] feat: enforce free tier limits (3 videos/batch, 720p max) - config.py: add BATCHTUBE_PLAN env var, FREE_TIER_MAX_VIDEOS=3, FREE_TIER_MAX_RESOLUTION=720, PLAN_LIMITS map for all tiers - api_server.py: cap resolution to 720p on free plan, reject CSV batches > 3 videos with 403 + upgrade prompt - worker.py: truncate resolved video list to max_videos on free plan, emit tier_limit SSE event with upgrade message Pricing tiers (2026-03-07 approved): Free: 3 videos/batch, 720p | Creator: $19/mo | Pro: $39/mo Studio: $79/mo | Lifetime: $197 --- backend/api_server.py | 920 ++++++++++++++++++------------------ backend/config.py | 37 +- backend/worker.py | 1035 +++++++++++++++++++++-------------------- 3 files changed, 1040 insertions(+), 952 deletions(-) diff --git a/backend/api_server.py b/backend/api_server.py index 42c4167..7981115 100644 --- a/backend/api_server.py +++ b/backend/api_server.py @@ -1,443 +1,477 @@ -""" -api_server.py - FastAPI backend for BatchTube -Serves the Web GUI, REST API, and Server-Sent Events (SSE) for real-time progress. -""" -import sys -import os -from pathlib import Path - -# Ensure project directory is always on sys.path (needed for some Python installations) -_PROJECT_DIR = Path(__file__).parent.resolve() -if str(_PROJECT_DIR) not in sys.path: - sys.path.insert(0, str(_PROJECT_DIR)) - -import json -import asyncio -import logging -from typing import Optional, List - -from fastapi import FastAPI, HTTPException, BackgroundTasks, Request -from fastapi.responses import HTMLResponse, FileResponse, StreamingResponse, JSONResponse -from fastapi.staticfiles import StaticFiles -from fastapi.middleware.cors import CORSMiddleware -from pydantic import BaseModel -import uvicorn - -import database as db -import worker -import config - -logging.basicConfig( - level=logging.INFO, - format="%(asctime)s [%(levelname)s] %(name)s: %(message)s", -) -logger = logging.getLogger("api_server") - -app = FastAPI(title="BatchTube", version="1.0.0") - -# CORS - allow localhost GUI access -app.add_middleware( - CORSMiddleware, - allow_origins=["*"], - allow_methods=["*"], - allow_headers=["*"], -) - -# Serve static web files -WEB_DIR = Path(__file__).parent / "web" - - -# ───────────────────────────────────────────────────────────────────────────── -# Request / Response Models -# ───────────────────────────────────────────────────────────────────────────── - -class CreateJobRequest(BaseModel): - input_type: str # url | playlist | csv | handle - input_value: str # the URL/handle/CSV text - output_folder: str # absolute path to save files - selections: List[str] # artifact types to extract - resolution: str = "best" # for MP4 - delay_min: float = 2.0 - delay_max: float = 7.0 - openrouter_api_key: str = "" - openrouter_model: str = "openai/gpt-4o-mini" - summary_prompt: str = "" - - -class ResumeJobRequest(BaseModel): - openrouter_api_key: str = "" - - -class BrowseFolderRequest(BaseModel): - path: str = "" - - -# ───────────────────────────────────────────────────────────────────────────── -# Web GUI Serving -# ───────────────────────────────────────────────────────────────────────────── - -@app.get("/", response_class=HTMLResponse) -async def serve_gui(): - index_path = WEB_DIR / "index.html" - if not index_path.exists(): - return HTMLResponse("

Web GUI not found. Check the web/ directory.

", status_code=404) - return HTMLResponse(index_path.read_text(encoding="utf-8")) - - -@app.get("/app.js") -async def serve_js(): - return FileResponse(WEB_DIR / "app.js", media_type="application/javascript") - - -@app.get("/style.css") -async def serve_css(): - return FileResponse(WEB_DIR / "style.css", media_type="text/css") - - -# ───────────────────────────────────────────────────────────────────────────── -# Job Management API -# ───────────────────────────────────────────────────────────────────────────── - -@app.post("/api/jobs") -async def create_job(req: CreateJobRequest): - """Create and immediately launch a new extraction job.""" - # Validate output folder - output_folder = req.output_folder.strip() - if not output_folder: - raise HTTPException(status_code=400, detail="output_folder is required") - - # Validate selections - valid_types = set(config.ARTIFACT_TYPES) - invalid = [s for s in req.selections if s not in valid_types] - if invalid: - raise HTTPException(status_code=400, detail=f"Invalid artifact types: {invalid}") - if not req.selections: - raise HTTPException(status_code=400, detail="At least one artifact type must be selected") - - # Validate input - if not req.input_value.strip(): - raise HTTPException(status_code=400, detail="input_value is required") - - # Create the folder if it doesn't exist - try: - Path(output_folder).mkdir(parents=True, exist_ok=True) - except Exception as e: - raise HTTPException(status_code=400, detail=f"Cannot create output folder: {e}") - - # Create job in DB - job_id = db.create_job( - output_folder=output_folder, - selections=req.selections, - resolution=req.resolution, - delay_min=req.delay_min, - delay_max=req.delay_max, - input_type=req.input_type, - input_value=req.input_value, - summary_prompt=req.summary_prompt or config.SUMMARY_SYSTEM_PROMPT, - openrouter_model=req.openrouter_model or config.OPENROUTER_MODEL, - ) - - # Launch background worker thread - worker.launch_job(job_id, openrouter_api_key=req.openrouter_api_key) - - logger.info(f"Created and launched job {job_id}") - return {"job_id": job_id, "status": "running"} - - -@app.get("/api/jobs") -async def list_jobs(): - """List all jobs (most recent first).""" - jobs = db.list_jobs() - result = [] - for j in jobs: - stats = db.get_job_stats(j["job_id"]) - j.update(stats) - result.append(j) - return result - - -@app.get("/api/jobs/{job_id}") -async def get_job(job_id: str): - """Get job details + stats.""" - job = db.get_job(job_id) - if not job: - raise HTTPException(status_code=404, detail="Job not found") - stats = db.get_job_stats(job_id) - job.update(stats) - return job - - -@app.post("/api/jobs/{job_id}/pause") -async def pause_job(job_id: str): - job = db.get_job(job_id) - if not job: - raise HTTPException(status_code=404, detail="Job not found") - if job["status"] not in ("running",): - raise HTTPException(status_code=400, detail=f"Cannot pause a job with status: {job['status']}") - worker.pause_job(job_id) - return {"job_id": job_id, "status": "paused"} - - -@app.post("/api/jobs/{job_id}/resume") -async def resume_job(job_id: str, req: ResumeJobRequest = ResumeJobRequest()): - job = db.get_job(job_id) - if not job: - raise HTTPException(status_code=404, detail="Job not found") - - if job["status"] == "paused": - # Job thread is still alive, just unpause it - worker.resume_job(job_id) - return {"job_id": job_id, "status": "running"} - - elif job["status"] in ("stopped", "failed", "done_with_errors"): - # Re-launch the job from where it left off - db.update_job_status(job_id, "pending") - worker.launch_job(job_id, openrouter_api_key=req.openrouter_api_key) - return {"job_id": job_id, "status": "running", "note": "Job re-launched to resume from last checkpoint"} - - else: - raise HTTPException(status_code=400, detail=f"Cannot resume job with status: {job['status']}") - - -@app.post("/api/jobs/{job_id}/stop") -async def stop_job(job_id: str): - job = db.get_job(job_id) - if not job: - raise HTTPException(status_code=404, detail="Job not found") - worker.stop_job(job_id) - return {"job_id": job_id, "status": "stopped"} - - -@app.post("/api/jobs/{job_id}/skip") -async def skip_video(job_id: str): - """Skip the currently processing video.""" - job = db.get_job(job_id) - if not job: - raise HTTPException(status_code=404, detail="Job not found") - worker.skip_video(job_id) - return {"job_id": job_id, "action": "skip_sent"} - - -# ───────────────────────────────────────────────────────────────────────────── -# Videos & Artifacts -# ───────────────────────────────────────────────────────────────────────────── - -@app.get("/api/jobs/{job_id}/videos") -async def get_videos(job_id: str): - videos = db.get_videos(job_id) - if not videos and not db.get_job(job_id): - raise HTTPException(status_code=404, detail="Job not found") - # Enrich with artifact statuses - result = [] - for v in videos: - arts = db.get_artifacts_for_video(job_id, v["video_id"]) - v["artifacts"] = arts - result.append(v) - return result - - -@app.get("/api/jobs/{job_id}/artifacts") -async def get_artifacts(job_id: str): - return db.get_all_artifacts(job_id) - - -# ───────────────────────────────────────────────────────────────────────────── -# Logging -# ───────────────────────────────────────────────────────────────────────────── - -@app.get("/api/jobs/{job_id}/logs") -async def get_logs(job_id: str, limit: int = 500): - if not db.get_job(job_id): - raise HTTPException(status_code=404, detail="Job not found") - return db.get_logs(job_id, limit=limit) - - -@app.get("/api/jobs/{job_id}/logs/csv") -async def download_log_csv(job_id: str): - """Download artifact log as CSV.""" - if not db.get_job(job_id): - raise HTTPException(status_code=404, detail="Job not found") - csv_content = db.get_log_csv(job_id) - return StreamingResponse( - iter([csv_content]), - media_type="text/csv", - headers={"Content-Disposition": f'attachment; filename="job_{job_id[:8]}_log.csv"'}, - ) - - -# ───────────────────────────────────────────────────────────────────────────── -# Server-Sent Events (SSE) for real-time updates -# ───────────────────────────────────────────────────────────────────────────── - -@app.get("/api/jobs/{job_id}/events") -async def job_events(job_id: str, request: Request): - """ - SSE endpoint. Client subscribes and receives real-time events - as the worker processes artifacts. - """ - if not db.get_job(job_id): - raise HTTPException(status_code=404, detail="Job not found") - - queue = worker.emitter.subscribe(job_id) - - async def event_generator(): - try: - # Send initial state snapshot - job = db.get_job(job_id) - stats = db.get_job_stats(job_id) - if job: - job.update(stats) - yield _sse("job_status", {"job_id": job_id, "status": job["status"], "stats": stats}) - - while True: - if await request.is_disconnected(): - break - - # Poll the queue with a short timeout - event = await asyncio.get_event_loop().run_in_executor( - None, lambda: queue.get(timeout=1.0) - ) - - if event: - yield _sse(event["type"], event["data"]) - else: - # Send a keepalive comment - yield ": keepalive\n\n" - - # Check if job is complete - job = db.get_job(job_id) - if job and job["status"] in ("done", "done_with_errors", "stopped", "failed"): - # Give a moment for final events to flush - await asyncio.sleep(0.2) - # Drain remaining queue events - while True: - event = await asyncio.get_event_loop().run_in_executor( - None, lambda: queue.get(timeout=0.1) - ) - if event: - yield _sse(event["type"], event["data"]) - else: - break - yield _sse("stream_end", {"job_id": job_id}) - break - - except asyncio.CancelledError: - pass - finally: - worker.emitter.unsubscribe(job_id, queue) - - return StreamingResponse( - event_generator(), - media_type="text/event-stream", - headers={ - "Cache-Control": "no-cache", - "X-Accel-Buffering": "no", - "Connection": "keep-alive", - }, - ) - - -def _sse(event_type: str, data: dict) -> str: - """Format a Server-Sent Event string.""" - return f"event: {event_type}\ndata: {json.dumps(data)}\n\n" - - -# ───────────────────────────────────────────────────────────────────────────── -# File System Browsing Helper -# ───────────────────────────────────────────────────────────────────────────── - -@app.get("/api/browse") -async def browse_folder(path: str = ""): - """ - Return directory listing for folder picking in the GUI. - Defaults to user home directory. - """ - if not path: - path = str(Path.home()) - - target = Path(path) - if not target.exists() or not target.is_dir(): - # Fallback to home - target = Path.home() - - try: - entries = [] - for entry in sorted(target.iterdir()): - try: - entries.append({ - "name": entry.name, - "path": str(entry.absolute()), - "is_dir": entry.is_dir(), - }) - except PermissionError: - pass - - # Add parent directory option - parent = str(target.parent.absolute()) - return { - "current": str(target.absolute()), - "parent": parent if parent != str(target.absolute()) else None, - "entries": entries, - } - except PermissionError: - raise HTTPException(status_code=403, detail="Permission denied") - - -@app.post("/api/mkdir") -async def make_dir(req: BrowseFolderRequest): - """Create a new directory.""" - if not req.path: - raise HTTPException(status_code=400, detail="path is required") - try: - Path(req.path).mkdir(parents=True, exist_ok=True) - return {"created": req.path} - except Exception as e: - raise HTTPException(status_code=400, detail=str(e)) - - -# ───────────────────────────────────────────────────────────────────────────── -# Config / Settings API -# ───────────────────────────────────────────────────────────────────────────── - -@app.get("/api/config") -async def get_config(): - """Return non-sensitive defaults for the GUI.""" - return { - "available_resolutions": config.AVAILABLE_RESOLUTIONS, - "artifact_types": config.ARTIFACT_TYPES, - "default_resolution": config.DEFAULT_VIDEO_RESOLUTION, - "delay_min": config.DELAY_MIN_SECONDS, - "delay_max": config.DELAY_MAX_SECONDS, - "max_retries": config.MAX_RETRIES, - "default_model": config.OPENROUTER_MODEL, - "summary_prompt": config.SUMMARY_SYSTEM_PROMPT, - "openrouter_base_url": config.OPENROUTER_BASE_URL, - } - - -# ───────────────────────────────────────────────────────────────────────────── -# Health Check -# ───────────────────────────────────────────────────────────────────────────── - -@app.get("/api/health") -async def health(): - return {"status": "ok", "version": "1.0.0"} - - -# ───────────────────────────────────────────────────────────────────────────── -# Entry Point -# ───────────────────────────────────────────────────────────────────────────── - -if __name__ == "__main__": - print(f"\n{'='*60}") - print(" BatchTube") - print(f" Web GUI: http://{config.API_HOST}:{config.API_PORT}") - print(f" API: http://{config.API_HOST}:{config.API_PORT}/api") - print(f"{'='*60}\n") - uvicorn.run( - "api_server:app", - host=config.API_HOST, - port=config.API_PORT, - reload=False, - log_level="info", - ) +""" +api_server.py - FastAPI backend for BatchTube +Serves the Web GUI, REST API, and Server-Sent Events (SSE) for real-time progress. +""" +import sys +import os +from pathlib import Path + +# Ensure project directory is always on sys.path (needed for some Python installations) +_PROJECT_DIR = Path(__file__).parent.resolve() +if str(_PROJECT_DIR) not in sys.path: + sys.path.insert(0, str(_PROJECT_DIR)) + +import json +import asyncio +import logging +from typing import Optional, List + +from fastapi import FastAPI, HTTPException, BackgroundTasks, Request +from fastapi.responses import HTMLResponse, FileResponse, StreamingResponse, JSONResponse +from fastapi.staticfiles import StaticFiles +from fastapi.middleware.cors import CORSMiddleware +from pydantic import BaseModel +import uvicorn + +import database as db +import worker +import config + +logging.basicConfig( + level=logging.INFO, + format="%(asctime)s [%(levelname)s] %(name)s: %(message)s", +) +logger = logging.getLogger("api_server") + +app = FastAPI(title="BatchTube", version="1.0.0") + +# CORS - allow localhost GUI access +app.add_middleware( + CORSMiddleware, + allow_origins=["*"], + allow_methods=["*"], + allow_headers=["*"], +) + +# Serve static web files +WEB_DIR = Path(__file__).parent / "web" + + +# ───────────────────────────────────────────────────────────────────────────── +# Request / Response Models +# ───────────────────────────────────────────────────────────────────────────── + +class CreateJobRequest(BaseModel): + input_type: str # url | playlist | csv | handle + input_value: str # the URL/handle/CSV text + output_folder: str # absolute path to save files + selections: List[str] # artifact types to extract + resolution: str = "best" # for MP4 + delay_min: float = 2.0 + delay_max: float = 7.0 + openrouter_api_key: str = "" + openrouter_model: str = "openai/gpt-4o-mini" + summary_prompt: str = "" + + +class ResumeJobRequest(BaseModel): + openrouter_api_key: str = "" + + +class BrowseFolderRequest(BaseModel): + path: str = "" + + +# ───────────────────────────────────────────────────────────────────────────── +# Web GUI Serving +# ───────────────────────────────────────────────────────────────────────────── + +@app.get("/", response_class=HTMLResponse) +async def serve_gui(): + index_path = WEB_DIR / "index.html" + if not index_path.exists(): + return HTMLResponse("

Web GUI not found. Check the web/ directory.

", status_code=404) + return HTMLResponse(index_path.read_text(encoding="utf-8")) + + +@app.get("/app.js") +async def serve_js(): + return FileResponse(WEB_DIR / "app.js", media_type="application/javascript") + + +@app.get("/style.css") +async def serve_css(): + return FileResponse(WEB_DIR / "style.css", media_type="text/css") + + +# ───────────────────────────────────────────────────────────────────────────── +# Job Management API +# ───────────────────────────────────────────────────────────────────────────── + +@app.post("/api/jobs") +async def create_job(req: CreateJobRequest): + """Create and immediately launch a new extraction job.""" + # Validate output folder + output_folder = req.output_folder.strip() + if not output_folder: + raise HTTPException(status_code=400, detail="output_folder is required") + + # Validate selections + valid_types = set(config.ARTIFACT_TYPES) + invalid = [s for s in req.selections if s not in valid_types] + if invalid: + raise HTTPException(status_code=400, detail=f"Invalid artifact types: {invalid}") + if not req.selections: + raise HTTPException(status_code=400, detail="At least one artifact type must be selected") + + # Validate input + if not req.input_value.strip(): + raise HTTPException(status_code=400, detail="input_value is required") + + + # ── Free Tier Enforcement ───────────────────────────────────────────────── + plan = config.BATCHTUBE_PLAN + limits = config.PLAN_LIMITS.get(plan, config.PLAN_LIMITS["free"]) + + # Cap resolution for free tier + resolution = req.resolution + max_res = limits.get("max_resolution") + if max_res and max_res != "best": + res_order = config.AVAILABLE_RESOLUTIONS # ["best", "4320", ..., "144"] + # Lower index = higher quality; cap if requested res is higher than allowed + def res_index(r): + try: + return res_order.index(r) + except ValueError: + return len(res_order) + if res_index(resolution) < res_index(max_res): + logger.info(f"Free tier: capping resolution from {resolution} to {max_res}") + resolution = max_res + else: + resolution = req.resolution + + # For CSV input: enforce max_videos limit upfront + max_videos = limits.get("max_videos") + if max_videos is not None and req.input_type == "csv": + csv_lines = [ln.strip() for ln in req.input_value.strip().splitlines() if ln.strip()] + if len(csv_lines) > max_videos: + raise HTTPException( + status_code=403, + detail=f"Free tier is limited to {max_videos} videos per batch. " + f"Upgrade to Creator or Pro for larger batches. " + f"({len(csv_lines)} URLs submitted)" + ) + + # Create the folder if it doesn't exist + try: + Path(output_folder).mkdir(parents=True, exist_ok=True) + except Exception as e: + raise HTTPException(status_code=400, detail=f"Cannot create output folder: {e}") + + # Create job in DB + job_id = db.create_job( + output_folder=output_folder, + selections=req.selections, + resolution=resolution, # may be capped by free tier enforcement + delay_min=req.delay_min, + delay_max=req.delay_max, + input_type=req.input_type, + input_value=req.input_value, + summary_prompt=req.summary_prompt or config.SUMMARY_SYSTEM_PROMPT, + openrouter_model=req.openrouter_model or config.OPENROUTER_MODEL, + ) + + # Launch background worker thread + worker.launch_job(job_id, openrouter_api_key=req.openrouter_api_key) + + logger.info(f"Created and launched job {job_id}") + return {"job_id": job_id, "status": "running"} + + +@app.get("/api/jobs") +async def list_jobs(): + """List all jobs (most recent first).""" + jobs = db.list_jobs() + result = [] + for j in jobs: + stats = db.get_job_stats(j["job_id"]) + j.update(stats) + result.append(j) + return result + + +@app.get("/api/jobs/{job_id}") +async def get_job(job_id: str): + """Get job details + stats.""" + job = db.get_job(job_id) + if not job: + raise HTTPException(status_code=404, detail="Job not found") + stats = db.get_job_stats(job_id) + job.update(stats) + return job + + +@app.post("/api/jobs/{job_id}/pause") +async def pause_job(job_id: str): + job = db.get_job(job_id) + if not job: + raise HTTPException(status_code=404, detail="Job not found") + if job["status"] not in ("running",): + raise HTTPException(status_code=400, detail=f"Cannot pause a job with status: {job['status']}") + worker.pause_job(job_id) + return {"job_id": job_id, "status": "paused"} + + +@app.post("/api/jobs/{job_id}/resume") +async def resume_job(job_id: str, req: ResumeJobRequest = ResumeJobRequest()): + job = db.get_job(job_id) + if not job: + raise HTTPException(status_code=404, detail="Job not found") + + if job["status"] == "paused": + # Job thread is still alive, just unpause it + worker.resume_job(job_id) + return {"job_id": job_id, "status": "running"} + + elif job["status"] in ("stopped", "failed", "done_with_errors"): + # Re-launch the job from where it left off + db.update_job_status(job_id, "pending") + worker.launch_job(job_id, openrouter_api_key=req.openrouter_api_key) + return {"job_id": job_id, "status": "running", "note": "Job re-launched to resume from last checkpoint"} + + else: + raise HTTPException(status_code=400, detail=f"Cannot resume job with status: {job['status']}") + + +@app.post("/api/jobs/{job_id}/stop") +async def stop_job(job_id: str): + job = db.get_job(job_id) + if not job: + raise HTTPException(status_code=404, detail="Job not found") + worker.stop_job(job_id) + return {"job_id": job_id, "status": "stopped"} + + +@app.post("/api/jobs/{job_id}/skip") +async def skip_video(job_id: str): + """Skip the currently processing video.""" + job = db.get_job(job_id) + if not job: + raise HTTPException(status_code=404, detail="Job not found") + worker.skip_video(job_id) + return {"job_id": job_id, "action": "skip_sent"} + + +# ───────────────────────────────────────────────────────────────────────────── +# Videos & Artifacts +# ───────────────────────────────────────────────────────────────────────────── + +@app.get("/api/jobs/{job_id}/videos") +async def get_videos(job_id: str): + videos = db.get_videos(job_id) + if not videos and not db.get_job(job_id): + raise HTTPException(status_code=404, detail="Job not found") + # Enrich with artifact statuses + result = [] + for v in videos: + arts = db.get_artifacts_for_video(job_id, v["video_id"]) + v["artifacts"] = arts + result.append(v) + return result + + +@app.get("/api/jobs/{job_id}/artifacts") +async def get_artifacts(job_id: str): + return db.get_all_artifacts(job_id) + + +# ───────────────────────────────────────────────────────────────────────────── +# Logging +# ───────────────────────────────────────────────────────────────────────────── + +@app.get("/api/jobs/{job_id}/logs") +async def get_logs(job_id: str, limit: int = 500): + if not db.get_job(job_id): + raise HTTPException(status_code=404, detail="Job not found") + return db.get_logs(job_id, limit=limit) + + +@app.get("/api/jobs/{job_id}/logs/csv") +async def download_log_csv(job_id: str): + """Download artifact log as CSV.""" + if not db.get_job(job_id): + raise HTTPException(status_code=404, detail="Job not found") + csv_content = db.get_log_csv(job_id) + return StreamingResponse( + iter([csv_content]), + media_type="text/csv", + headers={"Content-Disposition": f'attachment; filename="job_{job_id[:8]}_log.csv"'}, + ) + + +# ───────────────────────────────────────────────────────────────────────────── +# Server-Sent Events (SSE) for real-time updates +# ───────────────────────────────────────────────────────────────────────────── + +@app.get("/api/jobs/{job_id}/events") +async def job_events(job_id: str, request: Request): + """ + SSE endpoint. Client subscribes and receives real-time events + as the worker processes artifacts. + """ + if not db.get_job(job_id): + raise HTTPException(status_code=404, detail="Job not found") + + queue = worker.emitter.subscribe(job_id) + + async def event_generator(): + try: + # Send initial state snapshot + job = db.get_job(job_id) + stats = db.get_job_stats(job_id) + if job: + job.update(stats) + yield _sse("job_status", {"job_id": job_id, "status": job["status"], "stats": stats}) + + while True: + if await request.is_disconnected(): + break + + # Poll the queue with a short timeout + event = await asyncio.get_event_loop().run_in_executor( + None, lambda: queue.get(timeout=1.0) + ) + + if event: + yield _sse(event["type"], event["data"]) + else: + # Send a keepalive comment + yield ": keepalive\n\n" + + # Check if job is complete + job = db.get_job(job_id) + if job and job["status"] in ("done", "done_with_errors", "stopped", "failed"): + # Give a moment for final events to flush + await asyncio.sleep(0.2) + # Drain remaining queue events + while True: + event = await asyncio.get_event_loop().run_in_executor( + None, lambda: queue.get(timeout=0.1) + ) + if event: + yield _sse(event["type"], event["data"]) + else: + break + yield _sse("stream_end", {"job_id": job_id}) + break + + except asyncio.CancelledError: + pass + finally: + worker.emitter.unsubscribe(job_id, queue) + + return StreamingResponse( + event_generator(), + media_type="text/event-stream", + headers={ + "Cache-Control": "no-cache", + "X-Accel-Buffering": "no", + "Connection": "keep-alive", + }, + ) + + +def _sse(event_type: str, data: dict) -> str: + """Format a Server-Sent Event string.""" + return f"event: {event_type}\ndata: {json.dumps(data)}\n\n" + + +# ───────────────────────────────────────────────────────────────────────────── +# File System Browsing Helper +# ───────────────────────────────────────────────────────────────────────────── + +@app.get("/api/browse") +async def browse_folder(path: str = ""): + """ + Return directory listing for folder picking in the GUI. + Defaults to user home directory. + """ + if not path: + path = str(Path.home()) + + target = Path(path) + if not target.exists() or not target.is_dir(): + # Fallback to home + target = Path.home() + + try: + entries = [] + for entry in sorted(target.iterdir()): + try: + entries.append({ + "name": entry.name, + "path": str(entry.absolute()), + "is_dir": entry.is_dir(), + }) + except PermissionError: + pass + + # Add parent directory option + parent = str(target.parent.absolute()) + return { + "current": str(target.absolute()), + "parent": parent if parent != str(target.absolute()) else None, + "entries": entries, + } + except PermissionError: + raise HTTPException(status_code=403, detail="Permission denied") + + +@app.post("/api/mkdir") +async def make_dir(req: BrowseFolderRequest): + """Create a new directory.""" + if not req.path: + raise HTTPException(status_code=400, detail="path is required") + try: + Path(req.path).mkdir(parents=True, exist_ok=True) + return {"created": req.path} + except Exception as e: + raise HTTPException(status_code=400, detail=str(e)) + + +# ───────────────────────────────────────────────────────────────────────────── +# Config / Settings API +# ───────────────────────────────────────────────────────────────────────────── + +@app.get("/api/config") +async def get_config(): + """Return non-sensitive defaults for the GUI.""" + return { + "available_resolutions": config.AVAILABLE_RESOLUTIONS, + "artifact_types": config.ARTIFACT_TYPES, + "default_resolution": config.DEFAULT_VIDEO_RESOLUTION, + "delay_min": config.DELAY_MIN_SECONDS, + "delay_max": config.DELAY_MAX_SECONDS, + "max_retries": config.MAX_RETRIES, + "default_model": config.OPENROUTER_MODEL, + "summary_prompt": config.SUMMARY_SYSTEM_PROMPT, + "openrouter_base_url": config.OPENROUTER_BASE_URL, + } + + +# ───────────────────────────────────────────────────────────────────────────── +# Health Check +# ───────────────────────────────────────────────────────────────────────────── + +@app.get("/api/health") +async def health(): + return {"status": "ok", "version": "1.0.0"} + + +# ───────────────────────────────────────────────────────────────────────────── +# Entry Point +# ───────────────────────────────────────────────────────────────────────────── + +if __name__ == "__main__": + print(f"\n{'='*60}") + print(" BatchTube") + print(f" Web GUI: http://{config.API_HOST}:{config.API_PORT}") + print(f" API: http://{config.API_HOST}:{config.API_PORT}/api") + print(f"{'='*60}\n") + uvicorn.run( + "api_server:app", + host=config.API_HOST, + port=config.API_PORT, + reload=False, + log_level="info", + ) diff --git a/backend/config.py b/backend/config.py index 91b0d03..d98f66d 100644 --- a/backend/config.py +++ b/backend/config.py @@ -1,5 +1,5 @@ """ -config.py - Default configuration for YouTube SIPOC Tool +config.py - Default configuration for BatchTube """ import os @@ -32,7 +32,40 @@ SUMMARY_MAX_TOKENS = 1024 MAX_TITLE_LENGTH = 80 # characters - truncate long titles in filenames # ── Database ────────────────────────────────────────────────────────────────── -DB_PATH = os.environ.get("BATCHTUBE_DB_PATH", os.path.join(os.path.dirname(__file__), "youtube_sipoc.db")) +# Windows-safe: allow override via env var (avoids path issues in PyInstaller bundles) +DB_PATH = os.environ.get( + "BATCHTUBE_DB_PATH", + os.path.join(os.path.dirname(__file__), "youtube_sipoc.db") +) # ── Artifact Types ──────────────────────────────────────────────────────────── ARTIFACT_TYPES = ["title", "stats", "transcript", "summary", "thumbnail", "mp3", "mp4"] + +# ── Licensing / Plan Tiers ──────────────────────────────────────────────────── +# Plan is set via environment variable BATCHTUBE_PLAN. +# Valid values: free | creator | pro | studio | lifetime +# Defaults to 'free' — license validation layer overrides this at runtime. +BATCHTUBE_PLAN = os.environ.get("BATCHTUBE_PLAN", "free").lower() + +# ── Free Tier Limits (updated 2026-03-07) ───────────────────────────────────── +# Free tier: 3 videos/batch max (tightened from 5), 720p max resolution +FREE_TIER_MAX_VIDEOS = 3 # max videos per batch job on free plan +FREE_TIER_MAX_RESOLUTION = "720" # max resolution on free plan (was 1080p) + +# ── Plan Limits Map ─────────────────────────────────────────────────────────── +# Maps plan name -> {max_videos, max_resolution}; None = unlimited +PLAN_LIMITS = { + "free": {"max_videos": FREE_TIER_MAX_VIDEOS, "max_resolution": FREE_TIER_MAX_RESOLUTION}, + "creator": {"max_videos": 100, "max_resolution": "1080"}, + "pro": {"max_videos": None, "max_resolution": "best"}, + "studio": {"max_videos": None, "max_resolution": "best"}, + "lifetime": {"max_videos": None, "max_resolution": "best"}, +} + +# ── Pricing Reference (2026-03-07 approved) ─────────────────────────────────── +# Free: 3 videos/batch, 720p max +# Creator: $19/mo or $149/yr — 100 videos/batch, 1080p +# Pro: $39/mo or $299/yr — unlimited, 4K +# Studio: $79/mo — 3 seats, API access +# Lifetime: $197 one-time — unlimited, all features +# AppSumo: $79 one-time — limited launch deal diff --git a/backend/worker.py b/backend/worker.py index 42d72f8..bbd2da7 100644 --- a/backend/worker.py +++ b/backend/worker.py @@ -1,507 +1,528 @@ -""" -worker.py - Background job pipeline for YouTube SIPOC Tool -Handles pause/stop/resume, per-artifact saving, duplicate skipping, -random delays, retry logic, and resume-on-failure. -""" -import time -import random -import logging -import threading -from typing import Dict, Callable, Optional - -import database as db -import extractors -import config - -logger = logging.getLogger("worker") - -# ───────────────────────────────────────────────────────────────────────────── -# Job Control Registry -# Stores threading.Event objects per job_id for pause/stop/skip signals. -# ───────────────────────────────────────────────────────────────────────────── - -_job_controls: Dict[str, Dict] = {} -_registry_lock = threading.Lock() - - -def _get_controls(job_id: str) -> Dict: - with _registry_lock: - if job_id not in _job_controls: - _job_controls[job_id] = { - "pause": threading.Event(), # set = paused - "stop": threading.Event(), # set = stop requested - "skip": threading.Event(), # set = skip current video - } - return _job_controls[job_id] - - -def _cleanup_controls(job_id: str): - with _registry_lock: - _job_controls.pop(job_id, None) - - -def pause_job(job_id: str): - _get_controls(job_id)["pause"].set() - db.update_job_status(job_id, "paused") - db.add_log(job_id, "Job paused by user.", level="INFO") - - -def resume_job(job_id: str): - _get_controls(job_id)["pause"].clear() - db.update_job_status(job_id, "running") - db.add_log(job_id, "Job resumed by user.", level="INFO") - - -def stop_job(job_id: str): - _get_controls(job_id)["stop"].set() - _get_controls(job_id)["pause"].clear() # Unpause so thread can exit - db.update_job_status(job_id, "stopped") - db.add_log(job_id, "Job stopped by user.", level="INFO") - - -def skip_video(job_id: str): - """Signal the worker to skip the current video and move to the next.""" - _get_controls(job_id)["skip"].set() - db.add_log(job_id, "Skip requested for current video.", level="INFO") - - -# ───────────────────────────────────────────────────────────────────────────── -# SSE Event Emitter -# ───────────────────────────────────────────────────────────────────────────── - -class EventEmitter: - """Thread-safe event queue for Server-Sent Events.""" - - def __init__(self): - self._listeners: Dict[str, list] = {} - self._lock = threading.Lock() - - def subscribe(self, job_id: str) -> "EventQueue": - queue = EventQueue() - with self._lock: - if job_id not in self._listeners: - self._listeners[job_id] = [] - self._listeners[job_id].append(queue) - return queue - - def unsubscribe(self, job_id: str, queue: "EventQueue"): - with self._lock: - if job_id in self._listeners: - try: - self._listeners[job_id].remove(queue) - except ValueError: - pass - - def emit(self, job_id: str, event_type: str, data: dict): - with self._lock: - listeners = list(self._listeners.get(job_id, [])) - for q in listeners: - q.put(event_type, data) - - -class EventQueue: - """Per-subscriber event queue with thread-safe get.""" - - def __init__(self): - import queue - self._q = queue.Queue(maxsize=500) - - def put(self, event_type: str, data: dict): - try: - self._q.put_nowait({"type": event_type, "data": data}) - except Exception: - pass # Drop events if queue is full - - def get(self, timeout: float = 30.0): - try: - return self._q.get(timeout=timeout) - except Exception: - return None - - -# Global emitter instance -emitter = EventEmitter() - - -# ───────────────────────────────────────────────────────────────────────────── -# Worker Thread Entry Point -# ───────────────────────────────────────────────────────────────────────────── - -def run_job(job_id: str): - """ - Main worker function. Runs in a background thread. - Resolves inputs, iterates videos, processes each artifact, - emits SSE events, handles pause/stop/skip/retry. - """ - controls = _get_controls(job_id) - job = db.get_job(job_id) - if not job: - logger.error(f"Job {job_id} not found in database.") - return - - db.update_job_status(job_id, "running") - _emit(job_id, "job_status", {"job_id": job_id, "status": "running"}) - db.add_log(job_id, "Job started.", level="INFO") - - try: - # ── Step 1: Resolve input to video list ────────────────────────────── - _emit(job_id, "progress", {"message": "Resolving input URLs…"}) - db.add_log(job_id, f"Resolving input: type={job['input_type']}", level="INFO") - - try: - videos = extractors.resolve_input(job["input_type"], job["input_value"]) - except Exception as e: - db.add_log(job_id, f"Failed to resolve input: {e}", level="ERROR") - db.update_job_status(job_id, "failed") - _emit(job_id, "job_status", {"job_id": job_id, "status": "failed", "error": str(e)}) - return - - if not videos: - db.add_log(job_id, "No videos found for input.", level="WARN") - db.update_job_status(job_id, "done") - _emit(job_id, "job_status", {"job_id": job_id, "status": "done"}) - return - - # Deduplicate by video_id (preserve order) - seen = set() - unique_videos = [] - for v in videos: - if v["video_id"] not in seen: - seen.add(v["video_id"]) - unique_videos.append(v) - videos = unique_videos - - # ── Step 2: Register videos in DB ──────────────────────────────────── - for i, v in enumerate(videos): - db.upsert_video(job_id, v["video_id"], v["url"], v.get("title", ""), i) - db.update_job_progress(job_id, len(videos), 0) - _emit(job_id, "video_list", { - "job_id": job_id, - "total": len(videos), - "videos": [{"video_id": v["video_id"], "title": v.get("title", ""), "position": i} - for i, v in enumerate(videos)] - }) - db.add_log(job_id, f"Found {len(videos)} unique videos.", level="INFO") - - # ── Step 3: Process each video ──────────────────────────────────────── - selections = job["selections"] - done_count = 0 - - for idx, video in enumerate(videos): - # Check stop - if controls["stop"].is_set(): - db.add_log(job_id, "Job stopped before processing video.", level="INFO") - break - - # Handle pause (blocking) - _wait_if_paused(controls) - if controls["stop"].is_set(): - break - - # Reset skip flag for this video - controls["skip"].clear() - - video_id = video["video_id"] - db.add_log(job_id, f"Processing video {idx+1}/{len(videos)}: {video_id}", level="INFO") - _emit(job_id, "video_start", {"job_id": job_id, "video_id": video_id, "position": idx}) - - # ── Step 3a: Fetch metadata (once per video) ────────────────── - meta = {} - try: - _emit(job_id, "artifact_start", { - "job_id": job_id, "video_id": video_id, "artifact_type": "metadata" - }) - meta = extractors.fetch_metadata(video_id) - title = meta.get("title", "") - if title: - db.update_video_title(job_id, video_id, title) - _emit(job_id, "video_title", {"job_id": job_id, "video_id": video_id, "title": title}) - db.add_log(job_id, f"Metadata fetched for {video_id}: {title or '(no title)'}", level="INFO") - except Exception as e: - db.add_log(job_id, f"Failed to fetch metadata for {video_id}: {e}", level="ERROR") - _emit(job_id, "video_error", {"job_id": job_id, "video_id": video_id, "error": str(e)}) - db.update_video_status(job_id, video_id, "failed") - continue - - # ── Step 3b: Initialize artifact records ────────────────────── - for artifact_type in selections: - db.upsert_artifact(job_id, video_id, artifact_type) - - # ── Step 3c: Process each selected artifact ─────────────────── - video_failed = False - for artifact_type in selections: - # Check stop or skip - if controls["stop"].is_set(): - break - if controls["skip"].is_set(): - db.add_log(job_id, f"Skipping remaining artifacts for {video_id} (skip requested).", level="INFO") - break - - # Handle pause - _wait_if_paused(controls) - if controls["stop"].is_set(): - break - - # Skip if already done - if db.artifact_is_done(job_id, video_id, artifact_type): - db.add_log(job_id, f"[SKIP-DUP] {artifact_type} for {video_id} already done.", level="INFO") - _emit(job_id, "artifact_skip", { - "job_id": job_id, "video_id": video_id, "artifact_type": artifact_type, - "reason": "already_done" - }) - continue - - # Process with retry - success = _process_artifact_with_retry( - job_id=job_id, - video_id=video_id, - artifact_type=artifact_type, - meta=meta, - job=job, - controls=controls, - ) - if not success: - video_failed = True - - # Random delay between artifacts (bot avoidance) - if not controls["stop"].is_set() and not controls["skip"].is_set(): - delay = random.uniform(job["delay_min"], job["delay_max"]) - db.add_log(job_id, f"Waiting {delay:.1f}s before next artifact…", level="DEBUG") - _interruptible_sleep(delay, controls) - - # ── Step 3d: Mark video done/failed ────────────────────────── - if not controls["stop"].is_set(): - status = "failed" if video_failed else "done" - db.update_video_status(job_id, video_id, status) - done_count += 1 - db.update_job_progress(job_id, len(videos), done_count) - _emit(job_id, "video_done", { - "job_id": job_id, - "video_id": video_id, - "status": status, - "done": done_count, - "total": len(videos), - }) - - # Random delay between videos - if idx < len(videos) - 1 and not controls["stop"].is_set(): - delay = random.uniform(job["delay_min"], job["delay_max"]) - db.add_log(job_id, f"Waiting {delay:.1f}s before next video…", level="DEBUG") - _interruptible_sleep(delay, controls) - - # ── Step 4: Finalize job ────────────────────────────────────────────── - if controls["stop"].is_set(): - final_status = "stopped" - else: - stats = db.get_job_stats(job_id) - final_status = "done" if stats["artifact_failed"] == 0 else "done_with_errors" - - db.update_job_status(job_id, final_status) - stats = db.get_job_stats(job_id) - db.add_log(job_id, f"Job finished with status: {final_status}. Stats: {stats}", level="INFO") - _emit(job_id, "job_status", {"job_id": job_id, "status": final_status, "stats": stats}) - - except Exception as e: - logger.exception(f"Unexpected error in job {job_id}: {e}") - db.add_log(job_id, f"Unexpected error: {e}", level="ERROR") - db.update_job_status(job_id, "failed") - _emit(job_id, "job_status", {"job_id": job_id, "status": "failed", "error": str(e)}) - finally: - _cleanup_controls(job_id) - - -# ───────────────────────────────────────────────────────────────────────────── -# Artifact Processing with Retry -# ───────────────────────────────────────────────────────────────────────────── - -def _process_artifact_with_retry( - job_id: str, - video_id: str, - artifact_type: str, - meta: Dict, - job: Dict, - controls: Dict, -) -> bool: - """ - Attempt to process one artifact, retrying up to MAX_RETRIES times. - Returns True on success, False on final failure. - """ - max_retries = config.MAX_RETRIES - output_folder = job["output_folder"] - resolution = job.get("resolution", "best") - api_key = job.get("openrouter_model", "") # model field - # We store the API key in the job separately; retrieve from job dict - # (api_key is set in the job via the POST body - see api_server.py) - openrouter_api_key = job.get("_openrouter_api_key", "") # injected by api_server - openrouter_model = job.get("openrouter_model", config.OPENROUTER_MODEL) - summary_prompt = job.get("summary_prompt", config.SUMMARY_SYSTEM_PROMPT) - - for attempt in range(1, max_retries + 1): - if controls["stop"].is_set() or controls["skip"].is_set(): - return False - - db.mark_artifact_started(job_id, video_id, artifact_type) - _emit(job_id, "artifact_start", { - "job_id": job_id, - "video_id": video_id, - "artifact_type": artifact_type, - "attempt": attempt, - }) - db.add_log(job_id, f"[{artifact_type}] Starting (attempt {attempt}/{max_retries})", level="INFO", - video_id=video_id, artifact_type=artifact_type) - - try: - file_path = _run_extractor( - artifact_type=artifact_type, - video_id=video_id, - meta=meta, - output_folder=output_folder, - resolution=resolution, - openrouter_api_key=openrouter_api_key, - openrouter_model=openrouter_model, - summary_prompt=summary_prompt, - ) - db.update_artifact_status(job_id, video_id, artifact_type, "done", file_path=file_path) - db.add_log(job_id, f"[{artifact_type}] Done → {file_path}", level="INFO", - video_id=video_id, artifact_type=artifact_type) - _emit(job_id, "artifact_done", { - "job_id": job_id, - "video_id": video_id, - "artifact_type": artifact_type, - "file_path": file_path, - }) - return True - - except Exception as e: - err_msg = str(e) - logger.warning(f"[{job_id}][{video_id}][{artifact_type}] Attempt {attempt} failed: {e}") - db.add_log(job_id, f"[{artifact_type}] Attempt {attempt} failed: {e}", level="WARN", - video_id=video_id, artifact_type=artifact_type) - - if attempt < max_retries: - # Exponential backoff between retries - backoff = min(2 ** attempt * 3, 30) + random.uniform(0, 2) - db.add_log(job_id, f"[{artifact_type}] Retry in {backoff:.1f}s…", level="INFO", - video_id=video_id, artifact_type=artifact_type) - _interruptible_sleep(backoff, controls) - else: - db.update_artifact_status( - job_id, video_id, artifact_type, "failed", error_msg=err_msg - ) - db.add_log(job_id, f"[{artifact_type}] FAILED after {max_retries} attempts: {e}", level="ERROR", - video_id=video_id, artifact_type=artifact_type) - _emit(job_id, "artifact_failed", { - "job_id": job_id, - "video_id": video_id, - "artifact_type": artifact_type, - "error": err_msg, - "attempts": max_retries, - }) - return False - - return False - - -def _run_extractor( - artifact_type: str, - video_id: str, - meta: Dict, - output_folder: str, - resolution: str, - openrouter_api_key: str, - openrouter_model: str, - summary_prompt: str, -) -> str: - """Dispatch to the correct extractor function.""" - if artifact_type == "title": - file_path, _ = extractors.extract_title(video_id, meta, output_folder) - return file_path - elif artifact_type == "stats": - return extractors.extract_stats(video_id, meta, output_folder) - elif artifact_type == "transcript": - return extractors.extract_transcript(video_id, meta, output_folder) - elif artifact_type == "summary": - return extractors.extract_summary( - video_id, meta, output_folder, - api_key=openrouter_api_key, - model=openrouter_model, - system_prompt=summary_prompt, - ) - elif artifact_type == "thumbnail": - return extractors.extract_thumbnail(video_id, meta, output_folder) - elif artifact_type == "mp3": - return extractors.extract_mp3(video_id, meta, output_folder) - elif artifact_type == "mp4": - return extractors.extract_mp4(video_id, meta, output_folder, resolution=resolution) - else: - raise ValueError(f"Unknown artifact type: {artifact_type}") - - -# ───────────────────────────────────────────────────────────────────────────── -# Helpers -# ───────────────────────────────────────────────────────────────────────────── - -def _emit(job_id: str, event_type: str, data: dict): - """Emit an SSE event.""" - emitter.emit(job_id, event_type, data) - - -def _wait_if_paused(controls: Dict): - """Block worker thread while paused, check every 0.5s.""" - while controls["pause"].is_set() and not controls["stop"].is_set(): - time.sleep(0.5) - - -def _interruptible_sleep(seconds: float, controls: Dict): - """Sleep for `seconds` but wake early if stop/skip is set.""" - end = time.monotonic() + seconds - while time.monotonic() < end: - if controls["stop"].is_set() or controls["skip"].is_set(): - return - time.sleep(min(0.3, end - time.monotonic())) - - -# ───────────────────────────────────────────────────────────────────────────── -# Job Launcher -# ───────────────────────────────────────────────────────────────────────────── - -def launch_job(job_id: str, openrouter_api_key: str = ""): - """ - Launch job in a daemon background thread. - openrouter_api_key is injected at runtime (not stored in DB for security). - """ - # Inject API key into job dict via a side channel stored in controls - controls = _get_controls(job_id) - controls["_openrouter_api_key"] = openrouter_api_key - - # Patch the run_job function to use the injected key - def _patched_run(): - job = db.get_job(job_id) - if job: - job["_openrouter_api_key"] = openrouter_api_key - run_job_with_key(job_id, openrouter_api_key) - - t = threading.Thread(target=_patched_run, daemon=True, name=f"job-{job_id[:8]}") - t.start() - return t - - -def run_job_with_key(job_id: str, openrouter_api_key: str): - """Wrapper that injects the API key into job processing.""" - # Monkey-patch the job dict to include the API key during this run - original_get_job = db.get_job - - def patched_get_job(jid): - result = original_get_job(jid) - if result and jid == job_id: - result["_openrouter_api_key"] = openrouter_api_key - return result - - # Temporarily patch db.get_job for this thread - import database - _orig = database.get_job - database.get_job = patched_get_job - try: - run_job(job_id) - finally: - database.get_job = _orig +""" +worker.py - Background job pipeline for YouTube SIPOC Tool +Handles pause/stop/resume, per-artifact saving, duplicate skipping, +random delays, retry logic, and resume-on-failure. +""" +import time +import random +import logging +import threading +from typing import Dict, Callable, Optional + +import database as db +import extractors +import config + +logger = logging.getLogger("worker") + +# ───────────────────────────────────────────────────────────────────────────── +# Job Control Registry +# Stores threading.Event objects per job_id for pause/stop/skip signals. +# ───────────────────────────────────────────────────────────────────────────── + +_job_controls: Dict[str, Dict] = {} +_registry_lock = threading.Lock() + + +def _get_controls(job_id: str) -> Dict: + with _registry_lock: + if job_id not in _job_controls: + _job_controls[job_id] = { + "pause": threading.Event(), # set = paused + "stop": threading.Event(), # set = stop requested + "skip": threading.Event(), # set = skip current video + } + return _job_controls[job_id] + + +def _cleanup_controls(job_id: str): + with _registry_lock: + _job_controls.pop(job_id, None) + + +def pause_job(job_id: str): + _get_controls(job_id)["pause"].set() + db.update_job_status(job_id, "paused") + db.add_log(job_id, "Job paused by user.", level="INFO") + + +def resume_job(job_id: str): + _get_controls(job_id)["pause"].clear() + db.update_job_status(job_id, "running") + db.add_log(job_id, "Job resumed by user.", level="INFO") + + +def stop_job(job_id: str): + _get_controls(job_id)["stop"].set() + _get_controls(job_id)["pause"].clear() # Unpause so thread can exit + db.update_job_status(job_id, "stopped") + db.add_log(job_id, "Job stopped by user.", level="INFO") + + +def skip_video(job_id: str): + """Signal the worker to skip the current video and move to the next.""" + _get_controls(job_id)["skip"].set() + db.add_log(job_id, "Skip requested for current video.", level="INFO") + + +# ───────────────────────────────────────────────────────────────────────────── +# SSE Event Emitter +# ───────────────────────────────────────────────────────────────────────────── + +class EventEmitter: + """Thread-safe event queue for Server-Sent Events.""" + + def __init__(self): + self._listeners: Dict[str, list] = {} + self._lock = threading.Lock() + + def subscribe(self, job_id: str) -> "EventQueue": + queue = EventQueue() + with self._lock: + if job_id not in self._listeners: + self._listeners[job_id] = [] + self._listeners[job_id].append(queue) + return queue + + def unsubscribe(self, job_id: str, queue: "EventQueue"): + with self._lock: + if job_id in self._listeners: + try: + self._listeners[job_id].remove(queue) + except ValueError: + pass + + def emit(self, job_id: str, event_type: str, data: dict): + with self._lock: + listeners = list(self._listeners.get(job_id, [])) + for q in listeners: + q.put(event_type, data) + + +class EventQueue: + """Per-subscriber event queue with thread-safe get.""" + + def __init__(self): + import queue + self._q = queue.Queue(maxsize=500) + + def put(self, event_type: str, data: dict): + try: + self._q.put_nowait({"type": event_type, "data": data}) + except Exception: + pass # Drop events if queue is full + + def get(self, timeout: float = 30.0): + try: + return self._q.get(timeout=timeout) + except Exception: + return None + + +# Global emitter instance +emitter = EventEmitter() + + +# ───────────────────────────────────────────────────────────────────────────── +# Worker Thread Entry Point +# ───────────────────────────────────────────────────────────────────────────── + +def run_job(job_id: str): + """ + Main worker function. Runs in a background thread. + Resolves inputs, iterates videos, processes each artifact, + emits SSE events, handles pause/stop/skip/retry. + """ + controls = _get_controls(job_id) + job = db.get_job(job_id) + if not job: + logger.error(f"Job {job_id} not found in database.") + return + + db.update_job_status(job_id, "running") + _emit(job_id, "job_status", {"job_id": job_id, "status": "running"}) + db.add_log(job_id, "Job started.", level="INFO") + + try: + # ── Step 1: Resolve input to video list ────────────────────────────── + _emit(job_id, "progress", {"message": "Resolving input URLs…"}) + db.add_log(job_id, f"Resolving input: type={job['input_type']}", level="INFO") + + try: + videos = extractors.resolve_input(job["input_type"], job["input_value"]) + except Exception as e: + db.add_log(job_id, f"Failed to resolve input: {e}", level="ERROR") + db.update_job_status(job_id, "failed") + _emit(job_id, "job_status", {"job_id": job_id, "status": "failed", "error": str(e)}) + return + + if not videos: + db.add_log(job_id, "No videos found for input.", level="WARN") + db.update_job_status(job_id, "done") + _emit(job_id, "job_status", {"job_id": job_id, "status": "done"}) + return + + # Deduplicate by video_id (preserve order) + seen = set() + unique_videos = [] + for v in videos: + if v["video_id"] not in seen: + seen.add(v["video_id"]) + unique_videos.append(v) + videos = unique_videos + + # ── Free Tier: Enforce max videos per batch ─────────────────────────── + plan = config.BATCHTUBE_PLAN + limits = config.PLAN_LIMITS.get(plan, config.PLAN_LIMITS["free"]) + max_videos = limits.get("max_videos") + if max_videos is not None and len(videos) > max_videos: + db.add_log( + job_id, + f"Free tier limit: truncating batch from {len(videos)} to {max_videos} videos. " + f"Upgrade to Creator or Pro for larger batches.", + level="WARN" + ) + _emit(job_id, "tier_limit", { + "job_id": job_id, + "plan": plan, + "requested": len(videos), + "allowed": max_videos, + "message": f"Free tier: batch limited to {max_videos} videos. Upgrade for more." + }) + videos = videos[:max_videos] + + + # ── Step 2: Register videos in DB ──────────────────────────────────── + for i, v in enumerate(videos): + db.upsert_video(job_id, v["video_id"], v["url"], v.get("title", ""), i) + db.update_job_progress(job_id, len(videos), 0) + _emit(job_id, "video_list", { + "job_id": job_id, + "total": len(videos), + "videos": [{"video_id": v["video_id"], "title": v.get("title", ""), "position": i} + for i, v in enumerate(videos)] + }) + db.add_log(job_id, f"Found {len(videos)} unique videos.", level="INFO") + + # ── Step 3: Process each video ──────────────────────────────────────── + selections = job["selections"] + done_count = 0 + + for idx, video in enumerate(videos): + # Check stop + if controls["stop"].is_set(): + db.add_log(job_id, "Job stopped before processing video.", level="INFO") + break + + # Handle pause (blocking) + _wait_if_paused(controls) + if controls["stop"].is_set(): + break + + # Reset skip flag for this video + controls["skip"].clear() + + video_id = video["video_id"] + db.add_log(job_id, f"Processing video {idx+1}/{len(videos)}: {video_id}", level="INFO") + _emit(job_id, "video_start", {"job_id": job_id, "video_id": video_id, "position": idx}) + + # ── Step 3a: Fetch metadata (once per video) ────────────────── + meta = {} + try: + _emit(job_id, "artifact_start", { + "job_id": job_id, "video_id": video_id, "artifact_type": "metadata" + }) + meta = extractors.fetch_metadata(video_id) + title = meta.get("title", "") + if title: + db.update_video_title(job_id, video_id, title) + _emit(job_id, "video_title", {"job_id": job_id, "video_id": video_id, "title": title}) + db.add_log(job_id, f"Metadata fetched for {video_id}: {title or '(no title)'}", level="INFO") + except Exception as e: + db.add_log(job_id, f"Failed to fetch metadata for {video_id}: {e}", level="ERROR") + _emit(job_id, "video_error", {"job_id": job_id, "video_id": video_id, "error": str(e)}) + db.update_video_status(job_id, video_id, "failed") + continue + + # ── Step 3b: Initialize artifact records ────────────────────── + for artifact_type in selections: + db.upsert_artifact(job_id, video_id, artifact_type) + + # ── Step 3c: Process each selected artifact ─────────────────── + video_failed = False + for artifact_type in selections: + # Check stop or skip + if controls["stop"].is_set(): + break + if controls["skip"].is_set(): + db.add_log(job_id, f"Skipping remaining artifacts for {video_id} (skip requested).", level="INFO") + break + + # Handle pause + _wait_if_paused(controls) + if controls["stop"].is_set(): + break + + # Skip if already done + if db.artifact_is_done(job_id, video_id, artifact_type): + db.add_log(job_id, f"[SKIP-DUP] {artifact_type} for {video_id} already done.", level="INFO") + _emit(job_id, "artifact_skip", { + "job_id": job_id, "video_id": video_id, "artifact_type": artifact_type, + "reason": "already_done" + }) + continue + + # Process with retry + success = _process_artifact_with_retry( + job_id=job_id, + video_id=video_id, + artifact_type=artifact_type, + meta=meta, + job=job, + controls=controls, + ) + if not success: + video_failed = True + + # Random delay between artifacts (bot avoidance) + if not controls["stop"].is_set() and not controls["skip"].is_set(): + delay = random.uniform(job["delay_min"], job["delay_max"]) + db.add_log(job_id, f"Waiting {delay:.1f}s before next artifact…", level="DEBUG") + _interruptible_sleep(delay, controls) + + # ── Step 3d: Mark video done/failed ────────────────────────── + if not controls["stop"].is_set(): + status = "failed" if video_failed else "done" + db.update_video_status(job_id, video_id, status) + done_count += 1 + db.update_job_progress(job_id, len(videos), done_count) + _emit(job_id, "video_done", { + "job_id": job_id, + "video_id": video_id, + "status": status, + "done": done_count, + "total": len(videos), + }) + + # Random delay between videos + if idx < len(videos) - 1 and not controls["stop"].is_set(): + delay = random.uniform(job["delay_min"], job["delay_max"]) + db.add_log(job_id, f"Waiting {delay:.1f}s before next video…", level="DEBUG") + _interruptible_sleep(delay, controls) + + # ── Step 4: Finalize job ────────────────────────────────────────────── + if controls["stop"].is_set(): + final_status = "stopped" + else: + stats = db.get_job_stats(job_id) + final_status = "done" if stats["artifact_failed"] == 0 else "done_with_errors" + + db.update_job_status(job_id, final_status) + stats = db.get_job_stats(job_id) + db.add_log(job_id, f"Job finished with status: {final_status}. Stats: {stats}", level="INFO") + _emit(job_id, "job_status", {"job_id": job_id, "status": final_status, "stats": stats}) + + except Exception as e: + logger.exception(f"Unexpected error in job {job_id}: {e}") + db.add_log(job_id, f"Unexpected error: {e}", level="ERROR") + db.update_job_status(job_id, "failed") + _emit(job_id, "job_status", {"job_id": job_id, "status": "failed", "error": str(e)}) + finally: + _cleanup_controls(job_id) + + +# ───────────────────────────────────────────────────────────────────────────── +# Artifact Processing with Retry +# ───────────────────────────────────────────────────────────────────────────── + +def _process_artifact_with_retry( + job_id: str, + video_id: str, + artifact_type: str, + meta: Dict, + job: Dict, + controls: Dict, +) -> bool: + """ + Attempt to process one artifact, retrying up to MAX_RETRIES times. + Returns True on success, False on final failure. + """ + max_retries = config.MAX_RETRIES + output_folder = job["output_folder"] + resolution = job.get("resolution", "best") + api_key = job.get("openrouter_model", "") # model field + # We store the API key in the job separately; retrieve from job dict + # (api_key is set in the job via the POST body - see api_server.py) + openrouter_api_key = job.get("_openrouter_api_key", "") # injected by api_server + openrouter_model = job.get("openrouter_model", config.OPENROUTER_MODEL) + summary_prompt = job.get("summary_prompt", config.SUMMARY_SYSTEM_PROMPT) + + for attempt in range(1, max_retries + 1): + if controls["stop"].is_set() or controls["skip"].is_set(): + return False + + db.mark_artifact_started(job_id, video_id, artifact_type) + _emit(job_id, "artifact_start", { + "job_id": job_id, + "video_id": video_id, + "artifact_type": artifact_type, + "attempt": attempt, + }) + db.add_log(job_id, f"[{artifact_type}] Starting (attempt {attempt}/{max_retries})", level="INFO", + video_id=video_id, artifact_type=artifact_type) + + try: + file_path = _run_extractor( + artifact_type=artifact_type, + video_id=video_id, + meta=meta, + output_folder=output_folder, + resolution=resolution, + openrouter_api_key=openrouter_api_key, + openrouter_model=openrouter_model, + summary_prompt=summary_prompt, + ) + db.update_artifact_status(job_id, video_id, artifact_type, "done", file_path=file_path) + db.add_log(job_id, f"[{artifact_type}] Done → {file_path}", level="INFO", + video_id=video_id, artifact_type=artifact_type) + _emit(job_id, "artifact_done", { + "job_id": job_id, + "video_id": video_id, + "artifact_type": artifact_type, + "file_path": file_path, + }) + return True + + except Exception as e: + err_msg = str(e) + logger.warning(f"[{job_id}][{video_id}][{artifact_type}] Attempt {attempt} failed: {e}") + db.add_log(job_id, f"[{artifact_type}] Attempt {attempt} failed: {e}", level="WARN", + video_id=video_id, artifact_type=artifact_type) + + if attempt < max_retries: + # Exponential backoff between retries + backoff = min(2 ** attempt * 3, 30) + random.uniform(0, 2) + db.add_log(job_id, f"[{artifact_type}] Retry in {backoff:.1f}s…", level="INFO", + video_id=video_id, artifact_type=artifact_type) + _interruptible_sleep(backoff, controls) + else: + db.update_artifact_status( + job_id, video_id, artifact_type, "failed", error_msg=err_msg + ) + db.add_log(job_id, f"[{artifact_type}] FAILED after {max_retries} attempts: {e}", level="ERROR", + video_id=video_id, artifact_type=artifact_type) + _emit(job_id, "artifact_failed", { + "job_id": job_id, + "video_id": video_id, + "artifact_type": artifact_type, + "error": err_msg, + "attempts": max_retries, + }) + return False + + return False + + +def _run_extractor( + artifact_type: str, + video_id: str, + meta: Dict, + output_folder: str, + resolution: str, + openrouter_api_key: str, + openrouter_model: str, + summary_prompt: str, +) -> str: + """Dispatch to the correct extractor function.""" + if artifact_type == "title": + file_path, _ = extractors.extract_title(video_id, meta, output_folder) + return file_path + elif artifact_type == "stats": + return extractors.extract_stats(video_id, meta, output_folder) + elif artifact_type == "transcript": + return extractors.extract_transcript(video_id, meta, output_folder) + elif artifact_type == "summary": + return extractors.extract_summary( + video_id, meta, output_folder, + api_key=openrouter_api_key, + model=openrouter_model, + system_prompt=summary_prompt, + ) + elif artifact_type == "thumbnail": + return extractors.extract_thumbnail(video_id, meta, output_folder) + elif artifact_type == "mp3": + return extractors.extract_mp3(video_id, meta, output_folder) + elif artifact_type == "mp4": + return extractors.extract_mp4(video_id, meta, output_folder, resolution=resolution) + else: + raise ValueError(f"Unknown artifact type: {artifact_type}") + + +# ───────────────────────────────────────────────────────────────────────────── +# Helpers +# ───────────────────────────────────────────────────────────────────────────── + +def _emit(job_id: str, event_type: str, data: dict): + """Emit an SSE event.""" + emitter.emit(job_id, event_type, data) + + +def _wait_if_paused(controls: Dict): + """Block worker thread while paused, check every 0.5s.""" + while controls["pause"].is_set() and not controls["stop"].is_set(): + time.sleep(0.5) + + +def _interruptible_sleep(seconds: float, controls: Dict): + """Sleep for `seconds` but wake early if stop/skip is set.""" + end = time.monotonic() + seconds + while time.monotonic() < end: + if controls["stop"].is_set() or controls["skip"].is_set(): + return + time.sleep(min(0.3, end - time.monotonic())) + + +# ───────────────────────────────────────────────────────────────────────────── +# Job Launcher +# ───────────────────────────────────────────────────────────────────────────── + +def launch_job(job_id: str, openrouter_api_key: str = ""): + """ + Launch job in a daemon background thread. + openrouter_api_key is injected at runtime (not stored in DB for security). + """ + # Inject API key into job dict via a side channel stored in controls + controls = _get_controls(job_id) + controls["_openrouter_api_key"] = openrouter_api_key + + # Patch the run_job function to use the injected key + def _patched_run(): + job = db.get_job(job_id) + if job: + job["_openrouter_api_key"] = openrouter_api_key + run_job_with_key(job_id, openrouter_api_key) + + t = threading.Thread(target=_patched_run, daemon=True, name=f"job-{job_id[:8]}") + t.start() + return t + + +def run_job_with_key(job_id: str, openrouter_api_key: str): + """Wrapper that injects the API key into job processing.""" + # Monkey-patch the job dict to include the API key during this run + original_get_job = db.get_job + + def patched_get_job(jid): + result = original_get_job(jid) + if result and jid == job_id: + result["_openrouter_api_key"] = openrouter_api_key + return result + + # Temporarily patch db.get_job for this thread + import database + _orig = database.get_job + database.get_job = patched_get_job + try: + run_job(job_id) + finally: + database.get_job = _orig