space-tapePublic

Open-source X Spaces transcription.

Star

159 lines · 5.9 KB

python
1"""space-tape transcribe <url> — recorded X Spaces → cues.json."""
2
3from __future__ import annotations
4
5import argparse
6import json
7import sys
8from pathlib import Path
9
10from space_tape import __version__
11from space_tape.download import DownloadError, download, parse_url
12from space_tape.hydra import hydra_series, regions
13from space_tape.merge import attach_speakers
14from space_tape.render import RenderError, transcript_md, wav_for_asr, write_outputs
15from space_tape.transcribe import DEFAULT_MODEL, transcribe_chunks
16
17
18def main(argv: list[str] | None = None) -> int:
19 parser = argparse.ArgumentParser(
20 prog="space-tape",
21 description=(
22 "Open-source X Spaces transcription. "
23 "Whisper is the ear. Hydra is the speaker list."
24 ),
25 )
26 parser.add_argument("--version", action="version", version=f"space-tape {__version__}")
27 sub = parser.add_subparsers(dest="cmd", required=True)
28
29 t = sub.add_parser("transcribe", help="Download a recorded Space and emit cues + transcript")
30 t.add_argument("url", help="x.com/i/spaces/<id> or x.com/<user>/status/<id>")
31 t.add_argument("-o", "--out", default="./out", help="Output directory (default: ./out)")
32 t.add_argument("--model", default=DEFAULT_MODEL, help=f"Whisper model (default: {DEFAULT_MODEL})")
33 t.add_argument("--host", default=None, help="Host handle (baked over the Hydra 'host' token)")
34 t.add_argument("--title", default="Space", help="Title written into transcript.md")
35 t.add_argument("--chunk-seconds", type=int, default=300, help="ASR chunk length; 0 to disable")
36 t.add_argument("--word-timestamps", action="store_true", help="Ask Whisper for word-level timestamps")
37 t.add_argument("--device", default=None, help="transformers device (cpu, 0, …)")
38 t.add_argument("--keep-wav", action="store_true", help="Leave space.wav next to the outputs")
39
40 h = sub.add_parser("hydra", help="Parse Hydra ID3 from a raw replay file (no Whisper)")
41 h.add_argument("file", help="Raw yt-dlp replay (.m4a / .ts). Do not remux first.")
42 h.add_argument("--host", default=None, help="Host handle replacing the 'host' token")
43 h.add_argument("-o", "--out", default=None, help="Write regions JSON here instead of stdout")
44
45 args = parser.parse_args(argv)
46 try:
47 if args.cmd == "transcribe":
48 return _cmd_transcribe(args)
49 if args.cmd == "hydra":
50 return _cmd_hydra(args)
51 except (DownloadError, RenderError, RuntimeError) as exc:
52 print(f"space-tape: {exc}", file=sys.stderr)
53 return 1
54 return 2
55
56
57def _cmd_transcribe(args: argparse.Namespace) -> int:
58 info = parse_url(args.url)
59 host = (args.host or info.get("host") or "").lstrip("@") or None
60 out_dir = Path(args.out)
61 out_dir.mkdir(parents=True, exist_ok=True)
62
63 print(f"download {info['url']}", file=sys.stderr)
64 replay = download(info["url"], out_dir)
65 print(f"replay {replay.name} ({replay.stat().st_size} bytes)", file=sys.stderr)
66
67 print("hydra parse ID3 (speakers)", file=sys.stderr)
68 series = hydra_series(replay, host=host)
69 hydra_regions = regions(series)
70 if not series:
71 print(
72 "hydra no Hydra tags — speakers will be 'unknown'. "
73 "Did something remux this file?",
74 file=sys.stderr,
75 )
76
77 wav = out_dir / "space.wav"
78 chunk_seconds = args.chunk_seconds if args.chunk_seconds and args.chunk_seconds > 0 else None
79 print("ffmpeg 16 kHz mono wav", file=sys.stderr)
80 chunks = wav_for_asr(replay, wav, chunk_seconds=chunk_seconds)
81
82 print(f"whisper {args.model}", file=sys.stderr)
83 cues = transcribe_chunks(
84 chunks,
85 model=args.model,
86 word_timestamps=args.word_timestamps,
87 device=args.device,
88 )
89 merged = attach_speakers(cues, hydra_regions)
90
91 source_url = info["url"]
92 posted = source_url if info["kind"] == "status" else None
93 if info["kind"] == "status":
94 posted = source_url
95 # Keep a spaces URL in the transcript when we have one later; for a
96 # status link, yt-dlp followed it. Write the URL the user gave.
97 source_url = info["url"]
98
99 written = write_outputs(
100 out_dir,
101 cues=merged,
102 source_url=source_url,
103 posted_url=posted,
104 title=args.title,
105 audio_src=replay,
106 speakers=_speakers(merged, host),
107 )
108 if not args.keep_wav:
109 _cleanup_wav(out_dir)
110 print(f"wrote {written['cues']}", file=sys.stderr)
111 print(f"wrote {written['transcript']}", file=sys.stderr)
112 if "audio" in written:
113 print(f"wrote {written['audio']}", file=sys.stderr)
114 print(transcript_md(merged[:3], source_url=source_url, title=args.title) if merged else "(no cues)", file=sys.stderr)
115 return 0
116
117
118def _cmd_hydra(args: argparse.Namespace) -> int:
119 path = Path(args.file)
120 if not path.is_file():
121 raise RuntimeError(f"not a file: {path}")
122 series = hydra_series(path, host=args.host)
123 hydra_regions = regions(series)
124 payload = {"samples": len(series), "regions": hydra_regions}
125 text = json.dumps(payload, indent=2, ensure_ascii=False) + "\n"
126 if args.out:
127 Path(args.out).write_text(text, encoding="utf-8")
128 else:
129 sys.stdout.write(text)
130 return 0
131
132
133def _speakers(cues: list[dict], host: str | None) -> list[str]:
134 names: list[str] = []
135 if host:
136 names.append(host)
137 skip = {"unknown", "both", "overlap", "silence", "host", None, ""}
138 for cue in cues:
139 sp = cue.get("speaker")
140 if sp in skip or sp in names:
141 continue
142 names.append(sp)
143 return names
144
145
146def _cleanup_wav(out_dir: Path) -> None:
147 wav = out_dir / "space.wav"
148 if wav.exists():
149 wav.unlink()
150 chunks = out_dir / "chunks"
151 if chunks.is_dir():
152 for child in chunks.glob("*"):
153 child.unlink()
154 chunks.rmdir()
155
156
157if __name__ == "__main__":
158 raise SystemExit(main())