Monitor YouTube video stats with Python

"""Record views, likes and comments of YouTube videos and print the growth since the last check."""
import csv
import os
import re
import time
from datetime import datetime, timezone
from pathlib import Path

import requests

API = "https://api.everydata.io"
KEY = os.environ["EVERYDATA_API_KEY"]
CSV_FILE = Path("video_stats.csv")
VIDEOS = ["kJQP7kiw5Fk"]  # video IDs or full YouTube links


def api_get(path: str, params: dict) -> dict:
    """GET an everydata.io endpoint. Waits when rate limited (429) and retries a failed page load (5xx) once."""
    retried = False
    for _ in range(4):
        resp = requests.get(f"{API}{path}", params=params, headers={"x-api-key": KEY}, timeout=60)
        if resp.status_code == 429:
            # Too many requests at once. The answer says how long to wait: "... Try again in 12 seconds."
            wait = re.search(r"in (\d+) seconds", resp.text)
            time.sleep(int(wait.group(1)) + 1 if wait else 60)
            continue
        if resp.status_code >= 500 and not retried:
            retried = True  # 5xx answers are not counted against your quota
            time.sleep(10)
            continue
        resp.raise_for_status()  # 4xx (wrong key, quota used up, bad parameter) or a second 5xx
        return resp.json()
    raise RuntimeError("still rate limited after several waits – try again later")


def video_details(video: str) -> dict:
    return api_get("/ytb/youtube-video-details", {"id": video})


def last_views() -> dict[str, int]:
    """Most recent view count per video from earlier runs."""
    if not CSV_FILE.exists():
        return {}
    with CSV_FILE.open(newline="") as f:
        return {r["video_id"]: int(r["views"]) for r in csv.DictReader(f) if r["views"]}


def main() -> None:
    before = last_views()
    now = datetime.now(timezone.utc).isoformat(timespec="seconds")
    new_file = not CSV_FILE.exists()

    with CSV_FILE.open("a", newline="") as f:
        writer = csv.writer(f)
        if new_file:
            writer.writerow(["checked_at", "video_id", "title", "views", "likes", "comments"])
        for video in VIDEOS:
            try:
                v = video_details(video)
            except (requests.RequestException, RuntimeError) as exc:
                print(f"{video}: could not check this time ({exc}) – the next run tries again")
                continue  # one failed video does not stop the others
            video_id = v.get("videoId") or video
            views = v.get("viewCount")
            writer.writerow([now, video_id, v.get("title"), views or "",
                             v.get("likeCount") or "", v.get("commentCount") or ""])
            if views is not None and video_id in before:
                print(f"{v.get('title')}: +{views - before[video_id]:,} views since the last check")


if __name__ == "__main__":
    main()

def latest_uploads(channel: str) -> list[str]:
    """IDs of the newest regular videos of a channel (channel ID or URL such as https://www.youtube.com/@WWE)."""
    data = api_get("/ytb/youtube-channel-videos", {"id": channel, "filter": "videos_latest"})
    return [item["videoId"] for item in data.get("items", []) if item.get("type") == "video"]


# In main(): watch the five newest uploads of a channel as well as your fixed list
# for video in VIDEOS + latest_uploads("https://www.youtube.com/@WWE")[:5]:

0 6 * * * cd /home/you/yt-monitor && EVERYDATA_API_KEY=ds_your_key /usr/bin/python3 monitor.py >> monitor.log 2>&1