Freeze actual Google music metadata and precise speech timing diagnostics

tools/speech-eval/token_alignment.py

@@ -0,0 +1,54 @@
+"""Strict ordinary Whisper token-time sentence subdivision, never label driven."""
+import re
+
+POLICY = "ordinary-token-sentences-v1;punctuation-boundaries;exact-text;no-dtw;no-labels"
+# Pinned multilingual Whisper vocabulary: ordinary text ids precede EOT.
+TEXT_TOKEN_LIMIT = 50257
+BOUNDARY = re.compile(r'[.!?。！？]["\u201d\u2019\')\]]*$')
+
+
+def aligned_segments(document, offset_ms, duration_ms):
+    from run import segments
+    originals = segments(document, offset_ms, duration_ms)
+    result = []
+    raw_rows = [row for row in document["transcription"] if row["text"].strip()]
+    for index, (row, original) in enumerate(zip(raw_rows, originals)):
+        tokens = row.get("tokens")
+        if not isinstance(tokens, list) or not 0 < len(tokens) <= 4096:
+            raise ValueError("Missing or oversized token times")
+        accepted = []
+        previous = 0
+        for token in tokens:
+            token_id, text = token.get("id"), token.get("text")
+            if type(token_id) is not int or token_id < 0 or not isinstance(text, str):
+                raise ValueError("Invalid token identity")
+            # Structural tokens must never become recognized text candidates.
+            if token_id >= TEXT_TOKEN_LIMIT:
+                if not text.startswith("[_") or not text.endswith("]"):
+                    raise ValueError("Unrecognized special token")
+                continue
+            times = token.get("offsets", {})
+            start, end = times.get("from"), times.get("to")
+            if (type(start) is not int or type(end) is not int
+                    or not 0 <= start <= end <= duration_ms or start < previous
+                    or not row["offsets"]["from"] <= start <= end <= row["offsets"]["to"]):
+                raise ValueError("Invalid ordinary token timestamps")
+            previous = start
+            accepted.append((text, start, end))
+        if "".join(token[0] for token in accepted) != row["text"]:
+            raise ValueError("Token text differs from original recognized segment")
+        phrase = []
+        for token_index, token in enumerate(accepted):
+            phrase.append(token)
+            if BOUNDARY.search(token[0]) or token_index == len(accepted) - 1:
+                text = "".join(part[0] for part in phrase)
+                if text.strip():
+                    result.append({"start_ms": offset_ms + phrase[0][1],
+                                   "end_ms": offset_ms + phrase[-1][2], "text": text,
+                                   "original_segment_index": index,
+                                   "original_start_ms": original["start_ms"],
+                                   "original_end_ms": original["end_ms"]})
+                phrase = []
+    if len(result) > 10000:
+        raise ValueError("Oversized aligned passage pool")
+    return result