268 lines · 7.7 KB
python
| 1 | """Chop a transcript into clips and highlights. |
| 2 | |
| 3 | cues.json is the working set. A clip is [start, end] over that set. |
| 4 | A highlight is a cue worth jumping to — punchy, named, timed. |
| 5 | """ |
| 6 | |
| 7 | from __future__ import annotations |
| 8 | |
| 9 | import json |
| 10 | import re |
| 11 | from pathlib import Path |
| 12 | |
| 13 | from space_tape.render import RenderError, format_ts, _at, _run, _ffmpeg |
| 14 | |
| 15 | SKIP = {"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. |
| 19 | KEYWORDS = ( |
| 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 | |
| 42 | def 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 | |
| 68 | def 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 | |
| 99 | def 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 | |
| 116 | def 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 | |
| 135 | def 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 | |
| 168 | def 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 | |
| 202 | def 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 | |
| 233 | def 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 | |
| 259 | def 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 | |
| 266 | def _overlap(a0: float, a1: float, b0: float, b1: float) -> float: |
| 267 | return max(0.0, min(a1, b1) - max(a0, b0)) |