Open-source X Spaces transcription.

Star

268 lines · 7.7 KB

python
1"""Chop a transcript into clips and highlights.
2
3cues.json is the working set. A clip is [start, end] over that set.
4A highlight is a cue worth jumping to — punchy, named, timed.
5"""
6
7from __future__ import annotations
8
9import json
10import re
11from pathlib import Path
12
13from space_tape.render import RenderError, format_ts, _at, _run, _ffmpeg
14
15SKIP = {"silence", "unknown", "both", "overlap", "host", None, ""}
16
17# Moments people actually clip. Generic enough for any Space; the demo
18# (melted / insane / running for president) lights up on these.
19KEYWORDS = (
20 "melted",
21 "insane",
22 "sorry",
23 "broke the internet",
24 "half a million",
25 "running for president",
26 "decline",
27 "choice",
28 "announcement",
29 "never seen",
30 "servers",
31 "straining",
32 "historic",
33 "broke twitter",
34 "i am running",
35 "i'm here",
36 "teleprompter",
37 "q and a",
38 "q&a",
39)
40
41
42def parse_ts(value: str | float | int) -> float:
43 """Accept 18, 18.5, 00:18, 1:02:03, 00:01:36.2."""
44 if isinstance(value, (int, float)):
45 t = float(value)
46 if t < 0:
47 raise ValueError("timestamp must be >= 0")
48 return t
49 raw = str(value).strip()
50 if not raw:
51 raise ValueError("empty timestamp")
52 if re.fullmatch(r"\d+(\.\d+)?", raw):
53 return float(raw)
54 parts = raw.split(":")
55 if not 2 <= len(parts) <= 3:
56 raise ValueError(f"bad timestamp: {value!r}")
57 try:
58 nums = [float(p) for p in parts]
59 except ValueError as exc:
60 raise ValueError(f"bad timestamp: {value!r}") from exc
61 if len(nums) == 2:
62 m, s = nums
63 return m * 60 + s
64 h, m, s = nums
65 return h * 3600 + m * 60 + s
66
67
68def clip_cues(
69 cues: list[dict],
70 start: float,
71 end: float,
72 *,
73 trim: bool = True,
74) -> list[dict]:
75 """Cues that overlap [start, end]. Optionally trim the first/last to the window."""
76 lo, hi = (start, end) if end >= start else (end, start)
77 out: list[dict] = []
78 for cue in cues:
79 a = float(cue.get("start") or 0)
80 b = float(cue.get("end") or a)
81 if b <= lo or a >= hi:
82 continue
83 row = {
84 "start": max(a, lo) if trim else a,
85 "end": min(b, hi) if trim else b,
86 "speaker": cue.get("speaker") or "unknown",
87 "text": cue.get("text") or "",
88 }
89 if "words" in cue:
90 row["words"] = [
91 w
92 for w in cue["words"]
93 if float(w.get("end") or 0) > lo and float(w.get("start") or 0) < hi
94 ]
95 out.append(row)
96 return out
97
98
99def quote_clip(cues: list[dict], *, source_url: str | None = None) -> str:
100 """Markdown you can paste. Speaker first, timestamp, text. X link at the end."""
101 lines: list[str] = []
102 for cue in cues:
103 text = (cue.get("text") or "").strip()
104 if not text:
105 continue
106 speaker = _at(cue.get("speaker") or "unknown")
107 stamp = format_ts(float(cue.get("start") or 0))
108 lines.append(f"**{speaker}** [{stamp}] {text}")
109 lines.append("")
110 if source_url:
111 lines.append(source_url)
112 lines.append("")
113 return "\n".join(lines)
114
115
116def label_for(text: str, *, words: int = 6) -> str:
117 clean = re.sub(r"\s+", " ", (text or "").strip())
118 if not clean:
119 return "clip"
120 sentences = [s.strip() for s in re.split(r"(?<=[.!?])\s+", clean) if s.strip()]
121 chosen = ""
122 for sentence in sentences:
123 if any(kw in sentence.lower() for kw in KEYWORDS):
124 chosen = sentence
125 break
126 if not chosen:
127 chosen = sentences[0] if sentences else clean
128 chosen = chosen.rstrip(".,;:")
129 parts = chosen.split(" ")
130 if len(parts) <= 10:
131 return chosen
132 return " ".join(parts[:words]).rstrip(".,;:") + "…"
133
134
135def score_cue(cue: dict) -> float:
136 text = (cue.get("text") or "").strip()
137 speaker = cue.get("speaker")
138 if not text or speaker in SKIP:
139 return 0.0
140 start = float(cue.get("start") or 0)
141 end = float(cue.get("end") or start)
142 dur = end - start
143 if dur < 0.5:
144 return 0.0
145 lower = text.lower()
146 kw_hits = sum(1 for kw in KEYWORDS if kw in lower)
147 nwords = len(text.split())
148 score = 1.0 + min(kw_hits, 2) * 3.0
149 if nwords <= 8 and kw_hits:
150 score += 2.0
151 elif nwords <= 22:
152 score += 1.4
153 elif nwords > 50:
154 score -= 2.0
155 if "?" in text and nwords <= 30:
156 score += 0.5
157 if re.search(r"\bI am\b|\bI'm\b", text):
158 score += 0.8
159 if 2.0 <= dur <= 16.0:
160 score += 0.5
161 for sentence in re.split(r"(?<=[.!?])\s+", text):
162 if any(kw in sentence.lower() for kw in KEYWORDS) and len(sentence.split()) <= 14:
163 score += 1.6
164 break
165 return score
166
167
168def highlights(cues: list[dict], *, max_n: int = 8) -> list[dict]:
169 """Pick punchy, named cues. Ordered by time, not score."""
170 ranked = []
171 for i, cue in enumerate(cues):
172 s = score_cue(cue)
173 if s <= 0:
174 continue
175 ranked.append((s, i, cue))
176 ranked.sort(key=lambda row: (-row[0], row[1]))
177 picked: list[dict] = []
178 used: list[tuple[float, float]] = []
179 for s, _i, cue in ranked:
180 a = float(cue.get("start") or 0)
181 b = float(cue.get("end") or a)
182 if any(_overlap(a, b, u0, u1) > 0.5 * (b - a) for u0, u1 in used):
183 continue
184 text = (cue.get("text") or "").strip()
185 picked.append(
186 {
187 "start": round(a, 3),
188 "end": round(b, 3),
189 "speaker": cue.get("speaker") or "unknown",
190 "text": text,
191 "label": label_for(text),
192 "score": round(s, 3),
193 }
194 )
195 used.append((a, b))
196 if len(picked) >= max_n:
197 break
198 picked.sort(key=lambda h: h["start"])
199 return picked
200
201
202def write_clip(
203 cues: list[dict],
204 start: float,
205 end: float,
206 dest: str | Path,
207 *,
208 source_url: str | None = None,
209 audio_src: str | Path | None = None,
210 title: str = "clip",
211) -> dict[str, Path]:
212 chopped = clip_cues(cues, start, end)
213 dest_path = Path(dest)
214 dest_path.mkdir(parents=True, exist_ok=True)
215 json_path = dest_path / "clip.json"
216 md_path = dest_path / "clip.md"
217 json_path.write_text(json.dumps(chopped, indent=2, ensure_ascii=False) + "\n", encoding="utf-8")
218 header = [
219 f"# {title}",
220 "",
221 f"- Window: {format_ts(start)}–{format_ts(end)}",
222 ]
223 if source_url:
224 header.append(f"- Space: {source_url}")
225 header.append("")
226 md_path.write_text("\n".join(header) + quote_clip(chopped, source_url=None), encoding="utf-8")
227 written = {"cues": json_path, "transcript": md_path}
228 if audio_src is not None:
229 written["audio"] = write_audio_clip(audio_src, dest_path / "clip.mp3", start, end)
230 return written
231
232
233def write_audio_clip(src: str | Path, dest: str | Path, start: float, end: float) -> Path:
234 ffmpeg = _ffmpeg()
235 dest_path = Path(dest)
236 dest_path.parent.mkdir(parents=True, exist_ok=True)
237 if end <= start:
238 raise RenderError("clip end must be after start")
239 cmd = [
240 ffmpeg,
241 "-y",
242 "-i",
243 str(src),
244 "-ss",
245 f"{start:.3f}",
246 "-to",
247 f"{end:.3f}",
248 "-vn",
249 "-ac",
250 "1",
251 "-b:a",
252 "40k",
253 str(dest_path),
254 ]
255 _run(cmd, "ffmpeg could not write clip.mp3")
256 return dest_path
257
258
259def load_cues(path: str | Path) -> list[dict]:
260 data = json.loads(Path(path).read_text(encoding="utf-8"))
261 if not isinstance(data, list):
262 raise ValueError("cues.json must be a list")
263 return data
264
265
266def _overlap(a0: float, a1: float, b0: float, b1: float) -> float:
267 return max(0.0, min(a1, b1) - max(a0, b0))