diora-web/podcasts/chapters.py

"""Podcast chapters: Podlove Simple Chapters from the feed, and Podcasting 2.0
JSON chapters fetched from the URL a feed points to.

Both end up in one shape, a list sorted by start time:
    [{"start": 62.5, "title": "…", "url": "https://…", "img": "https://…"}]
url and img only when present.

The JSON file lives on a server the feed's author chose, so fetching it is an
SSRF surface like the WebDAV import, and it reuses that module's guards:
public addresses only (checked before connecting and again on the connected
socket), redirects followed by hand with each hop re-checked, and a size cap.
"""

import json
import re
from urllib.parse import urljoin

import requests

from accounts.webdav import WebDAVError, WebDAVInputError, _assert_peer_is_safe, _read_capped, assert_safe_url

MAX_CHAPTERS = 500
MAX_TITLE = 500
MAX_URL = 1000
CHAPTERS_TIMEOUT = 10          # seconds
CHAPTERS_MAX_BYTES = 1024 * 1024
CHAPTERS_MAX_REDIRECTS = 3

_SCHEMES = ('http://', 'https://')


def _clean_url(value):
    value = (value or '').strip() if isinstance(value, str) else ''
    if value.lower().startswith(_SCHEMES) and len(value) <= MAX_URL:
        return value
    return ''


def _chapter(start, title, url='', img=''):
    """One normalised chapter, or None if the start isn't a usable time."""
    try:
        start = float(start)
    except (TypeError, ValueError):
        return None
    if start != start or start < 0 or start == float('inf'):  # NaN, negative, inf
        return None
    ch = {'start': round(start, 3), 'title': ' '.join(str(title or '').split())[:MAX_TITLE]}
    url, img = _clean_url(url), _clean_url(img)
    if url:
        ch['url'] = url
    if img:
        ch['img'] = img
    return ch


def _finish(chapters):
    chapters = [c for c in chapters if c is not None]
    chapters.sort(key=lambda c: c['start'])
    return chapters[:MAX_CHAPTERS]


# ---------------------------------------------------------------------------
# Podlove Simple Chapters (inline in the feed)
# ---------------------------------------------------------------------------

# "1:02:03.500", "01:02:03", "02:03", "123.5" (normal play time, as PSC uses)
_NPT = re.compile(r'^\s*(?:(?:(\d+):)?(\d{1,2}):)?(\d+(?:\.\d+)?)\s*$')


def parse_npt(value):
    """Seconds (float) from a PSC start time, or None."""
    m = _NPT.match(str(value or ''))
    if not m:
        return None
    hours, minutes, seconds = m.groups()
    return int(hours or 0) * 3600 + int(minutes or 0) * 60 + float(seconds)


def from_psc(entry):
    """Chapters from feedparser's entry["psc_chapters"], [] if there are none."""
    raw = (entry.get('psc_chapters') or {}).get('chapters') or []
    return _finish(
        _chapter(parse_npt(c.get('start')), c.get('title'), c.get('href'), c.get('image'))
        for c in raw
    )


# ---------------------------------------------------------------------------
# Podcasting 2.0 JSON chapters (fetched)
# ---------------------------------------------------------------------------

def from_json(data):
    """Normalise a JSON chapters document (https://github.com/Podcastindex-org/
    podcast-namespace/blob/main/chapters/jsonChapters.md)."""
    raw = data.get('chapters') if isinstance(data, dict) else None
    if not isinstance(raw, list):
        raise ValueError('no chapters list')
    chapters = []
    for c in raw:
        if not isinstance(c, dict):
            continue
        # toc: false = "don't show this one to the user" (e.g. art changes).
        if c.get('toc') is False:
            continue
        chapters.append(_chapter(c.get('startTime'), c.get('title'), c.get('url'), c.get('img')))
    return _finish(chapters)


class ChaptersUnavailable(Exception):
    """The chapters can't be loaded right now; worth trying again later."""


class ChaptersInvalid(Exception):
    """The chapters URL is unusable for good (blocked, missing, not chapters)."""


def fetch_json(url):
    """Download and parse the JSON at `url`.

    Raises ChaptersInvalid for permanent problems and ChaptersUnavailable for
    ones that may go away (network trouble, server errors).
    """
    for _ in range(CHAPTERS_MAX_REDIRECTS + 1):
        try:
            assert_safe_url(url)
        except WebDAVInputError as e:
            raise ChaptersInvalid(str(e))
        try:
            r = requests.get(
                url, timeout=CHAPTERS_TIMEOUT, allow_redirects=False, stream=True,
                headers={'User-Agent': 'diora podcast chapters', 'Accept': 'application/json'},
            )
        except requests.RequestException as e:
            raise ChaptersUnavailable(e.__class__.__name__)
        try:
            _assert_peer_is_safe(r)
            if r.status_code in (301, 302, 303, 307, 308):
                location = r.headers.get('Location')
                if not location:
                    raise ChaptersInvalid('redirect without a location')
                url = urljoin(url, location)
                continue
            if r.status_code >= 500 or r.status_code == 429:
                raise ChaptersUnavailable(f'HTTP {r.status_code}')
            if r.status_code >= 400:
                raise ChaptersInvalid(f'HTTP {r.status_code}')
            body = _read_capped(r, CHAPTERS_MAX_BYTES, lambda: ChaptersInvalid('too large'))
        except WebDAVInputError as e:
            raise ChaptersInvalid(str(e))
        except WebDAVError as e:  # transfer broke off in _read_capped
            raise ChaptersUnavailable(str(e))
        finally:
            r.close()
        try:
            return json.loads(body)
        except ValueError:
            raise ChaptersInvalid('not JSON')
    raise ChaptersInvalid('too many redirects')


def fetch(url):
    """Normalised chapters from a Podcasting 2.0 chapters URL."""
    try:
        return from_json(fetch_json(url))
    except ValueError:
        raise ChaptersInvalid('not a chapters document')