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