Collect new YouTube videos on your topics into Airtable

from datetime import datetime, timezone, timedelta
from clickruntime import on_trigger, http, credentials, ai, state

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()

TOPICS = ["warehouse automation", "last-mile delivery"]

@on_trigger
def run(event):
    since = (datetime.now(timezone.utc) - timedelta(days=1)).strftime("%Y-%m-%dT%H:%M:%SZ")
    seen, records = set(state.get("seen") or []), []
    for topic in TOPICS:
        for v in yt("search", part="snippet", q=topic, type="video", publishedAfter=since, maxResults=25, order="relevance"):
            vid, s = v["id"]["videoId"], v["snippet"]
            if vid in seen:
                continue
            summary = ai.generate(f"{s['title']}\n{s['description']}", system="Summarise this video in 2 sentences.", max_output_tokens=100).text.strip()
            records.append({"fields": {"Topic": topic, "Title": s["title"], "Channel": s["channelTitle"],
                            "Link": f"https://youtu.be/{vid}", "Summary": summary}})
            seen.add(vid)
    for i in range(0, len(records), 10):
        at("POST", "Research", json={"records": records[i:i + 10], "typecast": True})
    state.set("seen", sorted(seen)[-5000:])
    return {"saved": len(records)}