1"""Florida federal tax lien daily new-filing feed actor.
2
3Florida's Division of Corporations publishes federal tax lien (FLR) data as
4four fixed-length ASCII files per business day. This actor pulls the filing,
5debtor, and secured-party files for each requested day, joins them on the
6document number, dedupes against filings delivered in previous runs (named
7key-value store), and pushes only what's new.
8
9Parsing is positional against the state's published record layout, not
10header-driven: the files carry no header row and no delimiter.
11Layout: https://dos.sunbiz.org/data-definitions/lien.html
12"""
13
14from datetime import date, datetime, timedelta, timezone
15
16import httpx
17from apify import Actor
18
19CHARGE_EVENT = "lien-filing"
20SEEN_STORE_NAME = "florida-lien-seen"
21SEEN_KEY = "document-numbers"
22
23
24
25BASE_URL = "https://sftp.floridados.gov/Public/doc/FLR"
26PUBLIC_USER = "Public"
27PUBLIC_PASSWORD = "PubAccess1845!"
28
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": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
38 "Accept-Language": "en-US,en;q=0.9",
39}
40
41
42FILING_LAYOUT = [
43 ("document_number", 1, 12),
44 ("filing_date", 13, 8),
45 ("pages", 21, 5),
46 ("total_pages", 26, 5),
47 ("filing_status", 31, 1),
48 ("filing_type", 32, 1),
49 ("assessment_date", 33, 8),
50 ("cancellation_date", 41, 8),
51 ("expiration_date", 49, 8),
52 ("transmitting_utility", 57, 1),
53 ("event_count", 58, 5),
54 ("total_debtors", 63, 5),
55 ("total_secured_parties", 68, 5),
56 ("current_debtors", 73, 5),
57 ("current_secured_parties", 78, 5),
58]
59FILING_RECORD_LENGTH = 82
60
61
62PARTY_LAYOUT = [
63 ("filing_type", 1, 1),
64 ("document_number", 2, 12),
65 ("name", 14, 55),
66 ("name_format", 69, 1),
67 ("address_1", 70, 44),
68 ("address_2", 114, 44),
69 ("city", 158, 28),
70 ("state", 186, 2),
71 ("zip_code", 188, 9),
72 ("country", 197, 2),
73 ("sequence", 199, 5),
74 ("relation_to_filing", 204, 1),
75 ("original_party", 205, 1),
76 ("filing_status", 206, 1),
77]
78PARTY_RECORD_LENGTH = 206
79
80DATE_FIELDS = {"filing_date", "assessment_date", "cancellation_date", "expiration_date"}
81COUNT_FIELDS = {
82 "pages",
83 "total_pages",
84 "event_count",
85 "total_debtors",
86 "total_secured_parties",
87 "current_debtors",
88 "current_secured_parties",
89 "sequence",
90}
91
92
93def parse_date(value: str) -> str:
94 """MMDDYYYY -> YYYY-MM-DD. Blank and all-zero dates become ''."""
95 value = value.strip()
96 if not value or set(value) <= {"0"}:
97 return ""
98 try:
99 return datetime.strptime(value, "%m%d%Y").date().isoformat()
100 except ValueError:
101 Actor.log.warning(f"Unparseable date {value!r}; passing through as-is.")
102 return value
103
104
105def parse_count(value: str) -> int | None:
106 value = value.strip()
107 return int(value) if value.isdigit() else None
108
109
110def parse_fixed(line: str, layout: list[tuple[str, int, int]]) -> dict:
111 record = {}
112 for name, start, length in layout:
113 raw = line[start - 1 : start - 1 + length].strip()
114 if name in DATE_FIELDS:
115 record[name] = parse_date(raw)
116 elif name in COUNT_FIELDS:
117 record[name] = parse_count(raw)
118 else:
119 record[name] = raw
120 return record
121
122
123def parse_lines(
124 name: str, text: str, layout: list[tuple[str, int, int]], expected_length: int
125) -> list[dict]:
126 """Parse fixed-width lines, warning once if the record length has drifted."""
127 records = []
128 warned = False
129 for lineno, line in enumerate(text.splitlines(), start=1):
130 if not line.strip():
131 continue
132 if len(line) != expected_length and not warned:
133 Actor.log.warning(
134 f"{name}: line {lineno} is {len(line)} chars, expected "
135 f"{expected_length}. The state's record layout may have "
136 "changed — see https://dos.sunbiz.org/data-definitions/lien.html"
137 )
138 warned = True
139 records.append(parse_fixed(line, layout))
140 return records
141
142
143def file_urls(day: date) -> dict[str, str]:
144 stamp = day.strftime("%Y%m%d")
145 return {
146 "filings": f"{BASE_URL}/FILINGS/{stamp}flrf.txt",
147 "debtors": f"{BASE_URL}/DEBTORS/{stamp}flrd.txt",
148 "secured": f"{BASE_URL}/SECURED/{stamp}flrs.txt",
149 }
150
151
152async def fetch_text(client: httpx.AsyncClient, url: str) -> str | None:
153 """Return file text, or None when the state published nothing for that day."""
154 resp = await client.get(url)
155 if resp.status_code == 404:
156 return None
157 if resp.status_code == 403 and b"challenges.cloudflare.com" in resp.content:
158 raise RuntimeError(
159 f"Cloudflare challenged the request to {url}. The state's file host "
160 "is blocking this IP/client fingerprint; retry later or route the "
161 "run through Apify residential proxy."
162 )
163 resp.raise_for_status()
164
165
166 return resp.content.decode("latin-1")
167
168
169async def load_day(client: httpx.AsyncClient, day: date) -> list[dict] | None:
170 """Fetch and join one business day's filings, debtors and secured parties."""
171 urls = file_urls(day)
172 filings_text = await fetch_text(client, urls["filings"])
173 if filings_text is None:
174 Actor.log.info(f"{day.isoformat()}: no filings file published; skipping.")
175 return None
176
177 filings = parse_lines(
178 urls["filings"], filings_text, FILING_LAYOUT, FILING_RECORD_LENGTH
179 )
180 by_doc = {}
181 for f in filings:
182 f["debtors"] = []
183 f["secured_parties"] = []
184 f["file_date"] = day.isoformat()
185 f["source_file"] = urls["filings"]
186 by_doc[f["document_number"]] = f
187
188 for key, field in (("debtors", "debtors"), ("secured", "secured_parties")):
189 text = await fetch_text(client, urls[key])
190 if text is None:
191 Actor.log.warning(
192 f"{day.isoformat()}: no {key} file; those parties will be empty."
193 )
194 continue
195 orphans = 0
196 for party in parse_lines(urls[key], text, PARTY_LAYOUT, PARTY_RECORD_LENGTH):
197 target = by_doc.get(party["document_number"])
198 if target is None:
199 orphans += 1
200 continue
201
202 target[field].append(
203 {k: v for k, v in party.items() if k != "filing_type"}
204 )
205 if orphans:
206 Actor.log.warning(
207 f"{day.isoformat()}: {orphans} {key} rows referenced a document "
208 "number not in that day's filings file; dropped."
209 )
210
211 for f in by_doc.values():
212 f["debtors"].sort(key=lambda p: p["sequence"] or 0)
213 f["secured_parties"].sort(key=lambda p: p["sequence"] or 0)
214
215 Actor.log.info(
216 f"{day.isoformat()}: {len(by_doc)} filings, "
217 f"{sum(len(f['debtors']) for f in by_doc.values())} debtors, "
218 f"{sum(len(f['secured_parties']) for f in by_doc.values())} secured parties."
219 )
220 return list(by_doc.values())
221
222
223def target_days(file_date: str, lookback_days: int) -> list[date]:
224 if file_date:
225 try:
226 return [datetime.strptime(file_date.strip(), "%Y%m%d").date()]
227 except ValueError as exc:
228 raise ValueError(
229 f"'fileDate' must be YYYYMMDD (got {file_date!r})."
230 ) from exc
231 today = datetime.now(timezone.utc).date()
232
233 return [today - timedelta(days=n) for n in range(lookback_days, -1, -1)]
234
235
236def matches(
237 record: dict,
238 states: set[str],
239 debtor_terms: list[str],
240 sp_terms: list[str],
241 statuses: set[str],
242) -> bool:
243 if statuses and record["filing_status"].upper() not in statuses:
244 return False
245 if states and not any(d["state"].upper() in states for d in record["debtors"]):
246 return False
247 if debtor_terms and not any(
248 t in d["name"].upper() for d in record["debtors"] for t in debtor_terms
249 ):
250 return False
251 if sp_terms and not any(
252 t in s["name"].upper() for s in record["secured_parties"] for t in sp_terms
253 ):
254 return False
255 return True
256
257
258async def load_seen() -> set[str]:
259 store = await Actor.open_key_value_store(name=SEEN_STORE_NAME)
260 return set(await store.get_value(SEEN_KEY) or [])
261
262
263async def save_seen(seen: set[str]) -> None:
264 store = await Actor.open_key_value_store(name=SEEN_STORE_NAME)
265 await store.set_value(SEEN_KEY, sorted(seen))
266
267
268async def charge_safely(count: int) -> None:
269 try:
270 await Actor.charge(CHARGE_EVENT, count)
271 except Exception as exc:
272 Actor.log.debug(f"PPE charge skipped: {exc}")
273
274
275async def deliver_batch(batch: list[dict]) -> tuple[int, bool]:
276 """Push one batch and charge for it, honoring the buyer's max-charge limit.
277
278 Records beyond the limit are not pushed at all — delivering them free would
279 respect the letter of the cap while giving the rest of the run away.
280 Returns (records delivered, limit reached).
281 """
282 limit_hit = False
283 try:
284 remaining = Actor.get_charging_manager(
285 ).calculate_max_event_charge_count_within_limit(CHARGE_EVENT)
286 except Exception:
287 remaining = None
288 if remaining is not None and remaining < len(batch):
289 batch = batch[:remaining]
290 limit_hit = True
291 if not batch:
292 return 0, True
293 await Actor.push_data(batch)
294 await charge_safely(len(batch))
295 return len(batch), limit_hit
296
297
298async def main() -> None:
299 async with Actor:
300 actor_input = await Actor.get_input() or {}
301 file_date = (actor_input.get("fileDate") or "").strip()
302 lookback_days = int(actor_input.get("lookbackDays", 7) or 0)
303 only_new = actor_input.get("onlyNew", True)
304 states = {
305 s.strip().upper()
306 for s in actor_input.get("debtorStateFilter", [])
307 if s.strip()
308 }
309 debtor_terms = [
310 s.strip().upper()
311 for s in actor_input.get("debtorNameContains", [])
312 if s.strip()
313 ]
314 sp_terms = [
315 s.strip().upper()
316 for s in actor_input.get("securedPartyContains", [])
317 if s.strip()
318 ]
319 statuses = {
320 s.strip().upper()
321 for s in actor_input.get("filingStatuses", [])
322 if s.strip()
323 }
324 max_records = int(actor_input.get("maxRecords", 0) or 0)
325
326 days = target_days(file_date, lookback_days)
327 Actor.log.info(
328 f"Checking {len(days)} day(s): {days[0].isoformat()} .. "
329 f"{days[-1].isoformat()}"
330 )
331
332 seen = await load_seen() if only_new else set()
333 initial_seen = len(seen)
334
335 pushed = 0
336 skipped_seen = 0
337 filtered_out = 0
338 limit_hit = False
339 batch: list[dict] = []
340 days_with_data = 0
341
342 async def flush_batch() -> None:
343 nonlocal pushed, limit_hit
344 delivered, limit_hit = await deliver_batch(batch)
345 pushed += delivered
346
347
348 for record in batch[delivered:]:
349 seen.discard(record["document_number"])
350 batch.clear()
351
352 async with httpx.AsyncClient(
353 timeout=httpx.Timeout(120.0, connect=30.0),
354 follow_redirects=True,
355 auth=httpx.BasicAuth(PUBLIC_USER, PUBLIC_PASSWORD),
356 headers=HTTP_HEADERS,
357 ) as client:
358 for day in days:
359 records = await load_day(client, day)
360 if records is None:
361 continue
362 days_with_data += 1
363 for record in records:
364 doc_no = record["document_number"]
365 if only_new and doc_no:
366 if doc_no in seen:
367 skipped_seen += 1
368 continue
369 seen.add(doc_no)
370 if not matches(record, states, debtor_terms, sp_terms, statuses):
371 filtered_out += 1
372 continue
373
374 batch.append(record)
375 if len(batch) >= 500:
376 await flush_batch()
377 if limit_hit or (
378 max_records and pushed + len(batch) >= max_records
379 ):
380 break
381 if limit_hit or (
382 max_records and pushed + len(batch) >= max_records
383 ):
384 break
385
386 if batch and not limit_hit:
387 await flush_batch()
388 elif batch:
389 for record in batch:
390 seen.discard(record["document_number"])
391 batch.clear()
392
393 if limit_hit:
394 Actor.log.warning(
395 "The run's maximum charge was reached; stopped delivering early. "
396 "Undelivered filings stay unmarked and will arrive with the "
397 "next run."
398 )
399
400 if only_new:
401 await save_seen(seen)
402 Actor.log.info(
403 f"Dedupe store: {initial_seen} previously seen, now {len(seen)}."
404 )
405
406 if days_with_data == 0:
407 Actor.log.warning(
408 "No filing files were published for any requested day. Florida "
409 "publishes on business days only; try a larger 'lookbackDays'."
410 )
411
412 Actor.log.info(
413 f"Done. Pushed {pushed} lien filings "
414 f"({skipped_seen} already delivered, {filtered_out} filtered out)."
415 )
