docs: drop stale MP3-only implementation plan

Superseded by the current architecture; described the old
three-module, MP3-only tool and carried a leftover internal
agent instruction that has no place in the public repo.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

.github/workflows/tests.yml

@@ -0,0 +1,25 @@
+name: Tests
+
+on:
+  push:
+    branches: [master]
+  pull_request:
+
+jobs:
+  test:
+    runs-on: ubuntu-latest
+    strategy:
+      matrix:
+        python-version: ["3.9", "3.10", "3.11", "3.12"]
+    steps:
+      - uses: actions/checkout@v4
+      - name: Set up Python ${{ matrix.python-version }}
+        uses: actions/setup-python@v5
+        with:
+          python-version: ${{ matrix.python-version }}
+      - name: Install dependencies
+        run: |
+          python -m pip install --upgrade pip
+          pip install -r requirements.txt -r requirements-dev.txt
+      - name: Run tests
+        run: pytest -m "not integration"

.gitignore

@@ -6,6 +6,7 @@ __pycache__/
 .Python
 env/
 venv/
+.venv/
 ENV/
 *.egg-info/
 dist/
@@ -25,14 +26,5 @@ htmlcov/
 # Downloads
 *.mp3
 *.m4a
-cookies.txt
-
-# Downloaded outputs (artifacts, not source)
 *.mp4
-*.mov
+cookies.txt
-*.wav
-*.txt
-!requirements*.txt
-space_*.json
-space_*.txt
-.DS_Store

AGENTS.md

@@ -1,27 +1,31 @@
 # X Space Downloader
 
-Three-module CLI tool that wraps yt-dlp to download X/Twitter Space recordings as MP3.
+CLI tool that downloads X/Twitter Space recordings and renders them as an MP4 replay (participant view + audio) by default, or as plain MP3 audio.
 
 ## Commands
 
 ```bash
 pip install -e .                          # Install (editable)
-x-space-dl <URL>                          # Run the CLI
+x-space-dl <URL>                          # Run the CLI (video mode by default)
+x-space-dl --mode audio <URL>             # Audio-only MP3 mode
 python -m src.main <URL>                  # Alternative invocation
-pytest                                    # Run all tests
+pytest                                     # Run all tests
-pytest -m "not integration"               # Unit tests only
+pytest -m "not integration"                # Unit tests only
-pytest -m integration                     # Integration tests only
+pytest -m integration                      # Integration tests only
 pytest tests/test_validator.py::test_valid_x_url  # Single test
 ```
 
 ## Architecture
 
-- **`src/main.py`** -- Click CLI entry point. Validates URL via validator, delegates to downloader. Registered as `x-space-dl` console script.
+- **`src/main.py`** -- Click CLI entry point. Validates URL via validator, then delegates to `download_space` (audio) or `download_space_video` (video, default). Registered as `x-space-dl` console script.
 - **`src/validator.py`** -- `validate_space_url()` returns `(is_valid, space_id, error_message)`. Accepts both `x.com` and `twitter.com` domains.
-- **`src/downloader.py`** -- `download_space()` configures yt-dlp with `bestaudio/best` format and FFmpeg MP3 postprocessor. Output: `space_{SPACE_ID}.mp3`. Raises `DownloadError` on failure.
+- **`src/downloader.py`** -- `download_space()` wraps yt-dlp for audio extraction. `download_space_video()` fetches Space metadata/timeline via `XApiClient`, downloads audio, then renders the MP4 via `video.render_space_video()`.
+- **`src/x_api.py`** -- `XApiClient` calls X's web GraphQL (`AudioSpaceById`) and Periscope replay chat endpoints to fetch title, host, participants, and a speaker timeline. Requires the `X_BEARER_TOKEN` environment variable.
+- **`src/video.py`** -- Renders the MP4 participant-view frames (Pillow) and pipes them through ffmpeg alongside the audio track.
+- **`src/models.py`** -- Shared dataclasses: `Participant`, `SpaceMetadata`, `SpeakerInterval`.
 
 ## Dependencies
 
-Runtime: `yt-dlp`, `click`. System: `ffmpeg`. Dev: `pytest`, `pytest-mock`.
+Runtime: `yt-dlp`, `click`, `requests`, `Pillow`. System: `ffmpeg`. Dev: `pytest`, `pytest-mock`.
 
-Tests mock `yt_dlp.YoutubeDL` -- no network calls in unit tests. Integration tests use `@pytest.mark.integration`.
+Tests mock `yt_dlp.YoutubeDL` and network calls -- no real network access in unit tests. Integration tests use `@pytest.mark.integration`.

CLAUDE.md

@@ -8,8 +8,9 @@ This file provides guidance to Claude Code (claude.ai/code) when working with co
 # Install (editable)
 pip install -e .
 
-# Run the CLI
+# Run the CLI (video mode by default)
 x-space-dl <URL>
+x-space-dl --mode audio <URL>
 python -m src.main <URL>
 
 # Run all tests
@@ -27,14 +28,17 @@ pytest -m integration
 
 ## Architecture
 
-Three-module CLI tool that wraps yt-dlp to download X/Twitter Space recordings as MP3:
+CLI tool that downloads X/Twitter Space recordings and renders them as an MP4 replay with a participant view, or as plain MP3 audio:
 
-- **`src/main.py`** — Click CLI entry point (`cli`). Validates URL via validator, then delegates to downloader. Registered as `x-space-dl` console script.
+- **`src/main.py`** — Click CLI entry point (`cli`). Validates the URL, then delegates to `download_space` (audio mode) or `download_space_video` (video mode, default). Registered as `x-space-dl` console script.
-- **`src/validator.py`** — `validate_space_url()` returns `(is_valid, space_id, error_message)` tuple. Accepts both `x.com` and `twitter.com` domains.
+- **`src/validator.py`** — `validate_space_url()` returns `(is_valid, space_id, error_message)`. Accepts both `x.com` and `twitter.com` domains.
-- **`src/downloader.py`** — `download_space()` configures yt-dlp with `bestaudio/best` format and FFmpeg MP3 postprocessor. Output files are named `space_{SPACE_ID}.mp3`. Raises `DownloadError` on failure.
+- **`src/downloader.py`** — `download_space()` wraps yt-dlp for audio extraction. `download_space_video()` fetches metadata/timeline through `XApiClient`, downloads the audio, and renders the MP4 via `video.render_space_video()`.
+- **`src/x_api.py`** — `XApiClient` calls X's web GraphQL (`AudioSpaceById`) and the Periscope replay chat endpoints to build participant metadata and a speaker timeline. Requires the `X_BEARER_TOKEN` environment variable (see README).
+- **`src/video.py`** — Renders MP4 frames (Pillow) of the participant view and pipes them into ffmpeg alongside the audio track.
+- **`src/models.py`** — Shared dataclasses: `Participant`, `SpaceMetadata`, `SpeakerInterval`.
 
-Tests mock `yt_dlp.YoutubeDL` — no network calls in unit tests. Integration tests use the `@pytest.mark.integration` marker.
+Tests mock `yt_dlp.YoutubeDL` and outbound HTTP calls — no real network access in unit tests. Integration tests use the `@pytest.mark.integration` marker.
 
 ## Dependencies
 
-Runtime: `yt-dlp`, `click`. System: `ffmpeg`. Dev: `pytest`, `pytest-mock`.
+Runtime: `yt-dlp`, `click`, `requests`, `Pillow`. System: `ffmpeg`. Dev: `pytest`, `pytest-mock`.

LICENSE

@@ -0,0 +1,21 @@
+MIT License
+
+Copyright (c) 2026 Dario Sevilla
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.

README.md

@@ -1,10 +1,17 @@
 # X Space Downloader
 
-A Python CLI tool that downloads X (Twitter) Space recordings and outputs them as MP3 files.
+[![Tests](https://github.com/Dasemu/x-space-downloader/actions/workflows/tests.yml/badge.svg)](https://github.com/Dasemu/x-space-downloader/actions/workflows/tests.yml)
+[![License: MIT](https://img.shields.io/badge/License-MIT-yellow.svg)](LICENSE)
+
+A Python CLI tool that downloads X (Twitter) Space recordings and renders an MP4 replay with audio and a compact participant view.
 
 ## Features
 
-- Download X Space recordings as MP3 files
+- Download X Space recordings as MP4 videos by default
+- Keep the original audio-only MP3 mode
+- Fetch participant metadata from X `AudioSpaceById` when cookies are provided
+- Download the replay chat timeline and highlight active speakers when cookies are provided
+- Accept an optional speaker timeline JSON file as a fallback/manual override
 - Support for authenticated downloads using cookies
 - Simple command-line interface
 - Clear error messages
@@ -14,7 +21,7 @@ A Python CLI tool that downloads X (Twitter) Space recordings and outputs them a
 ### Prerequisites
 
 - Python 3.8 or higher
-- ffmpeg (required by yt-dlp for audio conversion)
+- ffmpeg (required for audio conversion and MP4 rendering)
 
 Install ffmpeg:
 ```bash
@@ -31,7 +38,7 @@ sudo apt install ffmpeg
 ### Install the tool
 
 ```bash
-git clone https://github.com/synapz-org/x-space-downloader.git
+git clone https://github.com/Dasemu/x-space-downloader.git
 cd x-space-downloader
 pip install -e .
 ```
@@ -44,7 +51,15 @@ pip install -e .
 x-space-dl https://x.com/i/spaces/1ZkKzZWnVRRKv
 ```
 
-This will download the Space recording and save it as `space_1ZkKzZWnVRRKv.mp3` in the current directory.
+This will download the Space recording and save it as `space_1ZkKzZWnVRRKv.mp4` in the current directory.
+
+### Audio-only mode
+
+```bash
+x-space-dl --mode audio https://x.com/i/spaces/1ZkKzZWnVRRKv
+```
+
+This keeps the original behavior and writes `space_1ZkKzZWnVRRKv.mp3`.
 
 ### With authentication (recommended)
 
@@ -54,6 +69,39 @@ For better reliability and access to private Spaces, provide cookies from your l
 x-space-dl --cookies cookies.txt https://x.com/i/spaces/1ZkKzZWnVRRKv
 ```
 
+Cookies are also used to fetch the Space title, host, speakers, avatars, and replay chat timeline from X/Periscope metadata.
+
+### Environment variables
+
+Fetching metadata (required for the default video mode) needs a bearer token for X's web GraphQL API:
+
+```bash
+export X_BEARER_TOKEN="<token>"
+```
+
+Get this value from your own browser: open x.com while logged in, open DevTools → Network, and copy the `authorization: Bearer ...` header from any request to `x.com/i/api/graphql`. This is the public token issued to X's web client, not a personal secret, but it's not bundled with this tool — you provide your own.
+
+`X_AUDIO_SPACE_QUERY_ID` can optionally override the GraphQL query ID used for `AudioSpaceById` if X changes it.
+
+### Speaker timeline
+
+With cookies, the tool automatically calls the replay `history` endpoint and builds participant/speaker timing from Space microphone events. If you already have speaker intervals from another extractor, you can override that with JSON:
+
+```bash
+x-space-dl --cookies cookies.txt --speaker-timeline speakers.json https://x.com/i/spaces/1ZkKzZWnVRRKv
+```
+
+Supported JSON shape:
+
+```json
+[
+  {"start": 0.0, "end": 12.4, "speaker": "alice"},
+  {"start": 12.4, "end": 31.0, "speaker": "bob"}
+]
+```
+
+The `speaker` value should match the X handle without requiring `@`.
+
 ### Specify output directory
 
 ```bash
@@ -92,6 +140,8 @@ To download Spaces that require authentication, you need to export cookies from
 - Python 3.8+
 - yt-dlp
 - click
+- requests
+- Pillow
 - ffmpeg (system dependency)
 
 ## Supported URLs
@@ -127,10 +177,15 @@ x-space-downloader/
 ├── src/
 │   ├── main.py          # CLI entry point
 │   ├── downloader.py    # yt-dlp wrapper
+│   ├── x_api.py         # X GraphQL metadata client
+│   ├── video.py         # MP4 participant-view renderer
+│   ├── models.py        # Shared data models
 │   └── validator.py     # URL validation
 ├── tests/
 │   ├── test_main.py
 │   ├── test_downloader.py
+│   ├── test_video.py
+│   ├── test_x_api.py
 │   └── test_validator.py
 └── docs/
     └── plans/           # Design documents

requirements.txt

@@ -1,2 +1,4 @@
 yt-dlp>=2024.0.0
 click>=8.0.0
+requests>=2.31.0
+Pillow>=10.0.0

setup.py

@@ -3,10 +3,18 @@ from setuptools import setup, find_packages
 setup(
     name="x-space-downloader",
     version="0.1.0",
+    description="Download X (Twitter) Space recordings as MP4 replays or MP3 audio",
+    long_description=open("README.md", encoding="utf-8").read(),
+    long_description_content_type="text/markdown",
+    url="https://github.com/Dasemu/x-space-downloader",
+    author="Dario Sevilla",
+    license="MIT",
     packages=find_packages(),
     install_requires=[
         "yt-dlp>=2024.0.0",
         "click>=8.0.0",
+        "requests>=2.31.0",
+        "Pillow>=10.0.0",
     ],
     entry_points={
         "console_scripts": [

src/downloader.py

@@ -1,7 +1,11 @@
 import os
+from glob import glob
 import yt_dlp
 from typing import Optional
 from src.validator import validate_space_url
+from src.models import SpaceMetadata
+from src.video import RenderError, SpeakerTimeline, render_space_video
+from src.x_api import XApiClient, XApiError
 
 
 class DownloadError(Exception):
@@ -12,7 +16,8 @@ class DownloadError(Exception):
 def download_space(
     space_url: str,
     output_dir: str,
-    cookies_file: Optional[str] = None
+    cookies_file: Optional[str] = None,
+    audio_format: str = "mp3",
 ) -> str:
     """
     Downloads X Space recording using yt-dlp.
@@ -21,6 +26,7 @@ def download_space(
         space_url: URL of the X Space recording
         output_dir: Directory to save the MP3 file
         cookies_file: Optional path to cookies file for authentication
+        audio_format: Audio container/codec to write
 
     Returns:
         Path to the downloaded MP3 file
@@ -36,17 +42,20 @@ def download_space(
     if space_id is None:
         raise ValueError("Could not extract space ID from URL")
 
+    os.makedirs(output_dir, exist_ok=True)
+
     # Configure yt-dlp options
     ydl_opts = {
         'format': 'bestaudio/best',
         'outtmpl': os.path.join(output_dir, f'space_{space_id}.%(ext)s'),
-        'postprocessors': [{
-            'key': 'FFmpegExtractAudio',
-            'preferredcodec': 'mp3',
-        }],
         'quiet': False,
         'no_warnings': False,
     }
+    if audio_format != "native":
+        ydl_opts['postprocessors'] = [{
+            'key': 'FFmpegExtractAudio',
+            'preferredcodec': audio_format,
+        }]
 
     # Add cookies if provided
     if cookies_file:
@@ -60,4 +69,77 @@ def download_space(
         raise DownloadError(f"Failed to download Space: {str(e)}")
 
     # Return the expected output path
-    return os.path.join(output_dir, f'space_{space_id}.mp3')
+    if audio_format == "native":
+        return os.path.join(output_dir, f'space_{space_id}')
+    return os.path.join(output_dir, f'space_{space_id}.{audio_format}')
+
+
+def download_space_video(
+    space_url: str,
+    output_dir: str,
+    cookies_file: Optional[str] = None,
+    speaker_timeline_file: Optional[str] = None,
+    max_duration: Optional[float] = None,
+) -> str:
+    """
+    Download an X Space replay and render an MP4 participant view with audio.
+    """
+    is_valid, space_id, error_message = validate_space_url(space_url)
+    if not is_valid:
+        raise ValueError(f"Invalid space URL: {error_message}")
+    if space_id is None:
+        raise ValueError("Could not extract space ID from URL")
+
+    metadata = fetch_required_metadata(space_id, cookies_file)
+    timeline = SpeakerTimeline.from_file(speaker_timeline_file) if speaker_timeline_file else None
+    if timeline is None:
+        timeline = fetch_required_timeline(metadata, cookies_file)
+
+    audio_path = download_space(space_url, output_dir, cookies_file, audio_format="native")
+    if not os.path.exists(audio_path):
+        candidates = sorted(glob(os.path.join(output_dir, f"space_{space_id}.*")))
+        candidates = [path for path in candidates if not path.endswith(".mp4")]
+        if not candidates:
+            raise DownloadError("Downloaded audio file was not found")
+        audio_path = candidates[0]
+    output_path = os.path.join(output_dir, f"space_{space_id}.mp4")
+
+    try:
+        return render_space_video(
+            audio_path=audio_path,
+            output_path=output_path,
+            metadata=metadata,
+            timeline=timeline,
+            max_duration=max_duration,
+        )
+    except RenderError as exc:
+        raise DownloadError(f"Failed to render Space video: {exc}") from exc
+
+
+def fetch_required_metadata(space_id: str, cookies_file: Optional[str]) -> SpaceMetadata:
+    try:
+        return XApiClient(cookies_file=cookies_file).fetch_space_metadata(space_id)
+    except XApiError as exc:
+        hint = " Export logged-in X cookies and pass them with --cookies." if not cookies_file else ""
+        raise DownloadError(f"Failed to fetch Space metadata.{hint} Details: {exc}") from exc
+
+
+def fetch_required_timeline(metadata: SpaceMetadata, cookies_file: Optional[str]) -> SpeakerTimeline:
+    if not cookies_file:
+        raise DownloadError("Video mode requires --cookies to fetch the replay participant timeline.")
+    try:
+        intervals, first_seen, history_participants = XApiClient(cookies_file=cookies_file).fetch_space_timeline(metadata)
+    except XApiError as exc:
+        raise DownloadError(f"Failed to fetch Space replay timeline: {exc}") from exc
+    merge_history_participants(metadata, history_participants)
+    if not intervals and not first_seen:
+        raise DownloadError("Space replay timeline did not include participant or microphone events.")
+    return SpeakerTimeline(intervals, first_seen)
+
+
+def merge_history_participants(metadata: SpaceMetadata, history_participants) -> None:
+    existing = {participant.username.lower() for participant in metadata.participants}
+    for participant in history_participants:
+        if participant.username.lower() not in existing:
+            metadata.participants.append(participant)
+            existing.add(participant.username.lower())

src/main.py

@@ -1,17 +1,34 @@
 import click
-import os
 from click.exceptions import Exit
 from src.validator import validate_space_url
-from src.downloader import download_space, DownloadError
+from src.downloader import download_space, download_space_video, DownloadError
 
 
 @click.command()
 @click.argument('url')
 @click.option('--cookies', '-c', type=click.Path(exists=True), help='Path to cookies file')
 @click.option('--output', '-o', type=click.Path(), default='.', help='Output directory')
-def cli(url, cookies, output):
+@click.option(
+    '--mode',
+    type=click.Choice(['audio', 'video']),
+    default='video',
+    show_default=True,
+    help='Output type to generate',
+)
+@click.option(
+    '--speaker-timeline',
+    type=click.Path(exists=True),
+    help='JSON file with speaker intervals: [{"start": 0, "end": 5, "speaker": "handle"}]',
+)
+@click.option(
+    '--max-duration',
+    type=float,
+    hidden=True,
+    help='Render only the first N seconds; intended for tests and previews',
+)
+def cli(url, cookies, output, mode, speaker_timeline, max_duration):
     """
-    Download X (Twitter) Space recordings as MP3 files.
+    Download X (Twitter) Space recordings as MP3 or MP4 files.
 
     Example:
         x-space-dl https://x.com/i/spaces/1ZkKzZWnVRRKv
@@ -25,7 +42,10 @@ def cli(url, cookies, output):
     # Download the Space
     try:
         click.echo(f"Downloading Space: {space_id}")
-        output_path = download_space(url, output, cookies)
+        if mode == 'audio':
+            output_path = download_space(url, output, cookies)
+        else:
+            output_path = download_space_video(url, output, cookies, speaker_timeline, max_duration)
         click.echo(f"Successfully downloaded to: {output_path}")
     except DownloadError as e:
         click.echo(f"Error: {str(e)}", err=True)

src/models.py

@@ -0,0 +1,34 @@
+from dataclasses import dataclass, field
+from typing import List, Optional
+
+
+@dataclass(frozen=True)
+class Participant:
+    """A user visible in the Space participant tray."""
+
+    user_id: str
+    display_name: str
+    username: str
+    avatar_url: Optional[str] = None
+    role: str = "speaker"
+
+
+@dataclass(frozen=True)
+class SpaceMetadata:
+    """Metadata needed to render a Space replay video."""
+
+    space_id: str
+    title: str
+    participants: List[Participant] = field(default_factory=list)
+    media_key: Optional[str] = None
+    started_at_ms: Optional[int] = None
+    ended_at_ms: Optional[int] = None
+
+
+@dataclass(frozen=True)
+class SpeakerInterval:
+    """An interval where a known participant is speaking."""
+
+    start: float
+    end: float
+    speaker: str

src/validator.py

@@ -19,5 +19,5 @@ def validate_space_url(url: str) -> Tuple[bool, Optional[str], Optional[str]]:
     if not match:
         return False, None, "Invalid URL format. Expected: https://x.com/i/spaces/{SPACE_ID}"
 
-    space_id = match.group(2)  # Changed from group(1) to group(2)
+    space_id = match.group(2)
     return True, space_id, None

src/video.py

@@ -0,0 +1,456 @@
+import json
+import math
+import os
+import shutil
+import subprocess
+import tempfile
+from dataclasses import replace
+from io import BytesIO
+from pathlib import Path
+from typing import Dict, Iterable, List, Optional, Set, Tuple
+
+import requests
+from PIL import Image, ImageDraw, ImageFont
+
+from src.models import Participant, SpaceMetadata, SpeakerInterval
+
+
+class RenderError(Exception):
+    """Raised when the MP4 render fails."""
+
+
+class SpeakerTimeline:
+    """Lookup table for active speakers during a replay."""
+
+    def __init__(self, intervals: Iterable[SpeakerInterval], participant_first_seen: Optional[Dict[str, float]] = None):
+        self.intervals = sorted(intervals, key=lambda item: item.start)
+        self.participant_first_seen = {
+            username.lstrip("@").lower(): float(seconds)
+            for username, seconds in (participant_first_seen or {}).items()
+        }
+
+    @classmethod
+    def from_file(cls, path: str) -> "SpeakerTimeline":
+        with open(path, "r", encoding="utf-8") as handle:
+            payload = json.load(handle)
+
+        return cls.from_file_data(payload)
+
+    @classmethod
+    def from_file_data(cls, payload) -> "SpeakerTimeline":
+        rows = payload.get("speakers", payload) if isinstance(payload, dict) else payload
+        intervals = []
+        for row in rows:
+            intervals.append(
+                SpeakerInterval(
+                    start=float(row["start"]),
+                    end=float(row["end"]),
+                    speaker=str(row.get("speaker") or row.get("username") or row.get("handle")),
+                )
+            )
+        return cls(intervals)
+
+    def active_at(self, seconds: float) -> Optional[str]:
+        active = self.active_speakers_at(seconds)
+        return active[0] if active else None
+
+    def active_speakers_at(self, seconds: float) -> List[str]:
+        return [
+            interval.speaker.lstrip("@").lower()
+            for interval in self.intervals
+            if interval.start <= seconds < interval.end
+        ]
+
+    def first_seen_at(self, username: str) -> Optional[float]:
+        return self.participant_first_seen.get(username.lstrip("@").lower())
+
+
+def render_space_video(
+    audio_path: str,
+    output_path: str,
+    metadata: SpaceMetadata,
+    timeline: Optional[SpeakerTimeline] = None,
+    width: int = 720,
+    height: int = 720,
+    fps: int = 1,
+    max_duration: Optional[float] = None,
+) -> str:
+    """Render an MP4 that mimics the compact X Spaces participant view."""
+    if not os.path.exists(audio_path):
+        raise RenderError(f"Audio file not found: {audio_path}")
+    if shutil.which("ffmpeg") is None:
+        raise RenderError("ffmpeg is required to render the video")
+
+    duration = max_duration or probe_duration(audio_path)
+    if duration <= 0:
+        raise RenderError("Could not determine audio duration")
+
+    output = Path(output_path)
+    output.parent.mkdir(parents=True, exist_ok=True)
+
+    avatars = load_avatars(metadata.participants, avatar_size=168)
+
+    if timeline is None:
+        frame = render_frame(metadata, avatars, None, 0, width, height)
+        return render_looped_image_video(frame, audio_path, str(output), duration)
+
+    command = build_ffmpeg_command(audio_path, str(output), width, height, fps)
+
+    process = subprocess.Popen(command, stdin=subprocess.PIPE, stderr=subprocess.PIPE)
+    try:
+        total_frames = max(1, int(math.ceil(duration * fps)))
+        frame_cache: Dict[Tuple[Tuple[str, ...], Optional[str], int], bytes] = {}
+        for frame_index in range(total_frames):
+            seconds = frame_index / fps
+            participants = visible_participants(metadata.participants, timeline, seconds)
+            active = tuple(timeline.active_speakers_at(seconds)) if timeline else ()
+            cache_key = (tuple(participant.username.lower() for participant in participants), active)
+            frame_bytes = frame_cache.get(cache_key)
+            if frame_bytes is None:
+                frame = render_frame(
+                    replace(metadata, participants=participants),
+                    avatars,
+                    timeline,
+                    seconds,
+                    width,
+                    height,
+                    participants_are_filtered=True,
+                )
+                frame_bytes = frame.tobytes()
+                frame_cache[cache_key] = frame_bytes
+            process.stdin.write(frame_bytes)
+    except BrokenPipeError as exc:
+        raise RenderError("ffmpeg closed before all frames were written") from exc
+    finally:
+        if process.stdin:
+            process.stdin.close()
+
+    stderr = process.stderr.read().decode("utf-8", errors="replace") if process.stderr else ""
+    return_code = process.wait()
+    if return_code != 0:
+        raise RenderError(stderr.strip() or "ffmpeg failed")
+
+    return str(output)
+
+
+def render_looped_image_video(frame: Image.Image, audio_path: str, output_path: str, duration: float) -> str:
+    with tempfile.NamedTemporaryFile(suffix=".png", delete=False) as handle:
+        frame_path = handle.name
+    try:
+        frame.save(frame_path)
+        command = [
+            "ffmpeg",
+            "-y",
+            "-hide_banner",
+            "-loglevel",
+            "error",
+            "-loop",
+            "1",
+            "-framerate",
+            "1",
+            "-i",
+            frame_path,
+            "-i",
+            audio_path,
+            "-t",
+            str(duration),
+            "-c:v",
+            "libx264",
+            "-preset",
+            "veryfast",
+            "-pix_fmt",
+            "yuv420p",
+            "-c:a",
+            "copy",
+            "-shortest",
+            output_path,
+        ]
+        result = subprocess.run(command, capture_output=True, text=True, check=False)
+        if result.returncode != 0:
+            raise RenderError(result.stderr.strip() or "ffmpeg failed")
+        return output_path
+    finally:
+        try:
+            os.unlink(frame_path)
+        except FileNotFoundError:
+            pass
+
+
+def build_ffmpeg_command(audio_path: str, output_path: str, width: int, height: int, fps: int) -> List[str]:
+    return [
+        "ffmpeg",
+        "-y",
+        "-hide_banner",
+        "-loglevel",
+        "error",
+        "-f",
+        "rawvideo",
+        "-vcodec",
+        "rawvideo",
+        "-s",
+        f"{width}x{height}",
+        "-pix_fmt",
+        "rgb24",
+        "-r",
+        str(fps),
+        "-i",
+        "-",
+        "-i",
+        audio_path,
+        "-c:v",
+        "libx264",
+        "-preset",
+        "veryfast",
+        "-pix_fmt",
+        "yuv420p",
+        "-c:a",
+        "copy",
+        "-shortest",
+        output_path,
+    ]
+
+
+def probe_duration(path: str) -> float:
+    if shutil.which("ffprobe") is None:
+        raise RenderError("ffprobe is required to inspect audio duration")
+
+    command = [
+        "ffprobe",
+        "-v",
+        "error",
+        "-show_entries",
+        "format=duration",
+        "-of",
+        "default=noprint_wrappers=1:nokey=1",
+        path,
+    ]
+    result = subprocess.run(command, capture_output=True, text=True, check=False)
+    if result.returncode != 0:
+        raise RenderError(result.stderr.strip() or "ffprobe failed")
+    return float(result.stdout.strip())
+
+
+def load_avatars(participants: List[Participant], avatar_size: int) -> Dict[str, Image.Image]:
+    avatars: Dict[str, Image.Image] = {}
+    for participant in participants:
+        image = None
+        if participant.avatar_url:
+            try:
+                response = requests.get(participant.avatar_url, timeout=8)
+                response.raise_for_status()
+                image = Image.open(BytesIO(response.content)).convert("RGB")
+            except Exception:
+                image = None
+
+        if image is None:
+            image = placeholder_avatar(participant, avatar_size)
+        avatars[participant.username.lower()] = crop_circle(image, avatar_size)
+    return avatars
+
+
+def render_frame(
+    metadata: SpaceMetadata,
+    avatars: Dict[str, Image.Image],
+    timeline: Optional[SpeakerTimeline],
+    seconds: float,
+    width: int,
+    height: int,
+    participants_are_filtered: bool = False,
+) -> Image.Image:
+    participants = metadata.participants if participants_are_filtered else visible_participants(metadata.participants, timeline, seconds)
+    active = set(timeline.active_speakers_at(seconds)) if timeline else set()
+
+    frame = Image.new("RGB", (width, height), (18, 21, 24))
+    draw = ImageDraw.Draw(frame)
+    fonts = load_fonts()
+
+    draw_rounded_rectangle(draw, (24, 24, width - 24, height - 24), 28, fill=(9, 13, 17), outline=(39, 46, 54))
+    draw.text((54, 54), truncate(metadata.title, 42), fill=(238, 242, 246), font=fonts["title"])
+    subtitle = "X Space replay"
+    if timeline and timeline.intervals:
+        subtitle += " · speaker timeline"
+    draw.text((54, 92), subtitle, fill=(139, 152, 165), font=fonts["small"])
+
+    if not participants:
+        participants = [Participant("space", "Space audio", "space", role="speaker")]
+
+    layout = participant_layout(width, height, len(participants))
+    for participant, box in zip(participants, layout):
+        draw_participant(frame, draw, participant, avatars.get(participant.username.lower()), box, active, seconds, fonts)
+
+    return frame
+
+
+def visible_participants(
+    participants: List[Participant],
+    timeline: Optional[SpeakerTimeline] = None,
+    seconds: float = 0,
+) -> List[Participant]:
+    primary = [item for item in participants if item.role in {"host", "speaker"}]
+    if not primary:
+        primary = participants
+
+    if not timeline or not timeline.participant_first_seen:
+        return primary[:10]
+
+    active = set(timeline.active_speakers_at(seconds))
+    visible = []
+    for participant in primary:
+        first_seen = timeline.first_seen_at(participant.username)
+        if participant.role == "host" or first_seen is None and participant.username.lower() in active:
+            first_seen = 0
+        if first_seen is not None and first_seen <= seconds:
+            visible.append((first_seen, participant))
+
+    visible.sort(key=lambda item: (item[0], item[1].username.lower()))
+    selected = [item[1] for item in visible[-10:]]
+    for active_username in active:
+        if all(item.username.lower() != active_username for item in selected):
+            active_participant = next((item for item in primary if item.username.lower() == active_username), None)
+            if active_participant:
+                selected = (selected + [active_participant])[-10:]
+    return selected
+
+
+def participant_layout(width: int, height: int, count: int) -> List[Tuple[int, int, int, int]]:
+    top = 142
+    bottom = height - 28
+    area_width = width - 88
+    area_height = bottom - top
+
+    columns = 1 if count == 1 else 2
+    rows = int(math.ceil(count / columns))
+    cell_width = area_width // columns
+    cell_height = area_height // rows
+
+    boxes = []
+    for index in range(count):
+        col = index % columns
+        row = index // columns
+        x1 = 44 + col * cell_width
+        y1 = top + row * cell_height
+        x2 = x1 + cell_width - 18
+        y2 = y1 + cell_height - 10
+        boxes.append((x1, y1, x2, y2))
+    return boxes
+
+
+def draw_participant(
+    frame: Image.Image,
+    draw: ImageDraw.ImageDraw,
+    participant: Participant,
+    avatar: Optional[Image.Image],
+    box: Tuple[int, int, int, int],
+    active: Set[str],
+    seconds: float,
+    fonts: Dict[str, ImageFont.ImageFont],
+) -> None:
+    x1, y1, x2, y2 = box
+    is_active = participant.username.lower() in active
+    fill = (25, 30, 35) if not is_active else (20, 38, 34)
+    outline = (52, 63, 73) if not is_active else (29, 155, 120)
+    draw_rounded_rectangle(draw, box, 20, fill=fill, outline=outline, width=3 if is_active else 1)
+
+    cell_height = y2 - y1
+    avatar_size = min(118, max(44, int(cell_height * 0.42)))
+    avatar_x = x1 + ((x2 - x1) - avatar_size) // 2
+    avatar_y = y1 + max(8, int(cell_height * 0.08))
+    if avatar is None:
+        avatar = placeholder_avatar(participant, avatar_size)
+    resized_avatar = avatar.resize((avatar_size, avatar_size))
+    frame.paste(resized_avatar, (avatar_x, avatar_y), resized_avatar.getchannel("A"))
+
+    if is_active:
+        pulse = 7
+        draw.ellipse(
+            (
+                avatar_x - pulse,
+                avatar_y - pulse,
+                avatar_x + avatar_size + pulse,
+                avatar_y + avatar_size + pulse,
+            ),
+            outline=(29, 155, 120),
+            width=4,
+        )
+
+    text_y = avatar_y + avatar_size + 6
+    text_width = max(24, (x2 - x1) - 22)
+    name = ellipsize(draw, participant.display_name, fonts["name"], text_width)
+    username = ellipsize(draw, f"@{participant.username}", fonts["small"], text_width)
+    draw_centered_text(draw, name, (x1 + 10, text_y, x2 - 10, text_y + 22), fonts["name"], (238, 242, 246))
+    draw_centered_text(draw, username, (x1 + 10, text_y + 22, x2 - 10, text_y + 40), fonts["small"], (139, 152, 165))
+
+    if is_active:
+        badge = "Speaking"
+        text_width = draw.textlength(badge, font=fonts["small"])
+        badge_box = (x1 + 12, y1 + 10, x1 + int(text_width) + 34, y1 + 36)
+        draw_rounded_rectangle(draw, badge_box, 14, fill=(29, 155, 120))
+        draw.text((badge_box[0] + 13, badge_box[1] + 6), badge, fill=(3, 18, 14), font=fonts["small"])
+
+
+def placeholder_avatar(participant: Participant, size: int) -> Image.Image:
+    colors = [(29, 155, 240), (249, 24, 128), (255, 212, 0), (120, 86, 255), (0, 186, 124)]
+    index = sum(ord(char) for char in participant.username) % len(colors)
+    image = Image.new("RGB", (size, size), colors[index])
+    draw = ImageDraw.Draw(image)
+    font = load_fonts()["initial"]
+    initial = (participant.display_name or participant.username or "?")[0].upper()
+    bbox = draw.textbbox((0, 0), initial, font=font)
+    draw.text(((size - (bbox[2] - bbox[0])) / 2, (size - (bbox[3] - bbox[1])) / 2 - 6), initial, fill=(255, 255, 255), font=font)
+    return crop_circle(image, size)
+
+
+def crop_circle(image: Image.Image, size: int) -> Image.Image:
+    source = image.convert("RGB").resize((size, size))
+    mask = Image.new("L", (size, size), 0)
+    draw = ImageDraw.Draw(mask)
+    draw.ellipse((0, 0, size - 1, size - 1), fill=255)
+    output = Image.new("RGBA", (size, size), (0, 0, 0, 0))
+    output.paste(source, (0, 0), mask)
+    return output
+
+
+def load_fonts() -> Dict[str, ImageFont.ImageFont]:
+    candidates = [
+        "/usr/share/fonts/truetype/dejavu/DejaVuSans.ttf",
+        "/usr/share/fonts/truetype/liberation2/LiberationSans-Regular.ttf",
+    ]
+    bold_candidates = [
+        "/usr/share/fonts/truetype/dejavu/DejaVuSans-Bold.ttf",
+        "/usr/share/fonts/truetype/liberation2/LiberationSans-Bold.ttf",
+    ]
+    regular = next((path for path in candidates if os.path.exists(path)), None)
+    bold = next((path for path in bold_candidates if os.path.exists(path)), regular)
+    return {
+        "title": ImageFont.truetype(bold, 28) if bold else ImageFont.load_default(),
+        "name": ImageFont.truetype(bold, 16) if bold else ImageFont.load_default(),
+        "small": ImageFont.truetype(regular, 13) if regular else ImageFont.load_default(),
+        "initial": ImageFont.truetype(bold, 64) if bold else ImageFont.load_default(),
+    }
+
+
+def draw_centered_text(draw: ImageDraw.ImageDraw, text: str, box: Tuple[int, int, int, int], font, fill) -> None:
+    bbox = draw.textbbox((0, 0), text, font=font)
+    x = box[0] + ((box[2] - box[0]) - (bbox[2] - bbox[0])) / 2
+    y = box[1] + ((box[3] - box[1]) - (bbox[3] - bbox[1])) / 2
+    draw.text((x, y), text, fill=fill, font=font)
+
+
+def draw_rounded_rectangle(draw: ImageDraw.ImageDraw, box, radius: int, fill, outline=None, width: int = 1) -> None:
+    draw.rounded_rectangle(box, radius=radius, fill=fill, outline=outline, width=width)
+
+
+def truncate(value: str, max_length: int) -> str:
+    if len(value) <= max_length:
+        return value
+    return value[: max_length - 1] + "..."
+
+
+def ellipsize(draw: ImageDraw.ImageDraw, value: str, font, max_width: int) -> str:
+    if draw.textlength(value, font=font) <= max_width:
+        return value
+    suffix = "..."
+    trimmed = value
+    while trimmed and draw.textlength(trimmed + suffix, font=font) > max_width:
+        trimmed = trimmed[:-1]
+    return (trimmed or value[:1]) + suffix

src/x_api.py

@@ -0,0 +1,478 @@
+import json
+import os
+from dataclasses import replace
+from datetime import datetime
+from http.cookiejar import MozillaCookieJar
+from typing import Dict, Iterable, List, Optional, Tuple
+from urllib.parse import urlencode
+
+import requests
+
+from src.models import Participant, SpaceMetadata, SpeakerInterval
+
+
+DEFAULT_GRAPHQL_QUERY_IDS = (
+    "DovcWBliWe-v2TCRGFtibA",
+    "Uv5R_-Chxbn1FEkyUkSW2w",
+    "FJoTSHMVF7fMhGLc2t9cog",
+)
+
+GRAPHQL_FEATURES = {
+    "spaces_2022_h2_spaces_communities": True,
+    "spaces_2022_h2_clipping": True,
+    "creator_subscriptions_tweet_preview_api_enabled": True,
+    "responsive_web_graphql_skip_user_profile_image_extensions_enabled": False,
+}
+
+
+class XApiError(Exception):
+    """Raised when X metadata cannot be fetched or parsed."""
+
+
+def load_netscape_cookies(cookies_file: str) -> Dict[str, str]:
+    """Load a Netscape cookies.txt file into a simple name/value mapping."""
+    jar = MozillaCookieJar()
+    jar.load(cookies_file, ignore_discard=True, ignore_expires=True)
+    return {cookie.name: cookie.value for cookie in jar}
+
+
+def cookie_header(cookies: Dict[str, str]) -> str:
+    return "; ".join(f"{name}={value}" for name, value in cookies.items())
+
+
+class XApiClient:
+    """Minimal client for X web GraphQL metadata used by Spaces replays."""
+
+    def __init__(
+        self,
+        cookies_file: Optional[str] = None,
+        bearer_token: Optional[str] = None,
+        query_ids: Iterable[str] = DEFAULT_GRAPHQL_QUERY_IDS,
+        timeout: int = 20,
+    ):
+        self.cookies = load_netscape_cookies(cookies_file) if cookies_file else {}
+        self.bearer_token = bearer_token or os.getenv("X_BEARER_TOKEN")
+        if not self.bearer_token:
+            raise XApiError(
+                "No X API bearer token available. Set the X_BEARER_TOKEN environment "
+                "variable to a bearer token from x.com's web client (visible in browser "
+                "devtools as the 'authorization: Bearer ...' header on requests to "
+                "x.com/i/api)."
+            )
+        env_query_id = os.getenv("X_AUDIO_SPACE_QUERY_ID")
+        self.query_ids = (env_query_id, *tuple(query_ids)) if env_query_id else tuple(query_ids)
+        self.timeout = timeout
+        self.session = requests.Session()
+
+    def fetch_space_metadata(self, space_id: str) -> SpaceMetadata:
+        last_error: Optional[Exception] = None
+        for query_id in self.query_ids:
+            try:
+                payload = self._get_audio_space_by_id(query_id, space_id)
+                return parse_audio_space_metadata(payload, space_id)
+            except Exception as exc:
+                last_error = exc
+
+        raise XApiError(f"Could not fetch Space metadata: {last_error}")
+
+    def fetch_space_timeline(self, metadata: SpaceMetadata) -> Tuple[List[SpeakerInterval], Dict[str, float], List[Participant]]:
+        if not metadata.media_key:
+            raise XApiError("Space metadata does not include media_key")
+
+        status = self.fetch_live_video_status(metadata.media_key)
+        playlist_url = (status.get("source") or {}).get("location")
+        start_ms = fetch_playlist_start_ms(playlist_url, self.session) if playlist_url else None
+        if start_ms is None:
+            start_ms = metadata.started_at_ms
+        if start_ms is None:
+            raise XApiError("Could not determine replay start time")
+
+        end_ms = metadata.ended_at_ms
+        if end_ms is None:
+            end_ms = start_ms + int((status.get("duration") or 4 * 60 * 60) * 1000)
+
+        access = self.access_chat(status.get("chatToken"))
+        endpoint = access.get("replay_endpoint") or access.get("endpoint")
+        access_token = access.get("replay_access_token") or access.get("access_token")
+        if not endpoint or not access_token:
+            raise XApiError("Chat access response did not include replay endpoint/token")
+
+        messages = self.fetch_history_messages(endpoint, access_token, start_ms, end_ms)
+        return build_timeline_from_history(messages, start_ms)
+
+    def fetch_live_video_status(self, media_key: str) -> dict:
+        url = f"https://x.com/i/api/1.1/live_video_stream/status/{media_key}"
+        headers = self._x_headers()
+        response = self.session.get(
+            url,
+            params={
+                "client": "web",
+                "use_syndication_guest_id": "false",
+                "cookie_set_host": "x.com",
+            },
+            headers=headers,
+            timeout=self.timeout,
+        )
+        if response.status_code >= 400:
+            raise XApiError(f"live_video_stream/status returned HTTP {response.status_code}")
+        return response.json()
+
+    def access_chat(self, chat_token: Optional[str]) -> dict:
+        if not chat_token:
+            raise XApiError("live_video_stream/status did not include chatToken")
+
+        payload = {"chat_token": chat_token}
+        urls = (
+            "https://proxsee-cf.pscp.tv/api/v2/accessChat",
+            "https://proxsee.pscp.tv/api/v2/accessChat",
+            "https://proxsee.pscp.tv/api/v2/accessChatPublic",
+        )
+        last_error = None
+        for url in urls:
+            try:
+                response = self.session.post(url, json=payload, headers=chat_headers(), timeout=self.timeout)
+                if response.status_code < 400 and response.text:
+                    return response.json()
+                last_error = XApiError(f"{url} returned HTTP {response.status_code}")
+            except Exception as exc:
+                last_error = exc
+        raise XApiError(f"Could not access chat replay: {last_error}")
+
+    def fetch_history_messages(self, endpoint: str, access_token: str, start_ms: int, end_ms: int) -> List[dict]:
+        url = endpoint.rstrip("/") + "/chatapi/v1/history"
+        step_ns = 60 * 1_000_000_000
+        cursor_ns = start_ms * 1_000_000
+        end_ns = end_ms * 1_000_000
+        seen_cursors = set()
+        seen_messages = set()
+        messages: List[dict] = []
+
+        first_payload = self._history_page(url, access_token, "", None)
+        messages.extend(unique_messages(first_payload.get("messages") or [], seen_messages))
+        response_cursor = _to_int(first_payload.get("cursor"))
+        if response_cursor and response_cursor > cursor_ns:
+            cursor_ns = response_cursor
+
+        while cursor_ns <= end_ns:
+            if cursor_ns in seen_cursors:
+                cursor_ns += step_ns
+                continue
+            seen_cursors.add(cursor_ns)
+            payload = self._history_page(url, access_token, str(cursor_ns), None)
+            page_messages = unique_messages(payload.get("messages") or [], seen_messages)
+            messages.extend(page_messages)
+            response_cursor = _to_int(payload.get("cursor"))
+            if response_cursor and response_cursor > cursor_ns:
+                cursor_ns = response_cursor
+            else:
+                cursor_ns += step_ns
+
+        return messages
+
+    def _history_page(self, url: str, access_token: str, cursor: str, since: Optional[int]) -> dict:
+        response = self.session.post(
+            url,
+            json={
+                "access_token": access_token,
+                "cursor": cursor,
+                "limit": 1000,
+                "since": since,
+                "quick_get": True,
+            },
+            headers=chat_headers(),
+            timeout=self.timeout,
+        )
+        if response.status_code >= 400:
+            raise XApiError(f"history returned HTTP {response.status_code}")
+        return response.json()
+
+    def _x_headers(self) -> Dict[str, str]:
+        headers = {
+            "accept": "*/*",
+            "authorization": f"Bearer {self.bearer_token}",
+            "content-type": "application/json",
+            "referer": "https://x.com/",
+            "user-agent": (
+                "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 "
+                "(KHTML, like Gecko) Chrome/148.0.0.0 Safari/537.36"
+            ),
+            "x-twitter-active-user": "yes",
+            "x-twitter-auth-type": "OAuth2Session",
+            "x-twitter-client-language": "en",
+        }
+        if self.cookies:
+            headers["cookie"] = cookie_header(self.cookies)
+        if self.cookies.get("ct0"):
+            headers["x-csrf-token"] = self.cookies["ct0"]
+        return headers
+
+    def _get_audio_space_by_id(self, query_id: str, space_id: str) -> dict:
+        variables = {
+            "id": space_id,
+            "isMetatagsQuery": False,
+            "withReplays": True,
+            "withListeners": True,
+        }
+        params = urlencode(
+            {
+                "variables": json.dumps(variables, separators=(",", ":")),
+                "features": json.dumps(GRAPHQL_FEATURES, separators=(",", ":")),
+            }
+        )
+        url = f"https://x.com/i/api/graphql/{query_id}/AudioSpaceById?{params}"
+        headers = self._x_headers()
+
+        response = self.session.get(url, headers=headers, timeout=self.timeout)
+        if response.status_code >= 400:
+            raise XApiError(f"X returned HTTP {response.status_code}")
+
+        payload = response.json()
+        if payload.get("errors"):
+            raise XApiError(payload["errors"][0].get("message", "GraphQL error"))
+        return payload
+
+
+def parse_audio_space_metadata(payload: dict, space_id: str) -> SpaceMetadata:
+    audio_space = payload.get("data", {}).get("audioSpace") or {}
+    metadata = audio_space.get("metadata") or {}
+    if not metadata:
+        raise XApiError("AudioSpaceById response did not include metadata")
+
+    title = metadata.get("title") or f"X Space {space_id}"
+    participants = _extract_participants(audio_space)
+    creator = _participant_from_user_result(metadata.get("creator_results"), "host")
+    if creator:
+        participants = _dedupe_participants([creator, *participants])
+
+    return SpaceMetadata(
+        space_id=metadata.get("rest_id") or space_id,
+        title=title,
+        participants=participants,
+        media_key=metadata.get("media_key"),
+        started_at_ms=_timestamp_to_ms(metadata.get("started_at") or metadata.get("scheduled_start")),
+        ended_at_ms=_timestamp_to_ms(metadata.get("ended_at")),
+    )
+
+
+def chat_headers() -> Dict[str, str]:
+    return {
+        "accept": "*/*",
+        "content-type": "application/json",
+        "origin": "https://x.com",
+        "referer": "https://x.com/",
+        "user-agent": (
+            "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 "
+            "(KHTML, like Gecko) Chrome/148.0.0.0 Safari/537.36"
+        ),
+        "x-periscope-user-agent": "Twitter/m5",
+    }
+
+
+def fetch_playlist_start_ms(playlist_url: str, session: requests.Session) -> Optional[int]:
+    response = session.get(playlist_url, headers={"origin": "https://x.com", "referer": "https://x.com/"}, timeout=20)
+    if response.status_code >= 400:
+        return None
+    for line in response.text.splitlines():
+        if line.startswith("#EXT-X-PROGRAM-DATE-TIME:"):
+            return iso_to_ms(line.split(":", 1)[1].strip())
+    return None
+
+
+def build_timeline_from_history(messages: List[dict], start_ms: int) -> Tuple[List[SpeakerInterval], Dict[str, float], List[Participant]]:
+    participant_first_seen: Dict[str, float] = {}
+    participants: Dict[str, Participant] = {}
+    open_microphones: Dict[str, float] = {}
+    intervals: List[SpeakerInterval] = []
+
+    for message in messages:
+        decoded = decode_history_payload(message)
+        if not decoded:
+            continue
+
+        body = decoded.get("body")
+        if isinstance(body, str):
+            try:
+                body = json.loads(body)
+            except json.JSONDecodeError:
+                body = None
+        if not isinstance(body, dict):
+            continue
+
+        body_type = body.get("type")
+        if body_type == 40:
+            username = body.get("guestUsername")
+            event_ms = ns_to_ms(decoded.get("timestamp"))
+            if username and event_ms:
+                username_key = username.lower()
+                seconds = max(0.0, (event_ms - start_ms) / 1000)
+                event = body.get("guestBroadcastingEvent")
+                if event in {9, 17}:
+                    participant_first_seen.setdefault(username_key, seconds)
+                participants.setdefault(
+                    username_key,
+                    Participant(
+                        user_id=str(body.get("guestRemoteID") or username),
+                        display_name=str(username),
+                        username=str(username),
+                        role="speaker",
+                    ),
+                )
+                if event == 17:
+                    open_microphones[username_key] = seconds
+                elif event == 16:
+                    start = open_microphones.pop(username_key, None)
+                    if start is not None and seconds > start:
+                        intervals.append(SpeakerInterval(start=start, end=seconds, speaker=username_key))
+        elif body_type == 45:
+            username = body.get("username")
+            event_ms = iso_to_ms(body.get("programDateTime")) or _to_int(body.get("timestamp"))
+            if username and event_ms:
+                participants.setdefault(
+                    username.lower(),
+                    Participant(
+                        user_id=str(body.get("twitter_id") or body.get("user_id") or username),
+                        display_name=str(body.get("displayName") or username),
+                        username=str(username),
+                        role="speaker",
+                    ),
+                )
+
+    last_event_second = max((interval.end for interval in intervals), default=0)
+    for username, start in open_microphones.items():
+        if last_event_second > start:
+            intervals.append(SpeakerInterval(start=start, end=last_event_second, speaker=username))
+
+    return intervals, participant_first_seen, list(participants.values())
+
+
+def unique_messages(messages: List[dict], seen: set) -> List[dict]:
+    output = []
+    for message in messages:
+        key = message.get("signature") or message.get("payload")
+        if key in seen:
+            continue
+        seen.add(key)
+        output.append(message)
+    return output
+
+
+def decode_history_payload(message: dict) -> Optional[dict]:
+    payload = message.get("payload")
+    if not isinstance(payload, str):
+        return None
+    try:
+        return json.loads(payload)
+    except json.JSONDecodeError:
+        return None
+
+
+def iso_to_ms(value: Optional[str]) -> Optional[int]:
+    if not value:
+        return None
+    try:
+        normalized = value.replace("Z", "+00:00")
+        return int(datetime.fromisoformat(normalized).timestamp() * 1000)
+    except ValueError:
+        return None
+
+
+def ns_to_ms(value) -> Optional[int]:
+    number = _to_int(value)
+    if number is None:
+        return None
+    return number // 1_000_000
+
+
+def _extract_participants(audio_space: dict) -> List[Participant]:
+    participants_section = audio_space.get("participants") or {}
+    participants: List[Participant] = []
+
+    for key, role in (("admins", "host"), ("speakers", "speaker"), ("listeners", "listener")):
+        for item in participants_section.get(key) or []:
+            participant = _participant_from_participant_item(item, role)
+            if participant:
+                participants.append(participant)
+
+    return _dedupe_participants(participants)
+
+
+def _participant_from_participant_item(item: dict, role: str) -> Optional[Participant]:
+    username = item.get("twitter_screen_name")
+    display_name = item.get("display_name") or username
+    avatar_url = item.get("avatar_url")
+
+    user_results = item.get("user_results") or {}
+    user_id = user_results.get("rest_id") or item.get("periscope_user_id") or username
+    user = _participant_from_user_result(user_results, role)
+    if user:
+        return replace(
+            user,
+            display_name=display_name or user.display_name,
+            username=username or user.username,
+            avatar_url=avatar_url or user.avatar_url,
+            role=role,
+        )
+
+    if not user_id or not username:
+        return None
+    return Participant(
+        user_id=str(user_id),
+        display_name=str(display_name or username),
+        username=str(username),
+        avatar_url=avatar_url,
+        role=role,
+    )
+
+
+def _participant_from_user_result(user_results: Optional[dict], role: str) -> Optional[Participant]:
+    if not user_results:
+        return None
+
+    result = user_results.get("result") or user_results
+    legacy = result.get("legacy") or {}
+    core = result.get("core") or {}
+    avatar = result.get("avatar") or {}
+
+    username = legacy.get("screen_name") or core.get("screen_name")
+    display_name = legacy.get("name") or core.get("name") or username
+    avatar_url = legacy.get("profile_image_url_https") or avatar.get("image_url")
+    user_id = result.get("rest_id") or user_results.get("rest_id") or result.get("id")
+
+    if not user_id or not username:
+        return None
+    return Participant(
+        user_id=str(user_id),
+        display_name=str(display_name or username),
+        username=str(username),
+        avatar_url=avatar_url,
+        role=role,
+    )
+
+
+def _dedupe_participants(participants: List[Participant]) -> List[Participant]:
+    by_key: Dict[str, Participant] = {}
+    role_rank = {"host": 3, "speaker": 2, "listener": 1}
+    for participant in participants:
+        key = participant.username.lower() or participant.user_id
+        existing = by_key.get(key)
+        if not existing or role_rank.get(participant.role, 0) > role_rank.get(existing.role, 0):
+            by_key[key] = participant
+    return list(by_key.values())
+
+
+def _to_int(value) -> Optional[int]:
+    if value is None:
+        return None
+    try:
+        return int(value)
+    except (TypeError, ValueError):
+        return None
+
+
+def _timestamp_to_ms(value) -> Optional[int]:
+    number = _to_int(value)
+    if number is None:
+        return None
+    if number > 10_000_000_000:
+        return number
+    return number * 1000

tests/test_integration.py

@@ -18,7 +18,8 @@ def test_full_workflow_with_mocks(tmp_path):
         # Run CLI
         result = runner.invoke(cli, [
             'https://x.com/i/spaces/1ZkKzZWnVRRKv',
-            '--output', str(tmp_path)
+            '--output', str(tmp_path),
+            '--mode', 'audio',
         ])
 
         # Verify success

tests/test_main.py

@@ -15,7 +15,7 @@ def test_cli_valid_url(tmp_path):
         mock_validate.return_value = (True, "1ZkKzZWnVRRKv", None)
         mock_download.return_value = str(tmp_path / "space_1ZkKzZWnVRRKv.mp3")
 
-        result = runner.invoke(cli, ['https://x.com/i/spaces/1ZkKzZWnVRRKv'])
+        result = runner.invoke(cli, ['https://x.com/i/spaces/1ZkKzZWnVRRKv', '--mode', 'audio'])
 
         assert result.exit_code == 0
         assert "Successfully downloaded" in result.output
@@ -42,7 +42,7 @@ def test_cli_download_failure():
         mock_validate.return_value = (True, "1ZkKzZWnVRRKv", None)
         mock_download.side_effect = DownloadError("Video unavailable")
 
-        result = runner.invoke(cli, ['https://x.com/i/spaces/1ZkKzZWnVRRKv'])
+        result = runner.invoke(cli, ['https://x.com/i/spaces/1ZkKzZWnVRRKv', '--mode', 'audio'])
 
         assert result.exit_code == 2
         assert "Video unavailable" in result.output
@@ -64,7 +64,8 @@ def test_cli_with_cookies(tmp_path):
 
         result = runner.invoke(cli, [
             'https://x.com/i/spaces/1ZkKzZWnVRRKv',
-            '--cookies', str(cookies_file)
+            '--cookies', str(cookies_file),
+            '--mode', 'audio',
         ])
 
         assert result.exit_code == 0
@@ -74,3 +75,26 @@ def test_cli_with_cookies(tmp_path):
             '.',
             str(cookies_file)
         )
+
+
+def test_cli_video_mode(tmp_path):
+    """Test CLI default video mode."""
+    runner = CliRunner()
+
+    with patch('src.main.validate_space_url') as mock_validate, \
+         patch('src.main.download_space_video') as mock_download:
+
+        mock_validate.return_value = (True, "1ZkKzZWnVRRKv", None)
+        mock_download.return_value = str(tmp_path / "space_1ZkKzZWnVRRKv.mp4")
+
+        result = runner.invoke(cli, ['https://x.com/i/spaces/1ZkKzZWnVRRKv'])
+
+        assert result.exit_code == 0
+        assert "space_1ZkKzZWnVRRKv.mp4" in result.output
+        mock_download.assert_called_once_with(
+            'https://x.com/i/spaces/1ZkKzZWnVRRKv',
+            '.',
+            None,
+            None,
+            None,
+        )

tests/test_video.py

@@ -0,0 +1,47 @@
+import json
+
+from src.models import Participant, SpaceMetadata
+from src.video import SpeakerTimeline, build_ffmpeg_command, render_frame
+
+
+def test_speaker_timeline_reads_active_speaker(tmp_path):
+    timeline_file = tmp_path / "timeline.json"
+    timeline_file.write_text(
+        json.dumps([
+            {"start": 0, "end": 2.5, "speaker": "alice"},
+            {"start": 2.5, "end": 4, "speaker": "@bob"},
+        ]),
+        encoding="utf-8",
+    )
+
+    timeline = SpeakerTimeline.from_file(str(timeline_file))
+
+    assert timeline.active_at(1) == "alice"
+    assert timeline.active_at(3) == "bob"
+    assert timeline.active_at(5) is None
+
+
+def test_build_ffmpeg_command_uses_audio_and_raw_video():
+    command = build_ffmpeg_command("audio.mp3", "out.mp4", 720, 720, 24)
+
+    assert command[:2] == ["ffmpeg", "-y"]
+    assert "audio.mp3" in command
+    assert "out.mp4" in command
+    assert "720x720" in command
+
+
+def test_render_frame_returns_rgb_image():
+    metadata = SpaceMetadata(
+        space_id="1ZkKzZWnVRRKv",
+        title="Example Space",
+        participants=[
+            Participant("1", "Alice", "alice", role="host"),
+            Participant("2", "Bob", "bob", role="speaker"),
+        ],
+    )
+    timeline = SpeakerTimeline.from_file_data([{"start": 0, "end": 10, "speaker": "bob"}])
+
+    frame = render_frame(metadata, {}, timeline, 1.0, 720, 720)
+
+    assert frame.mode == "RGB"
+    assert frame.size == (720, 720)

tests/test_x_api.py

@@ -0,0 +1,89 @@
+import json
+
+from src.x_api import build_timeline_from_history, parse_audio_space_metadata
+
+
+def test_parse_audio_space_metadata_extracts_title_creator_and_speakers():
+    payload = {
+        "data": {
+            "audioSpace": {
+                "metadata": {
+                    "rest_id": "1ZkKzZWnVRRKv",
+                    "title": "Example Space",
+                    "media_key": "28_123",
+                    "creator_results": {
+                        "result": {
+                            "rest_id": "10",
+                            "legacy": {
+                                "name": "Host User",
+                                "screen_name": "host",
+                                "profile_image_url_https": "https://example.com/host.jpg",
+                            },
+                        }
+                    },
+                },
+                "participants": {
+                    "speakers": [
+                        {
+                            "display_name": "Speaker User",
+                            "twitter_screen_name": "speaker",
+                            "avatar_url": "https://example.com/speaker.jpg",
+                            "periscope_user_id": "20",
+                        }
+                    ]
+                },
+            }
+        }
+    }
+
+    metadata = parse_audio_space_metadata(payload, "1ZkKzZWnVRRKv")
+
+    assert metadata.title == "Example Space"
+    assert metadata.media_key == "28_123"
+    assert [participant.username for participant in metadata.participants] == ["host", "speaker"]
+    assert metadata.participants[0].role == "host"
+    assert metadata.participants[1].role == "speaker"
+
+
+def test_build_timeline_uses_unmute_mute_events_not_captions():
+    start_ms = 1_000_000
+    messages = [
+        _history_message("alice", 17, start_ms + 5_000),
+        _caption_message("bob", start_ms + 6_000),
+        _history_message("alice", 16, start_ms + 12_000),
+    ]
+
+    intervals, first_seen, participants = build_timeline_from_history(messages, start_ms)
+
+    assert len(intervals) == 1
+    assert intervals[0].speaker == "alice"
+    assert intervals[0].start == 5
+    assert intervals[0].end == 12
+    assert first_seen == {"alice": 5}
+    assert {participant.username for participant in participants} == {"alice", "bob"}
+
+
+def _history_message(username, event, timestamp_ms):
+    payload = {
+        "body": json.dumps({
+            "guestBroadcastingEvent": event,
+            "guestRemoteID": username,
+            "guestUsername": username,
+            "type": 40,
+        }),
+        "timestamp": timestamp_ms * 1_000_000,
+    }
+    return {"payload": json.dumps(payload)}
+
+
+def _caption_message(username, timestamp_ms):
+    payload = {
+        "body": json.dumps({
+            "type": 45,
+            "username": username,
+            "displayName": username.title(),
+            "timestamp": timestamp_ms,
+        }),
+        "timestamp": timestamp_ms * 1_000_000,
+    }
+    return {"payload": json.dumps(payload)}