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