commit b66b22923dd490449367dc64870e0682526362fe Author: Dasemu Date: Fri Jul 17 16:57:55 2026 +0200 chore: initial commit diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..d84789a --- /dev/null +++ b/.gitignore @@ -0,0 +1,30 @@ +# Python +__pycache__/ +*.py[cod] +*$py.class +*.so +.Python +env/ +venv/ +.venv/ +ENV/ +*.egg-info/ +dist/ +build/ + +# IDEs +.vscode/ +.idea/ +*.swp +*.swo + +# Testing +.pytest_cache/ +.coverage +htmlcov/ + +# Downloads +*.mp3 +*.m4a +*.mp4 +cookies.txt diff --git a/README.md b/README.md new file mode 100644 index 0000000..0e10b21 --- /dev/null +++ b/README.md @@ -0,0 +1,202 @@ +# X Space Downloader + +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 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 + +## Installation + +### Prerequisites + +- Python 3.8 or higher +- ffmpeg (required for audio conversion and MP4 rendering) + +Install ffmpeg: +```bash +# macOS +brew install ffmpeg + +# Ubuntu/Debian +sudo apt install ffmpeg + +# Windows +# Download from https://ffmpeg.org/download.html +``` + +### Install the tool + +```bash +git clone https://github.com/synapz-org/x-space-downloader.git +cd x-space-downloader +pip install -e . +``` + +## Usage + +### Basic usage + +```bash +x-space-dl https://x.com/i/spaces/1ZkKzZWnVRRKv +``` + +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) + +For better reliability and access to private Spaces, provide cookies from your logged-in X session: + +```bash +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. + +### 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 +x-space-dl --output ~/Downloads https://x.com/i/spaces/1ZkKzZWnVRRKv +``` + +### Full options + +```bash +x-space-dl --help +``` + +## Exporting Cookies + +To download Spaces that require authentication, you need to export cookies from your logged-in X session. + +### Method 1: Browser Extension (Easiest) + +1. Install a cookie export extension: + - Chrome/Edge: [Get cookies.txt LOCALLY](https://chrome.google.com/webstore/detail/get-cookiestxt-locally/cclelndahbckbenkjhflpdbgdldlbecc) + - Firefox: [cookies.txt](https://addons.mozilla.org/en-US/firefox/addon/cookies-txt/) + +2. Log in to X (twitter.com or x.com) +3. Click the extension icon and export cookies +4. Save as `cookies.txt` + +### Method 2: Manual Export (Advanced) + +1. Log in to X +2. Open Developer Tools (F12) +3. Go to Application/Storage → Cookies +4. Copy cookies in Netscape format to a file + +## Requirements + +- Python 3.8+ +- yt-dlp +- click +- requests +- Pillow +- ffmpeg (system dependency) + +## Supported URLs + +The tool supports X Space URLs in these formats: + +- `https://x.com/i/spaces/{SPACE_ID}` +- `https://twitter.com/i/spaces/{SPACE_ID}` + +**Note:** Only recordings are supported. Live Spaces must be recorded before downloading. + +## Error Messages + +| Error | Meaning | +|-------|---------| +| "Invalid URL format" | URL doesn't match X Space format | +| "Failed to download Space" | Recording not found or network error | +| "Authentication required" | Space requires login (use --cookies) | + +## Development + +### Running tests + +```bash +pip install pytest pytest-mock +pytest +``` + +### Project structure + +``` +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 +``` + +## License + +MIT License - See LICENSE file for details + +## Contributing + +Contributions are welcome! Please open an issue or submit a pull request. + +## Troubleshooting + +### "ffmpeg not found" +Install ffmpeg using your system's package manager (see Installation section). + +### "Video unavailable" +The Space recording may have been deleted or is private. Try using the --cookies option. + +### "Authentication required" +Export cookies from your logged-in X session and use the --cookies option. + +## Acknowledgments + +Built with: +- [yt-dlp](https://github.com/yt-dlp/yt-dlp) - Media downloader +- [Click](https://click.palletsprojects.com/) - CLI framework diff --git a/docs/plans/2025-10-26-x-space-downloader-design.md b/docs/plans/2025-10-26-x-space-downloader-design.md new file mode 100644 index 0000000..13194a4 --- /dev/null +++ b/docs/plans/2025-10-26-x-space-downloader-design.md @@ -0,0 +1,188 @@ +# X Space Downloader - Design Document + +**Date:** 2025-10-26 +**Status:** Approved + +## Overview + +A Python CLI tool that downloads X (Twitter) Space recordings and outputs them as MP3 files. Takes a Space URL as input and produces an audio file. + +## Requirements + +- **Input:** X Space recording URL (e.g., `https://x.com/i/spaces/1ZkKzZWnVRRKv`) +- **Output:** MP3 audio file +- **Interface:** Command-line tool +- **Authentication:** Supports X account credentials via cookies +- **Scope:** Recordings only (not live Spaces) +- **Error Handling:** Clear error messages for invalid or non-recording URLs + +## Architecture Overview + +### Technology Choices + +- **Language:** Python +- **Core Engine:** yt-dlp (handles X authentication and media extraction) +- **CLI Framework:** Click (argument parsing and user interface) +- **Pattern:** Thin wrapper around yt-dlp + +### Project Structure + +``` +x-space-downloader/ +├── src/ +│ ├── __init__.py +│ ├── main.py # CLI entry point with argument parsing +│ ├── downloader.py # Wraps yt-dlp functionality +│ └── validator.py # URL validation and error handling +├── tests/ +│ ├── test_validator.py +│ └── test_downloader.py +├── requirements.txt # Dependencies: yt-dlp, click +├── setup.py # Makes it installable +└── README.md +``` + +### User Workflow + +```bash +x-space-dl --cookies cookies.txt https://x.com/i/spaces/1ZkKzZWnVRRKv +# Output: space_1ZkKzZWnVRRKv.mp3 +``` + +## Component Design + +### validator.py + +**Responsibilities:** +- Validate URL format (must match X Spaces pattern: `https://x.com/i/spaces/{SPACE_ID}`) +- Check if URL points to a recording vs live Space +- Return clear error messages for invalid inputs + +**Key Interface:** +```python +def validate_space_url(url: str) -> Tuple[bool, Optional[str], Optional[str]]: + """ + Validates X Space URL. + + Returns: + (is_valid, space_id, error_message) + """ +``` + +### downloader.py + +**Responsibilities:** +- Configure yt-dlp options (format: mp3, quality, output template) +- Pass cookie file to yt-dlp for authentication +- Translate yt-dlp output/errors to user-friendly messages +- Show download progress + +**Key Interface:** +```python +def download_space( + space_url: str, + output_dir: str, + cookies_file: Optional[str] +) -> str: + """ + Downloads Space recording using yt-dlp. + + Returns: + Path to downloaded MP3 file + """ +``` + +**yt-dlp Configuration:** +- Output format: mp3 (audio only) +- Output template: `space_{space_id}.mp3` +- Post-processing: convert to mp3 if needed +- Progress hooks for user feedback + +### main.py + +**Responsibilities:** +- CLI argument parsing with Click +- Orchestrate validation → download flow +- Handle and display errors appropriately +- Exit with appropriate status codes + +**CLI Arguments:** +- `URL` (required): X Space URL +- `--cookies` (optional): Path to cookies file +- `--output` (optional): Output directory (default: current directory) + +**Exit Codes:** +- `0`: Success +- `1`: Invalid URL +- `2`: Download failed +- `3`: Authentication required + +## Data Flow + +1. User provides URL + optional cookies file +2. **Validator** checks URL format → extracts space_id +3. **Downloader** configures yt-dlp with: + - Output format: mp3 (audio only) + - Output template: `space_{space_id}.mp3` + - Cookies file if provided + - Post-processing: convert to mp3 if needed +4. yt-dlp fetches metadata, downloads audio stream, converts to mp3 +5. Return file path to user + +## Error Handling + +| Error Scenario | Detection | User Message | +|----------------|-----------|--------------| +| Invalid URL format | Validator regex check | "Invalid URL format. Expected: https://x.com/i/spaces/{SPACE_ID}" | +| Space doesn't exist | yt-dlp 404 error | "Recording not found. Check URL or Space may be deleted." | +| Authentication required | yt-dlp auth error | "Authentication required. Please provide cookies file with --cookies" | +| Network errors | yt-dlp timeout/connection | "Network error. Check connection and try again." | +| Space is live | Validator metadata check | "Error: This is a live Space, not a recording." | +| Disk space/permissions | File system errors | "Cannot write to output directory. Check permissions and disk space." | + +## Authentication + +**Cookie-based Authentication:** +- User exports cookies from logged-in X session +- Provides cookie file path via `--cookies` flag +- yt-dlp uses cookies for authenticated requests + +**Cookie Export Instructions:** +- Browser extensions: "Get cookies.txt", "cookies.txt" +- Manual: Export via browser DevTools +- Instructions included in README + +## Testing Strategy + +**Unit Tests:** +- Validator: Test various URL formats, valid/invalid patterns +- Error message generation + +**Integration Tests:** +- Full download flow with yt-dlp +- Use test Space URLs or mock yt-dlp calls +- Cookie authentication flow + +**Error Scenario Tests:** +- Invalid URLs +- Missing cookies +- Network failures +- Live Space detection + +## Dependencies + +**Core:** +- `yt-dlp`: Media downloading and extraction +- `click`: CLI framework + +**Development:** +- `pytest`: Testing framework +- `pytest-mock`: Mocking for tests + +## Future Considerations (Out of Scope) + +- Live Space recording +- Batch downloading multiple Spaces +- GUI interface +- Format options beyond MP3 +- Metadata extraction (speakers, title, description) diff --git a/pytest.ini b/pytest.ini new file mode 100644 index 0000000..775a52a --- /dev/null +++ b/pytest.ini @@ -0,0 +1,3 @@ +[pytest] +markers = + integration: marks tests as integration tests (deselect with '-m "not integration"') diff --git a/requirements-dev.txt b/requirements-dev.txt new file mode 100644 index 0000000..e175fdf --- /dev/null +++ b/requirements-dev.txt @@ -0,0 +1,2 @@ +pytest>=7.0.0 +pytest-mock>=3.10.0 diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..66fd468 --- /dev/null +++ b/requirements.txt @@ -0,0 +1,4 @@ +yt-dlp>=2024.0.0 +click>=8.0.0 +requests>=2.31.0 +Pillow>=10.0.0 diff --git a/setup.py b/setup.py new file mode 100644 index 0000000..153e5ea --- /dev/null +++ b/setup.py @@ -0,0 +1,19 @@ +from setuptools import setup, find_packages + +setup( + name="x-space-downloader", + version="0.1.0", + 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": [ + "x-space-dl=src.main:cli", + ], + }, + python_requires=">=3.8", +) diff --git a/src/__init__.py b/src/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/downloader.py b/src/downloader.py new file mode 100644 index 0000000..9f715fd --- /dev/null +++ b/src/downloader.py @@ -0,0 +1,163 @@ +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): + """Raised when download fails.""" + pass + + +def download_space( + space_url: str, + output_dir: str, + cookies_file: Optional[str] = None, + audio_format: str = "mp3", +) -> str: + """ + Downloads X Space recording using yt-dlp. + + Args: + 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 + + Raises: + DownloadError: If download fails + """ + # Extract space ID from URL using validator for robustness + 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") + + 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'), + 'quiet': False, + 'no_warnings': False, + } + if audio_format != "native": + ydl_opts['postprocessors'] = [{ + 'key': 'FFmpegExtractAudio', + 'preferredcodec': audio_format, + }] + + # Add cookies if provided + if cookies_file: + ydl_opts['cookiefile'] = cookies_file + + # Download with yt-dlp + try: + with yt_dlp.YoutubeDL(ydl_opts) as ydl: + ydl.extract_info(space_url, download=True) + except Exception as e: + raise DownloadError(f"Failed to download Space: {str(e)}") + + # Return the expected output path + 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_metadata_or_fallback(space_id: str, cookies_file: Optional[str]) -> SpaceMetadata: + try: + return XApiClient(cookies_file=cookies_file).fetch_space_metadata(space_id) + except XApiError: + return SpaceMetadata(space_id=space_id, title=f"X Space {space_id}", participants=[]) + + +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 fetch_timeline_or_none(metadata: SpaceMetadata, cookies_file: str) -> Optional[SpeakerTimeline]: + try: + intervals, first_seen, history_participants = XApiClient(cookies_file=cookies_file).fetch_space_timeline(metadata) + except XApiError: + return None + merge_history_participants(metadata, history_participants) + if not intervals and not first_seen: + return None + 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()) diff --git a/src/main.py b/src/main.py new file mode 100644 index 0000000..543e8ab --- /dev/null +++ b/src/main.py @@ -0,0 +1,56 @@ +import click +from click.exceptions import Exit +from src.validator import validate_space_url +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') +@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 or MP4 files. + + Example: + x-space-dl https://x.com/i/spaces/1ZkKzZWnVRRKv + """ + # Validate URL + is_valid, space_id, error = validate_space_url(url) + if not is_valid: + click.echo(f"Error: {error}", err=True) + raise Exit(1) + + # Download the Space + try: + click.echo(f"Downloading Space: {space_id}") + 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) + raise Exit(2) + + +if __name__ == '__main__': + cli() diff --git a/src/models.py b/src/models.py new file mode 100644 index 0000000..7eadb45 --- /dev/null +++ b/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 diff --git a/src/validator.py b/src/validator.py new file mode 100644 index 0000000..c55c655 --- /dev/null +++ b/src/validator.py @@ -0,0 +1,23 @@ +import re +from typing import Tuple, Optional + + +def validate_space_url(url: str) -> Tuple[bool, Optional[str], Optional[str]]: + """ + Validates X Space URL format. + + Args: + url: The URL to validate + + Returns: + Tuple of (is_valid, space_id, error_message) + """ + # Pattern for X Space URLs (supports both x.com and twitter.com) + pattern = r"https://(x|twitter)\.com/i/spaces/([a-zA-Z0-9]+)" + match = re.match(pattern, url) + + 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) + return True, space_id, None diff --git a/src/video.py b/src/video.py new file mode 100644 index 0000000..0aa9040 --- /dev/null +++ b/src/video.py @@ -0,0 +1,464 @@ +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 + + +def format_time(seconds: float) -> str: + minutes, secs = divmod(int(seconds), 60) + hours, minutes = divmod(minutes, 60) + if hours: + return f"{hours:d}:{minutes:02d}:{secs:02d}" + return f"{minutes:d}:{secs:02d}" diff --git a/src/x_api.py b/src/x_api.py new file mode 100644 index 0000000..ecc32f6 --- /dev/null +++ b/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 diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/test_downloader.py b/tests/test_downloader.py new file mode 100644 index 0000000..5e2517b --- /dev/null +++ b/tests/test_downloader.py @@ -0,0 +1,104 @@ +import pytest +from unittest.mock import Mock, patch +from src.downloader import download_space, DownloadError + + +def test_download_space_basic(tmp_path): + """Test basic download functionality.""" + space_url = "https://x.com/i/spaces/1ZkKzZWnVRRKv" + + with patch('src.downloader.yt_dlp.YoutubeDL') as mock_ytdl_class: + # Mock the YoutubeDL instance + mock_ytdl = Mock() + mock_ytdl_class.return_value.__enter__.return_value = mock_ytdl + mock_ytdl.extract_info.return_value = {'id': '1ZkKzZWnVRRKv'} + + # Call download_space + result = download_space( + space_url=space_url, + output_dir=str(tmp_path), + cookies_file=None + ) + + # Verify YoutubeDL was called + mock_ytdl.extract_info.assert_called_once_with(space_url, download=True) + + # Verify result contains expected filename + assert "space_1ZkKzZWnVRRKv" in result + + +def test_download_space_with_cookies(tmp_path): + """Test that cookies file is passed to yt-dlp when provided.""" + space_url = "https://x.com/i/spaces/1ZkKzZWnVRRKv" + cookies_path = "/path/to/cookies.txt" + + with patch('src.downloader.yt_dlp.YoutubeDL') as mock_ytdl_class: + # Mock the YoutubeDL instance + mock_ytdl = Mock() + mock_ytdl_class.return_value.__enter__.return_value = mock_ytdl + mock_ytdl.extract_info.return_value = {'id': '1ZkKzZWnVRRKv'} + + # Call download_space with cookies + result = download_space( + space_url=space_url, + output_dir=str(tmp_path), + cookies_file=cookies_path + ) + + # Verify YoutubeDL was initialized with correct options + call_args = mock_ytdl_class.call_args[0][0] + assert 'cookiefile' in call_args + assert call_args['cookiefile'] == cookies_path + + +def test_download_space_ytdlp_configuration(tmp_path): + """Test that yt-dlp is configured correctly with format, outtmpl, and postprocessors.""" + space_url = "https://x.com/i/spaces/1ZkKzZWnVRRKv" + + with patch('src.downloader.yt_dlp.YoutubeDL') as mock_ytdl_class: + # Mock the YoutubeDL instance + mock_ytdl = Mock() + mock_ytdl_class.return_value.__enter__.return_value = mock_ytdl + mock_ytdl.extract_info.return_value = {'id': '1ZkKzZWnVRRKv'} + + # Call download_space + result = download_space( + space_url=space_url, + output_dir=str(tmp_path), + cookies_file=None + ) + + # Verify YoutubeDL was initialized with correct options + call_args = mock_ytdl_class.call_args[0][0] + + # Check format + assert call_args['format'] == 'bestaudio/best' + + # Check outtmpl contains the space_id + assert 'outtmpl' in call_args + assert 'space_1ZkKzZWnVRRKv' in call_args['outtmpl'] + + # Check postprocessors for MP3 conversion + assert 'postprocessors' in call_args + assert len(call_args['postprocessors']) > 0 + assert call_args['postprocessors'][0]['key'] == 'FFmpegExtractAudio' + assert call_args['postprocessors'][0]['preferredcodec'] == 'mp3' + + +def test_download_space_error_handling(tmp_path): + """Test that download errors are properly caught and wrapped.""" + space_url = "https://x.com/i/spaces/invalid" + + with patch('src.downloader.yt_dlp.YoutubeDL') as mock_ytdl_class: + mock_ytdl = Mock() + mock_ytdl_class.return_value.__enter__.return_value = mock_ytdl + mock_ytdl.extract_info.side_effect = Exception("Video unavailable") + + with pytest.raises(DownloadError) as exc_info: + download_space( + space_url=space_url, + output_dir=str(tmp_path), + cookies_file=None + ) + + assert "Failed to download" in str(exc_info.value) diff --git a/tests/test_integration.py b/tests/test_integration.py new file mode 100644 index 0000000..3074f25 --- /dev/null +++ b/tests/test_integration.py @@ -0,0 +1,33 @@ +import pytest +from click.testing import CliRunner +from src.main import cli + + +@pytest.mark.integration +def test_full_workflow_with_mocks(tmp_path): + """Integration test for complete workflow.""" + runner = CliRunner() + + from unittest.mock import patch + + with patch('src.downloader.yt_dlp.YoutubeDL') as mock_ytdl_class: + # Mock successful download + mock_ytdl = mock_ytdl_class.return_value.__enter__.return_value + mock_ytdl.extract_info.return_value = {'id': '1ZkKzZWnVRRKv'} + + # Run CLI + result = runner.invoke(cli, [ + 'https://x.com/i/spaces/1ZkKzZWnVRRKv', + '--output', str(tmp_path), + '--mode', 'audio', + ]) + + # Verify success + assert result.exit_code == 0 + assert "Successfully downloaded" in result.output + + # Verify yt-dlp was called correctly + mock_ytdl.extract_info.assert_called_once() + call_args = mock_ytdl_class.call_args + assert 'bestaudio' in call_args[0][0]['format'] + assert 'mp3' in call_args[0][0]['postprocessors'][0]['preferredcodec'] diff --git a/tests/test_main.py b/tests/test_main.py new file mode 100644 index 0000000..b746f42 --- /dev/null +++ b/tests/test_main.py @@ -0,0 +1,100 @@ +import pytest +from click.testing import CliRunner +from unittest.mock import patch, Mock +from src.main import cli +from src.downloader import DownloadError + + +def test_cli_valid_url(tmp_path): + """Test CLI with valid URL.""" + runner = CliRunner() + + with patch('src.main.validate_space_url') as mock_validate, \ + patch('src.main.download_space') as mock_download: + + 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', '--mode', 'audio']) + + assert result.exit_code == 0 + assert "Successfully downloaded" in result.output + assert "space_1ZkKzZWnVRRKv.mp3" in result.output + + +def test_cli_invalid_url(): + """Test CLI with invalid URL.""" + runner = CliRunner() + + result = runner.invoke(cli, ['https://invalid.com/spaces/123']) + + assert result.exit_code == 1 + assert "Invalid URL format" in result.output + + +def test_cli_download_failure(): + """Test CLI when download fails.""" + runner = CliRunner() + + with patch('src.main.validate_space_url') as mock_validate, \ + patch('src.main.download_space') as mock_download: + + mock_validate.return_value = (True, "1ZkKzZWnVRRKv", None) + mock_download.side_effect = DownloadError("Video unavailable") + + result = runner.invoke(cli, ['https://x.com/i/spaces/1ZkKzZWnVRRKv', '--mode', 'audio']) + + assert result.exit_code == 2 + assert "Video unavailable" in result.output + + +def test_cli_with_cookies(tmp_path): + """Test CLI with cookies file.""" + runner = CliRunner() + + # Create a fake cookies file + cookies_file = tmp_path / "cookies.txt" + cookies_file.write_text("# Netscape HTTP Cookie File\n") + + with patch('src.main.validate_space_url') as mock_validate, \ + patch('src.main.download_space') as mock_download: + + 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', + '--cookies', str(cookies_file), + '--mode', 'audio', + ]) + + assert result.exit_code == 0 + # Verify the call was made with correct arguments including cookies_file + mock_download.assert_called_once_with( + 'https://x.com/i/spaces/1ZkKzZWnVRRKv', + '.', + 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, + ) diff --git a/tests/test_validator.py b/tests/test_validator.py new file mode 100644 index 0000000..e15ef55 --- /dev/null +++ b/tests/test_validator.py @@ -0,0 +1,38 @@ +import pytest +from src.validator import validate_space_url + + +def test_validate_space_url_valid(): + """Test that valid X Space URL is accepted.""" + url = "https://x.com/i/spaces/1ZkKzZWnVRRKv" + is_valid, space_id, error = validate_space_url(url) + + assert is_valid is True + assert space_id == "1ZkKzZWnVRRKv" + assert error is None + + +def test_validate_space_url_invalid_format(): + """Test that invalid URLs are rejected.""" + invalid_urls = [ + "https://x.com/spaces/1ZkKzZWnVRRKv", # Missing /i/ + "https://x.com/i/spaces/", # Missing space ID + "not a url", # Completely invalid + "", # Empty string + ] + + for url in invalid_urls: + is_valid, space_id, error = validate_space_url(url) + assert is_valid is False + assert space_id is None + assert "Invalid URL format" in error + + +def test_validate_space_url_twitter_domain(): + """Test that twitter.com URLs are also accepted.""" + url = "https://twitter.com/i/spaces/1ZkKzZWnVRRKv" + is_valid, space_id, error = validate_space_url(url) + + assert is_valid is True + assert space_id == "1ZkKzZWnVRRKv" + assert error is None diff --git a/tests/test_video.py b/tests/test_video.py new file mode 100644 index 0000000..cbd72bd --- /dev/null +++ b/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) diff --git a/tests/test_x_api.py b/tests/test_x_api.py new file mode 100644 index 0000000..4befd75 --- /dev/null +++ b/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)}