Spaces:
Running
Running
| """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()) | |