Merge daily CSV exports in Drive into one clean master file

import csv, io
from datetime import date
from google.oauth2.credentials import Credentials
from googleapiclient.discovery import build
from googleapiclient.http import MediaInMemoryUpload
from clickruntime import on_trigger, credentials, notify
FOLDER_ID = "your-exports-folder-id"
KEY = "order_id"

@on_trigger
def run(event):
    drive = build("drive", "v3", credentials=Credentials(credentials.connection("google", "work")), cache_discovery=False)
    q = f"'{FOLDER_ID}' in parents and mimeType='text/csv' and createdTime > '{date.today().isoformat()}T00:00:00'"
    files = drive.files().list(q=q, fields="files(id,name)").execute().get("files", [])
    seen, good, bad, counts = set(), [], [], []
    for f in files:
        rows = list(csv.DictReader(io.StringIO(drive.files().get_media(fileId=f["id"]).execute().decode())))
        counts.append(f"{f['name']}: {len(rows)}")
        for r in rows:
            if not r.get(KEY):
                bad.append(r)
            elif r[KEY] not in seen:
                seen.add(r[KEY]); good.append(r)
    for name, rows in ((f"master-{date.today()}.csv", good), (f"quarantine-{date.today()}.csv", bad)):
        if rows:
            buf = io.StringIO(); w = csv.DictWriter(buf, fieldnames=list(rows[0])); w.writeheader(); w.writerows(rows)
            drive.files().create(body={"name": name, "parents": [FOLDER_ID]},
                                 media_body=MediaInMemoryUpload(buf.getvalue().encode(), mimetype="text/csv")).execute()
    notify.slack(f"CSV recap: {', '.join(counts) or 'no files'}. Master {len(good)} rows, {len(bad)} set aside.", channel="#ops")
    return {"files": len(files), "rows": len(good), "set_aside": len(bad)}