mirror of
https://github.com/Dasemu/x-space-downloader.git
synced 2026-10-13 13:39:45 +00:00
chore: initial commit
This commit is contained in:
0
src/__init__.py
Normal file
0
src/__init__.py
Normal file
163
src/downloader.py
Normal file
163
src/downloader.py
Normal file
@@ -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())
|
||||
56
src/main.py
Normal file
56
src/main.py
Normal file
@@ -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()
|
||||
34
src/models.py
Normal file
34
src/models.py
Normal file
@@ -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
|
||||
23
src/validator.py
Normal file
23
src/validator.py
Normal file
@@ -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
|
||||
464
src/video.py
Normal file
464
src/video.py
Normal file
@@ -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}"
|
||||
478
src/x_api.py
Normal file
478
src/x_api.py
Normal file
@@ -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
|
||||
Reference in New Issue
Block a user