1"""FAA aircraft registry feed actor: new registrations and ownership changes.
2
3The FAA refreshes its Releasable Aircraft Registration Database nightly
4(11:30 pm Central) as a ~73 MB zip. The MASTER file's CERT ISSUE DATE is the
5date the current Certificate of Aircraft Registration was issued: it resets
6when an aircraft is registered or re-registered to a new owner, and does NOT
7move on renewals or address changes. Records whose certificate was issued in
8the lookback window are therefore exactly the new registrations plus ownership
9changes — the feed brokers, insurers, FBOs and MRO sales teams want.
10
11Keyed and deduped on (N-number, cert issue date) across runs via a named
12key-value store. Aircraft make/model/seat data is joined from ACFTREF.
13
14The zip is streamed to a temp file; MASTER (~194 MB of CSV) is read
15row-by-row so memory scales with the selection, not the file.
16"""
17
18import csv
19import io
20import os
21import tempfile
22import zipfile
23from datetime import datetime, timedelta, timezone
24
25import httpx
26from apify import Actor
27
28FAA_ZIP_URL = "https://registry.faa.gov/database/ReleasableAircraft.zip"
29
30CHARGE_EVENT = "aircraft-record"
31SEEN_STORE_NAME = "faa-aircraft-seen"
32SEEN_KEY = "cert-keys"
33
34
35HTTP_HEADERS = {
36 "User-Agent": (
37 "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 "
38 "(KHTML, like Gecko) Chrome/126.0.0.0 Safari/537.36"
39 ),
40 "Accept": "*/*",
41}
42
43REGISTRANT_TYPES = {
44 "1": "Individual",
45 "2": "Partnership",
46 "3": "Corporation",
47 "4": "Co-owned",
48 "5": "Government",
49 "7": "LLC",
50 "8": "Non-citizen corporation",
51 "9": "Non-citizen co-owned",
52}
53
54AIRCRAFT_TYPES = {
55 "1": "Glider",
56 "2": "Balloon",
57 "3": "Blimp/dirigible",
58 "4": "Fixed wing single engine",
59 "5": "Fixed wing multi engine",
60 "6": "Rotorcraft",
61 "7": "Weight-shift-control",
62 "8": "Powered parachute",
63 "9": "Gyroplane",
64 "H": "Hybrid lift",
65 "O": "Other",
66}
67
68
69def parse_date(value: str) -> str:
70 """YYYYMMDD -> YYYY-MM-DD; blanks stay ''."""
71 value = value.strip()
72 if not value:
73 return ""
74 try:
75 return datetime.strptime(value, "%Y%m%d").date().isoformat()
76 except ValueError:
77 return value
78
79
80def csv_rows(zf: zipfile.ZipFile, member: str):
81 """Yield stripped-value dict rows; the FAA pads fields and adds a BOM."""
82 with zf.open(member) as fh:
83 text = io.TextIOWrapper(fh, encoding="utf-8-sig", errors="replace")
84 for row in csv.DictReader(text):
85 yield {
86 (k or "").strip(): (v or "").strip()
87 for k, v in row.items()
88 if k is not None
89 }
90
91
92def load_aircraft_ref(zf: zipfile.ZipFile) -> dict[str, dict]:
93 """MFR MDL CODE -> make/model/seats reference from ACFTREF."""
94 ref = {}
95 for row in csv_rows(zf, "ACFTREF.txt"):
96 code = row.get("CODE", "")
97 if code:
98 ref[code] = {
99 "make": row.get("MFR", ""),
100 "model": row.get("MODEL", ""),
101 "seats": row.get("NO-SEATS", "").lstrip("0") or "",
102 "engines": row.get("NO-ENG", "").lstrip("0") or "",
103 "weight_class": row.get("AC-WEIGHT", ""),
104 }
105 return ref
106
107
108def master_to_record(row: dict, ref: dict[str, dict], source_url: str) -> dict:
109 n_number = row.get("N-NUMBER", "")
110 mdl_code = row.get("MFR MDL CODE", "")
111 aircraft = ref.get(mdl_code, {})
112 other_names = [
113 row[k]
114 for k in ("OTHER NAMES(1)", "OTHER NAMES(2)", "OTHER NAMES(3)",
115 "OTHER NAMES(4)", "OTHER NAMES(5)")
116 if row.get(k)
117 ]
118 reg_type = row.get("TYPE REGISTRANT", "")
119 ac_type = row.get("TYPE AIRCRAFT", "")
120 return {
121 "n_number": f"N{n_number}" if n_number else "",
122 "serial_number": row.get("SERIAL NUMBER", ""),
123 "year_mfr": row.get("YEAR MFR", ""),
124 "make": aircraft.get("make", ""),
125 "model": aircraft.get("model", ""),
126 "seats": aircraft.get("seats", ""),
127 "engines": aircraft.get("engines", ""),
128 "weight_class": aircraft.get("weight_class", ""),
129 "aircraft_type": ac_type,
130 "aircraft_type_label": AIRCRAFT_TYPES.get(ac_type, ac_type),
131 "registrant_type": reg_type,
132 "registrant_type_label": REGISTRANT_TYPES.get(reg_type, reg_type),
133 "registrant_name": row.get("NAME", ""),
134 "other_names": other_names,
135 "street": row.get("STREET", ""),
136 "street_2": row.get("STREET2", ""),
137 "city": row.get("CITY", ""),
138 "state": row.get("STATE", ""),
139 "zip_code": row.get("ZIP CODE", ""),
140 "country": row.get("COUNTRY", ""),
141 "cert_issue_date": parse_date(row.get("CERT ISSUE DATE", "")),
142 "last_action_date": parse_date(row.get("LAST ACTION DATE", "")),
143 "airworthiness_date": parse_date(row.get("AIR WORTH DATE", "")),
144 "expiration_date": parse_date(row.get("EXPIRATION DATE", "")),
145 "status_code": row.get("STATUS CODE", ""),
146 "mode_s_code_hex": row.get("MODE S CODE HEX", ""),
147 "fractional_owner": row.get("FRACT OWNER", "") == "Y",
148 "source_file": source_url,
149 }
150
151
152async def download_zip(url: str, path: str, attempts: int = 3) -> None:
153 async with httpx.AsyncClient(
154 timeout=httpx.Timeout(600.0, connect=60.0),
155 follow_redirects=True,
156 headers=HTTP_HEADERS,
157 ) as client:
158 for attempt in range(1, attempts + 1):
159 try:
160 async with client.stream("GET", url) as resp:
161 resp.raise_for_status()
162 expected = int(resp.headers.get("content-length") or 0)
163 total = 0
164 with open(path, "wb") as fh:
165 async for chunk in resp.aiter_bytes(1 << 20):
166 if total == 0 and chunk[:2] != b"PK":
167 raise RuntimeError(
168 f"Response is not a zip "
169 f"(starts {chunk[:8]!r})."
170 )
171 fh.write(chunk)
172 total += len(chunk)
173 if expected and total != expected:
174 raise httpx.HTTPError(
175 f"truncated: got {total} of {expected} bytes"
176 )
177 Actor.log.info(f"Downloaded {total / 1_000_000:.1f} MB from {url}")
178 return
179 except (httpx.HTTPError, RuntimeError) as exc:
180 Actor.log.warning(f"Attempt {attempt}/{attempts} failed: {exc}")
181 if attempt == attempts:
182 raise
183
184
185async def load_seen() -> set[str]:
186 store = await Actor.open_key_value_store(name=SEEN_STORE_NAME)
187 return set(await store.get_value(SEEN_KEY) or [])
188
189
190async def save_seen(seen: set[str]) -> None:
191 store = await Actor.open_key_value_store(name=SEEN_STORE_NAME)
192 await store.set_value(SEEN_KEY, sorted(seen))
193
194
195async def charge_safely(count: int) -> None:
196 try:
197 await Actor.charge(CHARGE_EVENT, count)
198 except Exception as exc:
199 Actor.log.debug(f"PPE charge skipped: {exc}")
200
201
202async def deliver_batch(batch: list[dict]) -> tuple[int, bool]:
203 """Push one batch and charge for it, honoring the buyer's max-charge limit.
204
205 Records beyond the limit are not pushed at all — delivering them free would
206 respect the letter of the cap while giving the rest of the run away.
207 Returns (records delivered, limit reached).
208 """
209 limit_hit = False
210 try:
211 remaining = Actor.get_charging_manager(
212 ).calculate_max_event_charge_count_within_limit(CHARGE_EVENT)
213 except Exception:
214 remaining = None
215 if remaining is not None and remaining < len(batch):
216 batch = batch[:remaining]
217 limit_hit = True
218 if not batch:
219 return 0, True
220 await Actor.push_data(batch)
221 await charge_safely(len(batch))
222 return len(batch), limit_hit
223
224
225async def main() -> None:
226 async with Actor:
227 actor_input = await Actor.get_input() or {}
228 lookback_days = int(actor_input.get("lookbackDays", 30) or 0)
229 only_new = actor_input.get("onlyNew", True)
230 states = {s.strip().upper() for s in actor_input.get("states", []) if s.strip()}
231 registrant_types = {
232 s.strip() for s in actor_input.get("registrantTypes", []) if s.strip()
233 }
234 aircraft_types = {
235 s.strip().upper()
236 for s in actor_input.get("aircraftTypes", [])
237 if s.strip()
238 }
239 make_terms = [
240 s.strip().upper() for s in actor_input.get("makeContains", []) if s.strip()
241 ]
242 max_records = int(actor_input.get("maxRecords", 0) or 0)
243 override_url = (actor_input.get("downloadUrl") or "").strip()
244 url = override_url or FAA_ZIP_URL
245
246 cutoff = (
247 datetime.now(timezone.utc).date() - timedelta(days=lookback_days)
248 ).isoformat()
249 Actor.log.info(
250 f"Delivering registrations with a certificate issued on or after "
251 f"{cutoff}."
252 )
253
254 seen = await load_seen() if only_new else set()
255 initial_seen = len(seen)
256
257 pushed = 0
258 scanned = 0
259 skipped_seen = 0
260 limit_hit = False
261 batch: list[dict] = []
262
263 tmp_dir = tempfile.mkdtemp(prefix="faa-aircraft-")
264 zip_path = os.path.join(tmp_dir, "aircraft.zip")
265 try:
266 await download_zip(url, zip_path)
267 with zipfile.ZipFile(zip_path) as zf:
268 members = set(zf.namelist())
269 for required in ("MASTER.txt", "ACFTREF.txt"):
270 if required not in members:
271 raise RuntimeError(
272 f"No {required} in the downloaded zip ({sorted(members)}). "
273 "The FAA file layout may have changed."
274 )
275 ref = load_aircraft_ref(zf)
276 Actor.log.info(f"Loaded {len(ref):,} aircraft model references.")
277
278 for row in csv_rows(zf, "MASTER.txt"):
279 scanned += 1
280 cert_date = parse_date(row.get("CERT ISSUE DATE", ""))
281 if not cert_date or cert_date < cutoff:
282 continue
283 record = master_to_record(row, ref, url)
284 key = f"{record['n_number']}:{record['cert_issue_date']}"
285 if only_new:
286 if key in seen:
287 skipped_seen += 1
288 continue
289 seen.add(key)
290 if states and record["state"].upper() not in states:
291 continue
292 if registrant_types and record["registrant_type"] not in registrant_types:
293 continue
294 if aircraft_types and record["aircraft_type"].upper() not in aircraft_types:
295 continue
296 if make_terms and not any(
297 t in record["make"].upper() or t in record["model"].upper()
298 for t in make_terms
299 ):
300 continue
301
302 batch.append(record)
303 if len(batch) >= 500:
304 delivered, limit_hit = await deliver_batch(batch)
305 pushed += delivered
306 for rec in batch[delivered:]:
307 seen.discard(
308 f"{rec['n_number']}:{rec['cert_issue_date']}"
309 )
310 batch = []
311 if limit_hit or (
312 max_records and pushed + len(batch) >= max_records
313 ):
314 break
315 finally:
316 if os.path.exists(zip_path):
317 os.remove(zip_path)
318 os.rmdir(tmp_dir)
319
320 if batch and not limit_hit:
321 delivered, limit_hit = await deliver_batch(batch)
322 pushed += delivered
323 for rec in batch[delivered:]:
324 seen.discard(f"{rec['n_number']}:{rec['cert_issue_date']}")
325 elif batch:
326 for rec in batch:
327 seen.discard(f"{rec['n_number']}:{rec['cert_issue_date']}")
328
329 if limit_hit:
330 Actor.log.warning(
331 "The run's maximum charge was reached; stopped delivering early. "
332 "Undelivered aircraft stay unmarked and will arrive with the "
333 "next run."
334 )
335
336 if only_new:
337 await save_seen(seen)
338 Actor.log.info(
339 f"Dedupe store: {initial_seen} previously seen, now {len(seen)}."
340 )
341
342 Actor.log.info(
343 f"Done. Scanned {scanned:,} registry records; pushed {pushed} "
344 f"({skipped_seen} already delivered)."
345 )
