1"""FMCSA new motor carrier authority daily feed actor.
2
3Every business day FMCSA grants operating authority to 100-200 motor carriers
4and brokers. Newly-authorized carriers are prime prospects for freight
5factoring, insurance, ELD/compliance and dispatch services — they legally
6cannot haul until insurance is filed, so their buying window is immediate.
7
8Source: US DOT's official open-data portal (Socrata), refreshed daily:
9 - Authority grants: Motus AuthHist "All With History" (yu5v-wbh6) —
10 op_auth_status='Active', reason='Granted', keyed by status_change_date.
11 - Contact enrichment: Motor Carrier Census File (az4n-8mr2) — legal/DBA
12 name, physical & mailing address, phone, cell phone, email, fleet size.
13
14FMCSA's legacy L&I "Register" report (li-public.fmcsa.dot.gov) was retired in
15May 2026 with the Motus migration; these datasets are its replacement.
16
17Grants already delivered are remembered (docket number + grant date) in a
18named key-value store, so a scheduled run behaves like an alert feed.
19"""
20
21from datetime import datetime, timedelta, timezone
22
23import httpx
24from apify import Actor
25
26CHARGE_EVENT = "carrier-lead"
27SEEN_STORE_NAME = "fmcsa-authority-seen"
28SEEN_KEY = "grant-keys"
29
30SODA_BASE = "https://datahub.transportation.gov/resource"
31AUTH_HIST_DATASET = "yu5v-wbh6"
32CENSUS_DATASET = "az4n-8mr2"
33
34ENRICH_BATCH = 100
35
36
37def parse_date(value: str) -> str:
38 """YYYYMMDD (optionally with a time suffix) -> YYYY-MM-DD."""
39 value = (value or "").strip().split(" ")[0]
40 if not value:
41 return ""
42 try:
43 return datetime.strptime(value, "%Y%m%d").date().isoformat()
44 except ValueError:
45 return value
46
47
48async def soda_get(
49 client: httpx.AsyncClient, dataset: str, params: dict
50) -> list[dict]:
51 resp = await client.get(f"{SODA_BASE}/{dataset}.json", params=params)
52 resp.raise_for_status()
53 return resp.json()
54
55
56async def grants_for_day(client: httpx.AsyncClient, stamp: str) -> list[dict]:
57 return await soda_get(
58 client,
59 AUTH_HIST_DATASET,
60 {
61 "$where": (
62 "op_auth_status='Active' AND upper(reason)='GRANTED' "
63 f"AND status_change_date='{stamp}'"
64 ),
65 "$limit": 50000,
66 },
67 )
68
69
70async def census_for_dots(
71 client: httpx.AsyncClient, dots: list[str]
72) -> dict[str, dict]:
73 """Census rows for a batch of DOT numbers, keyed by DOT number."""
74 quoted = ",".join(f"'{d}'" for d in dots)
75 rows = await soda_get(
76 client,
77 CENSUS_DATASET,
78 {"$where": f"dot_number in({quoted})", "$limit": len(dots) + 10},
79 )
80 return {row.get("dot_number", ""): row for row in rows}
81
82
83def build_record(grant: dict, census: dict) -> dict:
84 return {
85 "docket_number": grant.get("docket_number", ""),
86 "usdot_number": grant.get("usdot_number", ""),
87 "authority_type": grant.get("op_auth_type", ""),
88 "granted_date": parse_date(grant.get("status_change_date", "")),
89 "legal_name": census.get("legal_name", ""),
90 "dba_name": census.get("dba_name", ""),
91 "physical_address": {
92 "street": census.get("phy_street", ""),
93 "city": census.get("phy_city", ""),
94 "state": census.get("phy_state", ""),
95 "zip_code": census.get("phy_zip", ""),
96 "country": census.get("phy_country", ""),
97 },
98 "mailing_address": {
99 "street": census.get("carrier_mailing_street", ""),
100 "city": census.get("carrier_mailing_city", ""),
101 "state": census.get("carrier_mailing_state", ""),
102 "zip_code": census.get("carrier_mailing_zip", ""),
103 "country": census.get("carrier_mailing_country", ""),
104 },
105 "phone": census.get("phone", ""),
106 "cell_phone": census.get("cell_phone", ""),
107 "email": census.get("email_address", ""),
108 "company_officer": census.get("company_officer_1", ""),
109 "power_units": census.get("power_units", ""),
110 "total_drivers": census.get("total_drivers", ""),
111 "operation_class": census.get("classdef", ""),
112 "carrier_operation": census.get("carrier_operation", ""),
113 "mcs150_date": parse_date(census.get("mcs150_date", "")),
114 "census_add_date": parse_date(census.get("add_date", "")),
115 "census_found": bool(census),
116 "source": f"{SODA_BASE}/{AUTH_HIST_DATASET}.json",
117 }
118
119
120def matches(
121 record: dict,
122 states: set[str],
123 authority_terms: list[str],
124 require_contact: bool,
125) -> bool:
126 if states and record["physical_address"]["state"].upper() not in states:
127 return False
128 if authority_terms and not any(
129 t in record["authority_type"].upper() for t in authority_terms
130 ):
131 return False
132 if require_contact and not (record["phone"] or record["email"]):
133 return False
134 return True
135
136
137async def load_seen() -> set[str]:
138 store = await Actor.open_key_value_store(name=SEEN_STORE_NAME)
139 return set(await store.get_value(SEEN_KEY) or [])
140
141
142async def save_seen(seen: set[str]) -> None:
143 store = await Actor.open_key_value_store(name=SEEN_STORE_NAME)
144 await store.set_value(SEEN_KEY, sorted(seen))
145
146
147async def charge_safely(count: int) -> None:
148 try:
149 await Actor.charge(CHARGE_EVENT, count)
150 except Exception as exc:
151 Actor.log.debug(f"PPE charge skipped: {exc}")
152
153
154async def deliver_batch(batch: list[dict]) -> tuple[int, bool]:
155 """Push one batch and charge for it, honoring the buyer's max-charge limit.
156
157 Records beyond the limit are not pushed at all — delivering them free would
158 respect the letter of the cap while giving the rest of the run away.
159 Returns (records delivered, limit reached).
160 """
161 limit_hit = False
162 try:
163 remaining = Actor.get_charging_manager(
164 ).calculate_max_event_charge_count_within_limit(CHARGE_EVENT)
165 except Exception:
166 remaining = None
167 if remaining is not None and remaining < len(batch):
168 batch = batch[:remaining]
169 limit_hit = True
170 if not batch:
171 return 0, True
172 await Actor.push_data(batch)
173 await charge_safely(len(batch))
174 return len(batch), limit_hit
175
176
177def grant_key(grant: dict) -> str:
178 return (
179 f"{grant.get('docket_number', '')}:"
180 f"{grant.get('status_change_date', '')}"
181 )
182
183
184async def main() -> None:
185 async with Actor:
186 actor_input = await Actor.get_input() or {}
187 lookback_days = int(actor_input.get("lookbackDays", 7) or 0)
188 only_new = actor_input.get("onlyNew", True)
189 states = {
190 s.strip().upper() for s in actor_input.get("states", []) if s.strip()
191 }
192 authority_terms = [
193 s.strip().upper()
194 for s in actor_input.get("authorityTypeContains", [])
195 if s.strip()
196 ]
197 require_contact = actor_input.get("requireContact", False)
198 max_records = int(actor_input.get("maxRecords", 0) or 0)
199 app_token = (actor_input.get("socrataAppToken") or "").strip()
200
201 headers = {"Accept": "application/json"}
202 if app_token:
203 headers["X-App-Token"] = app_token
204
205 today = datetime.now(timezone.utc).date()
206
207
208 days = [today - timedelta(days=n) for n in range(lookback_days, -1, -1)]
209 Actor.log.info(
210 f"Checking {len(days)} day(s): {days[0].isoformat()} .. "
211 f"{days[-1].isoformat()}"
212 )
213
214 seen = await load_seen() if only_new else set()
215 initial_seen = len(seen)
216
217 grants: list[dict] = []
218 skipped_seen = 0
219 async with httpx.AsyncClient(
220 timeout=httpx.Timeout(120.0, connect=30.0),
221 follow_redirects=True,
222 headers=headers,
223 ) as client:
224 for day in days:
225 stamp = day.strftime("%Y%m%d")
226 day_grants = await grants_for_day(client, stamp)
227 if not day_grants:
228 continue
229 fresh = []
230 for grant in day_grants:
231 if only_new:
232 key = grant_key(grant)
233 if key in seen:
234 skipped_seen += 1
235 continue
236 seen.add(key)
237 fresh.append(grant)
238 Actor.log.info(
239 f"{day.isoformat()}: {len(day_grants)} authority grants "
240 f"({len(fresh)} new)."
241 )
242 grants.extend(fresh)
243
244 pushed = 0
245 filtered_out = 0
246 no_census = 0
247 limit_hit = False
248 batch: list[dict] = []
249
250 async def flush_batch() -> None:
251 nonlocal pushed, limit_hit
252 delivered, limit_hit = await deliver_batch(batch)
253 pushed += delivered
254
255
256 for record in batch[delivered:]:
257 seen.discard(
258 f"{record['docket_number']}:"
259 f"{record['granted_date'].replace('-', '')}"
260 )
261 batch.clear()
262
263 for start in range(0, len(grants), ENRICH_BATCH):
264 chunk = grants[start : start + ENRICH_BATCH]
265 dots = sorted(
266 {g.get("usdot_number", "") for g in chunk if g.get("usdot_number")}
267 )
268 census_by_dot = await census_for_dots(client, dots) if dots else {}
269 for grant in chunk:
270 census = census_by_dot.get(grant.get("usdot_number", ""), {})
271 if not census:
272 no_census += 1
273 record = build_record(grant, census)
274 if not matches(record, states, authority_terms, require_contact):
275 filtered_out += 1
276 continue
277 batch.append(record)
278 if len(batch) >= 500:
279 await flush_batch()
280 if limit_hit or (
281 max_records and pushed + len(batch) >= max_records
282 ):
283 break
284 if limit_hit or (
285 max_records and pushed + len(batch) >= max_records
286 ):
287 break
288
289 if batch and not limit_hit:
290 await flush_batch()
291 elif batch:
292 for record in batch:
293 seen.discard(
294 f"{record['docket_number']}:"
295 f"{record['granted_date'].replace('-', '')}"
296 )
297 batch.clear()
298
299 if limit_hit:
300 Actor.log.warning(
301 "The run's maximum charge was reached; stopped delivering early. "
302 "Undelivered grants stay unmarked and will arrive with the "
303 "next run."
304 )
305
306 if only_new:
307 await save_seen(seen)
308 Actor.log.info(
309 f"Dedupe store: {initial_seen} previously seen, now {len(seen)}."
310 )
311
312 if no_census:
313 Actor.log.info(
314 f"{no_census} grants had no census row yet (very new carriers "
315 "appear there within a few days); delivered with census fields "
316 "empty."
317 )
318
319 Actor.log.info(
320 f"Done. Pushed {pushed} carrier leads "
321 f"({skipped_seen} already delivered, {filtered_out} filtered out)."
322 )
