YouTube API: get all videos from a channel

import csv
import os
import re
import sys
import time

import requests

BASE = "https://api.toolzerhub.com"
HEADERS = {"x-api-key": os.environ.get("TOOLZERHUB_API_KEY", "")}
FIELDS = ["channel", "video_id", "url", "title", "published", "views", "duration", "members_only"]
ROUTES = {"videos": "videos", "shorts": "shorts", "live": "live-streams"}
calls = 0
FAILED = object()  # a 400, a 5xx or a 429/503 that outlasted the retries: different from "empty"


def cell(value):
    # Spreadsheets run text that starts with = + - or @ as a formula; an apostrophe keeps it text.
    if isinstance(value, str) and value[:1] in ("=", "+", "-", "@", "\t", "\r"):
        return "'" + value
    return value


def channel_param(ref):
    # An id (UC...), a /channel/UC... URL, an @handle or a youtube.com/@handle URL.
    m = re.search(r"(UC[\w-]{22})", ref)
    if m and (ref.startswith("UC") or "/channel/" in ref):
        return {"channel_id": m.group(1)}
    m = re.search(r"(@[\w.\-]+)", ref)
    return {"channel_username": m.group(1) if m else ref}


def get(route, params, tries=4):
    # 429 and 503 are retryable; the wait comes from the error body or the Retry-After header.
    global calls
    for attempt in range(tries):
        r = requests.get(f"{BASE}/v2/youtube/channel/{route}", headers=HEADERS, params=params, timeout=90)
        calls += 1
        if r.status_code == 400:  # bad input: skip this channel, keep the rest of the run
            print("bad request:", r.json()["error"]["message"].splitlines()[0])
            return FAILED
        if r.status_code not in (429, 503) or attempt == tries - 1:
            if r.status_code in (429, 503) or r.status_code >= 500:  # give up on this channel
                return FAILED
            if r.status_code >= 400:
                sys.exit(f"HTTP {r.status_code}: {r.text[:200]}")
            return r.json().get("data")  # None when there is nothing to return
        try:
            wait = r.json()["error"]["details"]["retryAfter"]
        except (ValueError, KeyError, TypeError):
            wait = None
        try:
            wait = int(wait or r.headers.get("Retry-After") or 5 * (attempt + 1))
        except ValueError:
            wait = 5 * (attempt + 1)
        time.sleep(min(wait, 60))


def export_channel(writer, ref, max_pages, sort, kind, seen):
    route = ROUTES[kind]
    params = {**channel_param(ref), "sort_order": sort}
    pages, rows, more, failed, stated, name = 0, 0, False, False, None, ref
    while pages < max_pages:
        data = get(route, params)
        if data is FAILED:
            failed = True
            break
        if not data or not data.get("data"):
            break  # null data or an empty list
        pages += 1
        meta = data.get("meta") or {}
        name = meta.get("title") or name
        stated = stated or meta.get("videosCountText")  # the channel block comes with sort newest only
        for v in data["data"]:
            vid = v.get("videoId")
            if not vid or vid in seen:  # keep one row per video id
                continue
            seen.add(vid)
            rows += 1
            writer.writerow({
                "channel": cell(name), "video_id": cell(vid),  # some IDs start with "-"
                "url": f"https://www.youtube.com/{'shorts/' if kind == 'shorts' else 'watch?v='}{vid}",
                "title": cell((v.get("title") or "").replace("\n", " ")),
                "published": v.get("publishDate") or "",  # as returned: older dates are approximate
                "views": v.get("viewCount") if v.get("viewCount") is not None else v.get("viewCountText") or "",
                "duration": v.get("lengthText") or "",
                "members_only": "yes" if v.get("isMemberOnly") else "",  # listed, but no view count
            })
        more = bool(data.get("page_cursor"))
        if not more:
            break
        params["page_cursor"] = data["page_cursor"]
        time.sleep(0.4)  # the free plan allows 3 requests a second
    if failed:
        status = " (gave up: bad request, server error or repeated 429/503)"
    elif not pages:
        status = " (nothing returned: no items on this tab)"
    elif more:
        status = " (stopped at the page cap, more available)"
    else:
        extra = f"; YouTube shows {stated} in all, Shorts and live streams included" if stated and kind == "videos" else ""
        status = f" (reached the end{extra})"
    print(f"{ref}: {rows} {kind}, {pages} pages{status}")
    return rows


def main(channels, max_pages="50", sort="newest", kind="videos", out="channel_videos.csv"):
    total, seen = 0, set()
    with open(out, "w", newline="", encoding="utf-8") as f:
        writer = csv.DictWriter(f, fieldnames=FIELDS)
        writer.writeheader()
        for ref in [c.strip() for c in channels.split(",") if c.strip()]:
            total += export_channel(writer, ref, int(max_pages), sort, kind, seen)
    print(f"wrote {total} rows to {out} in {calls} calls")


if __name__ == "__main__":
    args = sys.argv[1:]
    if not args or len(args) > 5 or (len(args) > 2 and args[2] not in ("newest", "oldest", "popular")) \
            or (len(args) > 3 and args[3] not in ROUTES):
        sys.exit("usage: python youtube_channel_videos.py @handle_or_url_or_id[,...] [max_pages] "
                 "[newest|oldest|popular] [videos|shorts|live] [file.csv]")
    if not HEADERS["x-api-key"]:
        sys.exit("set TOOLZERHUB_API_KEY first")
    main(*args)

export TOOLZERHUB_API_KEY=your_key
python youtube_channel_videos.py "@Apple,https://www.youtube.com/@SpaceX,UCLA_DiR1FfKNvjuUpBHmylQ" 8 newest videos channel_videos.csv

@Apple: 108 videos, 4 pages (reached the end; YouTube shows 221 videos in all, Shorts and live streams included)
https://www.youtube.com/@SpaceX: 214 videos, 8 pages (reached the end; YouTube shows 533 videos in all, Shorts and live streams included)
UCLA_DiR1FfKNvjuUpBHmylQ: 240 videos, 8 pages (stopped at the page cap, more available)
wrote 562 rows to channel_videos.csv