conductor(score_dynamics_giorgini): Phase 1 Acquire - transcript (1485 clean segments, 46.5KB) + 178MB mp4

2026-06-21 20:43:50 -04:00
parent 1dce32037a
commit 16fbf5619f
5 changed files with 22546 additions and 0 deletions
@@ -0,0 +1,95 @@
+"""Clean yt-dlp auto-sub VTT transcripts.
+
+yt-dlp auto-subs produce rolling captions where each segment extends the
+previous one with new words, sometimes triplicated. Algorithm:
+  1. Strip VTT tags (<00:00:00.560><c>...</c>) from raw text.
+  2. For each pair (prev, curr), compute the new suffix curr adds vs prev.
+     Keep only the new suffix.
+  3. Drop empty / duplicate results.
+  4. Skip segments that are pure repetition of the prior segment.
+"""
+import json
+import re
+from pathlib import Path
+
+VTT_TAG = re.compile(r"<[^>]+>")
+WS = re.compile(r"\s+")
+
+def strip_vtt(t: str) -> str:
+    return WS.sub(" ", VTT_TAG.sub("", t)).strip()
+
+def longest_common_prefix_len(a: str, b: str) -> int:
+    n = min(len(a), len(b))
+    i = 0
+    while i < n and a[i] == b[i]:
+        i += 1
+    return i
+
+def longest_common_suffix_len(a: str, b: str, max_k: int = 80) -> int:
+    n = min(len(a), len(b), max_k)
+    i = 0
+    while i < n and a[-1 - i] == b[-1 - i]:
+        i += 1
+    return i
+
+def main(track: str) -> None:
+    art = Path(f"conductor/tracks/{track}/artifacts")
+    p = art / "transcript.json"
+    data = json.loads(p.read_text(encoding="utf-8"))
+    raw_segments = data["segments"]
+
+    cleaned: list[dict] = []
+    prev_clean = ""
+    for raw in raw_segments:
+        text = strip_vtt(raw["text"])
+        if not text:
+            continue
+        if prev_clean and text == prev_clean:
+            continue
+        if prev_clean and text.startswith(prev_clean):
+            new = text[len(prev_clean):].strip()
+            if not new:
+                continue
+            cleaned.append({"start": raw["start"], "text": new})
+            prev_clean = text
+            continue
+        lcp = longest_common_prefix_len(prev_clean, text)
+        if lcp >= 5:
+            new = text[lcp:].strip()
+            if new:
+                cleaned.append({"start": raw["start"], "text": new})
+                prev_clean = text
+                continue
+        lcs = longest_common_suffix_len(prev_clean, text)
+        if lcs >= 5:
+            new = text[: -lcs if lcs < len(text) else 0].strip()
+            if new:
+                cleaned.append({"start": raw["start"], "text": new})
+                prev_clean = text
+                continue
+        cleaned.append({"start": raw["start"], "text": text})
+        prev_clean = text
+
+    deduped: list[dict] = []
+    seen: set[str] = set()
+    for seg in cleaned:
+        t = seg["text"]
+        if t in seen:
+            continue
+        seen.add(t)
+        deduped.append(seg)
+
+    plain = "\n".join(s["text"] for s in deduped)
+    (art / "transcript_clean.txt").write_text(plain, encoding="utf-8")
+    data["clean_segments"] = deduped
+    data["plain"] = plain
+    p.write_text(json.dumps(data, indent=2, ensure_ascii=False), encoding="utf-8")
+    print(f"{track}: {len(raw_segments)} raw -> {len(cleaned)} cleaned -> {len(deduped)} deduped")
+    print(f"plain length: {len(plain)} chars")
+    print("First 800 chars:")
+    print(plain[:800])
+
+
+if __name__ == "__main__":
+    import sys
+    main(sys.argv[1] if len(sys.argv) > 1 else "video_analysis_score_dynamics_giorgini_20260621")