Prepare bounded speech diagnostics and document mixed routing risks

tools/speech-eval/score.py

@@ -0,0 +1,53 @@
+"""Score externally ranked genuine transcript hits against unchanged media v1."""
+import argparse
+import hashlib
+import json
+from pathlib import Path
+from run import FIXTURE_SHA
+
+
+def score(fixture, rankings):
+    queries = [query for query in fixture["queries"] if query["category"] == "video-speech"]
+    if len(queries) != 8:
+        raise ValueError("Expected the eight frozen speech queries")
+    ids = {query["id"] for query in queries}
+    if set(rankings) != ids:
+        raise ValueError("Rankings must cover exactly eight frozen speech queries")
+    rows = []
+    for query in queries:
+        hits = rankings[query["id"]]
+        if not isinstance(hits, list) or len(hits) > 100:
+            raise ValueError("Invalid ranking length")
+        hit = hits[0] if hits else None
+        point = overlap = False
+        if hit:
+            start, end = hit["start_ms"], hit["end_ms"]
+            if type(start) is not int or type(end) is not int or not 0 <= start <= end:
+                raise ValueError("Invalid ranked timestamp")
+            intervals = query["intervals"].get(hit["asset"], [])
+            point = any(left <= start / 1000 <= right for left, right in intervals)
+            overlap = any(min(end / 1000, right) > max(start / 1000, left) for left, right in intervals)
+        rows.append({"query": query["id"], "point_hit": point, "overlap_hit": overlap})
+    return {"scope": "Externally ranked transcript top1, no oracle segment selection",
+            "queries": len(rows), "point_hits": sum(row["point_hit"] for row in rows),
+            "overlap_hits": sum(row["overlap_hit"] for row in rows), "rows": rows}
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--fixture", type=Path, default=Path("fixtures/media/fixture.json"))
+    parser.add_argument("--rankings", type=Path, required=True)
+    parser.add_argument("--output", type=Path, required=True)
+    args = parser.parse_args()
+    data = args.fixture.read_bytes()
+    rankings = json.loads(args.rankings.read_text(encoding="utf-8"))
+    digest = hashlib.sha256(data).hexdigest()
+    if digest != FIXTURE_SHA or rankings["fixture_sha256"] != digest or not rankings.get("asr_report_sha256") or not rankings.get("google_worker_sha256"):
+        raise ValueError("Missing/mismatched actual ASR/Google provenance")
+    report = score(json.loads(data), rankings["rankings"])
+    report.update({key: rankings[key] for key in ("fixture_sha256", "asr_report_sha256", "google_worker_sha256")})
+    args.output.write_text(json.dumps(report, indent=2) + "\n", encoding="utf-8")
+
+
+if __name__ == "__main__":
+    main()