1"""Florida new business registration daily feed actor.
2
3Florida's Division of Corporations publishes a fixed-width corporate data file
4every business day containing that day's processed entity filings — new LLCs,
5corporations, partnerships and nonprofits, with principal/mailing addresses,
6registered agent and up to six officers. This actor pulls the daily files for
7the requested days, parses them positionally against the state's published
8record layout, dedupes against entities delivered in previous runs (named
9key-value store), and pushes only what's new.
10
11Layout: https://dos.sunbiz.org/data-definitions/cor.html (1,440 chars/record)
12Portal: https://dos.fl.gov/sunbiz/other-services/data-downloads/
13"""
14
15import re
16from datetime import date, datetime, timedelta, timezone
17
18import httpx
19from apify import Actor
20
21CHARGE_EVENT = "business-registration"
22SEEN_STORE_NAME = "florida-new-business-seen"
23SEEN_KEY = "document-numbers"
24
25
26
27BASE_URL = "https://sftp.floridados.gov/Public/doc/cor"
28PUBLIC_USER = "Public"
29PUBLIC_PASSWORD = "PubAccess1845!"
30
31
32
33HTTP_HEADERS = {
34 "User-Agent": (
35 "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 "
36 "(KHTML, like Gecko) Chrome/126.0.0.0 Safari/537.36"
37 ),
38 "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
39 "Accept-Language": "en-US,en;q=0.9",
40}
41
42RECORD_LENGTH = 1440
43
44
45SCALAR_LAYOUT = [
46 ("document_number", 1, 12),
47 ("name", 13, 192),
48 ("status", 205, 1),
49 ("filing_type", 206, 15),
50 ("file_date", 473, 8),
51 ("fei_number", 481, 14),
52 ("more_than_six_officers", 495, 1),
53 ("last_transaction_date", 496, 8),
54 ("state_country", 504, 2),
55]
56
57ADDRESS_LAYOUTS = {
58
59 "principal_address": (221, 263, 305, 333, 335, 345),
60 "mailing_address": (347, 389, 431, 459, 461, 471),
61}
62ADDRESS_LENGTHS = (42, 42, 28, 2, 10, 2)
63
64AGENT_LAYOUT = [
65 ("name", 545, 42),
66 ("type", 587, 1),
67 ("address", 588, 42),
68 ("city", 630, 28),
69 ("state", 658, 2),
70 ("zip_code", 660, 9),
71]
72
73
74OFFICER_START = 669
75OFFICER_COUNT = 6
76OFFICER_BLOCK = [
77 ("title", 0, 4),
78 ("type", 4, 1),
79 ("name", 5, 42),
80 ("address", 47, 42),
81 ("city", 89, 28),
82 ("state", 117, 2),
83 ("zip_code", 119, 9),
84]
85OFFICER_BLOCK_LENGTH = 128
86
87DATE_FIELDS = {"file_date", "last_transaction_date"}
88
89
90
91FILING_TYPE_LABELS = {
92 "FLAL": "Florida limited liability company",
93 "FORL": "Foreign limited liability company",
94 "DOMP": "Domestic for-profit corporation",
95 "FORP": "Foreign for-profit corporation",
96 "DOMNP": "Domestic nonprofit corporation",
97 "FORNP": "Foreign nonprofit corporation",
98 "DOMLP": "Domestic limited partnership",
99 "FORLP": "Foreign limited partnership",
100 "AGENT": "Registered agent filing",
101 "TRUST": "Trust",
102}
103
104
105def slice_field(line: str, start: int, length: int) -> str:
106 return line[start - 1 : start - 1 + length].strip()
107
108
109def collapse_spaces(value: str) -> str:
110 """The state pads name parts into fixed sub-columns; collapse the runs."""
111 return re.sub(r" {2,}", " ", value).strip()
112
113
114def parse_date(value: str) -> str:
115 """MMDDYYYY -> YYYY-MM-DD. Blank and all-zero dates become ''."""
116 value = value.strip()
117 if not value or set(value) <= {"0"}:
118 return ""
119 try:
120 return datetime.strptime(value, "%m%d%Y").date().isoformat()
121 except ValueError:
122 Actor.log.warning(f"Unparseable date {value!r}; passing through as-is.")
123 return value
124
125
126def parse_address(line: str, starts: tuple[int, ...]) -> dict:
127 keys = ("address_1", "address_2", "city", "state", "zip_code", "country")
128 return {
129 key: slice_field(line, start, length)
130 for key, start, length in zip(keys, starts, ADDRESS_LENGTHS)
131 }
132
133
134def parse_record(line: str, feed_day: str, source_file: str) -> dict:
135 record: dict = {}
136 for name, start, length in SCALAR_LAYOUT:
137 raw = slice_field(line, start, length)
138 record[name] = parse_date(raw) if name in DATE_FIELDS else raw
139 record["name"] = collapse_spaces(record["name"])
140 record["filing_type_label"] = FILING_TYPE_LABELS.get(
141 record["filing_type"], record["filing_type"]
142 )
143 record["more_than_six_officers"] = record["more_than_six_officers"] == "Y"
144
145 for field, starts in ADDRESS_LAYOUTS.items():
146 record[field] = parse_address(line, starts)
147
148 agent = {
149 name: slice_field(line, start, length)
150 for name, start, length in AGENT_LAYOUT
151 }
152 agent["name"] = collapse_spaces(agent["name"])
153 record["registered_agent"] = agent
154
155 officers = []
156 for i in range(OFFICER_COUNT):
157 base = OFFICER_START + i * OFFICER_BLOCK_LENGTH
158 officer = {
159 name: slice_field(line, base + offset, length)
160 for name, offset, length in OFFICER_BLOCK
161 }
162 if not officer["name"]:
163 continue
164 officer["name"] = collapse_spaces(officer["name"])
165 officers.append(officer)
166 record["officers"] = officers
167
168 record["feed_day"] = feed_day
169 record["source_file"] = source_file
170 return record
171
172
173def file_url(day: date) -> str:
174 return f"{BASE_URL}/{day.strftime('%Y%m%d')}c.txt"
175
176
177async def fetch_text(client: httpx.AsyncClient, url: str) -> str | None:
178 """Return file text, or None when the state published nothing for that day."""
179 resp = await client.get(url)
180 if resp.status_code == 404:
181 return None
182 if resp.status_code == 403 and b"challenges.cloudflare.com" in resp.content:
183 raise RuntimeError(
184 f"Cloudflare challenged the request to {url}. The state's file host "
185 "is blocking this IP/client fingerprint; retry later or route the "
186 "run through Apify residential proxy."
187 )
188 resp.raise_for_status()
189
190
191 return resp.content.decode("latin-1")
192
193
194def parse_day(text: str, day: date, url: str) -> list[dict]:
195 records = []
196 warned = False
197 for lineno, line in enumerate(text.splitlines(), start=1):
198 if not line.strip():
199 continue
200 if len(line) != RECORD_LENGTH and not warned:
201 Actor.log.warning(
202 f"{url}: line {lineno} is {len(line)} chars, expected "
203 f"{RECORD_LENGTH}. The state's record layout may have changed "
204 "— see https://dos.sunbiz.org/data-definitions/cor.html"
205 )
206 warned = True
207 records.append(parse_record(line, day.isoformat(), url))
208 return records
209
210
211def target_days(file_date: str, lookback_days: int) -> list[date]:
212 if file_date:
213 try:
214 return [datetime.strptime(file_date.strip(), "%Y%m%d").date()]
215 except ValueError as exc:
216 raise ValueError(
217 f"'fileDate' must be YYYYMMDD (got {file_date!r})."
218 ) from exc
219 today = datetime.now(timezone.utc).date()
220
221 return [today - timedelta(days=n) for n in range(lookback_days, -1, -1)]
222
223
224def matches(
225 record: dict,
226 filing_types: set[str],
227 zip_prefixes: list[str],
228 name_terms: list[str],
229 agent_terms: list[str],
230 max_age_days: int,
231 feed_day: date,
232) -> bool:
233 if filing_types and record["filing_type"].upper() not in filing_types:
234 return False
235 if zip_prefixes:
236 zips = (
237 record["principal_address"]["zip_code"],
238 record["mailing_address"]["zip_code"],
239 )
240 if not any(z.startswith(p) for z in zips if z for p in zip_prefixes):
241 return False
242 if name_terms and not any(t in record["name"].upper() for t in name_terms):
243 return False
244 if agent_terms and not any(
245 t in record["registered_agent"]["name"].upper() for t in agent_terms
246 ):
247 return False
248 if max_age_days and record["file_date"]:
249 try:
250 filed = date.fromisoformat(record["file_date"])
251 except ValueError:
252 return True
253
254
255 if (feed_day - filed).days > max_age_days:
256 return False
257 return True
258
259
260async def load_seen() -> set[str]:
261 store = await Actor.open_key_value_store(name=SEEN_STORE_NAME)
262 return set(await store.get_value(SEEN_KEY) or [])
263
264
265async def save_seen(seen: set[str]) -> None:
266 store = await Actor.open_key_value_store(name=SEEN_STORE_NAME)
267 await store.set_value(SEEN_KEY, sorted(seen))
268
269
270async def charge_safely(count: int) -> None:
271 try:
272 await Actor.charge(CHARGE_EVENT, count)
273 except Exception as exc:
274 Actor.log.debug(f"PPE charge skipped: {exc}")
275
276
277async def deliver_batch(batch: list[dict]) -> tuple[int, bool]:
278 """Push one batch and charge for it, honoring the buyer's max-charge limit.
279
280 Records beyond the limit are not pushed at all — delivering them free would
281 respect the letter of the cap while giving the rest of the run away.
282 Returns (records delivered, limit reached).
283 """
284 limit_hit = False
285 try:
286 remaining = Actor.get_charging_manager(
287 ).calculate_max_event_charge_count_within_limit(CHARGE_EVENT)
288 except Exception:
289 remaining = None
290 if remaining is not None and remaining < len(batch):
291 batch = batch[:remaining]
292 limit_hit = True
293 if not batch:
294 return 0, True
295 await Actor.push_data(batch)
296 await charge_safely(len(batch))
297 return len(batch), limit_hit
298
299
300async def main() -> None:
301 async with Actor:
302 actor_input = await Actor.get_input() or {}
303 file_date = (actor_input.get("fileDate") or "").strip()
304 lookback_days = int(actor_input.get("lookbackDays", 7) or 0)
305 only_new = actor_input.get("onlyNew", True)
306 filing_types = {
307 s.strip().upper()
308 for s in actor_input.get("filingTypes", [])
309 if s.strip()
310 }
311 zip_prefixes = [
312 s.strip() for s in actor_input.get("zipPrefixes", []) if s.strip()
313 ]
314 name_terms = [
315 s.strip().upper()
316 for s in actor_input.get("nameContains", [])
317 if s.strip()
318 ]
319 agent_terms = [
320 s.strip().upper()
321 for s in actor_input.get("registeredAgentContains", [])
322 if s.strip()
323 ]
324 max_age_days = int(actor_input.get("maxFilingAgeDays", 30) or 0)
325 max_records = int(actor_input.get("maxRecords", 0) or 0)
326
327 days = target_days(file_date, lookback_days)
328 Actor.log.info(
329 f"Checking {len(days)} day(s): {days[0].isoformat()} .. "
330 f"{days[-1].isoformat()}"
331 )
332
333 seen = await load_seen() if only_new else set()
334 initial_seen = len(seen)
335
336 pushed = 0
337 skipped_seen = 0
338 filtered_out = 0
339 limit_hit = False
340 batch: list[dict] = []
341 days_with_data = 0
342
343 async def flush_batch() -> None:
344 nonlocal pushed, limit_hit
345 delivered, limit_hit = await deliver_batch(batch)
346 pushed += delivered
347
348
349 for record in batch[delivered:]:
350 seen.discard(record["document_number"])
351 batch.clear()
352
353 async with httpx.AsyncClient(
354 timeout=httpx.Timeout(120.0, connect=30.0),
355 follow_redirects=True,
356 auth=httpx.BasicAuth(PUBLIC_USER, PUBLIC_PASSWORD),
357 headers=HTTP_HEADERS,
358 ) as client:
359 for day in days:
360 url = file_url(day)
361 text = await fetch_text(client, url)
362 if text is None:
363 Actor.log.info(
364 f"{day.isoformat()}: no corporate file published; "
365 "skipping."
366 )
367 continue
368 days_with_data += 1
369 records = parse_day(text, day, url)
370 Actor.log.info(f"{day.isoformat()}: {len(records)} filings.")
371 for record in records:
372 doc_no = record["document_number"]
373 if only_new and doc_no:
374 if doc_no in seen:
375 skipped_seen += 1
376 continue
377 seen.add(doc_no)
378 if not matches(
379 record,
380 filing_types,
381 zip_prefixes,
382 name_terms,
383 agent_terms,
384 max_age_days,
385 day,
386 ):
387 filtered_out += 1
388 continue
389
390 batch.append(record)
391 if len(batch) >= 500:
392 await flush_batch()
393 if limit_hit or (
394 max_records and pushed + len(batch) >= max_records
395 ):
396 break
397 if limit_hit or (
398 max_records and pushed + len(batch) >= max_records
399 ):
400 break
401
402 if batch and not limit_hit:
403 await flush_batch()
404 elif batch:
405 for record in batch:
406 seen.discard(record["document_number"])
407 batch.clear()
408
409 if limit_hit:
410 Actor.log.warning(
411 "The run's maximum charge was reached; stopped delivering early. "
412 "Undelivered filings stay unmarked and will arrive with the "
413 "next run."
414 )
415
416 if only_new:
417 await save_seen(seen)
418 Actor.log.info(
419 f"Dedupe store: {initial_seen} previously seen, now {len(seen)}."
420 )
421
422 if days_with_data == 0:
423 Actor.log.warning(
424 "No corporate files were published for any requested day. "
425 "Florida publishes on business days only; try a larger "
426 "'lookbackDays'."
427 )
428
429 Actor.log.info(
430 f"Done. Pushed {pushed} business registrations "
431 f"({skipped_seen} already delivered, {filtered_out} filtered out)."
432 )
