Keep your YouTube videos and stats in Airtable

from clickruntime import on_trigger, http, credentials, ai, notify

YT = "https://www.googleapis.com/youtube/v3"

def yt(path, **params):
    return http.get(f"{YT}/{path}", params={**params, "key": credentials.secret("YOUTUBE_API_KEY")}).json().get("items", [])

AT = "https://api.airtable.com/v0/appXXXXXXXXXXXXXX"

def at(method, path, **kw):
    r = http.request(method, f"{AT}/{path}", headers={"Authorization": f"Bearer {credentials.connection('airtable', 'work')}"}, **kw)
    return r.raise_for_status() or r.json()

UPLOADS = "UUxxxxxxxxxxxxxxxxxxxxxx"

@on_trigger
def run(event):
    ids = [i["contentDetails"]["videoId"] for i in yt("playlistItems", part="contentDetails", playlistId=UPLOADS, maxResults=50)]
    records = []
    for v in yt("videos", part="snippet,statistics", id=",".join(ids)):
        s = v["statistics"]
        records.append({"fields": {"Video ID": v["id"], "Title": v["snippet"]["title"], "Published": v["snippet"]["publishedAt"][:10],
                        "Views": int(s.get("viewCount", 0)), "Likes": int(s.get("likeCount", 0)), "Comments": int(s.get("commentCount", 0))}})
    for i in range(0, len(records), 10):
        at("PATCH", "Videos", json={"performUpsert": {"fieldsToMergeOn": ["Video ID"]}, "records": records[i:i + 10]})
    try:
        top = sorted(records, key=lambda r: -r["fields"]["Views"])[:10]
        note = ai.generate("\n".join(f"{r['fields']['Title']}: {r['fields']['Views']} views" for r in top),
                           system="In 3 sentences, summarise how these videos are doing.", max_output_tokens=150).text
        notify.email_me("YouTube stats updated in Airtable", note)
    except ai.AIError:
        pass
    return {"synced": len(records)}