Screen CVs from email into an Airtable candidate list

from googleapiclient.discovery import build
from google.oauth2.credentials import Credentials
from clickruntime import on_trigger, credentials, state, ai, report_fetched, http

ROLE = "Shift manager: 2+ years in hospitality, team leadership, weekend availability."
SCHEMA = {"type": "object", "properties": {k: {"type": "string"} for k in ["Name", "Email", "Phone", "Experience", "Score"]}}

@on_trigger
def run(event):
    gmail = build("gmail", "v1", credentials=Credentials(credentials.connection("google", "work")), cache_discovery=False).users()
    seen = state.get("seen") or []
    msgs = [m for m in gmail.messages().list(userId="me", q="label:applications has:attachment newer_than:2d").execute().get("messages", [])
            if m["id"] not in seen]
    report_fetched(len(msgs))
    if not msgs:
        return {"summary": "No new applications"}
    records = []
    for m in msgs:
        msg = gmail.messages().get(userId="me", id=m["id"]).execute()
        part = next(p for p in msg["payload"]["parts"] if p.get("filename"))
        cv = gmail.messages().attachments().get(userId="me", messageId=m["id"], id=part["body"]["attachmentId"]).execute()["data"]
        r = ai.generate(f"Role: {ROLE}\nEmail: {msg['snippet']}\nCV (base64 text): {cv[:6000]}\nExtract fields and Score 1-10.", json_schema=SCHEMA, max_output_tokens=200)
        records.append({"fields": r.json})
        seen.append(m["id"])
    http.post("https://api.airtable.com/v0/appXXXXXXXX/Candidates", headers={"Authorization": f"Bearer {credentials.secret('AIRTABLE_API_KEY')}"},
              json={"records": records[:10]})
    state.set("seen", seen[-500:])
    return {"added": len(records)}