jacquie/batchtube
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: enforce free tier limits (3 videos/batch, 720p max)

Browse Source 
- 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
This commit is contained in:
Agent Developer
2026-03-08 00:54:00 +00:00
parent 069dfc0ae4
commit c9041bde01
3 changed files with 1040 additions and 952 deletions
Download Patch File Download Diff File Expand all files Collapse all files

920
backend/api_server.py
View File

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

37
backend/config.py
View File

@@ -1,5 +1,5 @@
""" """
config.py - Default configuration for YouTube SIPOC Tool config.py - Default configuration for BatchTube
""" """
import os import os
@@ -32,7 +32,40 @@ SUMMARY_MAX_TOKENS = 1024
MAX_TITLE_LENGTH = 80 # characters - truncate long titles in filenames MAX_TITLE_LENGTH = 80 # characters - truncate long titles in filenames
# ── Database ────────────────────────────────────────────────────────────────── # ── 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 ────────────────────────────────────────────────────────────
ARTIFACT_TYPES = ["title", "stats", "transcript", "summary", "thumbnail", "mp3", "mp4"] 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

1035
backend/worker.py
View File

File diff suppressed because it is too large Load Diff