""" VOD Transcriber — Hugging Face Spaces FastAPI + faster-whisper + yt-dlp """ import asyncio import glob import hashlib import hmac import json import os import secrets import shutil import threading import uuid from concurrent.futures import ThreadPoolExecutor from typing import Optional import yt_dlp from faster_whisper import WhisperModel from fastapi import FastAPI, File, Form, Request, UploadFile from fastapi.responses import HTMLResponse, RedirectResponse, StreamingResponse from starlette.middleware.base import BaseHTTPMiddleware # --------------------------------------------------------------------------- # Config # --------------------------------------------------------------------------- PASSWORD = os.getenv("APP_PASSWORD", "changeme") MAX_UPLOAD_BYTES = 500 * 1024 * 1024 # 500 MB WHISPER_MODEL = "base.en" COOKIE_NAME = "auth" _SECRET = os.urandom(32) # ephemeral signing key # --------------------------------------------------------------------------- # Auth helpers # --------------------------------------------------------------------------- def _make_token(pw: str) -> str: return hmac.new(_SECRET, pw.encode(), hashlib.sha256).hexdigest() def _is_authed(request: Request) -> bool: token = request.cookies.get(COOKIE_NAME, "") return secrets.compare_digest(token, _make_token(PASSWORD)) class AuthMiddleware(BaseHTTPMiddleware): async def dispatch(self, request: Request, call_next): if request.url.path in ("/login",): return await call_next(request) if not _is_authed(request): return RedirectResponse("/login", status_code=302) return await call_next(request) # --------------------------------------------------------------------------- # App # --------------------------------------------------------------------------- app = FastAPI() app.add_middleware(AuthMiddleware) executor = ThreadPoolExecutor(max_workers=2) _model: Optional[WhisperModel] = None _model_lock = threading.Lock() def get_model() -> WhisperModel: global _model if _model is None: with _model_lock: if _model is None: _model = WhisperModel(WHISPER_MODEL, device="cpu", compute_type="int8") return _model # --------------------------------------------------------------------------- # Login routes # --------------------------------------------------------------------------- LOGIN_HTML = """ VOD Transcriber — Login

VOD Transcriber

Enter password to continue

{err_msg}

""" @app.get("/login", response_class=HTMLResponse) async def login_page(): return LOGIN_HTML.replace("{err_class}", "").replace("{err_msg}", "") @app.post("/login") async def login(password: str = Form(...)): if secrets.compare_digest(password.encode(), PASSWORD.encode()): resp = RedirectResponse("/", status_code=302) resp.set_cookie(COOKIE_NAME, _make_token(PASSWORD), httponly=True, samesite="lax") return resp html = LOGIN_HTML.replace("{err_class}", "visible").replace("{err_msg}", "Incorrect password.") return HTMLResponse(html, status_code=401) # --------------------------------------------------------------------------- # Job store # --------------------------------------------------------------------------- jobs: dict[str, dict] = {} def new_job() -> str: jid = str(uuid.uuid4()) jobs[jid] = {"messages": [], "done": False, "transcript": None, "error": None} return jid def push(jid: str, msg: str): jobs[jid]["messages"].append(msg) # --------------------------------------------------------------------------- # Transcription worker (runs in thread) # --------------------------------------------------------------------------- def transcribe_file(jid: str, path: str, cleanup_paths: list[str]): try: push(jid, "Loading whisper model…") model = get_model() push(jid, f"Transcribing {os.path.basename(path)}…") segments, info = model.transcribe(path, language="en", beam_size=5) lines = [] for seg in segments: ts = f"[{seg.start:.1f}s – {seg.end:.1f}s]" lines.append(f"{ts} {seg.text.strip()}") push(jid, f"{ts} {seg.text.strip()}") jobs[jid]["transcript"] = "\n".join(lines) push(jid, "✓ Done") except Exception as e: jobs[jid]["error"] = str(e) push(jid, f"ERROR: {e}") finally: jobs[jid]["done"] = True for p in cleanup_paths: try: os.remove(p) except OSError: pass # --------------------------------------------------------------------------- # Routes # --------------------------------------------------------------------------- @app.get("/", response_class=HTMLResponse) async def index(): return HTML @app.post("/transcribe/file") async def transcribe_upload(file: UploadFile = File(...)): # Size check via content-length header size = int(file.headers.get("content-length", 0)) if size > MAX_UPLOAD_BYTES: raise HTTPException(413, f"File too large. Max {MAX_UPLOAD_BYTES // 1024 // 1024} MB.") jid = new_job() ext = os.path.splitext(file.filename or "video.mp4")[1] or ".mp4" tmp_path = f"/tmp/{jid}{ext}" push(jid, f"Saving upload ({file.filename})…") with open(tmp_path, "wb") as f: shutil.copyfileobj(file.file, f) # Double-check actual size actual = os.path.getsize(tmp_path) if actual > MAX_UPLOAD_BYTES: os.remove(tmp_path) raise HTTPException(413, f"File too large. Max {MAX_UPLOAD_BYTES // 1024 // 1024} MB.") executor.submit(transcribe_file, jid, tmp_path, [tmp_path]) return {"job_id": jid} @app.post("/transcribe/url") async def transcribe_url( url: str = Form(...), cookies: UploadFile = File(None), ): jid = new_job() push(jid, f"Fetching URL: {url}") # Save cookies file if provided cookie_path = None if cookies and cookies.filename: cookie_path = f"/tmp/{jid}_cookies.txt" with open(cookie_path, "wb") as f: shutil.copyfileobj(cookies.file, f) push(jid, "Cookies loaded.") def download_and_transcribe(): output_tpl = f"/tmp/{jid}.%(ext)s" downloaded = [] class Hook: def __call__(self, d): if d["status"] == "downloading": pct = d.get("_percent_str", "").strip() spd = d.get("_speed_str", "").strip() if pct: push(jid, f"Downloading {pct} at {spd}") elif d["status"] == "finished": downloaded.append(d["filename"]) push(jid, "Download complete.") opts = { "format": "bestaudio/best", "outtmpl": output_tpl, "progress_hooks": [Hook()], "quiet": True, "no_warnings": True, # Bypass YouTube datacenter IP blocks "extractor_args": {"youtube": {"player_client": ["ios", "web"]}}, "http_headers": { "User-Agent": ( "Mozilla/5.0 (Windows NT 10.0; Win64; x64) " "AppleWebKit/537.36 (KHTML, like Gecko) " "Chrome/124.0.0.0 Safari/537.36" ), }, "source_address": "0.0.0.0", # force IPv4 } if cookie_path: opts["cookiefile"] = cookie_path try: with yt_dlp.YoutubeDL(opts) as ydl: ydl.download([url]) # Find downloaded file (extension may differ from template) files = glob.glob(f"/tmp/{jid}.*") if not files: raise RuntimeError("Download produced no output file.") video_path = files[0] # Check size if os.path.getsize(video_path) > MAX_UPLOAD_BYTES: os.remove(video_path) raise RuntimeError(f"Downloaded file exceeds 500 MB limit.") transcribe_file(jid, video_path, files) except Exception as e: jobs[jid]["error"] = str(e) push(jid, f"ERROR: {e}") jobs[jid]["done"] = True finally: if cookie_path: try: os.remove(cookie_path) except OSError: pass executor.submit(download_and_transcribe) return {"job_id": jid} @app.get("/progress/{jid}") async def progress(jid: str): if jid not in jobs: raise HTTPException(404, "Job not found.") async def stream(): sent = 0 while True: job = jobs[jid] msgs = job["messages"] while sent < len(msgs): yield f"data: {json.dumps({'msg': msgs[sent]})}\n\n" sent += 1 if job["done"]: yield f"data: {json.dumps({'done': True, 'transcript': job['transcript'], 'error': job['error']})}\n\n" del jobs[jid] break await asyncio.sleep(0.4) return StreamingResponse(stream(), media_type="text/event-stream") # --------------------------------------------------------------------------- # HTML # --------------------------------------------------------------------------- HTML = r""" VOD Transcriber

VOD Transcriber

whisper · local processing · no data retained

File Upload
Video URL
🎬
Drop video or audio file here, or browse
MP4, MOV, MKV, WebM, MP3, WAV · max 500 MB

YouTube, Twitter/X, Vimeo, Twitch, and 1000+ sites via yt-dlp

▸ YouTube blocked? Add cookies
Processing…


  

    
Transcript ready

    

      Download .txt
    

  


  






"""