From 6823a0d2b900aa77092dd2cf5497b256446296f9 Mon Sep 17 00:00:00 2001 From: Jelmer de Wit <1598297+jdwit@users.noreply.github.com> Date: Tue, 26 May 2026 22:31:32 +0200 Subject: [PATCH 1/2] feat: add livestreams command for managing YouTube live broadcasts New 'ytstudio livestreams' typer app with list, show, schedule, start, stop, update subcommands driving the YouTube liveBroadcasts API. Inspired by the ParkerClelland fork but rewritten against the current codebase. Improvements over the fork's implementation: - 'show --ingest' fetches the bound liveStream and prints the RTMP URLs; the stream key is redacted unless '--show-key' is passed explicitly. - 'start --to testing|live' covers the monitor-stream transition that streamers use to verify a feed before going live. - 'update' computes a partial change set, only touching the contentDetails part when any contentDetails field is actually changed. - ASCII '-' for empty values (project style) and no redundant BroadcastStatus.to_api_value mapping. - Demo-mode support via _DemoLiveBroadcasts and _DemoLiveStreams plus broadcasts.json / live_streams.json fixtures so 'YTSTUDIO_DEMO=1' can drive the new commands offline. Tests cover list/filter/empty, show with ingest redaction and reveal, start to testing vs live, stop, schedule dry-run vs execute body, partial update body, and friendly error mapping for liveStreamingNotEnabled. 160 tests pass, ruff clean. --- README.md | 28 + src/ytstudio/commands/livestreams.py | 653 +++++++++++++++++++++++ src/ytstudio/demo.py | 80 +++ src/ytstudio/demo_data/broadcasts.json | 80 +++ src/ytstudio/demo_data/live_streams.json | 34 ++ src/ytstudio/main.py | 3 +- tests/test_livestreams.py | 396 ++++++++++++++ 7 files changed, 1273 insertions(+), 1 deletion(-) create mode 100644 src/ytstudio/commands/livestreams.py create mode 100644 src/ytstudio/demo_data/broadcasts.json create mode 100644 src/ytstudio/demo_data/live_streams.json create mode 100644 tests/test_livestreams.py 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..b28a8e7 --- /dev/null +++ b/src/ytstudio/commands/livestreams.py @@ -0,0 +1,653 @@ +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" + + +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 + + +@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 = "" + 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.", +} + + +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 {} + 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), + ) + + +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", ""), + 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: str = typer.Option("table", "--output", "-o", help="Output format: table, 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 == "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: str = typer.Option("table", "--output", "-o", help="Output format: table, 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 == "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) + 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" + ), + made_for_kids: bool | None = typer.Option( + None, "--made-for-kids/--not-made-for-kids", help="Made-for-kids designation" + ), + execute: bool = typer.Option(False, "--execute", help="Apply changes (default is dry-run)"), +): + """Update a broadcast's metadata or settings (partial update).""" + 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 + if made_for_kids is not None: + status_changes["selfDeclaredMadeForKids"] = made_for_kids + + 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, " + "--projection, or --made-for-kids.[/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) + + status_body: dict[str, Any] = { + "privacyStatus": current.privacy, + "selfDeclaredMadeForKids": current.made_for_kids, + } + 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") + content_body: dict[str, Any] = { + "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"), + "selfDeclaredMadeForKids": ("made for kids", "made_for_kids"), +} +_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..9b07050 --- /dev/null +++ b/src/ytstudio/demo_data/broadcasts.json @@ -0,0 +1,80 @@ +{ + "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" + } + }, + { + "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" + } + }, + { + "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" + } + } + ] +} 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..ad4c8a6 --- /dev/null +++ b/tests/test_livestreams.py @@ -0,0 +1,396 @@ +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, + 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", + } + 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", 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 + # 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" + + 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" From b1a6e5475e0fc608b3ae5f85d88a3c236cb59ece Mon Sep 17 00:00:00 2001 From: Jelmer de Wit <1598297+jdwit@users.noreply.github.com> Date: Tue, 26 May 2026 22:37:39 +0200 Subject: [PATCH 2/2] fix: address Codex review on livestreams - Preserve contentDetails.monitorStream (enableMonitorStream + broadcastStreamDelayMs) on update; YouTube rejects contentDetails updates that omit it. - Drop --made-for-kids from update; liveBroadcasts.update only accepts privacyStatus under status. Schedule keeps the flag (the resource is set on insert). - Parse and surface rtmpsBackupIngestionAddress in show --ingest. - Add OutputFormat enum so --output rejects anything other than table/json. - Expand _LIVESTREAM_ERRORS map with livePermissionBlocked, insufficientLivePermissions, errorExecutingTransition, userRequestsExceedRateLimit, concurrentBroadcastsExceedLimit, sharedIngestionBroadcastsExceedLimit. - Demo broadcasts.json fixtures carry monitorStream so demo mode mirrors the real resource shape. Tests cover monitorStream preservation through partial update, --output json redaction vs reveal, RTMPS backup display, and rejection of invalid --output values. --- src/ytstudio/commands/livestreams.py | 73 ++++++++++++++++++++------ src/ytstudio/demo_data/broadcasts.json | 9 ++-- tests/test_livestreams.py | 66 ++++++++++++++++++++++- 3 files changed, 128 insertions(+), 20 deletions(-) diff --git a/src/ytstudio/commands/livestreams.py b/src/ytstudio/commands/livestreams.py index b28a8e7..5a87403 100644 --- a/src/ytstudio/commands/livestreams.py +++ b/src/ytstudio/commands/livestreams.py @@ -57,6 +57,11 @@ class TransitionTarget(StrEnum): live = "live" +class OutputFormat(StrEnum): + table = "table" + json = "json" + + EMPTY = "-" @@ -81,6 +86,10 @@ class Broadcast: 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 @@ -91,6 +100,7 @@ class StreamIngest: 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 = "" @@ -108,6 +118,22 @@ class StreamIngest: "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." + ), } @@ -124,6 +150,7 @@ 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", ""), @@ -144,6 +171,8 @@ def _parse_broadcast(item: dict[str, Any]) -> Broadcast: 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), ) @@ -171,6 +200,7 @@ def _fetch_stream_ingest(service, stream_id: str) -> StreamIngest | None: 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", ""), @@ -197,7 +227,9 @@ def list_broadcasts( ), 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: str = typer.Option("table", "--output", "-o", help="Output format: table, json"), + output: OutputFormat = typer.Option( + OutputFormat.table, "--output", "-o", help="Output format: table or json" + ), ): """List your YouTube live broadcasts.""" service = get_data_service() @@ -219,7 +251,7 @@ def list_broadcasts( console.print("[yellow]No broadcasts found[/yellow]") return - if output == "json": + if output is OutputFormat.json: print( json.dumps( { @@ -270,7 +302,9 @@ def show( "--show-key", help="Reveal the bound stream key (implies --ingest). Treat output as a secret.", ), - output: str = typer.Option("table", "--output", "-o", help="Output format: table, json"), + output: OutputFormat = typer.Option( + OutputFormat.table, "--output", "-o", help="Output format: table or json" + ), ): """Show details for a specific broadcast.""" service = get_data_service() @@ -284,7 +318,7 @@ def show( if (ingest or show_key) and broadcast.bound_stream_id: stream = _fetch_stream_ingest(service, broadcast.bound_stream_id) - if output == "json": + if output is OutputFormat.json: payload: dict[str, Any] = {"broadcast": asdict(broadcast)} if stream: ingest_dump = asdict(stream) @@ -337,6 +371,8 @@ def show( 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) @@ -500,12 +536,14 @@ def update( projection: Projection = typer.Option( None, "--projection", help="Projection: rectangular or 360" ), - made_for_kids: bool | None = typer.Option( - None, "--made-for-kids/--not-made-for-kids", help="Made-for-kids designation" - ), execute: bool = typer.Option(False, "--execute", help="Apply changes (default is dry-run)"), ): - """Update a broadcast's metadata or settings (partial update).""" + """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 @@ -519,8 +557,6 @@ def update( status_changes: dict[str, Any] = {} if privacy is not None: status_changes["privacyStatus"] = privacy.value - if made_for_kids is not None: - status_changes["selfDeclaredMadeForKids"] = made_for_kids content_changes: dict[str, Any] = {} if auto_start is not None: @@ -545,7 +581,7 @@ def update( "[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, " - "--projection, or --made-for-kids.[/yellow]" + "or --projection.[/yellow]" ) raise typer.Exit(1) @@ -566,10 +602,9 @@ def update( snippet_body["scheduledEndTime"] = current.scheduled_end snippet_body.update(snippet_changes) - status_body: dict[str, Any] = { - "privacyStatus": current.privacy, - "selfDeclaredMadeForKids": current.made_for_kids, - } + # 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"] @@ -581,7 +616,14 @@ def update( 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, @@ -614,7 +656,6 @@ def update( } _STATUS_LABELS: dict[str, tuple[str, str]] = { "privacyStatus": ("privacy", "privacy"), - "selfDeclaredMadeForKids": ("made for kids", "made_for_kids"), } _CONTENT_LABELS: dict[str, tuple[str, str]] = { "enableAutoStart": ("auto start", "auto_start"), diff --git a/src/ytstudio/demo_data/broadcasts.json b/src/ytstudio/demo_data/broadcasts.json index 9b07050..32308f5 100644 --- a/src/ytstudio/demo_data/broadcasts.json +++ b/src/ytstudio/demo_data/broadcasts.json @@ -22,7 +22,8 @@ "recordFromStart": true, "closedCaptionsType": "closedCaptionsDisabled", "latencyPreference": "low", - "projection": "rectangular" + "projection": "rectangular", + "monitorStream": {"enableMonitorStream": true, "broadcastStreamDelayMs": 0} } }, { @@ -47,7 +48,8 @@ "recordFromStart": false, "closedCaptionsType": "closedCaptionsEmbedded", "latencyPreference": "normal", - "projection": "rectangular" + "projection": "rectangular", + "monitorStream": {"enableMonitorStream": true, "broadcastStreamDelayMs": 5000} } }, { @@ -73,7 +75,8 @@ "recordFromStart": true, "closedCaptionsType": "closedCaptionsDisabled", "latencyPreference": "normal", - "projection": "rectangular" + "projection": "rectangular", + "monitorStream": {"enableMonitorStream": false, "broadcastStreamDelayMs": 0} } } ] diff --git a/tests/test_livestreams.py b/tests/test_livestreams.py index ad4c8a6..d742fb6 100644 --- a/tests/test_livestreams.py +++ b/tests/test_livestreams.py @@ -20,6 +20,8 @@ def _broadcast_item( 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 = { @@ -32,6 +34,10 @@ def _broadcast_item( "closedCaptionsType": "closedCaptionsDisabled", "latencyPreference": "normal", "projection": "rectangular", + "monitorStream": { + "enableMonitorStream": monitor_enabled, + "broadcastStreamDelayMs": stream_delay_ms, + }, } if content_overrides: content.update(content_overrides) @@ -362,7 +368,14 @@ def test_dry_run_shows_diff_only(self): def test_partial_update_keeps_other_fields(self): service = _service( - [_broadcast_item(bound_stream_id="s1", content_overrides={"enableDvr": True})] + [ + _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( @@ -378,9 +391,17 @@ def test_partial_update_keeps_other_fields(self): 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()]) @@ -394,3 +415,46 @@ def test_metadata_only_update_skips_content_part(self): 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