"""Podcast chapters: Podlove Simple Chapters from the feed, Podcasting 2.0 JSON chapters fetched from the URL a feed points to, and the CHAP frames of the ID3v2 tag at the start of the audio file. 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 and the audio file live on servers the feed's author chose, so fetching them 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. Transcripts (transcripts.py) are fetched the same way, through `fetch_bytes`. """ import json import re import time from urllib.parse import urljoin import requests from accounts.webdav import WebDAVError, WebDAVInputError, _assert_peer_is_safe, _read_capped, assert_safe_url from . import id3 MAX_CHAPTERS = 500 MAX_TITLE = 500 MAX_URL = 1000 CHAPTERS_TIMEOUT = 10 # seconds CHAPTERS_MAX_BYTES = 1024 * 1024 CHAPTERS_MAX_REDIRECTS = 3 ID3_MAX_BYTES = 4 * 1024 * 1024 # of the audio file's start; a larger tag is read this far ID3_DEADLINE = 30 # seconds for reading it, however slowly it trickles in ID3_CHUNK = 64 * 1024 _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).""" # Generic names for the two kinds of failure, for other fetches (transcripts). FetchUnavailable = ChaptersUnavailable FetchInvalid = ChaptersInvalid def _open(url, headers, method='GET', timeout=CHAPTERS_TIMEOUT, **kwargs): """`method` (GET or POST) `url` with every redirect hop checked; the streamed response of the final hop, which the caller has to close. `kwargs` (data, files) go to requests; a POST only follows 307/308, which keep the method and body. Raises ChaptersInvalid for permanent problems and ChaptersUnavailable for ones that may go away (network trouble, server errors). """ redirects = (301, 302, 303, 307, 308) if method == 'GET' else (307, 308) for _ in range(CHAPTERS_MAX_REDIRECTS + 1): try: assert_safe_url(url) except WebDAVInputError as e: raise ChaptersInvalid(str(e)) try: send = requests.get if method == 'GET' else requests.post r = send(url, timeout=timeout, allow_redirects=False, stream=True, headers=headers, **kwargs) except requests.RequestException as e: raise ChaptersUnavailable(e.__class__.__name__) try: _assert_peer_is_safe(r) if r.status_code in redirects: location = r.headers.get('Location') if not location: raise ChaptersInvalid('redirect without a location') url = urljoin(url, location) r.close() continue if r.status_code >= 500 or r.status_code == 429: raise ChaptersUnavailable(f'HTTP {r.status_code}') if r.status_code >= 400 or r.status_code in (301, 302, 303): raise ChaptersInvalid(f'HTTP {r.status_code}') except WebDAVInputError as e: r.close() raise ChaptersInvalid(str(e)) except Exception: r.close() raise return r raise ChaptersInvalid('too many redirects') def fetch_bytes(url, max_bytes, accept='*/*', agent='diora podcast chapters'): """The body at `url`, at most `max_bytes` of it (more is ChaptersInvalid).""" r = _open(url, {'User-Agent': agent, 'Accept': accept}) try: return _read_capped(r, max_bytes, lambda: ChaptersInvalid('too large')) except WebDAVError as e: # transfer broke off in _read_capped raise ChaptersUnavailable(str(e)) finally: r.close() def fetch_json(url): """Download and parse the JSON at `url`.""" body = fetch_bytes(url, CHAPTERS_MAX_BYTES, accept='application/json') try: return json.loads(body) except ValueError: raise ChaptersInvalid('not JSON') 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') # --------------------------------------------------------------------------- # ID3v2 chapters (CHAP frames at the start of the audio file) # --------------------------------------------------------------------------- def read_id3_tag(url): """The ID3v2 tag at the start of the file at `url` (at most ID3_MAX_BYTES of it), or b'' if the file doesn't start with one. Asks for just that part with a Range request. Servers that ignore Range and send the whole file are read only as far as needed, then dropped. """ r = _open(url, { 'User-Agent': 'diora podcast chapters', 'Range': f'bytes=0-{ID3_MAX_BYTES - 1}', 'Accept-Encoding': 'identity', }) data = bytearray() need = None # bytes of the tag, known once its header is in deadline = time.monotonic() + ID3_DEADLINE try: for chunk in r.iter_content(chunk_size=ID3_CHUNK): data += chunk if need is None and len(data) >= id3.HEADER_SIZE: size = id3.tag_size(bytes(data[:id3.HEADER_SIZE])) if size is None: return b'' need = min(size, ID3_MAX_BYTES) if need is not None and len(data) >= need: break if time.monotonic() > deadline: raise ChaptersUnavailable('too slow') except requests.RequestException as e: raise ChaptersUnavailable(e.__class__.__name__) finally: r.close() if need is None: return b'' # shorter than a tag header return bytes(data[:need]) # may be less if the file ended early def fetch_id3(url): """Normalised chapters from the ID3v2 tag of the audio file at `url`; [] if it has none.""" return _finish( _chapter(start_ms / 1000, title, link) for start_ms, _end_ms, title, link in id3.chapters(read_id3_tag(url)) )