diff --git a/README.md b/README.md index 1d9cb5c..7ac344e 100644 --- a/README.md +++ b/README.md @@ -87,6 +87,34 @@ A plain `ytstudio login` (no `profile add`) authenticates the active profile, wh on a fresh setup. Existing single-channel installs are migrated to the `default` profile automatically on first run. +## Live broadcasts + +Manage your YouTube livestream broadcasts from the CLI. This drives YouTube's +`liveBroadcasts` API (scheduling, starting, stopping, updating); the actual +video push to YouTube still goes through your encoder of choice (OBS, etc.). + +```bash +ytstudio livestreams list --status upcoming # what is scheduled +ytstudio livestreams show # details for one broadcast +ytstudio livestreams show --ingest # also fetch the bound stream's RTMP URL +ytstudio livestreams show --show-key # reveal the stream key (treat as secret) +ytstudio livestreams schedule \ + --title "Title" --scheduled-start 2026-06-01T19:00:00+02:00 # dry-run +ytstudio livestreams schedule ... --execute # actually create +ytstudio livestreams start # transition to live +ytstudio livestreams start --to testing # transition to the monitor stream first +ytstudio livestreams stop # transition to complete +ytstudio livestreams update --privacy unlisted --no-dvr # dry-run by default +``` + +Notes: + +- `schedule` and `update` default to a dry-run preview; pass `--execute` to apply. +- `--show-key` on `show` reveals the bound stream key; without it the key is + redacted. Stream keys are credentials. +- `liveBroadcasts.insert`/`update` are write operations (~50 quota units each); + see [API quota](#api-quota) below. + ## API quota The YouTube Data API enforces a default quota of 10_000 units per project per day. Most read diff --git a/src/ytstudio/commands/livestreams.py b/src/ytstudio/commands/livestreams.py new file mode 100644 index 0000000..5a87403 --- /dev/null +++ b/src/ytstudio/commands/livestreams.py @@ -0,0 +1,694 @@ +import json +from dataclasses import asdict, dataclass +from enum import StrEnum +from typing import Any + +import typer +from googleapiclient.errors import HttpError + +from ytstudio.api import api, handle_api_error +from ytstudio.services import get_data_service +from ytstudio.ui import ( + console, + create_kv_table, + create_table, + dim, + success_message, + truncate, +) + +app = typer.Typer(help="Live broadcast management (schedule, start, stop, update)") + + +class BroadcastStatus(StrEnum): + all = "all" + active = "active" + completed = "completed" + upcoming = "upcoming" + + +class PrivacyStatus(StrEnum): + public = "public" + private = "private" + unlisted = "unlisted" + + +class ClosedCaptionsType(StrEnum): + disabled = "closedCaptionsDisabled" + http_post = "closedCaptionsHttpPost" + embedded = "closedCaptionsEmbedded" + + +class LatencyPreference(StrEnum): + normal = "normal" + low = "low" + ultra_low = "ultraLow" + + +class Projection(StrEnum): + rectangular = "rectangular" + three_sixty = "360" + + +class TransitionTarget(StrEnum): + """The two states a user typically asks the CLI to drive into.""" + + testing = "testing" + live = "live" + + +class OutputFormat(StrEnum): + table = "table" + json = "json" + + +EMPTY = "-" + + +@dataclass +class Broadcast: + id: str + title: str + lifecycle_status: str + scheduled_start: str + scheduled_end: str = "" + description: str = "" + privacy: str = "public" + actual_start: str = "" + actual_end: str = "" + bound_stream_id: str = "" + auto_start: bool = False + auto_stop: bool = False + dvr: bool = True + embed: bool = True + record_from_start: bool = True + closed_captions_type: str = "closedCaptionsDisabled" + latency_preference: str = "normal" + projection: str = "rectangular" + made_for_kids: bool = False + # YouTube requires both of these whenever contentDetails is included in an + # update, so we keep them on the parsed object and round-trip them. + monitor_stream_enabled: bool = True + broadcast_stream_delay_ms: int = 0 + + +@dataclass +class StreamIngest: + """Subset of liveStream.cdn.ingestionInfo we surface to the user.""" + + stream_id: str + ingestion_address: str = "" + backup_ingestion_address: str = "" + rtmps_ingestion_address: str = "" + rtmps_backup_ingestion_address: str = "" + stream_name: str = "" # the actual stream key, redacted unless explicitly shown + format: str = "" + frame_rate: str = "" + resolution: str = "" + + +_LIVESTREAM_ERRORS = { + "invalidTransition": "Cannot perform this transition; broadcast may not be in the right state.", + "redundantTransition": "Broadcast is already in the requested state.", + "liveStreamingNotEnabled": ( + "Live streaming is not enabled for this channel. Enable it at youtube.com/features." + ), + "errorStreamInactive": ( + "No active stream is bound to this broadcast. Start your encoder (OBS, etc.) " + "or bind a stream first." + ), + "liveBroadcastNotFound": "Broadcast not found.", + "errorExecutingTransition": ( + "YouTube could not execute this transition. Check encoder health and try again." + ), + "livePermissionBlocked": ( + "Live streaming is blocked for this channel (account standing, age, or strikes)." + ), + "insufficientLivePermissions": ( + "This channel does not have permission for live streaming yet." + ), + "userRequestsExceedRateLimit": "Too many requests; wait a moment and retry.", + "concurrentBroadcastsExceedLimit": ( + "You already have the maximum number of concurrent live broadcasts." + ), + "sharedIngestionBroadcastsExceedLimit": ( + "Too many broadcasts share this ingestion stream; stop one and try again." + ), +} + + +def _handle_livestream_error(e: HttpError): + detail = e.error_details[0] if getattr(e, "error_details", None) else {} + reason = detail.get("reason", "") if isinstance(detail, dict) else "" + if reason in _LIVESTREAM_ERRORS: + console.print(f"[red]{_LIVESTREAM_ERRORS[reason]}[/red]") + raise typer.Exit(1) + handle_api_error(e) + + +def _parse_broadcast(item: dict[str, Any]) -> Broadcast: + snippet = item.get("snippet") or {} + status = item.get("status") or {} + content = item.get("contentDetails") or {} + monitor = content.get("monitorStream") or {} + return Broadcast( + id=str(item["id"]), + title=snippet.get("title", ""), + lifecycle_status=status.get("lifeCycleStatus", ""), + scheduled_start=snippet.get("scheduledStartTime", ""), + scheduled_end=snippet.get("scheduledEndTime", ""), + description=snippet.get("description", ""), + privacy=status.get("privacyStatus", "public"), + actual_start=snippet.get("actualStartTime", ""), + actual_end=snippet.get("actualEndTime", ""), + bound_stream_id=content.get("boundStreamId", ""), + auto_start=content.get("enableAutoStart", False), + auto_stop=content.get("enableAutoStop", False), + dvr=content.get("enableDvr", True), + embed=content.get("enableEmbed", True), + record_from_start=content.get("recordFromStart", True), + closed_captions_type=content.get("closedCaptionsType", "closedCaptionsDisabled"), + latency_preference=content.get("latencyPreference", "normal"), + projection=content.get("projection", "rectangular"), + made_for_kids=status.get("selfDeclaredMadeForKids", False), + monitor_stream_enabled=monitor.get("enableMonitorStream", True), + broadcast_stream_delay_ms=int(monitor.get("broadcastStreamDelayMs", 0) or 0), + ) + + +def _fetch_broadcast(service, broadcast_id: str) -> Broadcast | None: + response = ( + api(service.liveBroadcasts().list(part="snippet,status,contentDetails", id=broadcast_id)) + or {} + ) + for item in response.get("items", []): + if item.get("id") == broadcast_id: + return _parse_broadcast(item) + return None + + +def _fetch_stream_ingest(service, stream_id: str) -> StreamIngest | None: + response = api(service.liveStreams().list(part="snippet,cdn,status", id=stream_id)) or {} + items = response.get("items", []) + if not items: + return None + item = items[0] + cdn = item.get("cdn") or {} + ingestion_info = cdn.get("ingestionInfo") or {} + return StreamIngest( + stream_id=str(item.get("id", stream_id)), + ingestion_address=ingestion_info.get("ingestionAddress", ""), + backup_ingestion_address=ingestion_info.get("backupIngestionAddress", ""), + rtmps_ingestion_address=ingestion_info.get("rtmpsIngestionAddress", ""), + rtmps_backup_ingestion_address=ingestion_info.get("rtmpsBackupIngestionAddress", ""), + stream_name=ingestion_info.get("streamName", ""), + format=cdn.get("format", ""), + frame_rate=cdn.get("frameRate", ""), + resolution=cdn.get("resolution", ""), + ) + + +def _redact_key(stream_name: str) -> str: + """Mask all but the last 4 chars of the stream key.""" + if not stream_name: + return "" + if len(stream_name) <= 4: + return "*" * len(stream_name) + return "*" * (len(stream_name) - 4) + stream_name[-4:] + + +@app.command("list") +def list_broadcasts( + status: BroadcastStatus = typer.Option( + BroadcastStatus.upcoming, + "--status", + "-s", + help="Filter: all, upcoming, active, completed", + ), + limit: int = typer.Option(20, "--limit", "-n", min=1, max=50, help="Number of broadcasts"), + page_token: str = typer.Option(None, "--page-token", "-p", help="Page token for pagination"), + output: OutputFormat = typer.Option( + OutputFormat.table, "--output", "-o", help="Output format: table or json" + ), +): + """List your YouTube live broadcasts.""" + service = get_data_service() + response = ( + api( + service.liveBroadcasts().list( + part="snippet,status,contentDetails", + broadcastStatus=status.value, + maxResults=limit, + pageToken=page_token, + ) + ) + or {} + ) + broadcasts = [_parse_broadcast(item) for item in response.get("items", [])] + broadcasts.sort(key=lambda b: b.scheduled_start or "", reverse=True) + + if not broadcasts: + console.print("[yellow]No broadcasts found[/yellow]") + return + + if output is OutputFormat.json: + print( + json.dumps( + { + "broadcasts": [asdict(b) for b in broadcasts], + "next_page_token": response.get("nextPageToken"), + "total_results": (response.get("pageInfo") or {}).get("totalResults", 0), + }, + indent=2, + ) + ) + return + + table = create_table() + table.add_column("ID", style="yellow") + table.add_column("Title", style="cyan") + table.add_column("Status") + table.add_column("Scheduled Start") + table.add_column("Privacy") + + for broadcast in broadcasts: + scheduled = ( + broadcast.scheduled_start[:16].replace("T", " ") if broadcast.scheduled_start else EMPTY + ) + table.add_row( + broadcast.id, + truncate(broadcast.title), + broadcast.lifecycle_status or EMPTY, + scheduled, + broadcast.privacy, + ) + + console.print(table) + + if response.get("nextPageToken"): + console.print(f"\nNext page: --page-token {response['nextPageToken']}") + + +@app.command() +def show( + broadcast_id: str = typer.Argument(..., help="Broadcast ID"), + ingest: bool = typer.Option( + False, + "--ingest", + help="Also fetch and display the bound stream's ingest URL (key is redacted by default)", + ), + show_key: bool = typer.Option( + False, + "--show-key", + help="Reveal the bound stream key (implies --ingest). Treat output as a secret.", + ), + output: OutputFormat = typer.Option( + OutputFormat.table, "--output", "-o", help="Output format: table or json" + ), +): + """Show details for a specific broadcast.""" + service = get_data_service() + broadcast = _fetch_broadcast(service, broadcast_id) + + if not broadcast: + console.print(f"[red]Broadcast not found: {broadcast_id}[/red]") + raise typer.Exit(1) + + stream: StreamIngest | None = None + if (ingest or show_key) and broadcast.bound_stream_id: + stream = _fetch_stream_ingest(service, broadcast.bound_stream_id) + + if output is OutputFormat.json: + payload: dict[str, Any] = {"broadcast": asdict(broadcast)} + if stream: + ingest_dump = asdict(stream) + if not show_key: + ingest_dump["stream_name"] = _redact_key(ingest_dump["stream_name"]) + payload["ingest"] = ingest_dump + print(json.dumps(payload, indent=2)) + return + + console.print(f"\n[bold]{broadcast.title}[/bold]\n") + + table = create_kv_table() + table.add_column("field", style="dim") + table.add_column("value") + table.add_row("status", broadcast.lifecycle_status or EMPTY) + table.add_row("privacy", broadcast.privacy) + table.add_row("scheduled start", broadcast.scheduled_start or EMPTY) + table.add_row("scheduled end", broadcast.scheduled_end or EMPTY) + table.add_row("actual start", broadcast.actual_start or EMPTY) + table.add_row("actual end", broadcast.actual_end or EMPTY) + table.add_row("stream bound", broadcast.bound_stream_id or "No") + table.add_row("auto start", "Yes" if broadcast.auto_start else "No") + table.add_row("auto stop", "Yes" if broadcast.auto_stop else "No") + table.add_row("dvr", "Yes" if broadcast.dvr else "No") + table.add_row("embed", "Yes" if broadcast.embed else "No") + table.add_row("record from start", "Yes" if broadcast.record_from_start else "No") + table.add_row("closed captions", broadcast.closed_captions_type) + table.add_row("latency", broadcast.latency_preference) + table.add_row("projection", broadcast.projection) + table.add_row("made for kids", "Yes" if broadcast.made_for_kids else "No") + table.add_row("description", broadcast.description or EMPTY) + console.print(table) + + if (ingest or show_key) and not broadcast.bound_stream_id: + console.print(dim("\nNo stream bound; nothing to fetch.")) + return + + if stream: + console.print() + console.print("[bold]Ingest[/bold]") + ingest_table = create_kv_table() + ingest_table.add_column("field", style="dim") + ingest_table.add_column("value") + ingest_table.add_row("stream id", stream.stream_id) + ingest_table.add_row( + "format", f"{stream.resolution or '?'} {stream.frame_rate or ''}".strip() + ) + ingest_table.add_row("rtmp", stream.ingestion_address or EMPTY) + if stream.backup_ingestion_address: + ingest_table.add_row("rtmp (backup)", stream.backup_ingestion_address) + if stream.rtmps_ingestion_address: + ingest_table.add_row("rtmps", stream.rtmps_ingestion_address) + if stream.rtmps_backup_ingestion_address: + ingest_table.add_row("rtmps (backup)", stream.rtmps_backup_ingestion_address) + key_display = stream.stream_name if show_key else _redact_key(stream.stream_name) + ingest_table.add_row("stream key", key_display or EMPTY) + console.print(ingest_table) + if show_key: + console.print(dim("\nstream key revealed; treat the line above as a secret.")) + + +@app.command() +def start( + broadcast_id: str = typer.Argument(..., help="Broadcast ID"), + to: TransitionTarget = typer.Option( + TransitionTarget.live, + "--to", + help="Target state: testing (monitor only) or live (publish to viewers).", + ), +): + """Transition a broadcast to testing or live.""" + service = get_data_service() + try: + response = api( + service.liveBroadcasts().transition( + broadcastStatus=to.value, + id=broadcast_id, + part="id,snippet,status", + ) + ) + except HttpError as e: + _handle_livestream_error(e) + return + + new_status = (response or {}).get("status", {}).get("lifeCycleStatus", to.value) + success_message(f"Broadcast transitioning to {to.value}: {broadcast_id}") + console.print(f"Current status: {new_status}") + console.print(dim("Note: transitions are asynchronous; status may take a moment to update.")) + + +@app.command() +def stop(broadcast_id: str = typer.Argument(..., help="Broadcast ID")): + """Stop a live broadcast (transition to complete).""" + service = get_data_service() + try: + response = api( + service.liveBroadcasts().transition( + broadcastStatus="complete", + id=broadcast_id, + part="id,snippet,status", + ) + ) + except HttpError as e: + _handle_livestream_error(e) + return + + new_status = (response or {}).get("status", {}).get("lifeCycleStatus", "complete") + success_message(f"Broadcast stopped: {broadcast_id}") + console.print(f"Current status: {new_status}") + + +@app.command() +def schedule( + title: str = typer.Option(..., "--title", "-t", help="Broadcast title"), + scheduled_start: str = typer.Option( + ..., + "--scheduled-start", + help="Scheduled start time, ISO 8601 (e.g. 2026-06-01T19:00:00+02:00)", + ), + scheduled_end: str = typer.Option( + "", + "--scheduled-end", + help="Scheduled end time, ISO 8601", + ), + description: str = typer.Option("", "--description", "-d", help="Broadcast description"), + privacy: PrivacyStatus = typer.Option( + PrivacyStatus.public, "--privacy", help="public, private, or unlisted" + ), + made_for_kids: bool = typer.Option( + False, + "--made-for-kids/--not-made-for-kids", + help="COPPA self-declaration; required by YouTube on every broadcast.", + ), + execute: bool = typer.Option( + False, "--execute", help="Create the broadcast (default is dry-run preview)" + ), +): + """Schedule a new live broadcast.""" + snippet_body: dict[str, Any] = { + "title": title, + "description": description, + "scheduledStartTime": scheduled_start, + } + if scheduled_end: + snippet_body["scheduledEndTime"] = scheduled_end + status_body = { + "privacyStatus": privacy.value, + "selfDeclaredMadeForKids": made_for_kids, + } + body = {"snippet": snippet_body, "status": status_body} + + if not execute: + console.print("[bold]Preview new broadcast:[/bold]\n") + preview = create_kv_table() + preview.add_column("field", style="dim") + preview.add_column("value") + preview.add_row("title", title) + preview.add_row("scheduled start", scheduled_start) + if scheduled_end: + preview.add_row("scheduled end", scheduled_end) + if description: + preview.add_row("description", description) + preview.add_row("privacy", privacy.value) + preview.add_row("made for kids", "Yes" if made_for_kids else "No") + console.print(preview) + console.print("\nRun with --execute to create.") + console.print(dim("Quota: liveBroadcasts.insert costs ~50 units against your daily quota.")) + return + + service = get_data_service() + try: + response = ( + api(service.liveBroadcasts().insert(part="snippet,status,contentDetails", body=body)) + or {} + ) + except HttpError as e: + _handle_livestream_error(e) + return + + broadcast_id = response.get("id", "unknown") + success_message(f"Broadcast created: {broadcast_id}") + console.print(f"Title: {title}") + console.print(dim("Next: bind a stream in YouTube Studio or start your encoder.")) + + +@app.command() +def update( + broadcast_id: str = typer.Argument(..., help="Broadcast ID"), + title: str = typer.Option(None, "--title", "-t", help="New title"), + description: str = typer.Option(None, "--description", "-d", help="New description"), + privacy: PrivacyStatus = typer.Option(None, "--privacy", help="New privacy status"), + scheduled_start: str = typer.Option( + None, "--scheduled-start", help="New scheduled start, ISO 8601" + ), + scheduled_end: str = typer.Option(None, "--scheduled-end", help="New scheduled end, ISO 8601"), + auto_start: bool | None = typer.Option( + None, "--auto-start/--no-auto-start", help="Auto-start when stream begins" + ), + auto_stop: bool | None = typer.Option( + None, "--auto-stop/--no-auto-stop", help="Auto-stop when stream ends" + ), + dvr: bool | None = typer.Option(None, "--dvr/--no-dvr", help="Enable DVR controls"), + embed: bool | None = typer.Option(None, "--embed/--no-embed", help="Allow embedding"), + record_from_start: bool | None = typer.Option( + None, + "--record-from-start/--no-record-from-start", + help="Record broadcast for archive", + ), + closed_captions: ClosedCaptionsType = typer.Option( + None, "--closed-captions", help="Closed-caption mode" + ), + latency: LatencyPreference = typer.Option( + None, "--latency", help="Latency: normal, low, ultraLow" + ), + projection: Projection = typer.Option( + None, "--projection", help="Projection: rectangular or 360" + ), + execute: bool = typer.Option(False, "--execute", help="Apply changes (default is dry-run)"), +): + """Update a broadcast's metadata or settings (partial update). + + Note: liveBroadcasts.update only accepts privacyStatus under status; the + made-for-kids designation is set at schedule time and managed on the + resulting video resource afterwards. + """ + snippet_changes: dict[str, Any] = {} + if title is not None: + snippet_changes["title"] = title + if description is not None: + snippet_changes["description"] = description + if scheduled_start is not None: + snippet_changes["scheduledStartTime"] = scheduled_start + if scheduled_end is not None: + snippet_changes["scheduledEndTime"] = scheduled_end + + status_changes: dict[str, Any] = {} + if privacy is not None: + status_changes["privacyStatus"] = privacy.value + + content_changes: dict[str, Any] = {} + if auto_start is not None: + content_changes["enableAutoStart"] = auto_start + if auto_stop is not None: + content_changes["enableAutoStop"] = auto_stop + if dvr is not None: + content_changes["enableDvr"] = dvr + if embed is not None: + content_changes["enableEmbed"] = embed + if record_from_start is not None: + content_changes["recordFromStart"] = record_from_start + if closed_captions is not None: + content_changes["closedCaptionsType"] = closed_captions.value + if latency is not None: + content_changes["latencyPreference"] = latency.value + if projection is not None: + content_changes["projection"] = projection.value + + if not snippet_changes and not status_changes and not content_changes: + console.print( + "[yellow]Nothing to update. Pass at least one of --title, --description, " + "--privacy, --scheduled-start, --scheduled-end, --auto-start, --auto-stop, " + "--dvr, --embed, --record-from-start, --closed-captions, --latency, " + "or --projection.[/yellow]" + ) + raise typer.Exit(1) + + service = get_data_service() + current = _fetch_broadcast(service, broadcast_id) + if not current: + console.print(f"[red]Broadcast not found: {broadcast_id}[/red]") + raise typer.Exit(1) + + # Snippet/status PUTs replace the whole part, so we merge changes onto the current + # values from a fresh read. contentDetails accepts the same pattern. + snippet_body: dict[str, Any] = { + "title": current.title, + "description": current.description, + "scheduledStartTime": current.scheduled_start, + } + if current.scheduled_end: + snippet_body["scheduledEndTime"] = current.scheduled_end + snippet_body.update(snippet_changes) + + # liveBroadcasts.update only accepts privacyStatus under status (per the API + # reference), so we deliberately omit selfDeclaredMadeForKids here. + status_body: dict[str, Any] = {"privacyStatus": current.privacy} + status_body.update(status_changes) + + parts = ["snippet", "status"] + body: dict[str, Any] = { + "id": broadcast_id, + "snippet": snippet_body, + "status": status_body, + } + + if content_changes: + parts.append("contentDetails") + # contentDetails is replace-on-update. Include both monitorStream fields + # (YouTube rejects the request otherwise) and round-trip the rest from + # the freshly-read broadcast so unchanged fields are preserved. + content_body: dict[str, Any] = { + "monitorStream": { + "enableMonitorStream": current.monitor_stream_enabled, + "broadcastStreamDelayMs": current.broadcast_stream_delay_ms, + }, + "enableAutoStart": current.auto_start, + "enableAutoStop": current.auto_stop, + "enableDvr": current.dvr, + "enableEmbed": current.embed, + "recordFromStart": current.record_from_start, + "closedCaptionsType": current.closed_captions_type, + "latencyPreference": current.latency_preference, + "projection": current.projection, + } + content_body.update(content_changes) + body["contentDetails"] = content_body + + if not execute: + _print_update_preview(current, snippet_changes, status_changes, content_changes) + return + + try: + api(service.liveBroadcasts().update(part=",".join(parts), body=body)) + except HttpError as e: + _handle_livestream_error(e) + return + success_message(f"Updated: {snippet_body['title']}") + + +_SNIPPET_LABELS: dict[str, tuple[str, str]] = { + "title": ("title", "title"), + "description": ("description", "description"), + "scheduledStartTime": ("scheduled start", "scheduled_start"), + "scheduledEndTime": ("scheduled end", "scheduled_end"), +} +_STATUS_LABELS: dict[str, tuple[str, str]] = { + "privacyStatus": ("privacy", "privacy"), +} +_CONTENT_LABELS: dict[str, tuple[str, str]] = { + "enableAutoStart": ("auto start", "auto_start"), + "enableAutoStop": ("auto stop", "auto_stop"), + "enableDvr": ("dvr", "dvr"), + "enableEmbed": ("embed", "embed"), + "recordFromStart": ("record from start", "record_from_start"), + "closedCaptionsType": ("closed captions", "closed_captions_type"), + "latencyPreference": ("latency", "latency_preference"), + "projection": ("projection", "projection"), +} + + +def _print_update_preview( + current: Broadcast, + snippet_changes: dict[str, Any], + status_changes: dict[str, Any], + content_changes: dict[str, Any], +) -> None: + console.print("[bold]Preview changes:[/bold]\n") + table = create_kv_table() + table.add_column("field", style="dim") + table.add_column("old") + table.add_column("new", style="green") + for changes, label_map in ( + (snippet_changes, _SNIPPET_LABELS), + (status_changes, _STATUS_LABELS), + (content_changes, _CONTENT_LABELS), + ): + for api_key, new_value in changes.items(): + label, attr = label_map.get(api_key, (api_key, api_key)) + old_value = getattr(current, attr, "") + old_display = str(old_value) if old_value not in ("", None) else EMPTY + table.add_row(label, old_display, str(new_value)) + console.print(table) + console.print("\nRun with --execute to apply.") diff --git a/src/ytstudio/demo.py b/src/ytstudio/demo.py index 2537f0c..3e221d5 100644 --- a/src/ytstudio/demo.py +++ b/src/ytstudio/demo.py @@ -4,6 +4,7 @@ import time from datetime import UTC, datetime, timedelta from pathlib import Path +from typing import ClassVar DEMO_MODE = os.environ.get("YTSTUDIO_DEMO", "").lower() in ("1", "true", "yes") @@ -81,6 +82,79 @@ def setModerationStatus(self, **kwargs): return DemoRequest({}) +class _DemoLiveBroadcasts: + _STATUS_MAP: ClassVar[dict[str, str]] = { + "ready": "upcoming", + "created": "upcoming", + "live": "active", + "testing": "active", + "complete": "completed", + } + + def _all(self) -> list[dict]: + return _load("broadcasts.json")["items"] + + def list(self, **kwargs): + id_param = kwargs.get("id") + status_param = kwargs.get("broadcastStatus", "all") + max_results = kwargs.get("maxResults", 50) + + items = self._all() + if id_param: + ids = {i.strip() for i in str(id_param).split(",") if i.strip()} + items = [b for b in items if b.get("id") in ids] + elif status_param and status_param != "all": + items = [ + b + for b in items + if self._STATUS_MAP.get(b.get("status", {}).get("lifeCycleStatus", "")) + == status_param + ] + return DemoRequest({"items": items[:max_results]}) + + def transition(self, **kwargs): + broadcast_id = kwargs.get("id", "") + target = kwargs.get("broadcastStatus", "live") + return DemoRequest( + { + "id": broadcast_id, + "snippet": {"title": f"Demo broadcast {broadcast_id}"}, + "status": {"lifeCycleStatus": target}, + } + ) + + def insert(self, **kwargs): + body = kwargs.get("body", {}) or {} + snippet = body.get("snippet", {}) + status = body.get("status", {}) + return DemoRequest( + { + "id": "demo-broadcast-new", + "snippet": snippet, + "status": { + "lifeCycleStatus": "created", + "privacyStatus": status.get("privacyStatus", "public"), + "selfDeclaredMadeForKids": status.get("selfDeclaredMadeForKids", False), + }, + "contentDetails": {}, + } + ) + + def update(self, **kwargs): + body = kwargs.get("body", {}) or {} + return DemoRequest(body) + + +class _DemoLiveStreams: + def list(self, **kwargs): + id_param = kwargs.get("id", "") + items = _load("live_streams.json")["items"] + if id_param: + ids = {i.strip() for i in str(id_param).split(",") if i.strip()} + items = [s for s in items if s.get("id") in ids] + return DemoRequest({"items": items}) + + class DemoDataService: def channels(self): return _DemoChannels() @@ -97,6 +171,12 @@ def comments(self): def commentThreads(self): return _DemoCommentThreads() + def liveBroadcasts(self): + return _DemoLiveBroadcasts() + + def liveStreams(self): + return _DemoLiveStreams() + class _DemoReports: def query(self, **params): diff --git a/src/ytstudio/demo_data/broadcasts.json b/src/ytstudio/demo_data/broadcasts.json new file mode 100644 index 0000000..32308f5 --- /dev/null +++ b/src/ytstudio/demo_data/broadcasts.json @@ -0,0 +1,83 @@ +{ + "items": [ + { + "id": "demo-broadcast-upcoming-1", + "snippet": { + "title": "Demo upcoming livestream", + "description": "Scheduled demo broadcast (offline preview)", + "scheduledStartTime": "2026-06-15T19:00:00Z", + "scheduledEndTime": "2026-06-15T20:30:00Z" + }, + "status": { + "lifeCycleStatus": "ready", + "privacyStatus": "public", + "selfDeclaredMadeForKids": false + }, + "contentDetails": { + "boundStreamId": "demo-stream-1", + "enableAutoStart": true, + "enableAutoStop": true, + "enableDvr": true, + "enableEmbed": true, + "recordFromStart": true, + "closedCaptionsType": "closedCaptionsDisabled", + "latencyPreference": "low", + "projection": "rectangular", + "monitorStream": {"enableMonitorStream": true, "broadcastStreamDelayMs": 0} + } + }, + { + "id": "demo-broadcast-active-1", + "snippet": { + "title": "Demo broadcast in progress", + "description": "Currently live demo broadcast", + "scheduledStartTime": "2026-05-20T10:00:00Z", + "actualStartTime": "2026-05-20T10:02:11Z" + }, + "status": { + "lifeCycleStatus": "live", + "privacyStatus": "unlisted", + "selfDeclaredMadeForKids": false + }, + "contentDetails": { + "boundStreamId": "demo-stream-2", + "enableAutoStart": false, + "enableAutoStop": false, + "enableDvr": true, + "enableEmbed": true, + "recordFromStart": false, + "closedCaptionsType": "closedCaptionsEmbedded", + "latencyPreference": "normal", + "projection": "rectangular", + "monitorStream": {"enableMonitorStream": true, "broadcastStreamDelayMs": 5000} + } + }, + { + "id": "demo-broadcast-completed-1", + "snippet": { + "title": "Demo broadcast already finished", + "description": "Past demo broadcast", + "scheduledStartTime": "2026-04-01T18:00:00Z", + "actualStartTime": "2026-04-01T18:01:00Z", + "actualEndTime": "2026-04-01T19:30:00Z" + }, + "status": { + "lifeCycleStatus": "complete", + "privacyStatus": "public", + "selfDeclaredMadeForKids": false + }, + "contentDetails": { + "boundStreamId": "demo-stream-1", + "enableAutoStart": false, + "enableAutoStop": false, + "enableDvr": true, + "enableEmbed": false, + "recordFromStart": true, + "closedCaptionsType": "closedCaptionsDisabled", + "latencyPreference": "normal", + "projection": "rectangular", + "monitorStream": {"enableMonitorStream": false, "broadcastStreamDelayMs": 0} + } + } + ] +} diff --git a/src/ytstudio/demo_data/live_streams.json b/src/ytstudio/demo_data/live_streams.json new file mode 100644 index 0000000..494ce66 --- /dev/null +++ b/src/ytstudio/demo_data/live_streams.json @@ -0,0 +1,34 @@ +{ + "items": [ + { + "id": "demo-stream-1", + "snippet": {"title": "Demo reusable stream"}, + "cdn": { + "format": "1080p", + "frameRate": "30fps", + "resolution": "1080p", + "ingestionInfo": { + "ingestionAddress": "rtmp://a.rtmp.youtube.com/live2", + "backupIngestionAddress": "rtmp://b.rtmp.youtube.com/live2?backup=1", + "rtmpsIngestionAddress": "rtmps://a.rtmps.youtube.com/live2", + "streamName": "demo-key-1234-abcd-zxcv-XYZ9" + } + }, + "status": {"streamStatus": "active"} + }, + { + "id": "demo-stream-2", + "snippet": {"title": "Demo backup stream"}, + "cdn": { + "format": "720p", + "frameRate": "60fps", + "resolution": "720p", + "ingestionInfo": { + "ingestionAddress": "rtmp://a.rtmp.youtube.com/live2", + "streamName": "demo-key-9999-zzzz-PEXX" + } + }, + "status": {"streamStatus": "active"} + } + ] +} diff --git a/src/ytstudio/main.py b/src/ytstudio/main.py index 54650c2..57339a3 100644 --- a/src/ytstudio/main.py +++ b/src/ytstudio/main.py @@ -4,7 +4,7 @@ from rich.console import Console from ytstudio.api import authenticate, get_status -from ytstudio.commands import analytics, comments, profile, videos +from ytstudio.commands import analytics, comments, livestreams, profile, videos from ytstudio.config import migrate_legacy_credentials, setup_credentials from ytstudio.version import get_current_version, is_update_available @@ -20,6 +20,7 @@ app.add_typer(videos.app, name="videos") app.add_typer(analytics.app, name="analytics") app.add_typer(comments.app, name="comments") +app.add_typer(livestreams.app, name="livestreams") app.add_typer(profile.app, name="profile") diff --git a/tests/test_livestreams.py b/tests/test_livestreams.py new file mode 100644 index 0000000..d742fb6 --- /dev/null +++ b/tests/test_livestreams.py @@ -0,0 +1,460 @@ +import json +from unittest.mock import MagicMock, patch + +import pytest +from googleapiclient.errors import HttpError +from typer.testing import CliRunner + +from ytstudio.commands.livestreams import _redact_key, app + +runner = CliRunner() + + +def _broadcast_item( + *, + broadcast_id: str = "abc123", + title: str = "My Stream", + lifecycle: str = "ready", + privacy: str = "public", + scheduled_start: str = "2026-06-01T19:00:00Z", + scheduled_end: str = "2026-06-01T20:00:00Z", + bound_stream_id: str = "", + made_for_kids: bool = False, + monitor_enabled: bool = True, + stream_delay_ms: int = 0, + content_overrides: dict | None = None, +) -> dict: + content = { + "boundStreamId": bound_stream_id, + "enableAutoStart": True, + "enableAutoStop": False, + "enableDvr": True, + "enableEmbed": True, + "recordFromStart": True, + "closedCaptionsType": "closedCaptionsDisabled", + "latencyPreference": "normal", + "projection": "rectangular", + "monitorStream": { + "enableMonitorStream": monitor_enabled, + "broadcastStreamDelayMs": stream_delay_ms, + }, + } + if content_overrides: + content.update(content_overrides) + return { + "id": broadcast_id, + "snippet": { + "title": title, + "description": "Demo", + "scheduledStartTime": scheduled_start, + "scheduledEndTime": scheduled_end, + }, + "status": { + "lifeCycleStatus": lifecycle, + "privacyStatus": privacy, + "selfDeclaredMadeForKids": made_for_kids, + }, + "contentDetails": content, + } + + +def _stream_item(stream_id: str = "s1", stream_name: str = "key-FULL-1234-ZXCV") -> dict: + return { + "id": stream_id, + "snippet": {"title": "Stream"}, + "cdn": { + "format": "1080p", + "frameRate": "30fps", + "resolution": "1080p", + "ingestionInfo": { + "ingestionAddress": "rtmp://primary", + "backupIngestionAddress": "rtmp://backup", + "rtmpsIngestionAddress": "rtmps://secure", + "streamName": stream_name, + }, + }, + "status": {"streamStatus": "active"}, + } + + +def _service(items=None, *, stream_items=None, transition_response=None, insert_response=None): + service = MagicMock() + + list_request = MagicMock() + list_request.execute.return_value = {"items": items or []} + service.liveBroadcasts.return_value.list.return_value = list_request + + transition_request = MagicMock() + transition_request.execute.return_value = transition_response or { + "id": "abc123", + "status": {"lifeCycleStatus": "live"}, + } + service.liveBroadcasts.return_value.transition.return_value = transition_request + + insert_request = MagicMock() + insert_request.execute.return_value = insert_response or {"id": "new-id"} + service.liveBroadcasts.return_value.insert.return_value = insert_request + + update_request = MagicMock() + update_request.execute.return_value = {} + service.liveBroadcasts.return_value.update.return_value = update_request + + stream_list_request = MagicMock() + stream_list_request.execute.return_value = {"items": stream_items or []} + service.liveStreams.return_value.list.return_value = stream_list_request + + return service + + +def _http_error(reason: str, status: int = 400) -> HttpError: + resp = MagicMock() + resp.status = status + err = HttpError(resp, b"{}") + err.error_details = [{"reason": reason}] + return err + + +# -------- redact helper -------- + + +class TestRedactKey: + @pytest.mark.parametrize( + ("key", "expected"), + [ + ("", ""), + ("abcd", "****"), + ("abcdef", "**cdef"), + ("super-long-stream-key-XYZ9", "**********************XYZ9"), + ], + ) + def test_redact(self, key, expected): + assert _redact_key(key) == expected + + +# -------- list -------- + + +class TestList: + def test_table_output(self): + service = _service([_broadcast_item()]) + with patch("ytstudio.commands.livestreams.get_data_service", return_value=service): + result = runner.invoke(app, ["list"]) + assert result.exit_code == 0 + assert "abc123" in result.stdout + assert "My Stream" in result.stdout + + def test_json_output_includes_fields(self): + service = _service([_broadcast_item(bound_stream_id="s1")]) + with patch("ytstudio.commands.livestreams.get_data_service", return_value=service): + result = runner.invoke(app, ["list", "--output", "json"]) + assert result.exit_code == 0 + payload = json.loads(result.stdout) + assert payload["broadcasts"][0]["id"] == "abc123" + assert payload["broadcasts"][0]["bound_stream_id"] == "s1" + + def test_empty_message(self): + service = _service([]) + with patch("ytstudio.commands.livestreams.get_data_service", return_value=service): + result = runner.invoke(app, ["list"]) + assert result.exit_code == 0 + assert "No broadcasts" in result.stdout + + def test_status_filter_passed_to_api(self): + service = _service([]) + with patch("ytstudio.commands.livestreams.get_data_service", return_value=service): + runner.invoke(app, ["list", "--status", "active"]) + service.liveBroadcasts.return_value.list.assert_called_with( + part="snippet,status,contentDetails", + broadcastStatus="active", + maxResults=20, + pageToken=None, + ) + + def test_limit_clamped_by_typer(self): + service = _service([]) + with patch("ytstudio.commands.livestreams.get_data_service", return_value=service): + result = runner.invoke(app, ["list", "--limit", "999"]) + # Typer's min/max=50 should reject this. + assert result.exit_code != 0 + + +# -------- show -------- + + +class TestShow: + def test_table(self): + service = _service([_broadcast_item(bound_stream_id="s1")]) + with patch("ytstudio.commands.livestreams.get_data_service", return_value=service): + result = runner.invoke(app, ["show", "abc123"]) + assert result.exit_code == 0 + assert "My Stream" in result.stdout + assert "s1" in result.stdout + + def test_not_found(self): + service = _service([]) + with patch("ytstudio.commands.livestreams.get_data_service", return_value=service): + result = runner.invoke(app, ["show", "missing"]) + assert result.exit_code == 1 + assert "not found" in result.stdout.lower() + + def test_ingest_redacted_by_default(self): + service = _service( + [_broadcast_item(bound_stream_id="s1")], + stream_items=[_stream_item("s1", "secret-key-ABCDEFGH")], + ) + with patch("ytstudio.commands.livestreams.get_data_service", return_value=service): + result = runner.invoke(app, ["show", "abc123", "--ingest"]) + assert result.exit_code == 0 + assert "secret-key-ABCDEFGH" not in result.stdout + assert "EFGH" in result.stdout # last four shown + assert "rtmp://primary" in result.stdout + + def test_show_key_reveals(self): + service = _service( + [_broadcast_item(bound_stream_id="s1")], + stream_items=[_stream_item("s1", "secret-key-ABCDEFGH")], + ) + with patch("ytstudio.commands.livestreams.get_data_service", return_value=service): + result = runner.invoke(app, ["show", "abc123", "--show-key"]) + assert result.exit_code == 0 + assert "secret-key-ABCDEFGH" in result.stdout + + def test_ingest_without_bound_stream(self): + service = _service([_broadcast_item(bound_stream_id="")]) + with patch("ytstudio.commands.livestreams.get_data_service", return_value=service): + result = runner.invoke(app, ["show", "abc123", "--ingest"]) + assert result.exit_code == 0 + assert "No stream bound" in result.stdout + service.liveStreams.return_value.list.assert_not_called() + + def test_json_output_redacted(self): + service = _service( + [_broadcast_item(bound_stream_id="s1")], + stream_items=[_stream_item("s1", "secret-key-ABCDEFGH")], + ) + with patch("ytstudio.commands.livestreams.get_data_service", return_value=service): + result = runner.invoke(app, ["show", "abc123", "--ingest", "--output", "json"]) + assert result.exit_code == 0 + payload = json.loads(result.stdout) + assert "secret-key-ABCDEFGH" not in result.stdout + assert payload["ingest"]["stream_name"].endswith("EFGH") + + +# -------- start / stop -------- + + +class TestStartStop: + def test_start_default_goes_live(self): + service = _service() + with patch("ytstudio.commands.livestreams.get_data_service", return_value=service): + result = runner.invoke(app, ["start", "abc123"]) + assert result.exit_code == 0 + service.liveBroadcasts.return_value.transition.assert_called_with( + broadcastStatus="live", + id="abc123", + part="id,snippet,status", + ) + + def test_start_to_testing(self): + service = _service(transition_response={"status": {"lifeCycleStatus": "testing"}}) + with patch("ytstudio.commands.livestreams.get_data_service", return_value=service): + result = runner.invoke(app, ["start", "abc123", "--to", "testing"]) + assert result.exit_code == 0 + assert "testing" in result.stdout + service.liveBroadcasts.return_value.transition.assert_called_with( + broadcastStatus="testing", + id="abc123", + part="id,snippet,status", + ) + + def test_stop_transitions_to_complete(self): + service = _service(transition_response={"status": {"lifeCycleStatus": "complete"}}) + with patch("ytstudio.commands.livestreams.get_data_service", return_value=service): + result = runner.invoke(app, ["stop", "abc123"]) + assert result.exit_code == 0 + service.liveBroadcasts.return_value.transition.assert_called_with( + broadcastStatus="complete", + id="abc123", + part="id,snippet,status", + ) + + def test_known_error_reasons_map_to_friendly_message(self): + service = _service() + service.liveBroadcasts.return_value.transition.return_value.execute.side_effect = ( + _http_error("liveStreamingNotEnabled") + ) + with patch("ytstudio.commands.livestreams.get_data_service", return_value=service): + result = runner.invoke(app, ["start", "abc123"]) + assert result.exit_code == 1 + assert "not enabled" in result.stdout.lower() + + +# -------- schedule -------- + + +class TestSchedule: + def test_dry_run_does_not_call_insert(self): + service = _service() + with patch("ytstudio.commands.livestreams.get_data_service", return_value=service): + result = runner.invoke( + app, + [ + "schedule", + "--title", + "Hello", + "--scheduled-start", + "2026-06-01T19:00:00+02:00", + ], + ) + assert result.exit_code == 0 + assert "Preview" in result.stdout + service.liveBroadcasts.return_value.insert.assert_not_called() + + def test_execute_creates_broadcast(self): + service = _service(insert_response={"id": "freshly-made"}) + with patch("ytstudio.commands.livestreams.get_data_service", return_value=service): + result = runner.invoke( + app, + [ + "schedule", + "--title", + "Hello", + "--scheduled-start", + "2026-06-01T19:00:00+02:00", + "--scheduled-end", + "2026-06-01T20:00:00+02:00", + "--privacy", + "unlisted", + "--made-for-kids", + "--execute", + ], + ) + assert result.exit_code == 0 + assert "freshly-made" in result.stdout + call_kwargs = service.liveBroadcasts.return_value.insert.call_args.kwargs + body = call_kwargs["body"] + assert body["snippet"]["title"] == "Hello" + assert body["snippet"]["scheduledStartTime"] == "2026-06-01T19:00:00+02:00" + assert body["snippet"]["scheduledEndTime"] == "2026-06-01T20:00:00+02:00" + assert body["status"]["privacyStatus"] == "unlisted" + assert body["status"]["selfDeclaredMadeForKids"] is True + + +# -------- update -------- + + +class TestUpdate: + def test_no_change_flags_exits(self): + service = _service([_broadcast_item()]) + with patch("ytstudio.commands.livestreams.get_data_service", return_value=service): + result = runner.invoke(app, ["update", "abc123"]) + assert result.exit_code == 1 + service.liveBroadcasts.return_value.update.assert_not_called() + + def test_not_found(self): + service = _service([]) + with patch("ytstudio.commands.livestreams.get_data_service", return_value=service): + result = runner.invoke(app, ["update", "missing", "--title", "x"]) + assert result.exit_code == 1 + assert "not found" in result.stdout.lower() + + def test_dry_run_shows_diff_only(self): + service = _service([_broadcast_item()]) + with patch("ytstudio.commands.livestreams.get_data_service", return_value=service): + result = runner.invoke(app, ["update", "abc123", "--title", "Renamed"]) + assert result.exit_code == 0 + assert "Renamed" in result.stdout + service.liveBroadcasts.return_value.update.assert_not_called() + + def test_partial_update_keeps_other_fields(self): + service = _service( + [ + _broadcast_item( + bound_stream_id="s1", + monitor_enabled=True, + stream_delay_ms=5000, + content_overrides={"enableDvr": True}, + ) + ] + ) + with patch("ytstudio.commands.livestreams.get_data_service", return_value=service): + result = runner.invoke( + app, + ["update", "abc123", "--no-dvr", "--execute"], + ) + assert result.exit_code == 0 + update = service.liveBroadcasts.return_value.update + update.assert_called_once() + kwargs = update.call_args.kwargs + assert "contentDetails" in kwargs["part"] + body = kwargs["body"] + assert body["contentDetails"]["enableDvr"] is False + # Other content fields keep their fetched-current values. + assert body["contentDetails"]["enableAutoStart"] is True + # monitorStream must be carried over verbatim; YouTube rejects updates + # to contentDetails that omit it. + monitor = body["contentDetails"]["monitorStream"] + assert monitor["enableMonitorStream"] is True + assert monitor["broadcastStreamDelayMs"] == 5000 + # Snippet must still have title/start/end to avoid clearing them via PUT. + assert body["snippet"]["title"] == "My Stream" + assert body["snippet"]["scheduledStartTime"] == "2026-06-01T19:00:00Z" + # status only carries privacyStatus; selfDeclaredMadeForKids is not a + # writable field on liveBroadcasts.update. + assert body["status"] == {"privacyStatus": "public"} + + def test_metadata_only_update_skips_content_part(self): + service = _service([_broadcast_item()]) + with patch("ytstudio.commands.livestreams.get_data_service", return_value=service): + result = runner.invoke( + app, + ["update", "abc123", "--title", "Renamed", "--execute"], + ) + assert result.exit_code == 0 + kwargs = service.liveBroadcasts.return_value.update.call_args.kwargs + assert "contentDetails" not in kwargs["part"] + assert "snippet" in kwargs["part"] + assert kwargs["body"]["snippet"]["title"] == "Renamed" + + +class TestOutputValidation: + def test_invalid_output_format_rejected(self): + result = runner.invoke(app, ["list", "--output", "yaml"]) + assert result.exit_code != 0 + + def test_show_json_redacted_without_show_key(self): + service = _service( + [_broadcast_item(bound_stream_id="s1")], + stream_items=[_stream_item("s1", "secret-key-FOUR-LAST")], + ) + with patch("ytstudio.commands.livestreams.get_data_service", return_value=service): + result = runner.invoke(app, ["show", "abc123", "--ingest", "-o", "json"]) + assert result.exit_code == 0 + payload = json.loads(result.stdout) + assert payload["ingest"]["stream_name"].endswith("LAST") + assert "secret-key" not in payload["ingest"]["stream_name"] + + def test_show_json_reveals_key_with_show_key(self): + service = _service( + [_broadcast_item(bound_stream_id="s1")], + stream_items=[_stream_item("s1", "full-secret-key-XYZ")], + ) + with patch("ytstudio.commands.livestreams.get_data_service", return_value=service): + result = runner.invoke(app, ["show", "abc123", "--ingest", "--show-key", "-o", "json"]) + assert result.exit_code == 0 + payload = json.loads(result.stdout) + assert payload["ingest"]["stream_name"] == "full-secret-key-XYZ" + + +class TestRtmpsBackup: + def test_rtmps_backup_url_rendered(self): + stream = _stream_item("s1") + stream["cdn"]["ingestionInfo"]["rtmpsBackupIngestionAddress"] = "rtmps://backup-secure" + service = _service( + [_broadcast_item(bound_stream_id="s1")], + stream_items=[stream], + ) + with patch("ytstudio.commands.livestreams.get_data_service", return_value=service): + result = runner.invoke(app, ["show", "abc123", "--ingest"]) + assert result.exit_code == 0 + assert "rtmps://backup-secure" in result.stdout