feat(magazine): journalist/guest voices for c't-style interviews

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

CHANGELOG.md

@@ -6,6 +6,8 @@ Follows [Keep a Changelog](https://keepachangelog.com/en/1.0.0/) · versioned wi
 ---
 
 ## [Unreleased]
+### Changed
+- **Magazine interviews: journalist/guest voices.** Interview labels such as `c't:` (also `c’t`, `ct`, OCR `Cc't`), `Redaktion`, `Interviewer`, `Frage` are the interviewing side and are read in the narrator voice; every other recurring label is a guest and gets the second voice. Unlabelled blocks continue the current speaker's turn; headings, decks and bylines stay narrator. Detection stays strict (label starts a block and recurs at least twice). Audio-timing records now store each segment's voice and speaker. Tests: `tests/test_magazine.py` (`InterviewJournalistTests`).
 
 ---
 

architecture/magazine-ingestion.md

@@ -963,3 +963,7 @@ gets `audio_stale: true` with `audio_stale_reason`. Audio and timing files are *
   `static/js/magazine.js` and is not fixed here. Until it is, new stand-ins can appear.
 * Reading order across a page that holds two articles relies on the existing per-page order. A
   wrong page order gives a wrong *previous entry*.
+
+## Interview voice slots
+
+`interview_tts_segments` returns `{speaker, text, slot}` per block (slot 0 = narrator voice, 1 = second voice). A label must start a block and recur at least twice. Labels in `INTERVIEW_JOURNALIST_LABELS` (`c't`, `ct`, `redaktion`, `interviewer`, `frage`; apostrophes normalised, OCR `Cc't` folded) get slot 0; all other recurring labels get slot 1. With no journalist label, exactly two recurring labels are required and the first seen gets slot 0. Unlabelled blocks continue the current speaker; title/heading/deck/byline/figure_description reset to narrator. `write_audio_timings` records `voice` and `speaker` per segment.

core/magazine.py

@@ -5651,20 +5651,31 @@ def split_sentences_for_tts(text: str) -> list[str]:
 
 
 _INTERVIEW_LABEL_RE = re.compile(
-    r"^\s*([A-ZÄÖÜ][A-Za-zÄÖÜäöüß.'’\-]+(?:\s+[A-ZÄÖÜ][A-Za-zÄÖÜäöüß.'’\-]+){0,3})\s*:\s+(.+)$",
+    r"^\s*((?:[A-ZÄÖÜ][A-Za-zÄÖÜäöüß.'’\-]+(?:\s+[A-ZÄÖÜ][A-Za-zÄÖÜäöüß.'’\-]+){0,3})|[Cc]{1,2}['’‘`´]t|ct)\s*:\s+(.+)$",
     re.S,
 )
+# Labels of the interviewing side (casefolded, apostrophes normalised to "'"). Spoken in the narrator voice;
+# every other recurring label is a guest and gets the second voice.
+INTERVIEW_JOURNALIST_LABELS = frozenset({"c't", "ct", "redaktion", "interviewer", "frage"})
+_NON_TURN_TYPES = frozenset({"title", "heading", "deck", "byline", "figure_description"})
 
 
-def interview_tts_segments(blocks: Sequence[dict[str, Any]]) -> list[dict[str, str | None]]:
+def _label_key(label: str) -> str:
-    """Return article text segments with speaker labels for a two-person interview.
+    key = re.sub(r"[’‘`´]", "'", label.strip().casefold())
+    return "c't" if key == "cc't" else key       # OCR doubles the drop-cap: "Cc't:"
 
-    Detection is intentionally strict: exactly two recurring labels must begin
+
-    paragraph blocks.  Colons in ordinary prose therefore do not suddenly
+def interview_tts_segments(blocks: Sequence[dict[str, Any]]) -> list[dict[str, Any]]:
-    switch voices.  Speaker labels themselves are omitted from speech because
+    """Return article text segments with speaker labels and voice slots for an interview.
-    the voice change already conveys who is talking.
+
+    Each segment: ``speaker`` (label or None), ``text`` (label stripped), ``slot`` (0 narrator voice,
+    1 second voice).  Detection is strict so colons in prose never switch voices: a label must start a
+    block and recur at least twice, and there must be a journalist label (``INTERVIEW_JOURNALIST_LABELS``)
+    or two distinct recurring labels.  Journalist -> slot 0, every other label -> slot 1; with no journalist
+    label the first label seen gets slot 0.  An unlabelled block continues the current speaker's turn;
+    titles/headings/decks/bylines and everything before the first label stay narrator.
     """
-    parsed: list[tuple[str | None, str]] = []
+    parsed: list[tuple[str | None, str, str, dict[str, Any]]] = []   # key, body, full text, block
     counts: dict[str, int] = {}
     labels: dict[str, str] = {}
     for block in blocks:
@@ -5673,22 +5684,32 @@ def interview_tts_segments(blocks: Sequence[dict[str, Any]]) -> list[dict[str, s
             continue
         match = _INTERVIEW_LABEL_RE.match(text)
         if not match:
-            parsed.append((None, text))
+            parsed.append((None, text, text, block))
             continue
         label, body = match.group(1).strip(), match.group(2).strip()
-        key = label.casefold()
+        key = _label_key(label)
         labels.setdefault(key, label)
         counts[key] = counts.get(key, 0) + 1
-        parsed.append((key, body))
+        parsed.append((key, body, text, block))
 
     recurring = [key for key, count in counts.items() if count >= 2]
-    if len(recurring) != 2:
+    journalist = [k for k in recurring if k in INTERVIEW_JOURNALIST_LABELS]
-        return [{"speaker": None, "text": text} for _, text in parsed]
+    if not journalist and len(recurring) != 2:
-    allowed = set(recurring)
+        return [{"speaker": None, "text": full, "slot": 0} for _, _, full, _ in parsed]
-    return [
+    slot_of = {k: (0 if k in journalist else 1) for k in recurring} if journalist else {k: i for i, k in enumerate(recurring)}
-        {"speaker": labels[speaker] if speaker in allowed else None, "text": text}
+    out: list[dict[str, Any]] = []
-        for speaker, text in parsed
+    speaker: str | None = None
-    ]
+    slot = 0
+    for key, body, full, block in parsed:
+        if key in slot_of:
+            speaker, slot = labels[key], slot_of[key]
+            out.append({"speaker": speaker, "text": body, "slot": slot})
+        elif block.get("type") in _NON_TURN_TYPES:
+            speaker, slot = None, 0
+            out.append({"speaker": None, "text": full, "slot": 0})
+        else:
+            out.append({"speaker": speaker, "text": full, "slot": slot})
+    return out
 
 
 def spoken_chars_for(blocks: Sequence[dict[str, Any]], timings: dict[str, Any] | None) -> list[int] | None:
@@ -5728,18 +5749,12 @@ def synthesize_article_audio(doc_dir: Path, article: dict[str, Any], layout: dic
         {"speaker": None, "text": b["text"], "spoken": b.get("spoken_text")}
         for b in blocks if (b.get("text") or "").strip()
     ]
-    speakers = []
-    for segment in segments:
-        speaker = segment.get("speaker")
-        if speaker and speaker not in speakers:
-            speakers.append(speaker)
-    voice_for = {speaker: (voice if i == 0 else interview_voice)
-                 for i, speaker in enumerate(speakers[:2])}
     utterances: list[tuple[str, str]] = []
     utterance_segment: list[int] = []        # which segment each utterance came from, for the timing record
     for n, segment in enumerate(segments):
         spoken = segment.get("spoken") or prepare_text_for_tts(str(segment["text"]), language, glossary)
-        segment_voice = voice_for.get(segment.get("speaker"), voice)
+        segment_voice = interview_voice if segment.get("slot") == 1 else voice
+        segment["voice"] = segment_voice
         for sentence in split_sentences_for_tts(spoken):
             utterances.append((sentence, segment_voice))
             utterance_segment.append(n)
@@ -5849,7 +5864,9 @@ def write_audio_timings(mp3_path: Path, segments: Sequence[dict[str, Any]], utte
     record = {
         "version": 1, "interview": bool(interview), "duration_ms": int(spans[-1][1]) if spans else 0,
         "segments": [{"chars": len(str(seg.get("text") or "")), "times": times,
-                      **({"onsets": on} if onsets else {})}
+                      **({"onsets": on} if onsets else {}),
+                      **({"voice": seg["voice"]} if seg.get("voice") else {}),
+                      **({"speaker": seg["speaker"]} if seg.get("speaker") else {})}
                      for seg, times, on in zip(segments, per_segment, per_onset)],
     }
     try:

tests/test_magazine.py

@@ -819,6 +819,69 @@ class InterviewVoiceTests(unittest.TestCase):
         self.assertTrue(all(s["speaker"] is None for s in interview_tts_segments(blocks)))
 
 
+class InterviewJournalistTests(unittest.TestCase):
+    @staticmethod
+    def _ct_blocks(label="c't"):
+        return [
+            {"text": "Interview mit Michael Gerlich", "type": "heading"},
+            {"text": "Einleitung zum Gespräch.", "type": "body"},
+            {"text": f"{label}: Was ist kognitives Offloading?"},
+            {"text": "Michael Gerlich: Wir lagern Denken an KI aus."},
+            {"text": "Ein zweiter Absatz der Antwort ohne Label."},
+            {"text": f"{label}: Ist das gefährlich?"},
+            {"text": "Michael Gerlich: Teilweise."},
+        ]
+
+    def test_ct_variants_map_journalist_to_narrator_and_guest_to_second_voice(self):
+        for label in ("c't", "c’t", "ct", "C't", "Cc't"):
+            with self.subTest(label=label):
+                segs = interview_tts_segments(self._ct_blocks(label))
+                self.assertEqual([s["slot"] for s in segs], [0, 0, 0, 1, 1, 0, 1])
+                self.assertEqual(segs[2]["text"], "Was ist kognitives Offloading?")
+                self.assertEqual(segs[4]["speaker"], "Michael Gerlich")
+                self.assertEqual(segs[4]["text"], "Ein zweiter Absatz der Antwort ohne Label.")
+                self.assertIsNone(segs[0]["speaker"])
+
+    def test_multiple_guests_all_use_second_voice(self):
+        blocks = [{"text": "c't: Frage eins?"}, {"text": "Anna Meyer: A."}, {"text": "c't: Frage zwei?"},
+                  {"text": "Anna Meyer: B."}, {"text": "c't: Frage drei?"}, {"text": "Tom Schulz: C."},
+                  {"text": "c't: Frage vier?"}, {"text": "Tom Schulz: D."}]
+        self.assertEqual([s["slot"] for s in interview_tts_segments(blocks)], [0, 1, 0, 1, 0, 1, 0, 1])
+
+    def test_two_plain_labels_keep_first_appearance_rule(self):
+        blocks = [{"text": "Zed Alpha: Eins."}, {"text": "Bob Beta: Zwei."}, {"text": "Zed Alpha: Drei."},
+                  {"text": "Bob Beta: Vier."}]
+        self.assertEqual([s["slot"] for s in interview_tts_segments(blocks)], [0, 1, 0, 1])
+
+    def test_single_colon_labels_are_not_speakers_and_keep_their_text(self):
+        blocks = [{"text": "Hinweis: Das gilt nur für Version 2."}, {"text": "Fazit: Es lohnt sich."},
+                  {"text": "Hinweis: Noch einmal."}]
+        segs = interview_tts_segments(blocks)
+        self.assertTrue(all(s["speaker"] is None and s["slot"] == 0 for s in segs))
+        self.assertEqual(segs[0]["text"], "Hinweis: Das gilt nur für Version 2.")
+
+    def test_one_journalist_label_alone_is_not_enough(self):
+        segs = interview_tts_segments([{"text": "c't: Einmal."}, {"text": "Text."}])
+        self.assertTrue(all(s["speaker"] is None for s in segs))
+
+    def test_timings_record_voice_and_speaker_and_stay_backward_compatible(self):
+        import tempfile
+        import json
+        from pathlib import Path
+        from core.magazine import write_audio_timings
+        segs = interview_tts_segments(self._ct_blocks())
+        for s in segs:
+            s["voice"] = "B" if s["slot"] else "A"
+        with tempfile.TemporaryDirectory() as d:
+            mp3 = Path(d) / "x.mp3"
+            write_audio_timings(mp3, segs, list(range(len(segs))), [(i, i + 1) for i in range(len(segs))], interview=True)
+            rec = json.loads((Path(d) / "x.timings.json").read_text())
+        self.assertEqual(rec["version"], 1)
+        self.assertEqual({s["voice"] for s in rec["segments"]}, {"A", "B"})
+        self.assertTrue(all("chars" in s and "times" in s for s in rec["segments"]))
+        self.assertEqual(rec["segments"][3]["speaker"], "Michael Gerlich")
+
+
 class OcrQualityScoreTests(unittest.TestCase):
     """Grounded in real output from both OCR paths on a scanned c't magazine.