#!/usr/bin/env python3 """Phase 0 verification: everything the plan assumes about the YouTube Data API. The plan (§13 Phase 0) rests on three answers that can only come from a live key. This runs all three, costs about 5 quota units of the 10,000/day budget, and prints the numbers to paste back into plan.md. python3 tools/verify_api.py --key AIza... Or, once the key is in the settings table: python3 tools/verify_api.py --key "$(sqlite3 /var/lib/ytstream/ytstream.db \ "select value from setting where key='youtube_api_key'")" Exit status is 0 only if all three checks pass, so this is safe to gate on. """ from __future__ import annotations import argparse import json import re import sys import urllib.error import urllib.parse import urllib.request API = "https://www.googleapis.com/youtube/v3" # The account being mirrored — @cflux1030, resolved via yt-dlp on 2026-08-12. BROTHER = "UCPcTWaLV8zwx4WP4QExHj4Q" # A channel with a known-large back catalogue, used to exercise pagination. SAMPLE_CHANNEL = "UCjCJ2LaOIsPzOoXUTMDI3wg" # Pitch Side ISO8601 = re.compile( r"^P(?:(\d+)D)?T?(?:(\d+)H)?(?:(\d+)M)?(?:(\d+)S)?$" ) class ApiError(Exception): def __init__(self, status, reason, body): super().__init__(f"HTTP {status} {reason}") self.status = status self.reason = reason self.body = body def call(endpoint: str, key: str, **params) -> dict: """One API call. Raises ApiError with the parsed reason on 4xx/5xx.""" params["key"] = key url = f"{API}/{endpoint}?" + urllib.parse.urlencode(params) try: with urllib.request.urlopen(url, timeout=30) as response: return json.load(response) except urllib.error.HTTPError as exc: raw = exc.read().decode("utf8", "replace") reason = "" try: errors = json.loads(raw).get("error", {}).get("errors", []) reason = errors[0].get("reason", "") if errors else "" except ValueError: pass raise ApiError(exc.code, reason, raw[:400]) from exc def iso8601_seconds(text: str) -> int | None: """PT1H2M3S -> 3723. Returns None for anything unparseable.""" match = ISO8601.match(text or "") if not match: return None days, hours, minutes, seconds = (int(g or 0) for g in match.groups()) return days * 86400 + hours * 3600 + minutes * 60 + seconds # -------------------------------------------------------------------------- results: list[tuple[str, bool, str]] = [] def record(name: str, ok: bool, detail: str) -> None: print(f" {'PASS' if ok else 'FAIL'} {name}") for line in detail.splitlines(): print(f" {line}") results.append((name, ok, detail)) def check_subscriptions(key: str, channel_id: str) -> None: """The one that decides whether the whole feature is possible.""" print("\n1. subscriptions.list on the mirrored account") try: page = call("subscriptions", key, part="snippet", channelId=channel_id, maxResults=50) except ApiError as exc: if exc.status == 403 and exc.reason == "subscriptionForbidden": record("subscriptions readable", False, "403 subscriptionForbidden -- subscriptions are still PRIVATE.\n" "Fix: youtube.com -> Settings -> Privacy -> uncheck\n" '"Keep all my subscriptions private". There is no workaround;\n' "the public HTML route no longer exists (plan.md §15).") else: record("subscriptions readable", False, f"{exc} reason={exc.reason!r}\n{exc.body}") return total = (page.get("pageInfo") or {}).get("totalResults") items = page.get("items") or [] titles = [i["snippet"]["title"] for i in items] # Paginate so the recorded total is the real one, not just page 1. seen = len(items) token = page.get("nextPageToken") pages = 1 while token and pages < 20: page = call("subscriptions", key, part="snippet", channelId=channel_id, maxResults=50, pageToken=token) batch = page.get("items") or [] seen += len(batch) titles.extend(i["snippet"]["title"] for i in batch) token = page.get("nextPageToken") pages += 1 suggested = max(10, -(-int(total or seen) // 5)) # ceil(total * 0.2) record("subscriptions readable", True, f"totalResults = {total}, fetched {seen} across {pages} page(s)\n" f"-> set subsync_max_new = {suggested} [max(10, ceil(total*0.2))]\n" f"first few: {', '.join(titles[:8])}" + (" ..." if len(titles) > 8 else "")) def check_uploads_playlist(key: str, channel_id: str) -> None: """Does playlistItems.list accept the undocumented UULF playlist id?""" print("\n2. playlistItems.list on UULF (long-form-only) vs UU (documented)") outcome = {} for kind, playlist_id in (("UULF", "UULF" + channel_id[2:]), ("UU", "UU" + channel_id[2:])): try: page = call("playlistItems", key, part="contentDetails", playlistId=playlist_id, maxResults=5) except ApiError as exc: outcome[kind] = (False, f"{exc} reason={exc.reason!r}") continue items = page.get("items") or [] stamps = [i["contentDetails"].get("videoPublishedAt") for i in items] exact = all(s and s.endswith("Z") and "T" in s for s in stamps) outcome[kind] = ( bool(items) and exact, f"{len(items)} items, videoPublishedAt exact={exact}, " f"e.g. {stamps[0] if stamps else 'n/a'}, " f"total={(page.get('pageInfo') or {}).get('totalResults')}", ) uulf_ok, uulf_detail = outcome["UULF"] uu_ok, uu_detail = outcome["UU"] if uulf_ok: record("uploads playlist usable", True, f"UULF WORKS -> use it, existing Shorts/livestream filtering carries over\n" f" UULF: {uulf_detail}\n UU: {uu_detail}") elif uu_ok: record("uploads playlist usable", True, f"UULF REJECTED -> take the UU fallback and filter by duration +\n" f"liveStreamingDetails (plan.md §3)\n" f" UULF: {uulf_detail}\n UU: {uu_detail}") else: record("uploads playlist usable", False, f"neither worked\n UULF: {uulf_detail}\n UU: {uu_detail}") def check_durations(key: str, channel_id: str) -> None: """Durations for a batch of ids -- what feeds .""" print("\n3. videos.list durations for a batch of ids") try: listing = call("playlistItems", key, part="contentDetails", playlistId="UU" + channel_id[2:], maxResults=50) except ApiError as exc: record("durations available", False, f"could not list ids: {exc}") return ids = [i["contentDetails"]["videoId"] for i in listing.get("items") or []] if not ids: record("durations available", False, "no video ids to test with") return try: page = call("videos", key, part="contentDetails", id=",".join(ids), maxResults=50) except ApiError as exc: record("durations available", False, f"{exc} reason={exc.reason!r}") return items = page.get("items") or [] parsed = [(i["id"], iso8601_seconds(i["contentDetails"].get("duration", ""))) for i in items] bad = [vid for vid, secs in parsed if secs is None] shorts = [vid for vid, secs in parsed if secs is not None and secs <= 120] record("durations available", not bad and len(items) == len(ids), f"asked for {len(ids)} ids in 1 call, got {len(items)} back, " f"{len(bad)} unparseable\n" f"{len(shorts)} of {len(items)} are <=120s (would be filtered as Shorts)\n" f"sample: " + ", ".join(f"{v}={s}s" for v, s in parsed[:5])) def main() -> int: parser = argparse.ArgumentParser() parser.add_argument("--key", required=True, help="YouTube Data API v3 key") parser.add_argument("--brother", default=BROTHER, help=f"channel id whose subscriptions to read (default {BROTHER})") parser.add_argument("--sample", default=SAMPLE_CHANNEL, help="channel id to exercise the playlist/duration calls against") args = parser.parse_args() print(f"Phase 0 verification against the live API (~5 quota units of 10,000/day)") check_subscriptions(args.key, args.brother) check_uploads_playlist(args.key, args.sample) check_durations(args.key, args.sample) failed = [name for name, ok, _ in results if not ok] print() if failed: print(f"{len(failed)} of {len(results)} FAILED: {', '.join(failed)}") return 1 print(f"ALL {len(results)} PASS -- paste the numbers above into plan.md §4.1 and §14") return 0 if __name__ == "__main__": sys.exit(main())