diff --git a/scripts/apply_speaker_names.py b/scripts/apply_speaker_names.py new file mode 100644 index 0000000..9eb371f --- /dev/null +++ b/scripts/apply_speaker_names.py @@ -0,0 +1,95 @@ +#!/usr/bin/env python3 +""" +Apply human speaker identifications to journal transcripts. + +The mapping lives in the journal item's metadata.json — journal-owned, so it +survives regeneration and records provenance: + + "parts": [{"video_id": "vjtYYbO9jCY", ..., + "speakers": {"SPEAKER_03": "Daniel Friedman", + "SPEAKER_05": "Omar Hashash"}}] + +This script replaces the diarization labels in transcript.txt (and the +"speaker" fields in transcript.json) with the mapped names. Idempotent: +already-replaced names are untouched; unmapped labels remain SPEAKER_NN and +are reported as to-dos. + +Usage: + python scripts/apply_speaker_names.py # all items with mappings + python scripts/apply_speaker_names.py --item GuestStream/GuestStream_128 +""" + +import argparse +import json +import re +import sys +from pathlib import Path + +REPO = Path(__file__).resolve().parent.parent +SRC_PREFIX = "data/video/activeinferenceinstitute" + + +def apply_to_item(item_dir: Path, meta: dict) -> dict: + mapping: dict[str, str] = {} + for part in meta.get("parts", []): + mapping.update(part.get("speakers") or {}) + if not mapping: + return {} + + replaced = 0 + tx_path = item_dir / "transcript.txt" + if tx_path.exists(): + text = tx_path.read_text(encoding="utf-8") + for label, name in mapping.items(): + text, n = re.subn(rf"^{re.escape(label)}:", f"{name}:", text, flags=re.M) + replaced += n + if replaced: + tx_path.write_text(text, encoding="utf-8") + + tj_path = item_dir / "transcript.json" + if tj_path.exists(): + blocks = json.loads(tj_path.read_text(encoding="utf-8")) + changed = False + for block in blocks if isinstance(blocks, list) else []: + for seg in block.get("segments", []) if isinstance(block, dict) else []: + if seg.get("speaker") in mapping: + seg["speaker"] = mapping[seg["speaker"]] + changed = True + if changed: + tj_path.write_text(json.dumps(blocks, ensure_ascii=False), encoding="utf-8") + + unmapped = sorted(set(re.findall(r"^(SPEAKER_\d+):", tx_path.read_text(encoding="utf-8"), re.M))) \ + if tx_path.exists() else [] + return {"replaced": replaced, "unmapped": unmapped} + + +def main() -> int: + parser = argparse.ArgumentParser(description=__doc__, + formatter_class=argparse.RawDescriptionHelpFormatter) + parser.add_argument("--journal", type=Path, default=REPO.parent / "ActiveInferenceJournal") + parser.add_argument("--item", help="single item (path relative to the source root)") + args = parser.parse_args() + + root = args.journal / SRC_PREFIX + targets = [root / args.item / "metadata.json"] if args.item \ + else sorted(root.rglob("metadata.json")) + + any_applied = False + for meta_path in targets: + if not meta_path.exists(): + print(f"no such item: {meta_path.parent}") + return 1 + meta = json.loads(meta_path.read_text(encoding="utf-8")) + result = apply_to_item(meta_path.parent, meta) + if result: + any_applied = True + rel = meta_path.parent.relative_to(root) + todo = f", unmapped: {', '.join(result['unmapped'])}" if result["unmapped"] else "" + print(f"{rel}: {result['replaced']} label(s) replaced{todo}") + if not any_applied: + print("no items with parts[].speakers mappings found") + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/tests/journal_utilities/test_apply_speakers.py b/tests/journal_utilities/test_apply_speakers.py new file mode 100644 index 0000000..e91ef8d --- /dev/null +++ b/tests/journal_utilities/test_apply_speakers.py @@ -0,0 +1,53 @@ +"""Tests for scripts/apply_speaker_names.py.""" + +import json +import sys +from pathlib import Path + +REPO = Path(__file__).resolve().parent.parent.parent +sys.path.insert(0, str(REPO / "scripts")) + +from apply_speaker_names import apply_to_item + + +def _item(tmp_path: Path, speakers: dict) -> tuple[Path, dict]: + item = tmp_path / "item" + item.mkdir() + meta = {"parts": [{"video_id": "vid00000001", "speakers": speakers}]} + (item / "transcript.txt").write_text( + "SPEAKER_00:\nhello\n\nSPEAKER_01:\nhi SPEAKER_00 there\n") + (item / "transcript.json").write_text(json.dumps( + [{"video_id": "vid00000001", + "segments": [{"speaker": "SPEAKER_00", "text": "hello"}, + {"speaker": "SPEAKER_01", "text": "hi"}]}])) + return item, meta + + +class TestApplySpeakers: + def test_replaces_only_label_headings(self, tmp_path): + item, meta = _item(tmp_path, {"SPEAKER_00": "Ada Lovelace"}) + result = apply_to_item(item, meta) + text = (item / "transcript.txt").read_text() + assert text.startswith("Ada Lovelace:\n") + # in-sentence mention is not a heading and stays untouched + assert "hi SPEAKER_00 there" in text + assert result["unmapped"] == ["SPEAKER_01"] + + def test_json_speakers_replaced(self, tmp_path): + item, meta = _item(tmp_path, {"SPEAKER_00": "Ada Lovelace", "SPEAKER_01": "Alan Turing"}) + apply_to_item(item, meta) + segs = json.loads((item / "transcript.json").read_text())[0]["segments"] + assert [s["speaker"] for s in segs] == ["Ada Lovelace", "Alan Turing"] + + def test_idempotent(self, tmp_path): + item, meta = _item(tmp_path, {"SPEAKER_00": "Ada Lovelace", "SPEAKER_01": "Alan Turing"}) + apply_to_item(item, meta) + first = (item / "transcript.txt").read_text() + result = apply_to_item(item, meta) + assert (item / "transcript.txt").read_text() == first + assert result["unmapped"] == [] + + def test_no_mapping_is_noop(self, tmp_path): + item, meta = _item(tmp_path, {}) + meta["parts"][0].pop("speakers") + assert apply_to_item(item, meta) == {}