Add opt-in quota-bounded video telemetry (#12)

Add a disabled-by-default videos.list telemetry collector with strict eligibility checks, write-ahead quota throttling, private local storage, bounded retention, independent system/user timers, CI coverage, and operator documentation.

Author: sumitake

.github/workflows/ci.yml

@@ -41,7 +41,7 @@ jobs:
         run: ruff check .
 
       - name: Compile scripts
-        run: python -m py_compile bin/youtube-autoencoder bin/youtube-autoencoder-api bin/youtube-autoencoder-test-pattern
+        run: python -m py_compile bin/youtube-autoencoder bin/youtube-autoencoder-api bin/youtube-autoencoder-test-pattern bin/youtube-autoencoder-telemetry
 
       - name: Run tests
         run: pytest -q
@@ -51,11 +51,19 @@ jobs:
           test -x bin/youtube-autoencoder
           test -x bin/youtube-autoencoder-api
           test -x bin/youtube-autoencoder-test-pattern
+          test -x bin/youtube-autoencoder-telemetry
 
       - name: Verify systemd units
         run: |
           sudo install -m 0755 bin/youtube-autoencoder /usr/local/bin/youtube-autoencoder
-          systemd-analyze verify systemd/youtube-autoencoder@.service systemd/user/youtube-autoencoder.service
+          sudo install -m 0755 bin/youtube-autoencoder-telemetry /usr/local/bin/youtube-autoencoder-telemetry
+          systemd-analyze verify \
+            systemd/youtube-autoencoder@.service \
+            systemd/youtube-autoencoder-telemetry@.service \
+            systemd/youtube-autoencoder-telemetry@.timer \
+            systemd/user/youtube-autoencoder.service \
+            systemd/user/youtube-autoencoder-telemetry.service \
+            systemd/user/youtube-autoencoder-telemetry.timer
 
   repository-metadata:
     name: Repository Metadata

CHANGELOG.md

@@ -2,6 +2,12 @@
 
 All notable project changes are recorded here. The README shows only the most recent entry.
 
+## 2026-07-11 - Optional Video Telemetry
+
+- Added an opt-in `videos.list` command and one-shot collector for aggregate live-stream timing, viewer, view, like, and comment metrics.
+- Added private local latest and daily JSONL samples, strict payload validation, a nonblocking lock, write-ahead quota throttling, and configurable retention.
+- Added independent system and user five-minute timers that remain disabled after installation; one selected timer consumes at most 288 quota units per 24 hours at the minimum interval and never controls stream recovery.
+
 ## 2026-07-11 - Pre-Ingest Broadcast Staging
 
 - Reordered every recovery, including cached public state, so one marked broadcast is reconciled, created unlisted if needed, and bound before FFmpeg starts ingest; startup API failures now fail closed.

README.md

@@ -39,6 +39,10 @@ OBS_PROFILE_DIR = pathlib.Path(
 OBS_SERVICE_FILE = pathlib.Path(os.environ.get("YTA_OBS_SERVICE_FILE", OBS_PROFILE_DIR / "service.json")).expanduser()
 SCOPES = "https://www.googleapis.com/auth/youtube"
 API_BASE = "https://www.googleapis.com/youtube/v3"
+VIDEO_METRICS_FIELDS = (
+    "items(id,liveStreamingDetails(actualStartTime,actualEndTime,scheduledStartTime,"
+    "scheduledEndTime,concurrentViewers),statistics(viewCount,likeCount,commentCount))"
+)
 
 child: subprocess.Popen[Any] | None = None
 stopping = False
@@ -154,6 +158,12 @@ def read_state() -> dict[str, Any]:
         return {}
 
 
+def read_state_snapshot() -> dict[str, Any]:
+    if not STATE_FILE.exists():
+        return {}
+    return read_json(STATE_FILE)
+
+
 @contextlib.contextmanager
 def mutation_lock(timeout: float | None = None):
     LOCK_FILE.parent.mkdir(parents=True, exist_ok=True)
@@ -357,6 +367,57 @@ def api(method: str, path: str, params: dict[str, str], body: dict[str, Any] | N
     return http_json(method, f"{API_BASE}{path}?{query}", token=token(), body=body)
 
 
+def metric_int(value: Any, field: str) -> int | None:
+    if value is None or value == "":
+        return None
+    if isinstance(value, bool) or not isinstance(value, int | str):
+        raise ReconciliationError(f"invalid {field} metric")
+    try:
+        parsed = int(value)
+    except (TypeError, ValueError) as exc:
+        raise ReconciliationError(f"invalid {field} metric") from exc
+    if parsed < 0:
+        raise ReconciliationError(f"invalid {field} metric")
+    return parsed
+
+
+def video_metrics(video_id: str) -> dict[str, Any]:
+    data = api(
+        "GET",
+        "/videos",
+        {
+            "id": video_id,
+            "part": "liveStreamingDetails,statistics",
+            "fields": VIDEO_METRICS_FIELDS,
+        },
+    )
+    items = data.get("items") or []
+    if len(items) != 1 or not isinstance(items[0], dict):
+        raise ReconciliationError(f"video not found: {video_id}")
+    item = items[0]
+    if str(item.get("id") or "") != video_id:
+        raise ReconciliationError("videos.list returned an unexpected video id")
+    live = item.get("liveStreamingDetails")
+    statistics = item.get("statistics")
+    if (live is not None and not isinstance(live, dict)) or (
+        statistics is not None and not isinstance(statistics, dict)
+    ):
+        raise ReconciliationError("videos.list returned malformed metrics")
+    live = live or {}
+    statistics = statistics or {}
+    return {
+        "video_id": video_id,
+        "actual_start_time": live.get("actualStartTime"),
+        "actual_end_time": live.get("actualEndTime"),
+        "scheduled_start_time": live.get("scheduledStartTime"),
+        "scheduled_end_time": live.get("scheduledEndTime"),
+        "concurrent_viewers": metric_int(live.get("concurrentViewers"), "concurrentViewers"),
+        "view_count": metric_int(statistics.get("viewCount"), "viewCount"),
+        "like_count": metric_int(statistics.get("likeCount"), "likeCount"),
+        "comment_count": metric_int(statistics.get("commentCount"), "commentCount"),
+    }
+
+
 def obs_stream_name() -> str:
     settings = read_json(OBS_SERVICE_FILE).get("settings") or {}
     stream_name = str(settings.get("key") or "")
@@ -968,6 +1029,17 @@ def state_command(_args: argparse.Namespace) -> int:
     return 0
 
 
+def video_metrics_command(args: argparse.Namespace) -> int:
+    video_id = args.video_id
+    if not video_id:
+        state = read_state_snapshot()
+        video_id = str(state.get("broadcast_id") or state.get("last_broadcast_id") or "")
+    if not video_id:
+        raise ValueError("no broadcast id provided and no previous state file found")
+    print(json.dumps(video_metrics(video_id), indent=2, sort_keys=True))
+    return 0
+
+
 def reconcile_command(args: argparse.Namespace) -> int:
     stream = ensure_stream(create_if_missing=args.create_stream)
     stream_id = str(stream["id"])
@@ -1151,6 +1223,10 @@ def main() -> int:
     state_p = sub.add_parser("state", help="Show non-secret persisted lifecycle state")
     state_p.set_defaults(func=state_command)
 
+    video_metrics_p = sub.add_parser("video-metrics", help="Show read-only aggregate metrics for a live video")
+    video_metrics_p.add_argument("video_id", nargs="?")
+    video_metrics_p.set_defaults(func=video_metrics_command)
+
     stream_status_p = sub.add_parser("stream-status", help="Show reusable liveStream ingest status")
     stream_status_p.add_argument("stream_id", nargs="?")
     stream_status_p.set_defaults(func=stream_status_command)