ytstream/tools/verify_api.py

#!/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, message="", activation_url=""):
        super().__init__(f"HTTP {status} {reason}")
        self.status = status
        self.reason = reason
        self.body = body
        self.message = message
        self.activation_url = activation_url


class ServiceDisabled(ApiError):
    """YouTube Data API v3 is not enabled on the key's project.

    Signalled by `accessNotConfigured` in errors[] or `SERVICE_DISABLED` in
    details[], and carries an activationUrl naming the project number.
    """


class KeyRestricted(ApiError):
    """The API is enabled, but this key's API restrictions exclude it.

    Signalled by `API_KEY_SERVICE_BLOCKED` in details[] with the message
    "Requests to this API youtube method ... are blocked". Distinct from
    ServiceDisabled and fixed in a completely different console screen, which is
    why the two are separate types.

    Both were observed from one key on 2026-08-12, in this order, and the
    ordering is the trap: YouTube Data API v3 does not appear in the key's API
    restriction picker until the API is enabled on the project. Enable first,
    restrict second, or the key is created blocking the very API it is for.
    """


def call(endpoint: str, key: str, **params) -> dict:
    """One API call. Raises ApiError (or ServiceDisabled) 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 = message = activation = ""
        detail_reasons = set()
        try:
            error = json.loads(raw).get("error", {})
            message = error.get("message", "")
            errors = error.get("errors", [])
            reason = errors[0].get("reason", "") if errors else ""
            for detail in error.get("details", []):
                meta = detail.get("metadata") or {}
                if meta.get("activationUrl"):
                    activation = meta["activationUrl"]
                if detail.get("reason"):
                    detail_reasons.add(detail["reason"])
        except ValueError:
            pass

        # `forbidden` is used for both of these, so the details[] reason is what
        # actually distinguishes them. Check the specific ones before it.
        if "API_KEY_SERVICE_BLOCKED" in detail_reasons:
            cls = KeyRestricted
        elif ("SERVICE_DISABLED" in detail_reasons
              or reason == "accessNotConfigured"
              or "has not been used in project" in message):
            cls = ServiceDisabled
        else:
            cls = ApiError
        raise cls(exc.code, reason or next(iter(detail_reasons), ""),
                  raw[:400], message, activation) 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 <durationinseconds>."""
    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 preflight(key: str) -> None:
    """One cheap call to separate "project not set up" from "assumption wrong".

    Without this, a disabled API fails all three checks with three different
    messages and none of them says what to do about it.
    """
    try:
        call("videos", key, part="id", id="dQw4w9WgXcQ")
    except ServiceDisabled as exc:
        print("\nSTOP: the key is valid, but YouTube Data API v3 is not enabled "
              "on its project.")
        print(f"\n  {exc.message}\n")
        if exc.activation_url:
            print("  Enable it here, then wait ~2 minutes and re-run:")
            print(f"    {exc.activation_url}\n")
        print("  Creating an API key and enabling the API are separate steps in "
              "the console;\n  only the first one has been done.")
        raise SystemExit(2)
    except KeyRestricted as exc:
        project = ""
        match = re.search(r"projects?/(\d+)", exc.body) or \
            re.search(r"project=(\d+)", exc.body)
        if match:
            project = f"?project={match.group(1)}"
        print("\nSTOP: the API is enabled, but this key's API restrictions block "
              "it (API_KEY_SERVICE_BLOCKED).")
        print(f"\n  {exc.message}\n")
        print("  Fix: Credentials -> click the key -> API restrictions -> either")
        print('  "Don\'t restrict key", or tick "YouTube Data API v3" in the list.')
        print(f"    https://console.cloud.google.com/apis/credentials{project}\n")
        print("  Note the ordering trap: YouTube Data API v3 is absent from that")
        print("  picker until the API is enabled on the project, so a key created")
        print("  and restricted first ends up blocking the API it was made for.")
        raise SystemExit(2)
    except ApiError as exc:
        if exc.status in (400, 403) and exc.reason in ("badRequest", "keyInvalid",
                                                       "API_KEY_INVALID"):
            print(f"\nSTOP: the key was rejected -- {exc.reason}: {exc.message}")
            raise SystemExit(2)
        # Anything else is worth letting the real checks characterise.
        print(f"\n  (preflight warning: {exc} reason={exc.reason!r})")


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)")
    preflight(args.key)
    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())