booth/booth/app.py

"""The Booth — a standing web server that renders drop-folders as ephemeral media booths.

Model (deliberately dead-simple, no database):
  * The data dir holds one subfolder per "booth". A booth is created by a CC
    session simply making a folder and dropping files in — there is no upload API.
  * GET /            -> index: a card per booth (scan of the data dir).
  * GET /b/<name>/   -> if <name>/index.html exists, serve it verbatim; otherwise
                        auto-render a gallery of the images / webm-videos / audio in it.
  * GET /b/<name>/<file> -> serve a file out of the booth (also feeds a custom index.html's assets).
  * 24h TTL: a background sweeper wipes any booth untouched for TTL hours. A booth's
    age is measured from the *newest* mtime in its tree, so it lives while it's being
    worked on and self-destructs TTL hours after the last activity.

State is the filesystem — `ls ~/booth-data` tells you everything. That is the whole point.
"""

from __future__ import annotations

import asyncio
import os
import shutil
import time
from contextlib import asynccontextmanager
from pathlib import Path
from urllib.parse import quote

from fastapi import FastAPI, HTTPException, Request
from fastapi.responses import (
    FileResponse,
    HTMLResponse,
    JSONResponse,
    RedirectResponse,
)
from fastapi.templating import Jinja2Templates

TEMPLATES_DIR = Path(__file__).parent / "templates"

# Browser-playable media buckets. Anything else renders as a download link.
IMAGE_EXTS = {".png", ".jpg", ".jpeg", ".gif", ".webp", ".avif", ".svg", ".bmp"}
VIDEO_EXTS = {".webm", ".mp4", ".ogv", ".m4v", ".mov"}
AUDIO_EXTS = {".mp3", ".wav", ".ogg", ".oga", ".flac", ".m4a", ".opus", ".aac"}

CAPTION_MAX = 800  # chars of a sidecar .txt caption we render


def classify(name: str) -> str:
    """image | video | audio | other, by extension."""
    ext = Path(name).suffix.lower()
    if ext in IMAGE_EXTS:
        return "image"
    if ext in VIDEO_EXTS:
        return "video"
    if ext in AUDIO_EXTS:
        return "audio"
    return "other"


def human_dur(seconds: float) -> str:
    s = int(seconds)
    if s <= 0:
        return "expired"
    h, rem = divmod(s, 3600)
    m, _ = divmod(rem, 60)
    if h and m:
        return f"{h}h {m}m"
    if h:
        return f"{h}h"
    if m:
        return f"{m}m"
    return "<1m"


def _newest_mtime(path: Path) -> float:
    """Newest mtime among a folder and everything under it."""
    try:
        newest = path.stat().st_mtime
    except OSError:
        return 0.0
    for p in path.rglob("*"):
        try:
            m = p.stat().st_mtime
        except OSError:
            continue
        if m > newest:
            newest = m
    return newest


def booth_age_seconds(path: Path, now: float | None = None) -> float:
    now = time.time() if now is None else now
    return now - _newest_mtime(path)


def is_expired(path: Path, ttl_seconds: float, now: float | None = None) -> bool:
    return booth_age_seconds(path, now) > ttl_seconds


def sweep_once(data_dir: Path, ttl_seconds: float, now: float | None = None) -> list[str]:
    """Wipe every direct-child booth older than the TTL. Returns names wiped.

    Only ever removes direct children of data_dir (never data_dir itself), and
    skips dotfolders so a stray control dir can opt out.
    """
    wiped: list[str] = []
    if not data_dir.is_dir():
        return wiped
    for child in data_dir.iterdir():
        if not child.is_dir() or child.name.startswith("."):
            continue
        try:
            if is_expired(child, ttl_seconds, now):
                shutil.rmtree(child)
                wiped.append(child.name)
        except OSError:
            pass
    return wiped


def list_booths(data_dir: Path, ttl_seconds: float, now: float | None = None) -> list[dict]:
    now = time.time() if now is None else now
    booths: list[dict] = []
    if not data_dir.is_dir():
        return booths
    for child in data_dir.iterdir():
        if not child.is_dir() or child.name.startswith("."):
            continue
        files = [p for p in child.rglob("*") if p.is_file() and not p.name.startswith(".")]
        kinds = {"image": 0, "video": 0, "audio": 0, "other": 0}
        thumb_url = None
        for f in files:
            k = classify(f.name)
            kinds[k] += 1
            if k == "image" and thumb_url is None:
                thumb_url = quote(f.relative_to(child).as_posix(), safe="/")
        mtime = _newest_mtime(child)
        booths.append(
            {
                "name": child.name,
                "name_url": quote(child.name, safe=""),
                "count": len(files),
                "kinds": kinds,
                "thumb_url": thumb_url,
                "has_index": (child / "index.html").is_file(),
                "expires_in": max(0.0, ttl_seconds - (now - mtime)),
                "mtime": mtime,
            }
        )
    booths.sort(key=lambda b: b["mtime"], reverse=True)
    return booths


def build_gallery(child: Path) -> list[dict]:
    """Files in a booth as render items, with caption sidecars folded in.

    A `<file>.txt` (e.g. `a.png.txt`) or a same-stem `<stem>.txt` (e.g. `a.txt`
    next to `a.png`) is consumed as that item's caption rather than shown itself —
    the natural way to label an A/B pair.
    """
    all_files = [p for p in child.rglob("*") if p.is_file() and not p.name.startswith(".")]
    by_rel = {p.relative_to(child).as_posix(): p for p in all_files}
    caption: dict[str, str] = {}
    sidecars: set[str] = set()

    for rel, p in by_rel.items():
        if not rel.lower().endswith(".txt"):
            continue
        target = None
        base_full = rel[:-4]  # strip ".txt"  -> "a.png.txt" => "a.png"
        if base_full in by_rel:
            target = base_full
        else:  # "a.txt" beside "a.png"
            parent = str(Path(rel).parent)
            stem = Path(rel).stem
            for q_rel, q in by_rel.items():
                if q_rel == rel:
                    continue
                if (
                    str(Path(q_rel).parent) == parent
                    and Path(q_rel).stem == stem
                    and classify(q.name) != "other"
                ):
                    target = q_rel
                    break
        if target is not None:
            try:
                caption[target] = p.read_text(errors="replace").strip()[:CAPTION_MAX]
            except OSError:
                pass
            sidecars.add(rel)

    items = []
    for rel in sorted(by_rel):
        if rel in sidecars:
            continue
        p = by_rel[rel]
        items.append(
            {
                "name": rel,
                "kind": classify(p.name),
                "url": quote(rel, safe="/"),
                "caption": caption.get(rel),
            }
        )
    return items


def create_app(
    data_dir,
    ttl_hours: float = 24.0,
    host_label: str = "",
    start_sweeper: bool = True,
    sweep_interval_s: int = 900,
) -> FastAPI:
    data_dir = Path(data_dir).expanduser().resolve()
    data_dir.mkdir(parents=True, exist_ok=True)
    ttl_seconds = ttl_hours * 3600.0

    templates = Jinja2Templates(directory=str(TEMPLATES_DIR))
    templates.env.filters["dur"] = human_dur

    @asynccontextmanager
    async def lifespan(app: FastAPI):
        task = None
        if start_sweeper:

            async def loop():
                while True:
                    try:
                        wiped = sweep_once(data_dir, ttl_seconds)
                        if wiped:
                            print(f"[booth] swept {len(wiped)} expired: {', '.join(wiped)}", flush=True)
                    except Exception as exc:  # never let the sweeper die
                        print(f"[booth] sweep error: {exc}", flush=True)
                    await asyncio.sleep(sweep_interval_s)

            task = asyncio.create_task(loop())
        try:
            yield
        finally:
            if task is not None:
                task.cancel()

    app = FastAPI(title="The Booth", lifespan=lifespan)

    ttl_display = int(ttl_hours) if float(ttl_hours).is_integer() else ttl_hours
    base_ctx = {"ttl_hours": ttl_display, "host": host_label, "data_dir": str(data_dir)}

    def resolve_booth(name: str) -> Path:
        if not name or name.startswith(".") or "/" in name or "\\" in name or ".." in name:
            raise HTTPException(status_code=404, detail="no such booth")
        candidate = data_dir / name
        try:
            resolved = candidate.resolve()
        except OSError:
            raise HTTPException(status_code=404, detail="no such booth")
        # resolved.parent must be the data dir itself — blocks symlink escape + nesting.
        if resolved.parent != data_dir or not resolved.is_dir():
            raise HTTPException(status_code=404, detail="no such booth")
        return resolved

    @app.get("/", response_class=HTMLResponse)
    def index(request: Request):
        return templates.TemplateResponse(
            request, "index.html", {**base_ctx, "booths": list_booths(data_dir, ttl_seconds)}
        )

    @app.get("/healthz")
    def healthz():
        return {"ok": True, "ttl_hours": ttl_hours, "booths": len(list_booths(data_dir, ttl_seconds))}

    @app.get("/b/{name}", include_in_schema=False)
    def booth_redirect(name: str):
        resolve_booth(name)
        return RedirectResponse(url=f"/b/{quote(name, safe='')}/", status_code=307)

    @app.get("/b/{name}/", response_class=HTMLResponse)
    def booth_view(request: Request, name: str):
        booth = resolve_booth(name)
        own_index = booth / "index.html"
        if own_index.is_file():
            return FileResponse(str(own_index), media_type="text/html")
        return templates.TemplateResponse(
            request,
            "booth.html",
            {
                **base_ctx,
                "name": name,
                "name_url": quote(name, safe=""),
                "items": build_gallery(booth),
                "expires_in": max(0.0, ttl_seconds - booth_age_seconds(booth)),
            },
        )

    @app.get("/b/{name}/{filepath:path}")
    def booth_file(name: str, filepath: str):
        booth = resolve_booth(name)
        try:
            target = (booth / filepath).resolve()
        except OSError:
            raise HTTPException(status_code=404, detail="no such file")
        if not str(target).startswith(str(booth) + os.sep) or not target.is_file():
            raise HTTPException(status_code=404, detail="no such file")
        return FileResponse(str(target))

    @app.post("/b/{name}/delete")
    def booth_delete_form(name: str):
        shutil.rmtree(resolve_booth(name))
        return RedirectResponse(url="/", status_code=303)

    @app.delete("/b/{name}")
    def booth_delete_api(name: str):
        shutil.rmtree(resolve_booth(name))
        return JSONResponse({"wiped": name})

    return app


def _from_env() -> FastAPI:
    data = os.environ.get("BOOTH_DATA_DIR", str(Path.home() / "booth-data"))
    ttl = float(os.environ.get("BOOTH_TTL_HOURS", "24"))
    host = os.environ.get("BOOTH_HOST_LABEL", "")
    interval = int(float(os.environ.get("BOOTH_SWEEP_INTERVAL_MIN", "15")) * 60)
    return create_app(data, ttl_hours=ttl, host_label=host, sweep_interval_s=interval)


app = _from_env()