diff --git a/docs/JOURNAL_SCHEMA.md b/docs/JOURNAL_SCHEMA.md index 9082e1c..af40c5f 100644 --- a/docs/JOURNAL_SCHEMA.md +++ b/docs/JOURNAL_SCHEMA.md @@ -27,8 +27,8 @@ excluded from coverage counts. Uncategorized videos live under data/video/activeinferenceinstitute//_/ metadata.json # canonical record (see below) — single source of truth README.md # generated human nav: title(s), date, links, contents - transcript.txt # clean text (per part: "## Part N" headers when multi-part) - transcript.json # timestamped segments (array; part-tagged) + transcript.txt # derived text view ("## " headers when multi-part) + transcript.json # raw timestamped segments (array; part-tagged) — see "Transcripts" captions/ # original-language *.srt translations/ # translated *.srt (one per language) — preserved verbatim assets/ @@ -54,6 +54,35 @@ data/video/activeinferenceinstitute//_/ } ``` +## Transcripts — raw vs derived + +`transcript.json` is the **immutable raw layer**; `transcript.txt` is a +**derived view**. Human speaker names live only in `metadata.json`. + +- **`transcript.json`** — raw diarization output: an array of + `{"video_id", "segments"}` blocks, each segment + `{start, end, text, speaker}` in seconds. Speaker labels stay machine + labels (`SPEAKER_NN`) forever — never rewritten with human names, so + re-mapping is always possible (two labels identified as the same person + remain distinguishable). Legacy AssemblyAI items carry that tool's response + verbatim inside the block instead; treat those as frozen. +- **`parts[].speakers`** (metadata.json, journal-owned) — the only place human + speaker identifications are recorded: `{"SPEAKER_NN": "Name", ...}` per video. +- **`transcript.txt`** — canonical human-facing text, regenerated from + `transcript.json` + `parts[].speakers`; unmapped labels remain `SPEAKER_NN`. + Multi-part items use `## ` headers; split sessions + `## _sessNN`. Items without diarization keep their YouTube-caption + text until WhisperX runs; original captions always remain under `captions/`. + +Provenance is encoded by file shape — no metadata flag: + +| Files present | Meaning | +|---|---| +| `transcript.json` with `speaker` fields in segments | WhisperX-diarized | +| `parts[].speakers` non-empty | names mapped by a human | +| `assets/csv/*.sentences.csv` (± AssemblyAI-shaped transcript.json) | legacy AssemblyAI diarization | +| `transcript.txt` only | YouTube captions (not yet diarized) | + ## Enrichment fields `scripts/enrich_metadata.py` owns the enrichment fields in `metadata.json` and keeps diff --git a/scripts/apply_speaker_names.py b/scripts/apply_speaker_names.py index 9eb371f..76f9684 100644 --- a/scripts/apply_speaker_names.py +++ b/scripts/apply_speaker_names.py @@ -1,22 +1,26 @@ #!/usr/bin/env python3 """ -Apply human speaker identifications to journal transcripts. +Regenerate journal transcript.txt from transcript.json + parts[].speakers. -The mapping lives in the journal item's metadata.json — journal-owned, so it -survives regeneration and records provenance: +Per docs/SCHEMA.md "Transcripts — raw vs derived": transcript.json is the +immutable raw diarization layer (labels stay SPEAKER_NN forever) and +transcript.txt is a derived view — segments grouped into speaker-labeled +paragraphs, labels replaced with human names where the journal-owned +``parts[].speakers`` mapping identifies them. Unmapped labels remain +SPEAKER_NN and are reported as to-dos. Re-running is idempotent; fixing a +wrong name is editing metadata.json and re-running. - "parts": [{"video_id": "vjtYYbO9jCY", ..., - "speakers": {"SPEAKER_03": "Daniel Friedman", - "SPEAKER_05": "Omar Hashash"}}] +When an item's existing transcript.txt has no speaker structure (YouTube +captions or legacy prose), it is salvaged to captions/youtube_captions.txt +before being replaced, so the caption text stays in the working tree. -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. +Items whose transcript.json is not the whisperx block format (legacy +AssemblyAI responses) are skipped and reported. Usage: - python scripts/apply_speaker_names.py # all items with mappings - python scripts/apply_speaker_names.py --item GuestStream/GuestStream_128 + python scripts/apply_speaker_names.py # dry run: show plan + python scripts/apply_speaker_names.py --apply # regenerate + python scripts/apply_speaker_names.py --item GuestStream/GuestStream_128 --apply """ import argparse @@ -27,40 +31,110 @@ REPO = Path(__file__).resolve().parent.parent SRC_PREFIX = "data/video/activeinferenceinstitute" +SALVAGE_NAME = "youtube_captions.txt" -def apply_to_item(item_dir: Path, meta: dict) -> dict: - mapping: dict[str, str] = {} +def load_blocks(tj_path: Path): + """Parse transcript.json; None when absent or not whisperx block format.""" + if not tj_path.exists(): + return None + try: + blocks = json.loads(tj_path.read_text(encoding="utf-8")) + except json.JSONDecodeError: + return None + if not isinstance(blocks, list): + return None + for block in blocks: + if not isinstance(block, dict) or not isinstance(block.get("segments"), list): + return None + if not any(isinstance(s, dict) and s.get("speaker") for s in block["segments"]): + return None # no diarization — nothing to derive + return blocks or None + + +def speaker_mapping(meta: dict, video_id: str) -> dict[str, str]: + """Mapping for one video block; sess-split ids fall back to the base video.""" + base = re.sub(r"_sess\d+$", "", video_id) for part in meta.get("parts", []): - mapping.update(part.get("speakers") or {}) - if not mapping: - return {} - - replaced = 0 + if part.get("video_id") in (video_id, base): + return part.get("speakers") or {} + merged: dict[str, str] = {} + for part in meta.get("parts", []): + merged.update(part.get("speakers") or {}) + return merged + + +def render_block(segments: list, mapping: dict[str, str]) -> str: + """Speaker-labeled text, grouping consecutive segments by mapped name. + + Mirrors TranscriptionService.output_text so regenerated text matches + freshly transcribed output byte-for-byte when no names are mapped. + """ + out, prev = "", None + for seg in segments: + text = (seg.get("text") or "").strip() + if not text: + continue + speaker = mapping.get(seg.get("speaker"), seg.get("speaker") or "UNKNOWN") + if speaker != prev: + out += f"\n{speaker}:\n" + prev = speaker + out += text + "\n\n" + return out.strip() + + +def render_txt(blocks: list, meta: dict) -> tuple[str, set[str]]: + """Full transcript.txt content + the set of unmapped labels.""" + sections, unmapped = [], set() + for block in blocks: + vid = block.get("video_id", "") + mapping = speaker_mapping(meta, vid) + unmapped |= {s.get("speaker") for s in block["segments"] + if s.get("speaker") and s["speaker"] not in mapping} + text = render_block(block["segments"], mapping) + sections.append(text if len(blocks) == 1 else f"## {vid}\n\n{text}") + return "\n\n".join(sections) + "\n", unmapped + + +def has_speaker_structure(text: str) -> bool: + """True when the text already carries 'Label:'-style speaker lines.""" + for line in text.splitlines()[:50]: + line = line.strip() + if not line or line.startswith("## "): + continue + return bool(re.fullmatch(r"\S[^:\n]{0,78}:", line)) + return False + + +def process_item(item_dir: Path, meta: dict, apply: bool) -> dict | None: + blocks = load_blocks(item_dir / "transcript.json") + if blocks is None: + if (item_dir / "transcript.json").exists(): + return {"action": "skip-legacy-json", "unmapped": set()} + return None + + new_txt, unmapped = render_txt(blocks, meta) 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} + old_txt = tx_path.read_text(encoding="utf-8") if tx_path.exists() else None + + if old_txt == new_txt: + return {"action": "unchanged", "unmapped": unmapped} + if old_txt is None: + action = "new" + elif has_speaker_structure(old_txt) or "SPEAKER_" in old_txt[:200_000]: + action = "update" + else: + action = "upgrade+salvage" + + if apply: + if action == "upgrade+salvage": + captions_dir = item_dir / "captions" + captions_dir.mkdir(exist_ok=True) + salvage = captions_dir / SALVAGE_NAME + if not salvage.exists(): + salvage.write_text(old_txt, encoding="utf-8") + tx_path.write_text(new_txt, encoding="utf-8") + return {"action": action, "unmapped": unmapped} def main() -> int: @@ -68,26 +142,37 @@ def main() -> int: 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)") + parser.add_argument("--apply", action="store_true", + help="write transcript.txt (default: dry-run plan)") 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 + counts: dict[str, int] = {} 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 + if meta.get("duplicate_of"): + continue + result = process_item(meta_path.parent, meta, args.apply) + if result is None: + continue + counts[result["action"]] = counts.get(result["action"], 0) + 1 + if result["action"] != "unchanged": 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") + todo = f" [unmapped: {', '.join(sorted(result['unmapped']))}]" \ + if result["unmapped"] and result["action"] != "skip-legacy-json" else "" + print(f"{result['action']:16} {rel}{todo}") + + print("\nsummary:", ", ".join(f"{k}: {v}" for k, v in sorted(counts.items())) or "nothing to do") + if not args.apply: + print("dry run — pass --apply to write") + else: + print("regenerate indexes + validate before committing (see SCHEMA.md)") return 0 diff --git a/scripts/recover_whisperx.py b/scripts/recover_whisperx.py new file mode 100644 index 0000000..09749bf --- /dev/null +++ b/scripts/recover_whisperx.py @@ -0,0 +1,172 @@ +#!/usr/bin/env python3 +""" +Recover pre-reorg WhisperX transcripts from journal git history. + +Before the June 2026 reorg, speaker-diarized WhisperX outputs lived as +``._.simple.txt/.json`` under per-item ``Metadata/`` +folders. The reorg kept YouTube captions as ``transcript.txt`` and dropped +those outputs from the working tree — but they survive in git history. + +This script extracts them from a pre-reorg commit and writes them per the +raw/derived transcript design: + +- ``transcript.json`` — immutable raw diarization (``SPEAKER_NN``), an array + of ``{"video_id", "segments"}`` blocks +- ``transcript.txt`` — speaker-labeled text (``SPEAKER_NN`` until names are + mapped via ``parts[].speakers`` + apply_speaker_names.py) + +Only caption-only items fully covered by whole-video outputs are written. +Items that are partial, already diarized, session-split, or already carry a +root transcript.json are reported and skipped. Extracted blobs are cached in +--work-dir (same layout as transcribe_worklist's cache). + +Usage: + python scripts/recover_whisperx.py # dry run: show plan + python scripts/recover_whisperx.py --apply # write into the journal + python scripts/recover_whisperx.py --item "GuestStream/GuestStream_040" +""" + +import argparse +import json +import re +import subprocess +import sys +from pathlib import Path + +REPO = Path(__file__).resolve().parent.parent +sys.path.insert(0, str(REPO / "scripts")) + +SRC_PREFIX = "data/video/activeinferenceinstitute" +PRE_REORG_COMMIT = "9da92662668f9cbc063b8aaeb0f3d40aad0bed00" +VIDEO_ID = re.compile(r"[A-Za-z0-9_-]{11}") + + +def git_show(journal: Path, commit: str, path: str) -> str: + return subprocess.run( + ["git", "-C", str(journal), "show", f"{commit}:{path}"], + check=True, capture_output=True, text=True).stdout + + +def index_old_outputs(journal: Path, commit: str) -> dict[str, dict]: + """Map video_id -> {"txt": path, "json": path} for whole-video outputs. + + Session-split files (``_sessNN``) are ignored — every item they + cover is already diarized in the current tree. + """ + listing = subprocess.run( + ["git", "-C", str(journal), "ls-tree", "-r", "--name-only", commit], + check=True, capture_output=True, text=True).stdout.splitlines() + index: dict[str, dict] = {} + for path in listing: + m = re.match(r"(.+)\.simple\.(txt|json)$", path) + if not m: + continue + stem, ext = m.groups() + if re.search(r"_sess\d+$", stem): + continue + if stem.endswith("]") and "[" in stem: + vid = stem[stem.rindex("[") + 1:-1] + else: + vid = stem[-11:] + if not VIDEO_ID.fullmatch(vid): + continue + index.setdefault(vid, {})[ext] = path + return {v: p for v, p in index.items() if "txt" in p and "json" in p} + + +def plan_items(journal: Path, old: dict[str, dict], only_item: str = "") -> list[dict]: + """Classify every non-duplicate item against the recovered outputs.""" + from transcription_status import item_status + + plan = [] + for meta_path in sorted((journal / SRC_PREFIX).rglob("metadata.json")): + item_dir = meta_path.parent + rel = str(item_dir.relative_to(journal / SRC_PREFIX)) + if only_item and rel != only_item: + continue + meta = json.loads(meta_path.read_text(encoding="utf-8")) + if meta.get("duplicate_of"): + continue + status = item_status(item_dir) + vids = [p["video_id"] for p in meta.get("parts", []) if p.get("video_id")] + have = [v for v in vids if v in old] + if status != "captions_only": + action = f"skip ({status})" if have else None + elif not have: + action = None + elif len(have) < len(vids): + action = "skip (partial: %d/%d parts recovered)" % (len(have), len(vids)) + elif (item_dir / "transcript.json").exists(): + action = "skip (transcript.json already present)" + else: + action = "recover" + if action: + plan.append({"rel": rel, "dir": item_dir, "vids": vids, "action": action}) + return plan + + +def extract_to_cache(journal: Path, commit: str, old: dict[str, dict], + vid: str, work_dir: Path) -> None: + for ext in ("txt", "json"): + dest = work_dir / f"{vid}.simple.{ext}" + if dest.exists(): + continue + content = git_show(journal, commit, old[vid][ext]) + if ext == "json": + segments = json.loads(content) + if not any(seg.get("speaker") for seg in segments): + raise ValueError(f"{old[vid]['json']}: no speaker fields") + elif "SPEAKER_" not in content: + raise ValueError(f"{old[vid]['txt']}: no SPEAKER_ labels") + dest.write_text(content, encoding="utf-8") + + +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("--commit", default=PRE_REORG_COMMIT, + help="pre-reorg commit holding the *.simple.* outputs") + parser.add_argument("--work-dir", type=Path, + default=REPO / "data/output/whisperx-recovered", + help="cache for extracted per-video outputs") + parser.add_argument("--apply", action="store_true", + help="write into the journal (default: dry-run plan)") + parser.add_argument("--item", default="", help="single item (path relative to source root)") + args = parser.parse_args() + + old = index_old_outputs(args.journal, args.commit) + print(f"whole-video WhisperX outputs at {args.commit[:7]}: {len(old)}") + + plan = plan_items(args.journal, old, args.item) + recover = [p for p in plan if p["action"] == "recover"] + skipped = [p for p in plan if p["action"] != "recover"] + print(f"recoverable items: {len(recover)} skipped-with-coverage: {len(skipped)}\n") + for p in skipped: + print(f" {p['action']:48} {p['rel']}") + for p in recover: + print(f" recover ({len(p['vids'])} video(s)) {p['rel']}") + + if not args.apply: + print("\ndry run — pass --apply to write transcript.json/.txt into the journal") + return 0 + + from transcribe_worklist import write_journal_transcript + + args.work_dir.mkdir(parents=True, exist_ok=True) + done = 0 + for p in recover: + try: + for vid in p["vids"]: + extract_to_cache(args.journal, args.commit, old, vid, args.work_dir) + write_journal_transcript(p["dir"], p["vids"], args.work_dir) + done += 1 + except (subprocess.CalledProcessError, ValueError, IOError) as exc: + print(f"FAILED {p['rel']}: {exc}") + print(f"\nwritten: {done}/{len(recover)} items — review the journal diff, " + "then regenerate indexes + validate before committing") + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/scripts/transcribe_worklist.py b/scripts/transcribe_worklist.py index eff213b..fdf6fd1 100644 --- a/scripts/transcribe_worklist.py +++ b/scripts/transcribe_worklist.py @@ -41,15 +41,26 @@ SRC_PREFIX = "data/video/activeinferenceinstitute" -def build_worklist(journal: Path, no_video: list = None) -> list[dict]: +def load_private_ids(journal: Path) -> set[str]: + """Video ids registered as private/unlisted — excluded from transcription.""" + path = journal / SRC_PREFIX / "private_videos.json" + if not path.exists(): + return set() + videos = json.loads(path.read_text(encoding="utf-8")).get("videos", []) + return {v["video_id"] for v in videos if v.get("video_id")} + + +def build_worklist(journal: Path, excluded: list = None) -> list[dict]: """Items lacking transcript.txt (excluding duplicates and future streams). - Items with no known video id (private/unlisted content) are appended to - ``no_video`` — their audio must be sourced manually. + Private/unlisted content is never transcribed: video ids registered in + private_videos.json are dropped, and items left with no public video are + appended to ``excluded``. """ from transcription_status import item_status - no_video = no_video if no_video is not None else [] + excluded = excluded if excluded is not None else [] + private_ids = load_private_ids(journal) today = date.today().isoformat() work = [] for meta_path in sorted((journal / SRC_PREFIX).rglob("metadata.json")): @@ -62,11 +73,12 @@ def build_worklist(journal: Path, no_video: list = None) -> list[dict]: if max(filter(None, dates), default="") > today: continue # scheduled — nothing to transcribe yet rel = str(meta_path.parent.relative_to(journal / SRC_PREFIX)) - vids = [p["video_id"] for p in meta.get("parts", []) if p.get("video_id")] + vids = [p["video_id"] for p in meta.get("parts", []) + if p.get("video_id") and p["video_id"] not in private_ids] if vids: work.append({"rel": rel, "dir": meta_path.parent, "vids": vids}) else: - no_video.append(rel) + excluded.append(rel) return work @@ -115,16 +127,16 @@ def main() -> int: load_dotenv(REPO / ".env") import os - no_video: list = [] - worklist = build_worklist(args.journal, no_video) + excluded: list = [] + worklist = build_worklist(args.journal, excluded) total_vids = sum(len(w["vids"]) for w in worklist) print(f"worklist: {len(worklist)} items / {total_vids} videos") for w in worklist: cached = sum(1 for v in w["vids"] if (args.work_dir / f"{v}.simple.txt").exists()) print(f" {w['rel']} ({len(w['vids'])} video(s), {cached} cached)") - if no_video: - print(f"no known video ({len(no_video)} items — audio must be sourced manually):") - for rel in no_video: + if excluded: + print(f"excluded ({len(excluded)} items — private/unlisted, no transcription by policy):") + for rel in excluded: print(f" {rel}") if not args.run: print("\ndry run — pass --run to transcribe") diff --git a/scripts/transcription_status.py b/scripts/transcription_status.py index ce6ea4a..4fc2da6 100644 --- a/scripts/transcription_status.py +++ b/scripts/transcription_status.py @@ -4,8 +4,10 @@ Status is computed, never stored: an item needs transcription when it lacks transcript.txt; it's a WhisperX-upgrade candidate when its transcript has no -speaker labels and no diarization CSVs. New channel videos without a journal -item (and registered private videos) are listed separately. +speaker labels and no diarization CSVs. Private/unlisted content (items with +no public video id, and ids registered in private_videos.json) is excluded — +no transcription is created for it, by policy. New channel videos without a +journal item are listed separately. Usage: python scripts/transcription_status.py # summary to stdout @@ -44,7 +46,13 @@ def main() -> int: today = _date.today().isoformat() root = args.journal / SRC_PREFIX - buckets = {"missing": [], "captions_only": [], "diarized": [], "scheduled": []} + private_path = root / "private_videos.json" + private = json.loads(private_path.read_text(encoding="utf-8")).get("videos", []) \ + if private_path.exists() else [] + private_ids = {v["video_id"] for v in private if v.get("video_id")} + + buckets = {"missing": [], "captions_only": [], "diarized": [], + "scheduled": [], "excluded": []} covered_ids: set[str] = set() for meta_path in sorted(root.rglob("metadata.json")): item_dir = meta_path.parent @@ -57,8 +65,12 @@ def main() -> int: continue # content lives at the curated item status = item_status(item_dir) dates = [meta.get("date", "")] + [p.get("date", "") for p in meta.get("parts", [])] - if status == "missing" and max(filter(None, dates), default="") > today: - status = "scheduled" # future stream — nothing to transcribe yet + if status == "missing": + if max(filter(None, dates), default="") > today: + status = "scheduled" # future stream — nothing to transcribe yet + elif not any(p.get("video_id") and p["video_id"] not in private_ids + for p in meta.get("parts", [])): + status = "excluded" # private/unlisted — no transcription by policy buckets[status].append(rel) manifest_videos = json.loads(MANIFEST.read_text(encoding="utf-8")).get("videos", []) @@ -67,27 +79,27 @@ def main() -> int: for v in manifest_videos if v.get("id") and v["id"] not in covered_ids ] - private_path = root / "private_videos.json" - private = json.loads(private_path.read_text(encoding="utf-8")).get("videos", []) \ - if private_path.exists() else [] - print(f"journal items: {sum(len(b) for b in buckets.values())}") print(f" diarized (done): {len(buckets['diarized'])}") print(f" captions only: {len(buckets['captions_only'])} (WhisperX upgrade candidates)") print(f" missing transcript: {len(buckets['missing'])} (needs transcription)") + print(f" excluded: {len(buckets['excluded'])} (private/unlisted — no transcription by policy)") print(f" scheduled (future): {len(buckets['scheduled'])} {buckets['scheduled']}") from datetime import date manifest_age = date.fromtimestamp(MANIFEST.stat().st_mtime).isoformat() print(f"channel videos w/o item:{len(no_item):>4} (manifest from {manifest_age} — re-enumerate for newer videos)") - print(f"private videos (manual):{len(private):>4}") + print(f"registered private ids: {len(private):>4}") for rel in buckets["missing"]: print(f" MISSING {rel}") + for rel in buckets["excluded"]: + print(f" EXCLUDED {rel}") for row in no_item: print(f" NO-ITEM {row['video_id']} {row['title'][:70]}") if args.report: args.report.write_text(json.dumps( - {"missing": buckets["missing"], "scheduled": buckets["scheduled"], + {"missing": buckets["missing"], "excluded": buckets["excluded"], + "scheduled": buckets["scheduled"], "captions_only": buckets["captions_only"], "diarized_count": len(buckets["diarized"]), "no_item": no_item, "private": private}, indent=2, ensure_ascii=False) + "\n", encoding="utf-8") diff --git a/tests/journal_utilities/test_apply_speaker_names.py b/tests/journal_utilities/test_apply_speaker_names.py new file mode 100644 index 0000000..79c3438 --- /dev/null +++ b/tests/journal_utilities/test_apply_speaker_names.py @@ -0,0 +1,117 @@ +"""Tests for scripts/apply_speaker_names.py (regenerate txt from raw json).""" + +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 (SALVAGE_NAME, has_speaker_structure, load_blocks, + process_item, render_txt) + + +def _blocks(vid="vid00000001", speakers=("SPEAKER_00", "SPEAKER_01")): + return [{"video_id": vid, "segments": [ + {"start": 0.0, "end": 1.0, "text": "hello there", "speaker": speakers[0]}, + {"start": 1.0, "end": 2.0, "text": "hi back", "speaker": speakers[1]}, + {"start": 2.0, "end": 3.0, "text": "more from me", "speaker": speakers[1]}, + ]}] + + +def _meta(vid="vid00000001", speakers=None): + return {"parts": [{"video_id": vid, "speakers": speakers}]} + + +class TestRenderTxt: + def test_unmapped_keeps_labels(self): + txt, unmapped = render_txt(_blocks(), _meta()) + assert txt.startswith("SPEAKER_00:\nhello there\n") + assert unmapped == {"SPEAKER_00", "SPEAKER_01"} + + def test_mapping_applied_and_reported(self): + txt, unmapped = render_txt(_blocks(), _meta(speakers={"SPEAKER_00": "Ada"})) + assert txt.startswith("Ada:\nhello there\n") + assert unmapped == {"SPEAKER_01"} + + def test_same_name_labels_merge(self): + txt, _ = render_txt(_blocks(), _meta( + speakers={"SPEAKER_00": "Ada", "SPEAKER_01": "Ada"})) + assert txt.count("Ada:") == 1 + + def test_multi_block_headers(self): + blocks = _blocks("vidaaaaaaaa") + _blocks("vidbbbbbbbb") + meta = {"parts": [{"video_id": "vidaaaaaaaa"}, {"video_id": "vidbbbbbbbb"}]} + txt, _ = render_txt(blocks, meta) + assert "## vidaaaaaaaa" in txt and "## vidbbbbbbbb" in txt + + def test_sess_split_uses_base_video_mapping(self): + blocks = [{"video_id": "vid00000001_sess01", "segments": _blocks()[0]["segments"]}] + txt, _ = render_txt(blocks, _meta(speakers={"SPEAKER_00": "Ada"})) + assert "Ada:" in txt + + +class TestLoadBlocks: + def test_rejects_assemblyai_shape(self, tmp_path): + tj = tmp_path / "transcript.json" + tj.write_text(json.dumps([{"video_id": "", "segments": {"id": "x"}}])) + assert load_blocks(tj) is None + + def test_rejects_undiarized(self, tmp_path): + tj = tmp_path / "transcript.json" + tj.write_text(json.dumps([{"video_id": "v", "segments": [{"text": "hi"}]}])) + assert load_blocks(tj) is None + + def test_accepts_whisperx_blocks(self, tmp_path): + tj = tmp_path / "transcript.json" + tj.write_text(json.dumps(_blocks())) + assert load_blocks(tj) is not None + + +class TestHasSpeakerStructure: + def test_speaker_label(self): + assert has_speaker_structure("SPEAKER_00:\nhello\n") + + def test_mapped_name(self): + assert has_speaker_structure("Fraser Paterson:\nhello\n") + + def test_header_then_name(self): + assert has_speaker_structure("## vid00000001\n\nAda Lovelace:\nhello\n") + + def test_captions_prose(self): + assert not has_speaker_structure("hello and welcome to the stream today\nwe will\n") + + +class TestProcessItem: + def _item(self, tmp_path, old_txt=None): + (tmp_path / "transcript.json").write_text(json.dumps(_blocks())) + if old_txt is not None: + (tmp_path / "transcript.txt").write_text(old_txt) + return tmp_path + + def test_upgrade_salvages_captions(self, tmp_path): + item = self._item(tmp_path, "caption prose without speakers\n") + result = process_item(item, _meta(), apply=True) + assert result["action"] == "upgrade+salvage" + assert (item / "captions" / SALVAGE_NAME).read_text() \ + == "caption prose without speakers\n" + assert (item / "transcript.txt").read_text().startswith("SPEAKER_00:") + + def test_idempotent_second_run(self, tmp_path): + item = self._item(tmp_path, "caption prose without speakers\n") + process_item(item, _meta(), apply=True) + assert process_item(item, _meta(), apply=True)["action"] == "unchanged" + + def test_dry_run_writes_nothing(self, tmp_path): + item = self._item(tmp_path, "caption prose without speakers\n") + result = process_item(item, _meta(), apply=False) + assert result["action"] == "upgrade+salvage" + assert (item / "transcript.txt").read_text() == "caption prose without speakers\n" + assert not (item / "captions").exists() + + def test_json_never_touched(self, tmp_path): + item = self._item(tmp_path, "caption prose\n") + before = (item / "transcript.json").read_text() + process_item(item, _meta(speakers={"SPEAKER_00": "Ada"}), apply=True) + assert (item / "transcript.json").read_text() == before + assert (item / "transcript.txt").read_text().startswith("Ada:") diff --git a/tests/journal_utilities/test_transcribe_worklist.py b/tests/journal_utilities/test_transcribe_worklist.py index 2d2a564..851e30f 100644 --- a/tests/journal_utilities/test_transcribe_worklist.py +++ b/tests/journal_utilities/test_transcribe_worklist.py @@ -35,6 +35,18 @@ def test_selects_only_missing_past_items(self, tmp_path): assert [w["rel"] for w in work] == ["A/needs_it"] assert work[0]["vids"] == ["vid00000001"] + def test_private_unlisted_never_transcribed(self, tmp_path): + _make_item(tmp_path, "A/empty_parts", []) + _make_item(tmp_path, "A/all_private", ["vidprivate1"]) + _make_item(tmp_path, "A/mixed", ["vidprivate1", "vid00000001"]) + (tmp_path / SRC_PREFIX / "private_videos.json").write_text( + json.dumps({"videos": [{"video_id": "vidprivate1"}]})) + excluded = [] + work = build_worklist(tmp_path, excluded) + assert [w["rel"] for w in work] == ["A/mixed"] + assert work[0]["vids"] == ["vid00000001"] + assert excluded == ["A/all_private", "A/empty_parts"] + class TestWriteJournalTranscript: def _cache(self, out_dir: Path, vid: str, text: str) -> None: