File size: 2,099 Bytes
f734d80 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 | """
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)
|