weircon-front/podcast.py

"""Private podcasts (Slægtens Stemmer): side, personlige feeds og lydfiler.

Podcast-apps kan ikke logge ind med Authentik, så hver lytter får en personlig,
hemmelig nøgle i adressen, som Patreon og Apple gør det. Nøglen hører til
Authentik-brugeren og kan trækkes tilbage i /admin/podcast uden at de andre
mærker noget. Både feedet og mp3-filerne ligger under nøglen.

Data (ikke i git, persondata):
    DATA/podcast/<show>/show.json               titel, beskrivelse, forfatter
    DATA/podcast/<show>/cover.jpg               kvadratisk, 1400-3000 px
    DATA/podcast/<show>/episodes/<slug>.mp3     selve episoden
    DATA/podcast/<show>/episodes/<slug>.json    titel, beskrivelse, nr, udgivet, kladde
    DATA/podcast_keys.json                      nøgle -> bruger, statistik
    DATA/podcast_faelles.json                   det fælles familielink (én nøgle, kan slås fra)

En episode med "kladde": true ses kun af administratorer (i deres eget feed og
på siden), så den kan høres i en rigtig podcast-app før familien får den.
"""
import email.utils
import functools
import json
import os
import re
import secrets
import time
from pathlib import Path
from xml.sax.saxutils import escape, quoteattr

import config

ROOT = config.DATA / "podcast"
KEYS = config.DATA / "podcast_keys.json"
DELT = config.DATA / "podcast_faelles.json"
SHOW = re.compile(r"^[a-z0-9][a-z0-9-]{0,63}$")
SLUG = re.compile(r"^[a-z0-9][a-z0-9._-]{0,120}$")
KEY = re.compile(r"^[A-Za-z0-9_-]{32,64}$")


# ---------------- shows og episoder ----------------

def shows() -> list[str]:
    if not ROOT.exists():
        return []
    return sorted(p.name for p in ROOT.iterdir() if p.is_dir() and SHOW.match(p.name) and (p / "show.json").exists())


def show(name: str) -> dict | None:
    if not SHOW.match(name or ""):
        return None
    try:
        s = json.loads((ROOT / name / "show.json").read_text())
    except (OSError, ValueError):
        return None
    s["slug"] = name
    s["har_cover"] = (ROOT / name / "cover.jpg").exists()
    return s


@functools.lru_cache(maxsize=256)
def _varighed(sti: str, mtime: float) -> int:
    try:
        from mutagen.mp3 import MP3
        return int(MP3(sti).info.length)
    except Exception:
        return 0


def episodes(name: str, med_kladder: bool) -> list[dict]:
    """Episoder, nyeste først. Kladder kun med med_kladder."""
    d = ROOT / name / "episodes"
    if not SHOW.match(name or "") or not d.exists():
        return []
    ud = []
    for meta in d.glob("*.json"):
        slug = meta.stem
        mp3 = d / f"{slug}.mp3"
        if not SLUG.match(slug) or not mp3.exists():
            continue
        try:
            e = json.loads(meta.read_text())
        except ValueError:
            continue
        if e.get("kladde") and not med_kladder:
            continue
        st = mp3.stat()
        e.update(slug=slug, bytes=st.st_size, varighed=_varighed(str(mp3), st.st_mtime))
        ud.append(e)
    return sorted(ud, key=lambda e: (e.get("nr", 0), e.get("udgivet", "")), reverse=True)


def episode_fil(name: str, slug: str) -> Path | None:
    if not SHOW.match(name or "") or not SLUG.match(slug or ""):
        return None
    p = ROOT / name / "episodes" / f"{slug}.mp3"
    return p if p.exists() else None


def episode_meta(name: str, slug: str) -> dict | None:
    if not episode_fil(name, slug):
        return None
    try:
        return json.loads((ROOT / name / "episodes" / f"{slug}.json").read_text())
    except (OSError, ValueError):
        return None


def saet_kladde(name: str, slug: str, kladde: bool) -> bool:
    e = episode_meta(name, slug)
    if e is None:
        return False
    e["kladde"] = kladde
    if not kladde and not e.get("udgivet"):
        e["udgivet"] = time.strftime("%Y-%m-%dT%H:%M:%S%z")
    _skriv(ROOT / name / "episodes" / f"{slug}.json", e)
    return True


def cover(name: str, lille: bool = False) -> Path | None:
    """Omslaget; lille = 300 px-udgaven til link-forhåndsvisninger (cover-lille.jpg)."""
    p = ROOT / name / ("cover-lille.jpg" if lille else "cover.jpg")
    return p if SHOW.match(name or "") and p.exists() else None


# ---------------- nøgler ----------------

def _skriv(p: Path, data) -> None:
    tmp = p.with_name(f".{p.name}.tmp")
    fd = os.open(tmp, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600)
    with os.fdopen(fd, "w") as f:
        json.dump(data, f, ensure_ascii=False, indent=1)
    os.replace(tmp, p)


def _noegler() -> dict:
    try:
        return json.loads(KEYS.read_text())
    except (OSError, ValueError):
        return {}


def noegle_til(user: str, admin: bool) -> str:
    """Brugerens aktive nøgle; laves første gang. Admin-status opdateres her,
    fordi den afgør om feedet viser kladder."""
    k = _noegler()
    for key, v in k.items():
        if v["user"] == user and not v.get("tilbagekaldt"):
            if v.get("admin") != admin:
                v["admin"] = admin
                _skriv(KEYS, k)
            return key
    key = secrets.token_urlsafe(32)
    k[key] = {"user": user, "admin": admin, "oprettet": int(time.time()), "hentninger": 0}
    _skriv(KEYS, k)
    return key


def ny_noegle(user: str, admin: bool) -> str:
    k = _noegler()
    for v in k.values():
        if v["user"] == user and not v.get("tilbagekaldt"):
            v["tilbagekaldt"] = int(time.time())
    _skriv(KEYS, k)
    return noegle_til(user, admin)


def slaa_op(key: str, ua: str = "", hvad: str = "") -> dict | None:
    """Gyldig nøgle -> dens post (og registrér hentningen), ellers None."""
    if not KEY.match(key or ""):
        return None
    k = _noegler()
    v = k.get(key)
    if not v or v.get("tilbagekaldt"):
        return None
    v["hentninger"] = v.get("hentninger", 0) + 1
    v["sidst"] = int(time.time())
    v["sidst_ua"] = (ua or "")[:160]
    if hvad:
        v.setdefault("hentet", {})
        v["hentet"][hvad] = v["hentet"].get(hvad, 0) + 1
    _skriv(KEYS, k)
    return v


def alle_noegler() -> list[dict]:
    return sorted(({"key": key, **v} for key, v in _noegler().items()),
                  key=lambda v: (bool(v.get("tilbagekaldt")), -(v.get("sidst") or 0)))


def tilbagekald(key: str, aktiv: bool) -> bool:
    k = _noegler()
    if key not in k:
        return False
    if aktiv:
        # Genaktivering: brugeren må kun have én aktiv nøgle.
        for other, v in k.items():
            if other != key and v["user"] == k[key]["user"] and not v.get("tilbagekaldt"):
                v["tilbagekaldt"] = int(time.time())
        k[key].pop("tilbagekaldt", None)
    else:
        k[key]["tilbagekaldt"] = int(time.time())
    _skriv(KEYS, k)
    return True


# ---------------- feed ----------------

def _rfc822(iso: str) -> str:
    try:
        t = time.mktime(time.strptime(iso[:19], "%Y-%m-%dT%H:%M:%S"))
    except (ValueError, TypeError):
        t = time.time()
    return email.utils.formatdate(t, localtime=True)


def _hms(sek: int) -> str:
    return f"{sek // 3600:d}:{sek // 60 % 60:02d}:{sek % 60:02d}"


def feed(name: str, rod: str, base: str, med_kladder: bool) -> str | None:
    """rod = stien foran showet, fx "/podcast/<nøgle>" eller "/podcast/f/<fælles nøgle>"."""
    s = show(name)
    if not s:
        return None
    pre = f"{base}{rod}/{name}"
    items = []
    for e in episodes(name, med_kladder):
        titel = ("[Kladde] " if e.get("kladde") else "") + e.get("titel", e["slug"])
        items.append(f"""    <item>
      <title>{escape(titel)}</title>
      <description>{escape(e.get("beskrivelse", ""))}</description>
      <guid isPermaLink="false">{escape(name)}/{escape(e["slug"])}</guid>
      <pubDate>{_rfc822(e.get("udgivet", ""))}</pubDate>
      <enclosure url={quoteattr(f"{pre}/{e['slug']}.mp3")} length="{e['bytes']}" type="audio/mpeg"/>
      <itunes:duration>{_hms(e["varighed"])}</itunes:duration>
      <itunes:episode>{int(e.get("nr", 0))}</itunes:episode>
      <itunes:episodeType>full</itunes:episodeType>
    </item>""")
    cover_url = f"{pre}/cover.jpg"
    return f"""<?xml version="1.0" encoding="UTF-8"?>
<rss version="2.0" xmlns:itunes="http://www.itunes.com/dtds/podcast-1.0.dtd"
     xmlns:podcast="https://podcastindex.org/namespace/1.0" xmlns:atom="http://www.w3.org/2005/Atom">
  <channel>
    <title>{escape(s.get("titel", name))}</title>
    <link>{escape(base)}/podcast</link>
    <atom:link href={quoteattr(f"{pre}.xml")} rel="self" type="application/rss+xml"/>
    <language>{escape(s.get("sprog", "da"))}</language>
    <description>{escape(s.get("beskrivelse", ""))}</description>
    <itunes:summary>{escape(s.get("beskrivelse", ""))}</itunes:summary>
    <itunes:author>{escape(s.get("forfatter", ""))}</itunes:author>
    <itunes:image href={quoteattr(cover_url)}/>
    <image><url>{escape(cover_url)}</url><title>{escape(s.get("titel", name))}</title><link>{escape(base)}/podcast</link></image>
    <itunes:category text="History"/>
    <itunes:explicit>false</itunes:explicit>
    <itunes:type>serial</itunes:type>
    <itunes:block>Yes</itunes:block>
    <podcast:locked>yes</podcast:locked>
{chr(10).join(items)}
  </channel>
</rss>
"""


# ---------------- fælles familielink ----------------
# Ét link uden login til fx familiens Messenger-gruppe: /podcast/f/<nøgle>/ med
# afspiller og feed. Viser aldrig kladder. Kan skiftes (det gamle dør med det
# samme) eller slås helt fra i /admin/podcast.

def faelles() -> dict:
    try:
        return json.loads(DELT.read_text())
    except (OSError, ValueError):
        return {}


def faelles_skift() -> str:
    d = faelles()
    d.update(key=secrets.token_urlsafe(32), aktiv=True, oprettet=int(time.time()), hentninger=0, hentet={})
    d.pop("sidst", None)
    _skriv(DELT, d)
    return d["key"]


def faelles_aktiv(aktiv: bool) -> None:
    d = faelles()
    if not d.get("key"):
        faelles_skift()
        d = faelles()
    d["aktiv"] = aktiv
    _skriv(DELT, d)


def faelles_slaa_op(key: str, hvad: str = "") -> bool:
    d = faelles()
    if not (KEY.match(key or "") and d.get("aktiv") and d.get("key")
            and secrets.compare_digest(key, d["key"])):
        return False
    if hvad:
        d["hentninger"] = d.get("hentninger", 0) + 1
        d["sidst"] = int(time.time())
        d.setdefault("hentet", {})
        d["hentet"][hvad] = d["hentet"].get(hvad, 0) + 1
        _skriv(DELT, d)
    return True


# ---------------- gæt på nøgler ----------------
# Nøglerne er 256 bit og kan ikke gættes, men en klient der bliver ved med at
# ramme forkerte nøgler, får 404 uden at vi overhovedet slår op (og samme svar
# som en ukendt sti, så den ikke røber noget).
_MISS: dict[str, list[float]] = {}
MISS_GRAENSE, MISS_VINDUE = 20, 600


def blokeret(ip: str) -> bool:
    nu = time.time()
    t = [x for x in _MISS.get(ip, []) if nu - x < MISS_VINDUE]
    _MISS[ip] = t
    return len(t) >= MISS_GRAENSE


def forkert(ip: str) -> None:
    _MISS.setdefault(ip, []).append(time.time())
    if len(_MISS) > 5000:          # hold hukommelsen nede ved en storm
        for k in list(_MISS)[:2500]:
            _MISS.pop(k, None)