Initial commit: Nat20 Notes — TTRPG session transcription & summarization

backend/app/pipeline/turns.py

@@ -0,0 +1,60 @@
+import json
+from pathlib import Path
+
+
+def fmt_time(seconds: float) -> str:
+    h, rem = divmod(int(seconds), 3600)
+    m, s = divmod(rem, 60)
+    return f"{h:02d}:{m:02d}:{s:02d}"
+
+
+def load_turns(transcript_json: Path, speaker_map: dict[str, str] | None = None) -> list[dict]:
+    """Parse whisperx json into merged (start, end, speaker, text) turns.
+    speaker_map maps raw labels (SPEAKER_00...) to display names; unmapped labels pass through raw."""
+    data = json.loads(transcript_json.read_text())
+    speaker_map = speaker_map or {}
+    turns = []
+    for seg in data.get("segments", []):
+        raw = seg.get("speaker", "UNKNOWN")
+        speaker = speaker_map.get(raw, raw)
+        text = (seg.get("text") or "").strip()
+        if not text:
+            continue
+        if turns and turns[-1]["speaker"] == speaker:
+            turns[-1]["text"] += " " + text
+            turns[-1]["end"] = seg["end"]
+        else:
+            turns.append({"start": seg["start"], "end": seg["end"], "speaker": speaker, "text": text, "raw_speaker": raw})
+    return turns
+
+
+def distinct_speakers(transcript_json: Path) -> list[dict]:
+    """Return each raw speaker label with a couple of sample lines, for the naming UI."""
+    turns = load_turns(transcript_json)  # no map -> raw labels
+    seen: dict[str, list[dict]] = {}
+    for t in turns:
+        seen.setdefault(t["raw_speaker"], []).append(t)
+    return [
+        {
+            "raw_label": label,
+            "samples": [{"start": t["start"], "text": t["text"][:140]} for t in items[:3]],
+        }
+        for label, items in sorted(seen.items())
+    ]
+
+
+def chunk_turns(turns: list[dict], word_target: int) -> list[list[dict]]:
+    chunks, current, word_count = [], [], 0
+    for turn in turns:
+        current.append(turn)
+        word_count += len(turn["text"].split())
+        if word_count >= word_target:
+            chunks.append(current)
+            current, word_count = [], 0
+    if current:
+        chunks.append(current)
+    return chunks
+
+
+def turns_to_text(turns: list[dict]) -> str:
+    return "\n".join(f"[{fmt_time(t['start'])}] {t['speaker']}: {t['text']}" for t in turns)