Blame for scripts/prepare_interview.py

"""Prepare a released interview end to end: video parts -> transcript -> brief -> back-test.

Run on a GPU machine (Colab, a lab workstation, or a GPU Space). Partner media
and catalogue records stay under private/ and are never committed.

    python scripts/prepare_interview.py \
        --id RG-50.030.0001 \
        --title "Oral history interview with ..." \
        --media private/RG-50.030.0001/media/*.mp4 \
        --catalogue private/RG-50.030.0001/catalogue.json \
        --donor private/RG-50.030.0001/donor.json \
        --device cuda

Steps
  1. media.prepare      checks parts (e.g. 01.06 ... 06.06), extracts and joins audio
  2. asr.transcribe     faster-whisper large-v3, per-segment language ID, flags
  3. transcript review  writes transcript.txt and STOPS unless --skip-review:
                        correct speaker roles and obvious errors, then rerun with --transcript
  4. pipeline           spatio-textual + period gazetteer + agent + validators
  5. backtest           suggested headings vs the catalogue's own subjects

Outputs go to data/interviews/<id>/ (brief.json, backtest.json). Only commit
them, or deploy them to a public Space, when the museum has confirmed the
interview's release terms allow it.
"""
from __future__ import annotations

import argparse
import json
import sys
from pathlib import Path

ROOT = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(ROOT))

from curatorbot import transcript as tx  # noqa: E402


def main() -> int:
    p = argparse.ArgumentParser()
    p.add_argument("--id", required=True)
    p.add_argument("--title", default="")
    p.add_argument("--media", nargs="*", default=[], help="Recordings, or folders holding them (searched at any depth)")
    p.add_argument("--transcript", help="Reviewed transcript (.txt Q/A or .json); skips transcription")
    p.add_argument("--catalogue", help="JSON with 'subjects' (and optionally 'title', 'summary')")
    p.add_argument("--donor", help="JSON of donor-reported fields, if any")
    p.add_argument("--note", default="", help="Shown above the brief; a default is chosen from --status")
    p.add_argument("--model", default="large-v3")
    p.add_argument("--device", default="cuda")
    p.add_argument("--prompt", default="Auschwitz, Birkenau, Bergen-Belsen, Theresienstadt, Treblinka, Majdanek, Łódź, Litzmannstadt, Lemberg, Judenrat, Kapo, Appell, UNRRA, HIAS")
    p.add_argument("--skip-review", action="store_true", help="Analyse the raw ASR transcript without stopping for review")
    p.add_argument("--status", choices=["released", "under_review"], default="released",
                   help="released: catalogued interview, back-tested; under_review: new submission (e.g. from Box)")
    p.add_argument("--progress-file", help="Also keep the latest progress here (e.g. in Google Drive, to check from another device)")
    p.add_argument("--source-url", help="Public catalogue page for a released interview")
    a = p.parse_args()

    if not a.note:
        a.note = ("Released interview, already catalogued: CuratorBot's suggestions are compared with the catalogue record."
                  if a.status == "released" else
                  "Submission under review: not yet catalogued. Suggestions are drafts for the reviewing curator.")
    out_dir = ROOT / "data" / "interviews" / a.id
    work = ROOT / "work" / a.id
    out_dir.mkdir(parents=True, exist_ok=True)
    work.mkdir(parents=True, exist_ok=True)
    catalogue = json.loads(Path(a.catalogue).read_text(encoding="utf-8")) if a.catalogue else None
    donor = json.loads(Path(a.donor).read_text(encoding="utf-8")) if a.donor else {}
    media_info = None
    from curatorbot.progress import Progress
    prog = Progress(a.progress_file or work / "progress.json")

    if a.transcript:
        src = Path(a.transcript)
        if src.suffix == ".json":
            d = json.loads(src.read_text(encoding="utf-8"))
            segments, media_info = d["segments"], d.get("media")
        else:
            segments = tx.parse_qa_text(src.read_text(encoding="utf-8"))
            side = work / "transcript.json"
            if side.exists():
                media_info = json.loads(side.read_text(encoding="utf-8")).get("media")
    else:
        if not a.media:
            p.error("Give --media files or a reviewed --transcript.")
        from curatorbot.asr import transcribe
        from curatorbot.intake import find_media
        from curatorbot.media import prepare
        files = [f for m in a.media for f in find_media(m)]
        if not files:
            p.error(f"No recordings found in: {', '.join(a.media)}")
        prog.start("Starting", f"{a.id}: {len(files)} recording(s): " + ", ".join(f.name for f in files))
        prep = prepare(files, workdir=work, progress=prog)
        for issue in prep["issues"]:
            print("WARNING:", issue)
        res = transcribe(prep["audio"], size=a.model, device=a.device, initial_prompt=a.prompt, progress=prog)
        segments = tx.from_asr(res["chunks"])
        media_info = {
            "parts": [{"part": x["part"], "duration": x["duration"], "audio": x.get("audio_codec") or ""} for x in prep["parts"]],
            "duration": prep["duration"], "issues": prep["issues"], "simulated": False,
            "languages": _language_spans(res["chunks"], prep["duration"]),
            "flags": [{"at": tx.fmt_time(f["start"]), "text": f["reason"]} for f in res["flags"]][:50],
        }
        (work / "transcript.json").write_text(json.dumps({"media": media_info, "segments": segments}, ensure_ascii=False, indent=1), encoding="utf-8")
        (work / "transcript.txt").write_text(tx.to_qa_text(segments), encoding="utf-8")
        prog.start("Done", f"transcript {work / 'transcript.txt'}: {len(segments)} turns, {prep['duration']} of audio, "
                           f"{len(res['flags'])} listening flag(s)")
        if not a.skip_review:
            print("Review speaker roles and errors, then rerun with --transcript", work / "transcript.txt")
            return 0

    from curatorbot.pipeline import build_brief
    brief = build_brief(segments, interview_id=a.id, title=a.title or (catalogue or {}).get("title") or a.id,
                        donor=donor, note=a.note, media=media_info, status=a.status,
                        source={"url": a.source_url or (catalogue or {}).get("url", "")}, progress=prog)
    (out_dir / "brief.json").write_text(json.dumps(brief, ensure_ascii=False), encoding="utf-8")
    print("Brief:", out_dir / "brief.json", brief["agent"].get("validation"))
    print("Places:", brief.get("placeCoverage"))
    if catalogue:
        from curatorbot.backtest import score_brief
        bt = score_brief(brief, catalogue)
        (out_dir / "backtest.json").write_text(json.dumps(bt, ensure_ascii=False, indent=2), encoding="utf-8")
        print(f"Back-test: recall {bt['recall_exact']:.0%} exact, {bt['recall_head']:.0%} by main heading; "
              f"{len(bt['missed'])} catalogue headings missed.")
        print(f"Model only: precision@10 {bt['precision_exact']:.0%}; R-precision {bt['model_r_precision']:.0%} "
              f"(top {len(catalogue.get('subjects', []))} of {bt['model_suggested_all']} suggestions).")
        w = bt["with_rules"]
        print(f"With form headings by rule: R-precision {w['r_precision']:.0%} (top {w['r']}), "
              f"all {w['suggested']} suggestions: precision {w['precision_exact']:.0%}, recall {w['recall_exact']:.0%} exact.")
    return 0


def _language_spans(chunks, duration):
    spans = []
    for c in chunks:
        lang = c.get("lang") or "en"
        if spans and spans[-1]["code"] == lang:
            spans[-1]["to"] = tx.fmt_time(c["end"])
        else:
            spans.append({"lang": lang, "code": lang, "from": tx.fmt_time(c["start"]), "to": tx.fmt_time(c["end"])})
    if spans:
        spans[0]["from"], spans[-1]["to"] = "00:00:00", duration
    return spans


if __name__ == "__main__":
    raise SystemExit(main())