from __future__ import annotations from dotenv import load_dotenv load_dotenv() from concurrent.futures import ThreadPoolExecutor from pathlib import Path from threading import Lock from typing import Any from urllib.parse import quote import html import shutil import uuid import gradio as gr from fastapi import File, Form, HTTPException, UploadFile from fastapi.responses import FileResponse, HTMLResponse, JSONResponse from pipeline import generate_roast_movie APP_TITLE = "YeOldeCut" APP_TAGLINE = "A tiny theatre for texts that deserved better." ROOT_DIR = Path(__file__).resolve().parent OUTPUTS_DIR = ROOT_DIR / "outputs" UPLOAD_DIR = OUTPUTS_DIR / "_uploads" EXAMPLES = { "job": "Thanks for applying. We enjoyed learning about your background, but we have decided to move forward with candidates whose experience more closely matches the role.", "dating": "I think you are great, but I am not really looking for anything serious right now. I hope you understand.", "investor": "We loved the energy, but the fund is going to pass for now. The market feels a little early and we want to see more traction before revisiting.", } SAMPLE_VIDEO_URL = "https://res.cloudinary.com/dzhtwka6d/video/upload/v1781563310/a7992042-2a8d-44eb-a805-1543c1d0805a_alpzpg.mp4" SAMPLE_PROMPT = "I've been lying to myself for months, pretending you were worth my time. You're not. You're selfish, draining, and honestly? A complete embarrassment. The sex was terrible, your 'ambition' is a joke, and I've never met anyone so desperate for validation. Don't text me again — I'm disgusted I ever touched you." JOBS: dict[str, dict[str, Any]] = {} JOBS_LOCK = Lock() EXECUTOR = ThreadPoolExecutor(max_workers=1) def _artifact_url(path: str | Path) -> str: resolved = Path(path).resolve() try: relative = resolved.relative_to(ROOT_DIR) except ValueError: relative = resolved.relative_to(OUTPUTS_DIR) relative = Path("outputs") / relative return f"/artifact/{quote(relative.as_posix())}" def _set_job(job_id: str, **updates: Any) -> None: with JOBS_LOCK: JOBS.setdefault(job_id, {}).update(updates) def _get_job(job_id: str) -> dict[str, Any]: with JOBS_LOCK: job = JOBS.get(job_id) if not job: raise HTTPException(status_code=404, detail="Performance not found.") return dict(job) def _save_upload(job_id: str, upload: UploadFile | None) -> str | None: if not upload or not upload.filename: return None UPLOAD_DIR.mkdir(parents=True, exist_ok=True) suffix = Path(upload.filename).suffix or ".png" path = UPLOAD_DIR / f"{job_id}{suffix}" with path.open("wb") as file: shutil.copyfileobj(upload.file, file) return str(path) def _run_job( job_id: str, message: str, screenshot_path: str | None, runtime_seconds: int, scene_count: int, ) -> None: def progress(value: float, line: str) -> None: _set_job(job_id, status="running", progress=value, line=line) try: progress(0.04, "The house lights fall.") result = generate_roast_movie( rejection_text=message, image=screenshot_path, runtime_seconds=runtime_seconds, scene_count=scene_count, progress_callback=progress, ) script = result.get("script") or {} downloads = [ {"label": "Final film", "url": _artifact_url(result["video_path"])}, {"label": "Share caption", "url": _artifact_url(result["script_path"])}, {"label": "Narration", "url": _artifact_url(result["voice_path"])}, {"label": "Score", "url": _artifact_url(result["music_path"])}, ] for index, image_path in enumerate(result.get("image_paths") or [], start=1): downloads.append({"label": f"Still {index}", "url": _artifact_url(image_path)}) _set_job( job_id, status="done", progress=1.0, line="The curtain opens.", result={ "title": script.get("movie_title") or "The Court Has Spoken", "logline": script.get("logline") or "A private text became public theatre.", "caption": result.get("share_caption") or "", "video_url": _artifact_url(result["video_path"]), "downloads": downloads, }, ) except Exception as exc: _set_job( job_id, status="error", progress=1.0, line="The performance stumbled before the final bow.", error=str(exc), ) def create_server() -> gr.Server: server = gr.Server( title=APP_TITLE, docs_url=None, redoc_url=None, openapi_url=None, ) @server.get("/", response_class=HTMLResponse) def theatre() -> str: return THEATRE_HTML @server.post("/api/perform") async def perform( message: str = Form(""), runtime_seconds: int = Form(60), scene_count: int = Form(6), screenshot: UploadFile | None = File(None), ) -> JSONResponse: clean_message = (message or "").strip() if not clean_message and not (screenshot and screenshot.filename): raise HTTPException( status_code=400, detail="Bring a message or screenshot to the stage first.", ) job_id = uuid.uuid4().hex[:12] screenshot_path = _save_upload(job_id, screenshot) _set_job( job_id, status="queued", progress=0.01, line="The theatre inhales.", ) EXECUTOR.submit( _run_job, job_id, clean_message, screenshot_path, int(runtime_seconds), int(scene_count), ) return JSONResponse({"job_id": job_id}) @server.get("/api/perform/{job_id}") def perform_status(job_id: str) -> JSONResponse: return JSONResponse(_get_job(job_id)) @server.get("/artifact/{path:path}") def artifact(path: str) -> FileResponse: target = (ROOT_DIR / path).resolve() outputs_root = OUTPUTS_DIR.resolve() if not str(target).startswith(str(outputs_root)) or not target.exists(): raise HTTPException(status_code=404, detail="Artifact not found.") return FileResponse(target) return server def launch_app() -> None: server.launch( server_name="0.0.0.0", show_error=True, _frontend=False, ) def _example_attr(key: str) -> str: return html.escape(EXAMPLES[key], quote=True) def _sample_prompt_attr() -> str: return html.escape(SAMPLE_PROMPT, quote=True) def _sample_prompt_html() -> str: return html.escape(SAMPLE_PROMPT) THEATRE_HTML = f""" {APP_TITLE}
Now performing Admission: one terrible text
The Royal Theatre of Rejections

{APP_TITLE}

{APP_TAGLINE}

The verdict has arrived.

The reel is sealed, the balcony is silent, and the court is prepared to be ridiculous.

Selected screenshot preview
Borrow a dramatic prompt

""" server = create_server() if __name__ == "__main__": launch_app()