mrwnslz
38d7f6c643
- 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>
282 lines
10 KiB
Python
282 lines
10 KiB
Python
"""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')
|