""" airtable.py — email gate logging to Airtable. Required env vars: AIRTABLE_BASE_ID — e.g. "appXXXXXXXXXXXXXX" AIRTABLE_API_KEY — personal access token ("pat...") Table name is hardcoded as "RoboGen Leads". Change TABLE_NAME below if needed. Failures are swallowed — caller receives a (success: bool, message: str) tuple and must allow the download even if logging fails. """ import os import json import datetime from typing import Tuple, Optional try: import requests _REQUESTS_OK = True except ImportError: _REQUESTS_OK = False TABLE_NAME = "RoboGen Leads" _BASE_ID = os.environ.get("AIRTABLE_BASE_ID", "") _API_KEY = os.environ.get("AIRTABLE_API_KEY", "") def log_email( email: str, robot: str, task: str, n_episodes: int, quality_score: float, band: str, ) -> Tuple[bool, str]: """ POST one record to Airtable. Returns (True, "Logged") on success, (False, reason) on failure. Caller MUST still enable the download on failure. """ if not _REQUESTS_OK: return False, "requests library not installed" if not _BASE_ID or not _API_KEY: return False, "AIRTABLE_BASE_ID / AIRTABLE_API_KEY not set" endpoint = f"https://api.airtable.com/v0/{_BASE_ID}/{TABLE_NAME.replace(' ', '%20')}" headers = { "Authorization": f"Bearer {_API_KEY}", "Content-Type": "application/json", } payload = { "fields": { "Email": email.strip(), "Robot": robot, "Task": task, "Episodes": n_episodes, "Quality Score": quality_score, "Band": band, "Timestamp": datetime.datetime.utcnow().isoformat() + "Z", } } try: resp = requests.post(endpoint, headers=headers, json=payload, timeout=6) if resp.status_code in (200, 201): return True, "Logged" return False, f"HTTP {resp.status_code}: {resp.text[:200]}" except Exception as exc: return False, str(exc)