Media-Server-Setup/bin/dumpd.py

#!/usr/bin/env python3
"""Dump-tonight endpoint. Moves one streamer's night into trash (soft delete).

POST /dump   identity from X-Authentik-Username / X-Authentik-Groups (set by Caddy after Authentik)
             body (optional JSON): {"night":"YYYY-MM-DD","user":"name"}
GET  /dump   -> lists what a dump would move (dry run)
GET  /me/stream-key -> the caller's MediaMTX credentials (Go Live)

"Night" = everything the user recorded from 17:00 Eastern on <night> until 02:00 Eastern the next day.
Default night: today if it is 17:00 or later, else yesterday (so 00:00-02:00 still counts as last night).
Users may dump only their own prefix; members of the admins / media-admins groups may pass "user" to dump anyone, or "all".
"""
import json, os, re, shutil, sys
from datetime import datetime, timedelta, time
from zoneinfo import ZoneInfo
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path

ROOT   = Path(os.environ.get("MEDIA_ROOT", "/srv/media"))
SRC_DIRS = [ROOT / "streams", ROOT / "clips"]
TRASH  = ROOT / "trash"
MARK   = ROOT / ".markers"
STREAMERS_FILE = Path(os.environ.get("STREAMERS_FILE", "/etc/vanguard/streamers.json"))   # {"techy": "pass", ...} written by configure.py
AUTH_USER_HEADER = os.environ.get("AUTH_USER_HEADER", "X-Authentik-Username")
AUTH_GROUPS_HEADER = os.environ.get("AUTH_GROUPS_HEADER", "X-Authentik-Groups")
ADMIN_GROUPS = set(g.strip() for g in os.environ.get("DUMP_ADMIN_GROUPS", "admins,media-admins").split(",") if g.strip())
LISTEN = ("127.0.0.1", 8080)
TZ = ZoneInfo(os.environ.get("DUMP_TZ", "America/New_York"))
NIGHT_START = time(17, 0)   # 5 pm
NIGHT_END   = time(2, 0)    # 2 am next day

def identity(headers):
    """(username, is_admin) from the headers Caddy sets after Authentik; (None, False) if absent.
    dumpd binds to loopback only, so these headers can only come from Caddy."""
    who = (headers.get(AUTH_USER_HEADER) or "").strip().lower()
    if not who:
        return None, False
    groups = {g.strip() for g in re.split(r"[|,]", headers.get(AUTH_GROUPS_HEADER) or "") if g.strip()}
    return who, bool(groups & ADMIN_GROUPS)

def default_night(now=None):
    now = now or datetime.now(TZ)
    d = now.date() if now.time() >= NIGHT_START else (now - timedelta(days=1)).date()
    return d.isoformat()

def night_window(night):
    d = datetime.strptime(night, "%Y-%m-%d").date()
    return (datetime.combine(d, NIGHT_START, tzinfo=TZ),
            datetime.combine(d + timedelta(days=1), NIGHT_END, tzinfo=TZ))

def candidates(user, night):
    start, end = night_window(night)
    out = []
    for base in SRC_DIRS:
        udirs = [p for p in base.iterdir() if p.is_dir()] if user == "all" else [base / user]
        for ud in udirs:
            if not ud.is_dir():
                continue
            for f in ud.rglob("*"):
                if not f.is_file():
                    continue
                mt = datetime.fromtimestamp(f.stat().st_mtime, TZ)
                if start <= mt <= end:
                    out.append(f)
    return out

def do_move(files, night):
    moved, size = 0, 0
    for f in files:
        rel = f.relative_to(ROOT)
        dst = TRASH / night / rel
        dst.parent.mkdir(parents=True, exist_ok=True)
        try:
            size += f.stat().st_size
            shutil.move(str(f), str(dst))
            moved += 1
        except FileNotFoundError:
            pass
    # drop now-empty title folders
    for base in SRC_DIRS:
        for d in sorted((p for p in base.rglob("*") if p.is_dir()), key=lambda p: -len(p.parts)):
            if d.parent != base and not any(d.iterdir()):
                d.rmdir()
    return moved, size

def load_streamers():
    try:
        return json.loads(STREAMERS_FILE.read_text())
    except (OSError, json.JSONDecodeError):
        return {}

class H(BaseHTTPRequestHandler):
    server_version = "dumpd/1"
    def _json(self, code, obj):
        body = json.dumps(obj).encode()
        self.send_response(code)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)
    def _deny(self):
        self._json(401, {"error": "not authenticated (no identity header from the proxy)"})
    def _parse(self):
        n = int(self.headers.get("Content-Length") or 0)
        body = json.loads(self.rfile.read(n) or b"{}") if n else {}
        night = body.get("night")
        if night:
            datetime.strptime(night, "%Y-%m-%d")  # validate
        return night, body.get("user")
    def _resolve(self, night, tgt):
        """Explicit night wins. Otherwise: tonight, falling back to last night if tonight is empty."""
        if night:
            return night, candidates(tgt, night)
        n = default_night()
        files = candidates(tgt, n)
        if not files:
            prev = (datetime.strptime(n, "%Y-%m-%d") - timedelta(days=1)).strftime("%Y-%m-%d")
            pf = candidates(tgt, prev)
            if pf:
                return prev, pf
        return n, files
    def _target(self, auth_user, is_admin, req_user):
        if is_admin:
            return req_user or auth_user          # admins default to their own night; pass "all" or a name for others
        return auth_user
    def do_GET(self):
        if self.path.startswith("/me/stream-key"):
            # Only reachable through Caddy behind Authentik forward-auth, which injects the username header.
            # dumpd binds to loopback, so a request carrying this header is one Caddy already authenticated.
            who = (self.headers.get(AUTH_USER_HEADER) or "").strip().lower()
            if not who:
                return self._json(401, {"error": "not authenticated"})
            streamers = load_streamers()
            pw = streamers.get(who)
            if not pw:
                return self._json(403, {"error": f"no stream key for '{who}' — ask the admin to add you as a streamer with that exact name"})
            return self._json(200, {"user": who, "pass": pw})
        if not self.path.startswith("/dump"):
            return self._json(404, {"error": "not found"})
        u, is_admin = identity(self.headers)
        if not u:
            return self._deny()
        night, req_user = self._parse()
        tgt = self._target(u, is_admin, req_user)
        night, files = self._resolve(night, tgt)
        self._json(200, {"night": night, "user": tgt, "dry_run": True,
                         "files": [str(f.relative_to(ROOT)) for f in files],
                         "gb": round(sum(f.stat().st_size for f in files) / 1e9, 2)})
    def do_POST(self):
        if not self.path.startswith("/dump"):
            return self._json(404, {"error": "not found"})
        u, is_admin = identity(self.headers)
        if not u:
            return self._deny()
        try:
            night, req_user = self._parse()
        except Exception as e:
            return self._json(400, {"error": str(e)})
        tgt = self._target(u, is_admin, req_user)
        night, files = self._resolve(night, tgt)
        moved, size = do_move(files, night)
        MARK.mkdir(exist_ok=True)
        (MARK / f"{night}-{tgt}.dumped").write_text(f"{datetime.now(TZ).isoformat()} by {u}\n")
        print(f"{datetime.now(TZ):%F %T} dump night={night} user={tgt} by={u} files={moved} gb={size/1e9:.2f}", flush=True)
        self._json(200, {"night": night, "user": tgt, "moved": moved, "gb": round(size / 1e9, 2),
                         "undo_until": (datetime.now(TZ) + timedelta(days=3)).strftime("%F")})
    def log_message(self, *a):  # quiet
        pass

if __name__ == "__main__":
    TRASH.mkdir(exist_ok=True); MARK.mkdir(exist_ok=True)
    print(f"dumpd listening on {LISTEN[0]}:{LISTEN[1]}  tz={TZ.key}  night={NIGHT_START:%H:%M}-{NIGHT_END:%H:%M}", flush=True)
    ThreadingHTTPServer(LISTEN, H).serve_forever()