space-tapePublic

Open-source X Spaces transcription.

Star

208 lines · 5.3 KB

python
1"""Write cues.json, transcript.md, and a speech-bitrate mp3."""
2
3from __future__ import annotations
4
5import json
6import shutil
7import subprocess
8from pathlib import Path
9
10
11class RenderError(RuntimeError):
12 pass
13
14
15def write_outputs(
16 out_dir: str | Path,
17 *,
18 cues: list[dict],
19 source_url: str,
20 posted_url: str | None = None,
21 title: str = "Space",
22 audio_src: str | Path | None = None,
23 speakers: list[str] | None = None,
24) -> dict[str, Path]:
25 dest = Path(out_dir)
26 dest.mkdir(parents=True, exist_ok=True)
27 json_path = dest / "cues.json"
28 md_path = dest / "transcript.md"
29 json_path.write_text(
30 json.dumps(_public_cues(cues), indent=2, ensure_ascii=False) + "\n",
31 encoding="utf-8",
32 )
33 md_path.write_text(
34 transcript_md(
35 cues,
36 source_url=source_url,
37 posted_url=posted_url,
38 title=title,
39 speakers=speakers,
40 ),
41 encoding="utf-8",
42 )
43 written: dict[str, Path] = {"cues": json_path, "transcript": md_path}
44 if audio_src is not None:
45 mp3 = write_mp3(audio_src, dest / "audio.mp3")
46 written["audio"] = mp3
47 return written
48
49
50def _public_cues(cues: list[dict]) -> list[dict]:
51 out = []
52 for cue in cues:
53 row = {
54 "start": round(float(cue.get("start") or 0), 3),
55 "end": round(float(cue.get("end") or 0), 3),
56 "speaker": cue.get("speaker") or "unknown",
57 "text": cue.get("text") or "",
58 }
59 if "words" in cue:
60 row["words"] = cue["words"]
61 out.append(row)
62 return out
63
64
65def transcript_md(
66 cues: list[dict],
67 *,
68 source_url: str,
69 posted_url: str | None = None,
70 title: str = "Space",
71 speakers: list[str] | None = None,
72) -> str:
73 names = speakers or _speakers_of(cues)
74 lines = [
75 f"# {title}",
76 "",
77 f"- Space: {source_url}",
78 ]
79 if posted_url and posted_url != source_url:
80 lines.append(f"- Posted: {posted_url}")
81 if names:
82 shown = ", ".join(_at(n) for n in names)
83 lines.append(f"- Speakers: {shown}")
84 lines.append("")
85 for cue in cues:
86 text = (cue.get("text") or "").strip()
87 if not text:
88 continue
89 speaker = _at(cue.get("speaker") or "unknown")
90 stamp = format_ts(float(cue.get("start") or 0))
91 lines.append(f"**[{stamp}] {speaker}** {text}")
92 lines.append("")
93 return "\n".join(lines)
94
95
96def format_ts(t: float) -> str:
97 if t < 0:
98 t = 0.0
99 total = int(round(t))
100 h, rem = divmod(total, 3600)
101 m, s = divmod(rem, 60)
102 return f"{h:02d}:{m:02d}:{s:02d}"
103
104
105def _at(name: str) -> str:
106 if name in {"unknown", "both", "overlap", "silence", "host"}:
107 return name
108 return name if name.startswith("@") else f"@{name}"
109
110
111def _speakers_of(cues: list[dict]) -> list[str]:
112 seen: list[str] = []
113 skip = {"unknown", "both", "overlap", "silence", None, ""}
114 for cue in cues:
115 sp = cue.get("speaker")
116 if sp in skip or sp in seen:
117 continue
118 seen.append(sp)
119 return seen
120
121
122def wav_for_asr(
123 src: str | Path,
124 dest: str | Path,
125 *,
126 chunk_seconds: int | None = None,
127) -> list[tuple[Path, float]]:
128 """Convert a copy for Whisper. Original replay stays tagged.
129
130 Returns [(wav_path, offset_seconds)]. One entry if unchunked.
131 """
132 ffmpeg = _ffmpeg()
133 dest_path = Path(dest)
134 dest_path.parent.mkdir(parents=True, exist_ok=True)
135 wav = dest_path if dest_path.suffix.lower() == ".wav" else dest_path.with_suffix(".wav")
136 cmd = [
137 ffmpeg,
138 "-y",
139 "-i",
140 str(src),
141 "-ac",
142 "1",
143 "-ar",
144 "16000",
145 "-c:a",
146 "pcm_s16le",
147 str(wav),
148 ]
149 _run(cmd, "ffmpeg could not convert the replay to wav")
150 if not chunk_seconds:
151 return [(wav, 0.0)]
152 chunk_dir = wav.parent / "chunks"
153 chunk_dir.mkdir(parents=True, exist_ok=True)
154 pattern = str(chunk_dir / "c%03d.wav")
155 split = [
156 ffmpeg,
157 "-y",
158 "-i",
159 str(wav),
160 "-f",
161 "segment",
162 "-segment_time",
163 str(chunk_seconds),
164 "-reset_timestamps",
165 "1",
166 pattern,
167 ]
168 _run(split, "ffmpeg could not chunk the wav")
169 files = sorted(chunk_dir.glob("c*.wav"))
170 return [(path, idx * float(chunk_seconds)) for idx, path in enumerate(files)]
171
172
173def write_mp3(src: str | Path, dest: str | Path) -> Path:
174 ffmpeg = _ffmpeg()
175 dest_path = Path(dest)
176 dest_path.parent.mkdir(parents=True, exist_ok=True)
177 cmd = [
178 ffmpeg,
179 "-y",
180 "-i",
181 str(src),
182 "-vn",
183 "-ac",
184 "1",
185 "-b:a",
186 "40k",
187 str(dest_path),
188 ]
189 _run(cmd, "ffmpeg could not write audio.mp3")
190 return dest_path
191
192
193def _ffmpeg() -> str:
194 path = shutil.which("ffmpeg")
195 if not path:
196 raise RenderError(
197 "ffmpeg is not on PATH. Install ffmpeg, then re-run."
198 )
199 return path
200
201
202def _run(cmd: list[str], message: str) -> None:
203 proc = subprocess.run(cmd, check=False, capture_output=True, text=True)
204 if proc.returncode != 0:
205 err = (proc.stderr or proc.stdout or "").strip().splitlines()
206 tail = err[-1] if err else "ffmpeg failed"
207 raise RenderError(f"{message}: {tail[:300]}")