1"""FAA Releasable Airmen Certification Database actor.
2
3Downloads the FAA's monthly public airmen bulk file, joins basic records with
4certificate records, applies user filters, and pushes one dataset item per
5airman. The FAA files carry a header row, so parsing is header-driven rather
6than positional — resilient to column reordering between releases.
7
8The bulk file is large (~57 MB zipped, ~270 MB of CSV), so the zip is streamed
9to a temp file and the CSV members are read row-by-row. Basic-record filters
10(state, address) are applied during the first pass so that only matching airmen
11are held in memory for the certificate join.
12"""
13
14import csv
15import os
16import tempfile
17import zipfile
18from collections.abc import Iterator
19from datetime import datetime, timedelta, timezone
20
21import httpx
22from apify import Actor
23
24FAA_URL_TEMPLATE = "https://registry.faa.gov/database/CS{month:02d}{year}.zip"
25
26MONTHS_BACK = 4
27
28CHARGE_EVENT = "airman-record"
29
30
31
32HTTP_HEADERS = {
33 "User-Agent": (
34 "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 "
35 "(KHTML, like Gecko) Chrome/126.0.0.0 Safari/537.36"
36 ),
37 "Accept": "*/*",
38}
39
40
41def candidate_urls() -> list[str]:
42 urls = []
43 cursor = datetime.now(timezone.utc).replace(day=1)
44 for _ in range(MONTHS_BACK):
45 urls.append(FAA_URL_TEMPLATE.format(month=cursor.month, year=cursor.year))
46 cursor = (cursor - timedelta(days=1)).replace(day=1)
47 return urls
48
49
50def normalize_key(key: str) -> str:
51 return key.strip().upper().replace(" ", "_")
52
53
54def rating_sort_key(key: str, prefix: str) -> tuple[int, str]:
55 """Sort RATING1..RATING11 numerically, not lexicographically.
56
57 Plain string sorting orders RATING10 before RATING2, which scrambles the
58 FAA's ordering (the ratings are listed in a meaningful sequence).
59 """
60 suffix = key[len(prefix) :]
61 return (int(suffix), "") if suffix.isdigit() else (10**6, key)
62
63
64def member_names(zf: zipfile.ZipFile, stem: str) -> list[str]:
65 """Zip members whose basename starts with stem, e.g. PILOT_BASIC.
66
67 Matching is anchored to the start of the basename on purpose: a plain
68 substring test for "PILOT_BASIC" also matches "NONPILOT_BASIC.csv", which
69 silently folded non-pilots into every run.
70 """
71 names = []
72 for member in zf.namelist():
73 base = member.rsplit("/", 1)[-1].upper()
74 if base.startswith(stem) and base.endswith((".CSV", ".TXT")):
75 names.append(member)
76 return names
77
78
79def iter_csv_members(zf: zipfile.ZipFile, stem: str) -> Iterator[dict]:
80 """Yield normalized-key row dicts from every CSV member matching stem."""
81 for member in member_names(zf, stem):
82 with zf.open(member) as fh:
83
84 text = (line.decode("utf-8", errors="replace") for line in fh)
85 reader = csv.DictReader(text)
86 for row in reader:
87 yield {
88 normalize_key(k): (v or "").strip()
89 for k, v in row.items()
90 if k
91 }
92
93
94def basic_to_record(b: dict, source_file: str) -> dict | None:
95 uid = b.get("UNIQUE_ID") or b.get("UNIQUE_ID_NUMBER")
96 if not uid:
97 return None
98 return {
99 "unique_id": uid,
100 "first_name": b.get("FIRST_NAME", ""),
101 "last_name": b.get("LAST_NAME", ""),
102 "street_1": b.get("STREET_1", ""),
103 "street_2": b.get("STREET_2", ""),
104 "city": b.get("CITY", ""),
105 "state": b.get("STATE", ""),
106 "zip_code": b.get("ZIP_CODE", ""),
107 "country": b.get("COUNTRY", ""),
108 "region": b.get("REGION", ""),
109 "medical_class": b.get("MED_CLASS", ""),
110 "medical_date": b.get("MED_DATE", ""),
111 "medical_expire_date": b.get("MED_EXP_DATE", ""),
112 "basic_med_course_date": b.get("BASIC_MED_COURSE_DATE", ""),
113 "basic_med_cmec_date": b.get("BASIC_MED_CMEC_DATE", ""),
114 "certificates": [],
115 "source_file": source_file,
116 }
117
118
119def cert_from_row(c: dict) -> dict:
120 ratings = [
121 v
122 for k, v in sorted(
123 c.items(), key=lambda kv: rating_sort_key(kv[0], "RATING")
124 )
125 if k.startswith("RATING") and v
126 ]
127 type_ratings = [
128 v
129 for k, v in sorted(
130 c.items(), key=lambda kv: rating_sort_key(kv[0], "TYPERATING")
131 )
132 if k.startswith("TYPERATING") and v
133 ]
134 return {
135 "type": c.get("TYPE", ""),
136 "level": c.get("LEVEL", ""),
137 "expire_date": c.get("EXPIRE_DATE", ""),
138 "ratings": ratings,
139 "type_ratings": type_ratings,
140 }
141
142
143def basic_filters_ok(record: dict, states: set[str], require_address: bool) -> bool:
144 """Filters that depend only on the basic record, applied before the join."""
145 if states and record["state"].upper() not in states:
146 return False
147 if require_address and not record["street_1"]:
148 return False
149 return True
150
151
152def cert_filters_ok(record: dict, cert_types: set[str], cert_levels: set[str]) -> bool:
153 if not cert_types and not cert_levels:
154 return True
155 for cert in record["certificates"]:
156 type_ok = not cert_types or cert["type"].upper() in cert_types
157 level_ok = not cert_levels or cert["level"].upper() in cert_levels
158 if type_ok and level_ok:
159 return True
160 return False
161
162
163async def charge_safely(count: int) -> None:
164 """Charge PPE events; a no-op when the actor runs without PPE configured."""
165 try:
166 await Actor.charge(CHARGE_EVENT, count)
167 except Exception as exc:
168 Actor.log.debug(f"PPE charge skipped: {exc}")
169
170
171async def deliver_batch(batch: list[dict]) -> tuple[int, bool]:
172 """Push one batch and charge for it, honoring the buyer's max-charge limit.
173
174 Records beyond the limit are not pushed at all — delivering them free would
175 respect the letter of the cap while giving the rest of the run away.
176 Returns (records delivered, limit reached).
177 """
178 limit_hit = False
179 try:
180 remaining = Actor.get_charging_manager(
181 ).calculate_max_event_charge_count_within_limit(CHARGE_EVENT)
182 except Exception:
183 remaining = None
184 if remaining is not None and remaining < len(batch):
185 batch = batch[:remaining]
186 limit_hit = True
187 if not batch:
188 return 0, True
189 await Actor.push_data(batch)
190 await charge_safely(len(batch))
191 return len(batch), limit_hit
192
193
194async def try_download(client: httpx.AsyncClient, url: str, path: str) -> bool:
195 """Stream one URL to path. False means 'not available, move on'."""
196 async with client.stream("GET", url) as resp:
197 if resp.status_code != 200:
198 Actor.log.warning(f"HTTP {resp.status_code} at {url}.")
199 return False
200 expected = int(resp.headers.get("content-length") or 0)
201 total = 0
202 with open(path, "wb") as fh:
203 async for chunk in resp.aiter_bytes(1 << 20):
204 if total == 0 and chunk[:2] != b"PK":
205 Actor.log.warning(
206 f"Response at {url} is not a zip (starts {chunk[:8]!r})."
207 )
208 return False
209 fh.write(chunk)
210 total += len(chunk)
211 if expected and total != expected:
212
213
214 raise httpx.HTTPError(f"truncated: got {total} of {expected} bytes")
215 Actor.log.info(f"Downloaded {total / 1_000_000:.1f} MB from {url}")
216 return True
217
218
219async def download_bulk_file(urls: list[str], path: str, attempts: int = 3) -> str:
220 """Stream the first URL that yields a complete zip to path; return that URL."""
221 async with httpx.AsyncClient(
222 timeout=httpx.Timeout(600.0, connect=60.0),
223 follow_redirects=True,
224 headers=HTTP_HEADERS,
225 ) as client:
226 for url in urls:
227 Actor.log.info(f"Trying FAA bulk file: {url}")
228 for attempt in range(1, attempts + 1):
229 try:
230 if await try_download(client, url, path):
231 return url
232 break
233 except httpx.HTTPError as exc:
234 Actor.log.warning(
235 f"Attempt {attempt}/{attempts} failed for {url}: {exc}"
236 )
237 if attempt == attempts:
238 Actor.log.warning(
239 f"Giving up on {url} after {attempts} attempts; "
240 "falling back to the previous month."
241 )
242
243 raise RuntimeError(
244 "Could not download the FAA airmen bulk file from any candidate URL. "
245 "Set the 'downloadUrl' input to the current file listed at "
246 "https://www.faa.gov/licenses_certificates/airmen_certification/"
247 "releasable_airmen_download"
248 )
249
250
251async def main() -> None:
252 async with Actor:
253 actor_input = await Actor.get_input() or {}
254 states = {s.strip().upper() for s in actor_input.get("states", []) if s.strip()}
255 cert_types = {
256 s.strip().upper() for s in actor_input.get("certificateTypes", []) if s.strip()
257 }
258 cert_levels = {
259 s.strip().upper() for s in actor_input.get("certificateLevels", []) if s.strip()
260 }
261 include_non_pilots = actor_input.get("includeNonPilots", False)
262 require_address = actor_input.get("requireAddress", False)
263 max_records = actor_input.get("maxRecords", 10000) or 0
264 override_url = (actor_input.get("downloadUrl") or "").strip()
265
266 urls = [override_url] if override_url else candidate_urls()
267
268 tmp_dir = tempfile.mkdtemp(prefix="faa-airmen-")
269 zip_path = os.path.join(tmp_dir, "airmen.zip")
270 try:
271 source_url = await download_bulk_file(urls, zip_path)
272
273 with zipfile.ZipFile(zip_path) as zf:
274 Actor.log.info(f"Zip members: {zf.namelist()}")
275
276 basic_stems = ["PILOT_BASIC"]
277 cert_stems = ["PILOT_CERT"]
278 if include_non_pilots:
279 basic_stems.append("NONPILOT_BASIC")
280 cert_stems.append("NONPILOT_CERT")
281
282 for stem in basic_stems + cert_stems:
283 if not member_names(zf, stem):
284 raise RuntimeError(
285 f"No {stem}* member in the downloaded zip "
286 f"({zf.namelist()}). The FAA file layout may have "
287 "changed."
288 )
289
290
291
292 by_id: dict[str, dict] = {}
293 scanned = 0
294 for stem in basic_stems:
295 for row in iter_csv_members(zf, stem):
296 scanned += 1
297 record = basic_to_record(row, source_url)
298 if record is None:
299 continue
300 if not basic_filters_ok(record, states, require_address):
301 continue
302
303
304 by_id.setdefault(record["unique_id"], record)
305 Actor.log.info(
306 f"Scanned {scanned:,} basic rows; {len(by_id):,} match the "
307 "state/address filters."
308 )
309
310
311 cert_rows = 0
312 attached = 0
313 for stem in cert_stems:
314 for row in iter_csv_members(zf, stem):
315 cert_rows += 1
316 uid = row.get("UNIQUE_ID") or row.get("UNIQUE_ID_NUMBER")
317 rec = by_id.get(uid)
318 if rec is None:
319 continue
320 rec["certificates"].append(cert_from_row(row))
321 attached += 1
322 Actor.log.info(
323 f"Scanned {cert_rows:,} certificate rows; attached {attached:,}."
324 )
325 finally:
326 if os.path.exists(zip_path):
327 os.remove(zip_path)
328 os.rmdir(tmp_dir)
329
330 pushed = 0
331 limit_hit = False
332 batch: list[dict] = []
333 for record in by_id.values():
334 if not cert_filters_ok(record, cert_types, cert_levels):
335 continue
336 batch.append(record)
337 if len(batch) >= 500:
338 delivered, limit_hit = await deliver_batch(batch)
339 pushed += delivered
340 batch = []
341 if limit_hit or (max_records and pushed + len(batch) >= max_records):
342 break
343
344 if batch and not limit_hit:
345 delivered, limit_hit = await deliver_batch(batch)
346 pushed += delivered
347
348 if limit_hit:
349 Actor.log.warning(
350 "The run's maximum charge was reached; stopped delivering early. "
351 "Raise the run's charge limit to receive the remaining records."
352 )
353 Actor.log.info(f"Done. Pushed {pushed} airman records.")
