Podcasts: Kapitel aus dem ID3-Tag der Audiodatei, Transkripte

- Ohne Kapitel im Feed (kein PSC, keine chapters_url oder dort nichts)
  liest die Folgen-Detailansicht einmalig den ID3v2-Tag am Anfang der
  Audiodatei (Range-Anfrage, höchstens 4 MB, Weiterleitungen mit
  SSRF-Prüfung) und übernimmt dessen CHAP-Frames (v2.3/v2.4,
  Unsynchronisation, erweiterter Header, Synchsafe-Größen, CTOC-Reihenfolge).
  Eigener Parser nach der ID3-Spezifikation; Ergebnis wird wie die
  JSON-Kapitel zwischengespeichert, vorübergehende Fehler nicht.
- Neu: GET /podcasts/episodes/<id>/transcript/ holt das beste Transkript
  (JSON vor VTT vor SRT, Deutsch bzw. ohne Sprachangabe vor Englisch),
  zerlegt es in Abschnitte {start, end, speaker, text}, fasst kurze
  Schnipsel desselben Sprechers bis zum Satzende bzw. ~30 s zusammen und
  speichert das Ergebnis (eigene Tabelle EpisodeTranscript). HTML: 404
  {"error": "unsupported"}.
- Folgen-Detail liefert zusätzlich has_transcript.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

podcasts/chapters.py

@@ -1,30 +1,39 @@
-"""Podcast chapters: Podlove Simple Chapters from the feed, and Podcasting 2.0
+"""Podcast chapters: Podlove Simple Chapters from the feed, Podcasting 2.0
-JSON chapters fetched from the URL a feed points to.
+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 lives on a server the feed's author chose, so fetching it is an
+The JSON file and the audio file live on servers the feed's author chose, so
-SSRF surface like the WebDAV import, and it reuses that module's guards:
+fetching them is an SSRF surface like the WebDAV import, and it reuses that
-public addresses only (checked before connecting and again on the connected
+module's guards: public addresses only (checked before connecting and again
-socket), redirects followed by hand with each hop re-checked, and a size cap.
+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://')
 
@@ -114,8 +123,14 @@ class ChaptersInvalid(Exception):
     """The chapters URL is unusable for good (blocked, missing, not chapters)."""
 
 
-def fetch_json(url):
+# Generic names for the two kinds of failure, for other fetches (transcripts).
-    """Download and parse the JSON at `url`.
+FetchUnavailable = ChaptersUnavailable
+FetchInvalid = ChaptersInvalid
+
+
+def _open(url, headers):
+    """GET `url` with every redirect hop checked; the streamed response of
+    the final hop, which the caller has to close.
 
     Raises ChaptersInvalid for permanent problems and ChaptersUnavailable for
     ones that may go away (network trouble, server errors).
@@ -126,10 +141,7 @@ def fetch_json(url):
         except WebDAVInputError as e:
             raise ChaptersInvalid(str(e))
         try:
-            r = requests.get(
+            r = requests.get(url, timeout=CHAPTERS_TIMEOUT, allow_redirects=False, stream=True, headers=headers)
-                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:
@@ -139,23 +151,40 @@ def fetch_json(url):
                 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:
                 raise ChaptersInvalid(f'HTTP {r.status_code}')
-            body = _read_capped(r, CHAPTERS_MAX_BYTES, lambda: ChaptersInvalid('too large'))
         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')
-    raise ChaptersInvalid('too many redirects')
 
 
 def fetch(url):
@@ -164,3 +193,52 @@ def fetch(url):
         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))
+    )

podcasts/id3.py

@@ -0,0 +1,281 @@
+"""Chapters embedded in the ID3v2 tag at the start of an MP3 file.
+
+Written from the ID3v2.3.0 and ID3v2.4.0 specifications and the ID3v2
+Chapter Frame Addendum (CHAP and CTOC frames). Only what chapters need is
+read: the tag header, an optional extended header (skipped), and the frames.
+Of those, CHAP frames give the chapters, with the title from an embedded TIT2
+frame and a link from WXXX or another W*** frame; CTOC frames, when there
+are any, give the order (which matters only for chapters starting at the
+same time, as callers sort by start). Pictures (APIC) and everything else are
+skipped.
+
+Pure parsing, no I/O: the caller hands over the first bytes of the file
+(`tag_size` says how many are needed).
+"""
+
+import re
+import struct
+import zlib
+
+HEADER_SIZE = 10
+MAX_FRAME_BYTES = 4 * 1024 * 1024   # a decompressed frame can't grow past this
+
+_FRAME_ID = re.compile(rb'[A-Z0-9]{4}\Z')
+
+# Header flags
+_UNSYNC = 0x80
+_EXTENDED = 0x40
+_FOOTER = 0x10
+
+
+def _synchsafe(raw):
+    """Integer from bytes that carry 7 bits each (the high bit is always 0)."""
+    value = 0
+    for b in raw:
+        value = (value << 7) | (b & 0x7F)
+    return value
+
+
+def tag_size(header):
+    """Bytes the ID3v2 tag at the start of a file takes up, header and footer
+    included, from the file's first 10 bytes; None if there's no such tag."""
+    if len(header) < HEADER_SIZE or header[:3] != b'ID3':
+        return None
+    major, revision, flags = header[3], header[4], header[5]
+    if major == 0xFF or revision == 0xFF or any(b & 0x80 for b in header[6:10]):
+        return None
+    size = HEADER_SIZE + _synchsafe(header[6:10])
+    if major >= 4 and flags & _FOOTER:
+        size += HEADER_SIZE
+    return size
+
+
+def _deunsync(data):
+    """Undo unsynchronisation: every $FF $00 was an $FF before."""
+    return data.replace(b'\xff\x00', b'\xff')
+
+
+# ---------------------------------------------------------------------------
+# Frames
+# ---------------------------------------------------------------------------
+
+def _plausible_frame_start(buf, at):
+    """True if a frame (or padding, or the end of `buf`) can begin at `at`."""
+    if at == len(buf):
+        return True
+    if at > len(buf):
+        return False
+    head = buf[at:at + 4]
+    return bool(_FRAME_ID.match(head)) or not head.strip(b'\x00')
+
+
+def _v24_frame_size(buf, pos, raw):
+    """Size of a v2.4 frame. The spec says synchsafe; some writers (older
+    iTunes among them) store a plain integer, so where the two differ, take
+    the one after which the next frame starts."""
+    plain = int.from_bytes(raw, 'big')
+    if any(b & 0x80 for b in raw):
+        return plain  # can't be synchsafe
+    safe = _synchsafe(raw)
+    if safe == plain or _plausible_frame_start(buf, pos + HEADER_SIZE + safe):
+        return safe
+    if _plausible_frame_start(buf, pos + HEADER_SIZE + plain):
+        return plain
+    return safe
+
+
+def _frame_payload(data, fmt, major, unsync_all):
+    """The frame's content after its format flags are dealt with, or None if
+    it can't be read (encrypted, broken compression)."""
+    if major == 3:
+        compressed, encrypted = fmt & 0x80, fmt & 0x40
+        # Added after the frame header, in flag order: decompressed size (4),
+        # encryption method (1), group id (1).
+        skip = (4 if compressed else 0) + (1 if encrypted else 0) + (1 if fmt & 0x20 else 0)
+        unsync = False  # v2.3 unsynchronises the whole tag, undone already
+    else:
+        compressed, encrypted = fmt & 0x08, fmt & 0x04
+        # Group id (1), encryption method (1), data length indicator (4).
+        skip = (1 if fmt & 0x40 else 0) + (1 if encrypted else 0) + (4 if fmt & 0x01 else 0)
+        unsync = unsync_all or fmt & 0x02
+    if encrypted:
+        return None
+    data = data[skip:]
+    if unsync:
+        data = _deunsync(data)
+    if compressed:
+        try:
+            data = zlib.decompressobj().decompress(data, MAX_FRAME_BYTES)
+        except zlib.error:
+            return None
+    return data
+
+
+def _frames(buf, pos, major, unsync_all=False):
+    """(frame id, content) for each complete frame in buf[pos:]."""
+    while pos + HEADER_SIZE <= len(buf):
+        frame_id = buf[pos:pos + 4]
+        if not _FRAME_ID.match(frame_id):
+            return  # padding, or something we can't make sense of
+        raw_size = buf[pos + 4:pos + 8]
+        size = _v24_frame_size(buf, pos, raw_size) if major >= 4 else int.from_bytes(raw_size, 'big')
+        fmt = buf[pos + 9]
+        start = pos + HEADER_SIZE
+        if start + size > len(buf):
+            return  # cut off (the caller read only part of a large tag)
+        payload = _frame_payload(buf[start:start + size], fmt, major, unsync_all)
+        if payload is not None:
+            yield frame_id.decode('ascii'), payload
+        pos = start + size
+
+
+# ---------------------------------------------------------------------------
+# Frame contents
+# ---------------------------------------------------------------------------
+
+_CODECS = {0: 'latin-1', 1: 'utf-16', 2: 'utf-16-be', 3: 'utf-8'}
+
+
+def _decode(raw, encoding):
+    codec = _CODECS.get(encoding)
+    if codec is None:
+        return ''
+    if encoding in (1, 2) and len(raw) % 2:
+        raw = raw[:-1]
+    return raw.decode(codec, errors='replace')
+
+
+def _text_frame(data):
+    """The (first) string of a T*** frame."""
+    if not data:
+        return ''
+    text = _decode(data[1:], data[0])
+    # v2.4 allows several strings, separated by NUL; the first is the title.
+    return text.split('\x00')[0].replace('', '').strip()
+
+
+def _wxxx_frame(data):
+    """The URL of a WXXX frame: encoding, description, NUL, URL (latin-1)."""
+    if not data:
+        return ''
+    encoding, rest = data[0], data[1:]
+    if encoding in (1, 2):  # the description ends with a two-byte NUL
+        i = 0
+        while i + 1 < len(rest) and not (rest[i] == 0 and rest[i + 1] == 0):
+            i += 2
+        url = rest[i + 2:]
+    else:
+        i = rest.find(b'\x00')
+        url = rest[i + 1:] if i >= 0 else b''
+    return url.split(b'\x00')[0].decode('latin-1').strip()
+
+
+def _url_frame(data):
+    """The URL of any other W*** frame, which is nothing but the URL."""
+    return data.split(b'\x00')[0].decode('latin-1').strip()
+
+
+def _chap(data, major):
+    """(element id, start ms, end ms, title, url) of a CHAP frame, or None."""
+    nul = data.find(b'\x00')
+    if nul < 0 or len(data) < nul + 17:
+        return None
+    element_id = data[:nul]
+    start_ms, end_ms, _start_offset, _end_offset = struct.unpack('>IIII', data[nul + 1:nul + 17])
+    title = url = ''
+    # Embedded frames, in the same format as the tag's own. Unsynchronisation
+    # of the whole tag has already been undone on `data`.
+    for frame_id, payload in _frames(data, nul + 17, major):
+        if frame_id == 'TIT2' and not title:
+            title = _text_frame(payload)
+        elif frame_id == 'WXXX' and not url:
+            url = _wxxx_frame(payload)
+        elif frame_id.startswith('W') and not url:
+            url = _url_frame(payload)
+    return element_id, start_ms, end_ms, title, url
+
+
+def _ctoc(data):
+    """(element id, top level?, child element ids) of a CTOC frame, or None."""
+    nul = data.find(b'\x00')
+    if nul < 0 or len(data) < nul + 3:
+        return None
+    element_id, flags, count = data[:nul], data[nul + 1], data[nul + 2]
+    children, pos = [], nul + 3
+    for _ in range(count):
+        end = data.find(b'\x00', pos)
+        if end < 0:
+            break
+        children.append(data[pos:end])
+        pos = end + 1
+    return element_id, bool(flags & 0x02), children
+
+
+def _toc_order(chaps, tocs):
+    """Element ids of the chapters the table of contents lists, in its order
+    (top-level CTOCs first, nested ones walked depth first); [] if there's no
+    usable table of contents."""
+    if not tocs:
+        return []
+    nested = {child for _, _, children in tocs for child in children}
+    roots = [t for t in tocs if t[1]] or [t for t in tocs if t[0] not in nested]
+    by_id = {t[0]: t for t in tocs}
+    order, seen = [], set()
+
+    def walk(toc):
+        if toc[0] in seen:
+            return  # a loop
+        seen.add(toc[0])
+        for child in toc[2]:
+            if child in chaps:
+                if child not in order:
+                    order.append(child)
+            elif child in by_id:
+                walk(by_id[child])
+
+    for root in roots:
+        walk(root)
+    return order
+
+
+def chapters(data):
+    """[(start ms, end ms, title, url)] of the chapters in the ID3v2 tag at
+    the start of `data`, in table-of-contents order where there is one, else
+    in tag order.
+    [] if there is no tag or it has no chapters; a tag cut short yields the
+    chapters in the part that's there."""
+    size = tag_size(data)
+    if size is None:
+        return []
+    major, flags = data[3], data[5]
+    if major not in (3, 4):
+        return []  # v2.2 has no chapter frames
+    body = data[HEADER_SIZE:min(len(data), size)]
+    unsync = bool(flags & _UNSYNC)
+    if major == 3 and unsync:
+        body = _deunsync(body)
+    pos = 0
+    if flags & _EXTENDED and len(body) >= 4:
+        if major == 3:  # size of what follows the size field
+            pos = 4 + int.from_bytes(body[:4], 'big')
+        else:  # synchsafe size of the whole extended header
+            pos = _synchsafe(body[:4])
+
+    chaps, order, tocs = {}, [], []
+    for frame_id, payload in _frames(body, pos, major, unsync_all=(major == 4 and unsync)):
+        if frame_id == 'CHAP':
+            chap = _chap(payload, major)
+            if chap and chap[0] not in chaps:
+                chaps[chap[0]] = chap
+                order.append(chap[0])
+        elif frame_id == 'CTOC':
+            toc = _ctoc(payload)
+            if toc:
+                tocs.append(toc)
+
+    # The table of contents' order first; chapters it doesn't list are kept
+    # (after it) rather than dropped, since writers don't always list all.
+    toc = _toc_order(chaps, tocs)
+    listed = set(toc)
+    order = toc + [e for e in order if e not in listed]
+    return [chaps[e][1:] for e in order]

podcasts/migrations/0007_episode_transcript_cache.py

@@ -0,0 +1,25 @@
+# Generated by Django 6.1 on 2026-10-02 07:44
+
+import django.db.models.deletion
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('podcasts', '0006_episodeprogress_favorite'),
+    ]
+
+    operations = [
+        migrations.CreateModel(
+            name='EpisodeTranscript',
+            fields=[
+                ('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
+                ('sources', models.JSONField(default=list)),
+                ('data', models.JSONField(blank=True, null=True)),
+                ('error', models.CharField(blank=True, max_length=20)),
+                ('fetched_at', models.DateTimeField(auto_now=True)),
+                ('episode', models.OneToOneField(on_delete=django.db.models.deletion.CASCADE, related_name='transcript_cache', to='podcasts.podcastepisode')),
+            ],
+        ),
+    ]

podcasts/models.py

@@ -37,9 +37,11 @@ class PodcastEpisode(models.Model):
     # Podcasting 2.0 <podcast:chapters url=…>: a JSON file, loaded on demand.
     chapters_url = models.URLField(max_length=1000, blank=True)
     # [{start: seconds, title, url?, img?}], from Podlove Simple Chapters in
-    # the feed, or the JSON at chapters_url once it has been fetched.
+    # the feed, or once fetched: the JSON at chapters_url, else (or if that
+    # has none) the CHAP frames of the audio file's ID3v2 tag.
     chapters = models.JSONField(default=list, blank=True)
-    # When chapters_url was fetched into `chapters`; null = not yet.
+    # When chapters were fetched into `chapters` (from chapters_url or the
+    # audio file); null = not yet.
     chapters_fetched_at = models.DateTimeField(null=True, blank=True)
     # [{url, type, language}] from <podcast:transcript>.
     transcripts = models.JSONField(default=list, blank=True)
@@ -52,6 +54,27 @@ class PodcastEpisode(models.Model):
         return self.title
 
 
+class EpisodeTranscript(models.Model):
+    """An episode's transcript, fetched and parsed once (transcripts.py).
+
+    A table of its own rather than fields on PodcastEpisode: a parsed
+    transcript runs to hundreds of KB, and episode lists load whole rows.
+    """
+    episode = models.OneToOneField(PodcastEpisode, on_delete=models.CASCADE, related_name='transcript_cache')
+    # URLs of the transcripts it was made from, best first. When the feed
+    # offers different ones, the cache is stale.
+    sources = models.JSONField(default=list)
+    # {"segments": [...], "source": {...}}, or null with an error.
+    data = models.JSONField(null=True, blank=True)
+    # '' when data is there; else why there's none for good: 'unsupported'
+    # (no format with timing), 'invalid' (missing, broken, too large).
+    error = models.CharField(max_length=20, blank=True)
+    fetched_at = models.DateTimeField(auto_now=True)
+
+    def __str__(self):
+        return f'Transcript of {self.episode}'
+
+
 # An episode counts as played this close to its end (AntennaPod's default),
 # the same rule as the apps' "als gehört zählen".
 PLAYED_BEFORE_END_SECONDS = 30

podcasts/tests.py

@@ -487,6 +487,7 @@ class EpisodeDetailTests(TestCase):
                                               'rel="nofollow noopener noreferrer">x</a></p>')
         self.assertEqual(data['timestamps'], [{'seconds': 62, 'label': 'Thema x'}])
         self.assertEqual(data['transcripts'], self.ep.transcripts)
+        self.assertIs(data['has_transcript'], True)
         self.assertEqual(data['feed'], {'id': self.feed.id, 'title': 'Feed',
                                         'artwork_url': 'https://example.org/art.jpg'})
         self.assertEqual((data['position_seconds'], data['played'], data['in_queue']), (42, False, False))
@@ -524,6 +525,7 @@ class EpisodeDetailTests(TestCase):
 
     def test_private_chapters_url_is_refused_and_remembered(self):
         self.ep.chapters_url = 'http://127.0.0.1/c.json'
+        self.ep.audio_url = 'http://127.0.0.1/a.mp3'  # the ID3 fallback is refused too
         self.ep.save()
         with patch('podcasts.chapters.requests.get', self.fake_get):
             r = self.client.get(f'/podcasts/episodes/{self.ep.id}/')
@@ -894,3 +896,473 @@ class OpmlExportTests(TestCase):
     def test_login_required(self):
         self.client.logout()
         self.assertEqual(self.client.get('/podcasts/feeds/export.opml').status_code, 401)
+
+
+# ---------------------------------------------------------------------------
+# ID3v2 chapters (CHAP frames in the audio file)
+# ---------------------------------------------------------------------------
+
+def _ss(n):
+    """Synchsafe 4-byte integer."""
+    return bytes([(n >> 21) & 0x7F, (n >> 14) & 0x7F, (n >> 7) & 0x7F, n & 0x7F])
+
+
+def _frame(frame_id, data, v=3, fmt=0, plain_size=False):
+    size = _ss(len(data)) if v == 4 and not plain_size else len(data).to_bytes(4, 'big')
+    return frame_id.encode() + size + bytes([0, fmt]) + data
+
+
+def _text(s, enc=3):
+    codec = {0: 'latin-1', 1: 'utf-16', 3: 'utf-8'}[enc]
+    return bytes([enc]) + s.encode(codec) + (b'\0\0' if enc == 1 else b'\0')
+
+
+def _chap(element_id, start_ms, end_ms, subframes=b''):
+    import struct
+    return element_id + b'\0' + struct.pack('>IIII', start_ms, end_ms, 0xFFFFFFFF, 0xFFFFFFFF) + subframes
+
+
+def _ctoc(element_id, children, top=True):
+    return element_id + b'\0' + bytes([0x03 if top else 0x01, len(children)]) + b''.join(c + b'\0' for c in children)
+
+
+def _unsync(data):
+    """ID3 unsynchronisation: a $00 after every $FF followed by %111xxxxx or $00 (or at the end)."""
+    out = bytearray()
+    for i, b in enumerate(data):
+        out.append(b)
+        if b == 0xFF and (i + 1 == len(data) or data[i + 1] == 0 or data[i + 1] >= 0xE0):
+            out.append(0)
+    return bytes(out)
+
+
+def _tag(body, v=3, flags=0, padding=64):
+    body = body + b'\0' * padding
+    return b'ID3' + bytes([v, 0, flags]) + _ss(len(body)) + body
+
+
+_APIC = b'\0image/jpeg\0\x03\0' + b'\xff\xd8\xff\xe0' + b'p' * 300
+
+# A v2.3 tag as podcast tools write it: title, cover, table of contents, chapters.
+ID3V23 = _tag(
+    _frame('TIT2', _text('Folge 1'))
+    + _frame('APIC', _APIC * 10)
+    + _frame('CTOC', _ctoc(b'toc', [b'ch1', b'ch2', b'ch3']))
+    + _frame('CHAP', _chap(b'ch1', 0, 60000, _frame('TIT2', _text('Intro', 0))))
+    + _frame('CHAP', _chap(b'ch2', 60000, 125500,
+                           _frame('TIT2', _text('Zweiter Teil – ü', 1))
+                           + _frame('WXXX', b'\x00Link\x00https://x.org/2')
+                           + _frame('APIC', _APIC)))
+    + _frame('CHAP', _chap(b'ch3', 125500, 200000))
+)
+ID3V23_CHAPTERS = [
+    {'start': 0.0, 'title': 'Intro'},
+    {'start': 60.0, 'title': 'Zweiter Teil – ü', 'url': 'https://x.org/2'},
+    {'start': 125.5, 'title': ''},
+]
+
+
+class ID3ParserTests(TestCase):
+    def chapters(self, data):
+        from .chapters import _chapter, _finish
+        from .id3 import chapters
+        return _finish(_chapter(s / 1000, t, u) for s, _e, t, u in chapters(data))
+
+    def test_v23(self):
+        self.assertEqual(self.chapters(ID3V23), ID3V23_CHAPTERS)
+
+    def test_v23_unsynchronised_with_extended_header(self):
+        ext = (6).to_bytes(4, 'big') + b'\0\0' + b'\0\0\0\0'
+        frames = (
+            _frame('CHAP', _chap(b'a', 0xFFE0, 70000, _frame('TIT2', _text('Ünsynchron', 1))))
+            + _frame('CHAP', _chap(b'b', 70000, 80000, _frame('TIT2', _text('Zwei')) + _frame('WOAF', b'https://x.org/w')))
+        )
+        body = _unsync(ext + frames + b'\0' * 16)
+        self.assertNotEqual(body, ext + frames + b'\0' * 16)  # there really was something to unsynchronise
+        tag = b'ID3' + bytes([3, 0, 0xC0]) + _ss(len(body)) + body
+        self.assertEqual(self.chapters(tag), [
+            {'start': 65.504, 'title': 'Ünsynchron'},
+            {'start': 70.0, 'title': 'Zwei', 'url': 'https://x.org/w'},
+        ])
+
+    def test_v24_synchsafe_sizes_toc_and_frame_flags(self):
+        # The cover makes the CHAP frame longer than 127 bytes, where synchsafe
+        # and plain sizes differ.
+        long_chap = _chap(b'c2', 30000, 60000, _frame('APIC', _APIC, v=4) + _frame('TIT2', _text('Mit Bild'), v=4))
+        self.assertGreater(len(long_chap), 127)
+        # Unsynchronised frame with a data length indicator (flags %00000011).
+        raw = _chap(b'c3', 0x0001FFE0, 200000, _frame('TIT2', _text('ÿÿ Flags', 0), v=4))
+        flagged = b'CHAP' + _ss(4 + len(_unsync(raw))) + bytes([0, 0x03]) + _ss(len(raw)) + _unsync(raw)
+        tag = _tag(
+            _frame('CTOC', _ctoc(b'top', [b'sub', b'c3']), v=4)
+            + _frame('CTOC', _ctoc(b'sub', [b'c2', b'c1'], top=False), v=4)
+            + _frame('CHAP', _chap(b'c1', 0, 30000, _frame('TIT2', _text('Eins'), v=4)), v=4)
+            + _frame('CHAP', long_chap, v=4)
+            + flagged
+            + _frame('CHAP', _chap(b'extra', 0, 1000, _frame('TIT2', _text('Nicht im TOC'), v=4)), v=4),
+            v=4,
+        )
+        from .id3 import chapters
+        # Table-of-contents order (nested first), then what it doesn't list.
+        self.assertEqual([c[2] for c in chapters(tag)], ['Mit Bild', 'Eins', 'ÿÿ Flags', 'Nicht im TOC'])
+        self.assertEqual(self.chapters(tag), [
+            {'start': 0.0, 'title': 'Eins'},
+            {'start': 0.0, 'title': 'Nicht im TOC'},
+            {'start': 30.0, 'title': 'Mit Bild'},
+            {'start': 131.04, 'title': 'ÿÿ Flags'},
+        ])
+
+    def test_v24_with_plain_frame_sizes(self):
+        # Some writers put v2.3-style sizes into v2.4 tags.
+        long_chap = _chap(b'c1', 5000, 60000, _frame('APIC', _APIC, v=4, plain_size=True)
+                          + _frame('TIT2', _text('Plain'), v=4, plain_size=True))
+        tag = _tag(_frame('CHAP', long_chap, v=4, plain_size=True)
+                   + _frame('CHAP', _chap(b'c2', 60000, 70000, _frame('TIT2', _text('Danach'), v=4)), v=4), v=4)
+        self.assertEqual(self.chapters(tag), [{'start': 5.0, 'title': 'Plain'}, {'start': 60.0, 'title': 'Danach'}])
+
+    def test_no_tag_or_no_chapters(self):
+        from .id3 import chapters, tag_size
+        self.assertEqual(chapters(b'\xff\xfb\x90\x00' * 100), [])
+        self.assertEqual(chapters(_tag(_frame('TIT2', _text('Nur Titel')))), [])
+        self.assertIsNone(tag_size(b'ID3\x03\x00\x00\x80\x00\x00\x00'))  # not synchsafe
+        self.assertEqual(tag_size(_tag(b'', padding=300)), 310)
+
+
+class _StreamResponse(_FakeResponse):
+    """Hands out the body in the chunks asked for, counting what was taken."""
+
+    def __init__(self, body=b'', status=200, headers=None):
+        super().__init__(body, status, headers)
+        self.sent = 0
+
+    def iter_content(self, chunk_size=1):
+        for i in range(0, len(self.body), chunk_size):
+            chunk = self.body[i:i + chunk_size]
+            self.sent += len(chunk)
+            yield chunk
+
+
+class ID3ChapterFetchTests(TestCase):
+    def setUp(self):
+        self.user = User.objects.create_user(username='alice', password='pw12345678')
+        self.client.force_login(self.user)
+        self.feed = PodcastFeed.objects.create(user=self.user, rss_url='https://example.org/feed', title='Feed')
+        self.ep = PodcastEpisode.objects.create(feed=self.feed, guid='a', title='A',
+                                                audio_url='https://example.org/a.mp3')
+        self.calls = []
+        self.responses = {}
+
+    def fake_get(self, url, **kwargs):
+        self.calls.append((url, kwargs.get('headers') or {}))
+        r = self.responses[url]
+        return r() if callable(r) else r
+
+    def get(self):
+        with patch('podcasts.chapters.assert_safe_url'), patch('podcasts.chapters.requests.get', self.fake_get):
+            return self.client.get(f'/podcasts/episodes/{self.ep.id}/')
+
+    def test_range_request_answered_with_206(self):
+        from .chapters import ID3_MAX_BYTES
+        self.responses['https://example.org/a.mp3'] = lambda: _StreamResponse(ID3V23 + b'\xff\xfb' * 100, status=206)
+        self.assertEqual(self.get().json()['chapters'], ID3V23_CHAPTERS)
+        self.assertEqual(self.calls[0][1]['Range'], f'bytes=0-{ID3_MAX_BYTES - 1}')
+        # Cached: the second detail doesn't touch the file again.
+        self.assertEqual(self.get().json()['chapters'], ID3V23_CHAPTERS)
+        self.assertEqual(len(self.calls), 1)
+        self.ep.refresh_from_db()
+        self.assertEqual(self.ep.chapters, ID3V23_CHAPTERS)
+        self.assertIsNotNone(self.ep.chapters_fetched_at)
+
+    def test_server_ignoring_range_is_read_only_as_far_as_the_tag(self):
+        from .chapters import ID3_CHUNK
+        whole_file = _StreamResponse(ID3V23 + b'\xff\xfb' * (3 * 1024 * 1024), status=200)
+        self.responses['https://example.org/a.mp3'] = whole_file
+        self.assertEqual(self.get().json()['chapters'], ID3V23_CHAPTERS)
+        self.assertLessEqual(whole_file.sent, ID3_CHUNK)
+
+    def test_tag_larger_than_the_cap(self):
+        from .chapters import ID3_CHUNK, ID3_MAX_BYTES
+        tag = _tag(
+            _frame('CHAP', _chap(b'ch1', 0, 1000, _frame('TIT2', _text('Vorne'))))
+            + _frame('APIC', b'\0image/png\0\x03\0' + b'p' * (ID3_MAX_BYTES + 1024))
+            + _frame('CHAP', _chap(b'ch2', 1000, 2000, _frame('TIT2', _text('Hinter der Grenze')))),
+        )
+        response = _StreamResponse(tag + b'\xff\xfb' * 1024, status=206)
+        self.responses['https://example.org/a.mp3'] = response
+        self.assertEqual(self.get().json()['chapters'], [{'start': 0.0, 'title': 'Vorne'}])
+        self.assertLessEqual(response.sent, ID3_MAX_BYTES + ID3_CHUNK)
+
+    def test_file_without_tag_is_remembered(self):
+        self.responses['https://example.org/a.mp3'] = lambda: _StreamResponse(b'\0\0\0\x20ftypM4A ' + b'x' * 5000)
+        self.assertEqual(self.get().json()['chapters'], [])
+        self.assertEqual(self.get().json()['chapters'], [])
+        self.assertEqual(len(self.calls), 1)
+        self.ep.refresh_from_db()
+        self.assertIsNotNone(self.ep.chapters_fetched_at)
+
+    def test_empty_json_chapters_fall_back_to_the_file(self):
+        self.ep.chapters_url = 'https://x.org/c.json'
+        self.ep.save()
+        self.responses['https://x.org/c.json'] = _FakeResponse(b'{"chapters": []}')
+        self.responses['https://example.org/a.mp3'] = _StreamResponse(ID3V23, status=206)
+        self.assertEqual(self.get().json()['chapters'], ID3V23_CHAPTERS)
+        self.assertEqual([c[0] for c in self.calls], ['https://x.org/c.json', 'https://example.org/a.mp3'])
+
+    def test_json_chapters_win(self):
+        self.ep.chapters_url = 'https://x.org/c.json'
+        self.ep.save()
+        self.responses['https://x.org/c.json'] = _FakeResponse(json.dumps(CHAPTERS_JSON).encode())
+        self.assertEqual(len(self.get().json()['chapters']), 3)
+        self.assertEqual([c[0] for c in self.calls], ['https://x.org/c.json'])
+
+    def test_redirect_is_followed_and_rechecked(self):
+        self.responses['https://example.org/a.mp3'] = _FakeResponse(
+            status=302, headers={'Location': 'https://cdn.example.org/a.mp3'})
+        self.responses['https://cdn.example.org/a.mp3'] = _StreamResponse(ID3V23, status=206)
+        with patch('podcasts.chapters.assert_safe_url') as check, patch('podcasts.chapters.requests.get', self.fake_get):
+            r = self.client.get(f'/podcasts/episodes/{self.ep.id}/')
+        self.assertEqual(r.json()['chapters'], ID3V23_CHAPTERS)
+        self.assertEqual([c.args[0] for c in check.call_args_list],
+                         ['https://example.org/a.mp3', 'https://cdn.example.org/a.mp3'])
+        self.assertEqual(self.calls[1][1]['Range'], self.calls[0][1]['Range'])
+
+    def test_temporary_failure_is_not_cached(self):
+        self.responses['https://example.org/a.mp3'] = _FakeResponse(status=503)
+        self.assertEqual(self.get().json()['chapters'], [])
+        self.ep.refresh_from_db()
+        self.assertIsNone(self.ep.chapters_fetched_at)
+        self.responses['https://example.org/a.mp3'] = _StreamResponse(ID3V23, status=206)
+        self.assertEqual(self.get().json()['chapters'], ID3V23_CHAPTERS)
+
+    def test_inline_chapters_need_no_fetch(self):
+        self.ep.chapters = [{'start': 0.0, 'title': 'Inline'}]
+        self.ep.save()
+        self.assertEqual(self.get().json()['chapters'], [{'start': 0.0, 'title': 'Inline'}])
+        self.assertEqual(self.calls, [])
+
+
+class ChapterCacheRefreshTests(TestCase):
+    def setUp(self):
+        self.user = User.objects.create_user(username='alice', password='pw12345678')
+        self.feed = PodcastFeed.objects.create(user=self.user, rss_url='https://example.org/feed', title='Feed')
+
+    def refresh(self, rss):
+        with patch('podcasts.views._fetch_feed', lambda url: _parse(rss)):
+            _refresh_feed(self.feed)
+
+    def test_chapters_from_the_file_survive_a_refresh(self):
+        self.refresh(_rich_rss([('a', '')]))
+        fetched = datetime(2026, 1, 2, tzinfo=timezone.utc)
+        PodcastEpisode.objects.update(chapters=[{'start': 0.0, 'title': 'ID3'}], chapters_fetched_at=fetched)
+        self.refresh(_rich_rss([('a', '')]))
+        ep = PodcastEpisode.objects.get()
+        self.assertEqual((ep.chapters, ep.chapters_fetched_at), ([{'start': 0.0, 'title': 'ID3'}], fetched))
+        # A chapters URL turning up means fetching that.
+        self.refresh(_rich_rss([('a', '<podcast:chapters url="https://x.org/c.json" type="application/json+chapters"/>')]))
+        ep.refresh_from_db()
+        self.assertEqual((ep.chapters, ep.chapters_fetched_at), ([], None))
+
+    def test_inline_chapters_going_away_means_a_fetch(self):
+        self.refresh(_rich_rss([('a', PSC)]))
+        self.assertEqual(len(PodcastEpisode.objects.get().chapters), 4)
+        self.refresh(_rich_rss([('a', '')]))
+        ep = PodcastEpisode.objects.get()
+        self.assertEqual((ep.chapters, ep.chapters_fetched_at), ([], None))
+
+
+# ---------------------------------------------------------------------------
+# Transcripts
+# ---------------------------------------------------------------------------
+
+VTT = (
+    'WEBVTT\r\nKind: captions\r\n\r\n'
+    'NOTE Ein Kommentar\r\nüber zwei Zeilen\r\n\r\n'
+    'STYLE\r\n::cue { color: red }\r\n\r\n'
+    '1\r\n00:00.000 --> 00:02.000\r\n<v Anna>Hallo und</v>\r\n\r\n'
+    'intro-2\r\n00:02.000 --> 00:04.500 align:start position:10%\r\n<v Anna>willkommen &amp; so.</v>\r\n\r\n'
+    '00:00:04.500 --> 00:00:06.000\r\n<v.loud Ben>Danke!\r\n<v Anna>Gern.\r\n\r\n'
+    '01:00:00.000 --> 01:00:02.000\r\n<i>Ende</i> <00:00:01.000><c>gut</c>\r\n'
+)
+
+SRT = (
+    '1\n00:00:01,000 --> 00:00:03,500\nErste Zeile\nzweite Zeile\n\n'
+    '2\n00:00:03,500 --> 00:00:05,000\n<i>weiter.</i>\n\n'
+    '3\n00:00:40,000 --> 00:00:42,000\nNach der Pause\n'
+    '4\n00:00:42,000 --> 00:00:43,000\nohne Leerzeile davor\n'
+)
+
+TRANSCRIPT_JSON = {
+    'version': '1.0.0',
+    'segments': [
+        {'speaker': 'Anna', 'startTime': 0, 'endTime': 0.5, 'body': 'Hallo'},
+        {'speaker': 'Anna', 'startTime': 0.5, 'endTime': 1.2, 'body': 'zusammen.'},
+        {'speaker': 'Ben', 'startTime': 1.3, 'endTime': 2, 'body': 'Hi'},
+        {'speaker': '', 'startTime': 2.5, 'body': 'ohne Ende'},
+        {'startTime': 3, 'endTime': 4, 'body': 'und &amp; weiter'},
+        {'startTime': 'kaputt', 'body': 'weg'},
+        'kein Objekt',
+    ],
+}
+
+
+class TranscriptParsingTests(TestCase):
+    def test_vtt(self):
+        from .transcripts import merge, parse
+        self.assertEqual(merge(parse(VTT.encode(), 'vtt')), [
+            {'start': 0.0, 'end': 4.5, 'speaker': 'Anna', 'text': 'Hallo und willkommen & so.'},
+            {'start': 4.5, 'end': 6.0, 'speaker': 'Ben', 'text': 'Danke!'},
+            {'start': 4.5, 'end': 6.0, 'speaker': 'Anna', 'text': 'Gern.'},
+            {'start': 3600.0, 'end': 3602.0, 'speaker': None, 'text': 'Ende gut'},
+        ])
+
+    def test_srt(self):
+        from .transcripts import merge, parse
+        self.assertEqual(merge(parse(SRT.encode('cp1252'), 'srt')), [
+            {'start': 1.0, 'end': 5.0, 'speaker': None, 'text': 'Erste Zeile zweite Zeile weiter.'},
+            {'start': 40.0, 'end': 43.0, 'speaker': None, 'text': 'Nach der Pause ohne Leerzeile davor'},
+        ])
+
+    def test_json(self):
+        from .transcripts import merge, parse
+        self.assertEqual(merge(parse(json.dumps(TRANSCRIPT_JSON).encode(), 'json')), [
+            {'start': 0.0, 'end': 1.2, 'speaker': 'Anna', 'text': 'Hallo zusammen.'},
+            {'start': 1.3, 'end': 2.0, 'speaker': 'Ben', 'text': 'Hi'},
+            {'start': 2.5, 'end': 4.0, 'speaker': None, 'text': 'ohne Ende und & weiter'},
+        ])
+
+    def test_merging(self):
+        from .transcripts import merge
+        # One-second fragments without a sentence end: cut at 30 s.
+        words = [(float(i), i + 1.0, 'A', f'w{i}') for i in range(40)]
+        self.assertEqual([(s['start'], s['end']) for s in merge(words)], [(0.0, 30.0), (30.0, 40.0)])
+        # A sentence end closes a segment once it's 5 s long, not before.
+        frags = [(0.0, 1.0, None, 'Ja.'), (1.0, 6.0, None, 'Genau, so ist es.'), (6.0, 7.0, None, 'Weiter')]
+        self.assertEqual([s['text'] for s in merge(frags)], ['Ja. Genau, so ist es.', 'Weiter'])
+
+    def test_html_is_unsupported(self):
+        from .transcripts import Unsupported, parse
+        with self.assertRaises(Unsupported):
+            parse(b'<!DOCTYPE html><html><body><p>Hallo</p></body></html>', 'vtt')
+
+    def test_choice(self):
+        from .transcripts import candidates
+        ts = [
+            {'url': 'https://x.org/a.srt', 'type': 'application/x-subrip', 'language': 'de'},
+            {'url': 'https://x.org/a.html', 'type': 'text/html', 'language': 'de'},
+            {'url': 'https://x.org/en.json', 'type': 'application/json', 'language': 'en'},
+            {'url': 'https://x.org/a.vtt', 'type': 'text/vtt', 'language': 'de-DE'},
+            {'url': 'https://x.org/fr.json', 'type': 'application/json', 'language': 'fr'},
+            {'url': 'https://x.org/b.vtt', 'type': '', 'language': ''},
+        ]
+        self.assertEqual([t['url'] for t in candidates(ts)], [
+            'https://x.org/a.vtt', 'https://x.org/b.vtt', 'https://x.org/a.srt',
+            'https://x.org/en.json', 'https://x.org/fr.json',
+        ])
+
+
+class TranscriptEndpointTests(TestCase):
+    VTT_T = {'url': 'https://example.org/a.vtt', 'type': 'text/vtt', 'language': 'de'}
+    JSON_T = {'url': 'https://example.org/a.json', 'type': 'application/json', 'language': 'de'}
+    HTML_T = {'url': 'https://example.org/a.html', 'type': 'text/html', 'language': 'de'}
+
+    def setUp(self):
+        self.user = User.objects.create_user(username='alice', password='pw12345678')
+        self.client.force_login(self.user)
+        self.feed = PodcastFeed.objects.create(user=self.user, rss_url='https://example.org/feed', title='Feed')
+        self.ep = PodcastEpisode.objects.create(
+            feed=self.feed, guid='a', title='A', audio_url='https://example.org/a.mp3',
+            chapters=[{'start': 0.0, 'title': 'Inline'}],  # no chapter fetch in these tests
+            transcripts=[self.VTT_T, self.JSON_T, self.HTML_T],
+        )
+        self.requests = []
+        self.responses = {
+            self.VTT_T['url']: lambda: _FakeResponse(VTT.encode()),
+            self.JSON_T['url']: lambda: _FakeResponse(json.dumps(TRANSCRIPT_JSON).encode()),
+        }
+
+    def fake_get(self, url, **kwargs):
+        self.requests.append(url)
+        return self.responses[url]()
+
+    def get(self, ep=None, what='transcript/'):
+        with patch('podcasts.chapters.assert_safe_url'), patch('podcasts.chapters.requests.get', self.fake_get):
+            return self.client.get(f'/podcasts/episodes/{(ep or self.ep).id}/{what}')
+
+    def test_best_transcript_is_fetched_once(self):
+        r = self.get()
+        self.assertEqual(r.status_code, 200)
+        data = r.json()
+        self.assertEqual(data['source'], self.JSON_T)
+        self.assertEqual(data['segments'][0], {'start': 0.0, 'end': 1.2, 'speaker': 'Anna', 'text': 'Hallo zusammen.'})
+        self.assertEqual(self.get().json(), data)
+        self.assertEqual(self.requests, [self.JSON_T['url']])
+        self.assertTrue(self.get(what='').json()['has_transcript'])
+
+    def test_falls_back_to_the_next_one(self):
+        self.responses[self.JSON_T['url']] = lambda: _FakeResponse(status=404)
+        data = self.get().json()
+        self.assertEqual(data['source'], self.VTT_T)
+        self.assertEqual(data['segments'][1]['speaker'], 'Ben')
+        self.get()
+        self.assertEqual(self.requests, [self.JSON_T['url'], self.VTT_T['url']])
+
+    def test_temporary_failure_is_not_cached(self):
+        self.responses[self.JSON_T['url']] = lambda: _FakeResponse(status=503)
+        self.responses[self.VTT_T['url']] = lambda: _FakeResponse(status=503)
+        r = self.get()
+        self.assertEqual((r.status_code, r.json()), (502, {'error': 'unavailable'}))
+        # The better one failing for now: the other one is shown, but not kept.
+        self.responses[self.VTT_T['url']] = lambda: _FakeResponse(VTT.encode())
+        self.assertEqual(self.get().json()['source'], self.VTT_T)
+        self.responses[self.JSON_T['url']] = lambda: _FakeResponse(json.dumps(TRANSCRIPT_JSON).encode())
+        self.assertEqual(self.get().json()['source'], self.JSON_T)
+        self.get()
+        self.assertEqual(self.requests.count(self.JSON_T['url']), 3)
+
+    def test_html_only_is_unsupported(self):
+        self.ep.transcripts = [self.HTML_T]
+        self.ep.save()
+        r = self.get()
+        self.assertEqual((r.status_code, r.json()), (404, {'error': 'unsupported'}))
+        self.assertEqual(self.requests, [])
+        self.assertFalse(self.get(what='').json()['has_transcript'])
+
+    def test_html_served_under_another_type(self):
+        self.ep.transcripts = [self.VTT_T]
+        self.ep.save()
+        self.responses[self.VTT_T['url']] = lambda: _FakeResponse(b'<html><body>Hallo</body></html>')
+        self.assertTrue(self.get(what='').json()['has_transcript'])
+        r = self.get()
+        self.assertEqual((r.status_code, r.json()), (404, {'error': 'unsupported'}))
+        self.assertEqual(self.get().status_code, 404)
+        self.assertEqual(self.requests, [self.VTT_T['url']])
+        self.assertFalse(self.get(what='').json()['has_transcript'])
+
+    def test_no_transcript(self):
+        self.ep.transcripts = []
+        self.ep.save()
+        r = self.get()
+        self.assertEqual((r.status_code, r.json()), (404, {'error': 'no transcript'}))
+        self.assertFalse(self.get(what='').json()['has_transcript'])
+
+    def test_new_transcripts_in_the_feed_replace_the_cache(self):
+        self.ep.transcripts = [self.VTT_T]
+        self.ep.save()
+        self.assertEqual(self.get().json()['source'], self.VTT_T)
+        self.ep.transcripts = [self.VTT_T, self.JSON_T]
+        self.ep.save()
+        self.assertEqual(self.get().json()['source'], self.JSON_T)
+        self.assertEqual(self.requests, [self.VTT_T['url'], self.JSON_T['url']])
+
+    def test_other_users_episode_is_not_found(self):
+        bob = User.objects.create_user(username='bob', password='pw12345678')
+        feed = PodcastFeed.objects.create(user=bob, rss_url='https://example.org/b', title='B')
+        ep = PodcastEpisode.objects.create(feed=feed, guid='x', title='X', audio_url='https://example.org/x.mp3',
+                                           transcripts=[self.VTT_T])
+        r = self.get(ep)
+        self.assertEqual((r.status_code, r.json()), (404, {'error': 'not found'}))
+        self.assertEqual(self.requests, [])
+
+    def test_login_required(self):
+        self.client.logout()
+        self.assertEqual(self.get().status_code, 401)

podcasts/transcripts.py

@@ -0,0 +1,282 @@
+"""Podcast transcripts (<podcast:transcript>): pick the best one an episode
+offers, fetch it with the same guards as the chapters, and turn it into
+readable segments:
+
+    {"segments": [{"start": 1.5, "end": 9.25, "speaker": "Anna" | None, "text": "…"}],
+     "source": {"url": "…", "type": "text/vtt", "language": "de"}}
+
+Formats: Podcasting 2.0 JSON, WebVTT and SubRip (SRT). HTML transcripts have
+no timing and aren't supported. Cues are often a few words each; consecutive
+ones of the same speaker are merged up to a sentence end or about 30 s.
+"""
+
+import html
+import json
+import re
+
+from . import chapters as fetching
+
+MAX_BYTES = 2 * 1024 * 1024
+MAX_SEGMENTS = 20_000
+MAX_TEXT = 5_000       # characters of one segment
+MAX_SPEAKER = 200
+
+# Merging
+MERGE_MAX_SECONDS = 30   # a segment never spans more than this
+MERGE_MIN_SECONDS = 5    # ... and ends at a sentence end once it's this long
+MERGE_MAX_GAP = 5        # silence that always starts a new segment
+
+JSON, VTT, SRT, HTML = 'json', 'vtt', 'srt', 'html'
+_KIND_RANK = {JSON: 0, VTT: 1, SRT: 2}
+_ACCEPT = {JSON: 'application/json', VTT: 'text/vtt', SRT: 'application/x-subrip'}
+
+
+class Unsupported(Exception):
+    """The transcript is in a format without timing (HTML) or an unknown one."""
+
+
+def kind(transcript):
+    """json, vtt, srt, html or '' (unknown) from the type, else the URL."""
+    t = (transcript.get('type') or '').lower().split(';')[0].strip()
+    if 'json' in t:
+        return JSON
+    if 'vtt' in t:
+        return VTT
+    if 'srt' in t or 'subrip' in t:
+        return SRT
+    if 'html' in t:
+        return HTML
+    path = (transcript.get('url') or '').lower().split('?')[0].split('#')[0]
+    for ext, k in (('.json', JSON), ('.vtt', VTT), ('.srt', SRT), ('.html', HTML), ('.htm', HTML)):
+        if path.endswith(ext):
+            return k
+    return ''
+
+
+def _language_rank(language):
+    lang = (language or '').lower().split('-')[0].split('_')[0]
+    if lang in ('de', ''):  # no language given: most likely the episode's own
+        return 0
+    if lang == 'en':
+        return 1
+    return 2
+
+
+def candidates(transcripts):
+    """The episode's transcripts in a format we read, best first: German (or
+    no language given) before English before others, then JSON before VTT
+    before SRT."""
+    usable = [(i, t) for i, t in enumerate(transcripts or []) if kind(t) in _KIND_RANK]
+    usable.sort(key=lambda it: (_language_rank(it[1].get('language')), _KIND_RANK[kind(it[1])], it[0]))
+    return [t for _, t in usable]
+
+
+# ---------------------------------------------------------------------------
+# Parsers: each returns fragments [(start, end, speaker, text)]
+# ---------------------------------------------------------------------------
+
+def _decode(body):
+    if body.startswith((b'\xff\xfe', b'\xfe\xff')):
+        return body.decode('utf-16', errors='replace')
+    try:
+        return body.decode('utf-8-sig')
+    except UnicodeDecodeError:
+        return body.decode('cp1252', errors='replace')
+
+
+def _clean(text):
+    text = re.sub(r'<[^>]*>', ' ', text)
+    return ' '.join(html.unescape(text).split())
+
+
+def _seconds(value):
+    try:
+        value = float(value)
+    except (TypeError, ValueError):
+        return None
+    if value != value or value < 0 or value == float('inf'):
+        return None
+    return value
+
+
+def parse_json(text):
+    """Podcasting 2.0 JSON: {"segments": [{"startTime", "endTime", "speaker", "body"}]}."""
+    try:
+        data = json.loads(text)
+    except ValueError:
+        raise fetching.FetchInvalid('not JSON')
+    segments = data.get('segments') if isinstance(data, dict) else None
+    if not isinstance(segments, list):
+        raise fetching.FetchInvalid('no segments list')
+    out = []
+    for s in segments:
+        if not isinstance(s, dict):
+            continue
+        start = _seconds(s.get('startTime'))
+        if start is None:
+            continue
+        end = _seconds(s.get('endTime'))
+        speaker = s.get('speaker')
+        speaker = ' '.join(speaker.split()) if isinstance(speaker, str) else None
+        body = s.get('body')
+        out.append((start, end, speaker or None, _clean(body) if isinstance(body, str) else ''))
+    return out
+
+
+# 00:01:02.500 / 01:02.500 (VTT), 00:01:02,500 (SRT)
+_TIME = r'(?:(\d+):)?(\d{1,2}):(\d{1,2})(?:[.,](\d{1,3}))?'
+_TIMING = re.compile(r'^\s*' + _TIME + r'\s*-->\s*' + _TIME + r'(?:\s|$)')
+_VOICE = re.compile(r'<v(?:\.[^\s>]*)?(?:\s+([^>]*))?>', re.IGNORECASE)
+
+
+def _time(hours, minutes, seconds, fraction):
+    return int(hours or 0) * 3600 + int(minutes) * 60 + int(seconds) + (int(fraction.ljust(3, '0')) / 1000 if fraction else 0)
+
+
+def _cue_fragments(start, end, lines):
+    """Split a cue's text at its voice tags (<v Anna>…) into fragments."""
+    text = '\n'.join(lines)
+    out = []
+    speaker, pos = None, 0
+    for m in _VOICE.finditer(text):
+        out.append((start, end, speaker, _clean(text[pos:m.start()])))
+        speaker = ' '.join((m.group(1) or '').split())[:MAX_SPEAKER] or None
+        pos = m.end()
+    out.append((start, end, speaker, _clean(text[pos:])))
+    return [f for f in out if f[3]]
+
+
+def parse_cues(text):
+    """WebVTT or SRT. Lines before a cue's timing line (VTT cue ids, SRT
+    numbers) are dropped, as is everything outside cues (the WEBVTT header,
+    NOTE, STYLE and REGION blocks)."""
+    cues, current = [], None
+    for line in text.replace('\r\n', '\n').replace('\r', '\n').split('\n'):
+        m = _TIMING.match(line)
+        if m:
+            if current is not None and current[2] and current[2][-1].strip().isdigit():
+                current[2].pop()  # an SRT number with no blank line before it
+            g = m.groups()
+            current = (_time(*g[:4]), _time(*g[4:]), [])
+            cues.append(current)
+        elif not line.strip():
+            current = None
+        elif current is not None:
+            current[2].append(line)
+    out = []
+    for start, end, lines in cues:
+        out.extend(_cue_fragments(start, end, lines))
+    return out
+
+
+def parse(body, declared_kind):
+    """Fragments from a downloaded transcript; the content decides the format
+    when it's obvious, the declared type otherwise."""
+    text = _decode(body)
+    head = text.lstrip()[:200].lower()
+    if head.startswith(('<!doctype', '<html', '<?xml', '<body', '<p', '<div')):
+        raise Unsupported('HTML')
+    if head.startswith(('{', '[')):
+        return parse_json(text)
+    if head.startswith('webvtt') or declared_kind in (VTT, SRT):
+        return parse_cues(text)
+    if declared_kind == JSON:
+        return parse_json(text)
+    raise Unsupported('unknown format')
+
+
+# ---------------------------------------------------------------------------
+# Merging
+# ---------------------------------------------------------------------------
+
+_SENTENCE_END = re.compile(r'[.!?…]["\'”’»)\]]*$')
+
+
+def merge(fragments):
+    """Readable segments from cue fragments: consecutive ones of the same
+    speaker are joined until a sentence ends (once the segment is at least
+    MERGE_MIN_SECONDS long), MERGE_MAX_SECONDS are reached, or there's a
+    pause of more than MERGE_MAX_GAP."""
+    frags = sorted((f for f in fragments if f[3]), key=lambda f: f[0])[:MAX_SEGMENTS * 10]
+    # Fragments without a usable end run until the next one starts.
+    fixed = []
+    for i, (start, end, speaker, text) in enumerate(frags):
+        if end is None or end <= start:
+            nxt = frags[i + 1][0] if i + 1 < len(frags) else start
+            end = max(nxt, start)
+        fixed.append((start, end, speaker, text))
+
+    segments = []
+    for start, end, speaker, text in fixed:
+        last = segments[-1] if segments else None
+        if (
+            last is not None
+            and last['speaker'] == speaker
+            and start - last['end'] <= MERGE_MAX_GAP
+            and end - last['start'] <= MERGE_MAX_SECONDS
+            and len(last['text']) + len(text) < MAX_TEXT
+            and not (_SENTENCE_END.search(last['text']) and last['end'] - last['start'] >= MERGE_MIN_SECONDS)
+        ):
+            last['text'] += ' ' + text
+            last['end'] = max(last['end'], end)
+            continue
+        if len(segments) >= MAX_SEGMENTS:
+            break
+        segments.append({'start': start, 'end': end, 'speaker': speaker, 'text': text[:MAX_TEXT]})
+    for s in segments:
+        s['start'], s['end'] = round(s['start'], 3), round(s['end'], 3)
+    return segments
+
+
+# ---------------------------------------------------------------------------
+# Fetching
+# ---------------------------------------------------------------------------
+
+def fetch(transcript):
+    """The parsed transcript {"segments", "source"} from one of the episode's
+    <podcast:transcript> entries. Raises Unsupported, FetchInvalid or
+    FetchUnavailable."""
+    k = kind(transcript)
+    if k not in _KIND_RANK:
+        raise Unsupported(transcript.get('type') or 'unknown type')
+    body = fetching.fetch_bytes(
+        transcript['url'], MAX_BYTES, accept=f'{_ACCEPT[k]}, */*;q=0.1', agent='diora podcast transcripts',
+    )
+    segments = merge(parse(body, k))
+    if not segments:
+        raise fetching.FetchInvalid('no segments')
+    return {
+        'segments': segments,
+        'source': {
+            'url': transcript['url'],
+            'type': transcript.get('type') or '',
+            'language': transcript.get('language') or '',
+        },
+    }
+
+
+def load(transcripts):
+    """(parsed transcript, final) from the best candidate that works.
+
+    final is False when a better candidate failed for now only, so the
+    result shouldn't be kept for good. Raises FetchUnavailable when nothing
+    worked and something may work later, Unsupported when no candidate is in
+    a usable format, FetchInvalid otherwise.
+    """
+    retry_later = False
+    unsupported_only = True
+    for t in candidates(transcripts):
+        try:
+            return fetch(t), not retry_later
+        except fetching.FetchUnavailable:
+            retry_later = True
+            unsupported_only = False
+        except Unsupported:
+            pass
+        except fetching.FetchInvalid:
+            unsupported_only = False
+    if retry_later:
+        raise fetching.FetchUnavailable('try again later')
+    if unsupported_only:
+        raise Unsupported('no usable transcript')
+    raise fetching.FetchInvalid('no usable transcript')

podcasts/urls.py

@@ -15,6 +15,7 @@ urlpatterns = [
     path('feeds/<int:pk>/episodes/', views.feed_episodes, name='podcast_feed_episodes'),
     path('episodes/', views.episode_list, name='podcast_episode_list'),
     path('episodes/<int:pk>/', views.episode_detail, name='podcast_episode_detail'),
+    path('episodes/<int:pk>/transcript/', views.episode_transcript, name='podcast_episode_transcript'),
     path('episodes/<int:pk>/favorite/', views.set_favorite, name='podcast_set_favorite'),
     path('stats/', views.podcast_stats, name='podcast_stats'),
     path('queue/', views.queue_get, name='podcast_queue_get'),

podcasts/views.py

@@ -19,7 +19,10 @@ from django.views.decorators.http import require_http_methods
 
 from . import chapters as chapters_mod
 from . import feedext
-from .models import EpisodeProgress, PodcastEpisode, PodcastFeed, PodcastQueue, PodcastSettings, counts_as_played
+from . import transcripts as transcripts_mod
+from .models import (
+    EpisodeProgress, EpisodeTranscript, PodcastEpisode, PodcastFeed, PodcastQueue, PodcastSettings, counts_as_played,
+)
 from .shownotes import extract_timestamps, sanitize_html
 
 feedext.install()
@@ -179,8 +182,10 @@ def _entry_transcripts(entry):
 def _apply_extras(episode, description, chapters_url, psc_chapters, transcripts):
     """Copy what the feed says now onto `episode`; True if anything changed.
 
-    Inline (Podlove) chapters win. Without them, chapters fetched from
+    Inline (Podlove) chapters win. Without them, fetched chapters (from
-    chapters_url stay cached for as long as the URL stays the same.
+    chapters_url, or from the audio file when there's no URL) stay cached for
+    as long as the URL stays the same; inline chapters going away means a
+    fresh fetch.
     """
     changed = []
     if episode.description != description:
@@ -190,8 +195,8 @@ def _apply_extras(episode, description, chapters_url, psc_chapters, transcripts)
         episode.transcripts = transcripts
         changed.append('transcripts')
     if psc_chapters:
-        chapters, fetched_at = psc_chapters, episode.chapters_fetched_at
+        chapters, fetched_at = psc_chapters, None
-    elif chapters_url and chapters_url == episode.chapters_url and episode.chapters_fetched_at:
+    elif chapters_url == episode.chapters_url and episode.chapters_fetched_at:
         chapters, fetched_at = episode.chapters, episode.chapters_fetched_at
     else:
         chapters, fetched_at = [], None
@@ -602,20 +607,29 @@ def feed_episodes(request, pk):
 # ---------------------------------------------------------------------------
 
 def _load_chapters(episode):
-    """Fetch the JSON chapters of `episode` once and cache them on it.
+    """Fetch the chapters the feed doesn't carry inline, once, and cache them
+    on `episode`: the JSON at chapters_url, else (no URL, or nothing there)
+    the CHAP frames in the ID3v2 tag at the start of the audio file.
 
     Problems that may go away (timeouts, server errors) aren't cached, so the
     next request tries again; anything else is remembered as "no chapters".
     """
     url = episode.chapters_url
+    chapters = []
+    try:
+        if url:
             try:
                 chapters = chapters_mod.fetch(url)
+            except chapters_mod.ChaptersInvalid as e:
+                log.info('chapters of episode %s unusable: %s', episode.id, e)
+        if not chapters and episode.audio_url:
+            try:
+                chapters = chapters_mod.fetch_id3(episode.audio_url)
+            except chapters_mod.ChaptersInvalid as e:
+                log.info('ID3 chapters of episode %s unusable: %s', episode.id, e)
     except chapters_mod.ChaptersUnavailable as e:
         log.info('chapters of episode %s not available: %s', episode.id, e)
         return []
-    except chapters_mod.ChaptersInvalid as e:
-        log.info('chapters of episode %s unusable: %s', episode.id, e)
-        chapters = []
     now = timezone.now()
     # Only if a refresh hasn't changed the URL in the meantime.
     PodcastEpisode.objects.filter(pk=episode.pk, chapters_url=url).update(
@@ -636,7 +650,7 @@ def episode_detail(request, pk):
         return JsonResponse({'error': 'not found'}, status=404)
 
     chapters = ep.chapters
-    if not chapters and ep.chapters_url and ep.chapters_fetched_at is None:
+    if not chapters and ep.chapters_fetched_at is None:
         chapters = _load_chapters(ep)
 
     progress = EpisodeProgress.objects.filter(user=request.user, episode=ep).first()
@@ -647,6 +661,7 @@ def episode_detail(request, pk):
         'timestamps': extract_timestamps(ep.description, ep.duration_seconds),
         'chapters': chapters,
         'transcripts': ep.transcripts,
+        'has_transcript': _has_transcript(ep),
         'feed': {'id': ep.feed.id, 'title': ep.feed.title, 'artwork_url': ep.feed.artwork_url},
         'audio_url': ep.audio_url,
         'artwork_url': ep.artwork_url,
@@ -662,6 +677,65 @@ def episode_detail(request, pk):
     })
 
 
+def _transcript_sources(ep):
+    return [t['url'] for t in transcripts_mod.candidates(ep.transcripts)]
+
+
+def _has_transcript(ep):
+    """True if the episode offers a transcript we can read, and it hasn't
+    turned out to be unusable."""
+    sources = _transcript_sources(ep)
+    if not sources:
+        return False
+    cached = EpisodeTranscript.objects.filter(episode=ep).values_list('sources', 'error').first()
+    return not (cached and cached[0] == sources and cached[1])
+
+
+@csrf_exempt
+@require_http_methods(['GET'])
+def episode_transcript(request, pk):
+    """The episode's transcript as {"segments": [{start, end, speaker, text}],
+    "source": {url, type, language}}, fetched and parsed on first use, then
+    kept. 404 {"error": "no transcript" | "unsupported" | "invalid"} when
+    there's none to show; 502 {"error": "unavailable"} when it couldn't be
+    loaded right now."""
+    if not request.user.is_authenticated:
+        return JsonResponse({'error': 'authentication required'}, status=401)
+    try:
+        ep = PodcastEpisode.objects.get(pk=pk, feed__user=request.user)
+    except PodcastEpisode.DoesNotExist:
+        return JsonResponse({'error': 'not found'}, status=404)
+    if not ep.transcripts:
+        return JsonResponse({'error': 'no transcript'}, status=404)
+    sources = _transcript_sources(ep)
+    if not sources:
+        return JsonResponse({'error': 'unsupported'}, status=404)
+
+    cached = EpisodeTranscript.objects.filter(episode=ep).first()
+    if cached is not None and cached.sources == sources:
+        data, error = cached.data, cached.error
+    else:
+        try:
+            data, final = transcripts_mod.load(ep.transcripts)
+            error = ''
+        except chapters_mod.FetchUnavailable as e:
+            log.info('transcript of episode %s not available: %s', ep.id, e)
+            return JsonResponse({'error': 'unavailable'}, status=502)
+        except transcripts_mod.Unsupported as e:
+            log.info('transcript of episode %s unsupported: %s', ep.id, e)
+            data, error, final = None, 'unsupported', True
+        except chapters_mod.FetchInvalid as e:
+            log.info('transcript of episode %s unusable: %s', ep.id, e)
+            data, error, final = None, 'invalid', True
+        if final:
+            EpisodeTranscript.objects.update_or_create(
+                episode=ep, defaults={'sources': sources, 'data': data, 'error': error},
+            )
+    if error:
+        return JsonResponse({'error': error}, status=404)
+    return JsonResponse(data)
+
+
 # ---------------------------------------------------------------------------
 # Favorites
 # ---------------------------------------------------------------------------