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

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"
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) -> Path | None:
    p = ROOT / name / "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, key: str, base: str, med_kladder: bool) -> str | None:
    s = show(name)
    if not s:
        return None
    pre = f"{base}/podcast/{key}/{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>
"""