"""Ingest OH SoS new-filings reports (5 types) into `oh_business_filings`. Uses playwright-stealth to defeat the ohiosos.gov Cloudflare bot filter. Files are "recent filings" snapshots — not the full historical registry. """ from __future__ import annotations import csv import re from pathlib import Path import pandas as pd from playwright.sync_api import sync_playwright from playwright_stealth import Stealth from shapely.geometry import Point from sqlalchemy import create_engine, text import geopandas as gpd DATA_DIR = Path("/home/m3ac/genoa-entwuerfe.com") CACHE = DATA_DIR / "gis_scripts" / ".cache" / "ohsos-biz" CACHE.mkdir(parents=True, exist_ok=True) # each = (file id, entity_type shown in DB) BIZ_FILES = [ ("WI0100R.TXT", "Domestic LLC"), ("WI0050R.TXT", "Domestic Corp (for-profit)"), ("WI0070R.TXT", "Domestic Corp (nonprofit)"), ("WI0090R.TXT", "Foreign LLC"), ("WI0220R.TXT", "Dissolution"), ] def _env(k): for line in (DATA_DIR / ".env.gis").read_text().splitlines(): if line.startswith(f"{k}="): return line.split("=", 1)[1].strip() raise SystemExit(f"{k} missing") def _fetch_all() -> None: with Stealth().use_sync(sync_playwright()) as p: b = p.chromium.launch( headless=True, args=["--no-sandbox", "--disable-blink-features=AutomationControlled"], ) ctx = b.new_context( user_agent="Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/126.0.0.0 Safari/537.36", viewport={"width": 1280, "height": 800}, ) pg = ctx.new_page() # warm cookies from the reports landing page pg.goto("https://www.ohiosos.gov/business/business-reports", timeout=45000, wait_until="domcontentloaded") pg.wait_for_timeout(3000) for fn, _ in BIZ_FILES: out = CACHE / fn if out.exists() and out.stat().st_size > 5000: print(f"[cache] {fn}") continue url = f"https://publicfiles.ohiosos.gov/free/OnlineBusinessReports/{fn}" pg.goto(url, timeout=60000, wait_until="domcontentloaded") pg.wait_for_timeout(1200) txt = pg.evaluate("document.body.innerText") if not txt or len(txt) < 500: print(f" FAIL {fn}: {len(txt) if txt else 0} chars"); continue out.write_text(txt) print(f" OK {fn}: {len(txt):,} chars") b.close() def _normalize_addr(s: str) -> str: if not s: return "" s = s.upper().strip() s = re.sub(r"\.\s*", "", s) # drop periods s = re.sub(r",\s*", " ", s) # drop commas s = re.sub(r"\s+", " ", s) # collapse whitespace # street-type synonyms → canonical subs = { r"\bSTREET\b":"ST", r"\bAVENUE\b":"AVE", r"\bBOULEVARD\b":"BLVD", r"\bROAD\b":"RD", r"\bDRIVE\b":"DR", r"\bLANE\b":"LN", r"\bCOURT\b":"CT", r"\bCIRCLE\b":"CIR", r"\bPLACE\b":"PL", r"\bTERRACE\b":"TER", r"\bPARKWAY\b":"PKWY", r"\bHIGHWAY\b":"HWY", r"\bNORTH\b":"N", r"\bSOUTH\b":"S", r"\bEAST\b":"E", r"\bWEST\b":"W", } for k, v in subs.items(): s = re.sub(k, v, s) return s.strip() def _load() -> None: engine = create_engine( f"postgresql+psycopg2://gisapp:{_env('GISPASS')}@127.0.0.1:5432/westerville" ) rows = [] for fn, kind in BIZ_FILES: p = CACHE / fn if not p.exists() or p.stat().st_size < 500: print(f" skip {fn} (missing / small)"); continue with open(p) as fh: for r in csv.DictReader(fh): zip5 = (r.get("FILING ZIP") or "")[:5] rows.append({ "doc_num": (r.get("DOCUMENT NUMBER") or "").strip(), "entity_type": kind, "effective_date": (r.get("EFFECTIVE DATE") or "").strip(), "business_name": (r.get("BUSINESS NAME") or "").strip(), "trans_desc": (r.get("TRANSACTION CODE DESCRIPTION") or "").strip(), "addr1": (r.get("FILING ADDRESS 1") or "").strip(), "addr2": (r.get("FILING ADDRESS 2") or "").strip(), "city": (r.get("FILING CITY") or "").strip(), "state": (r.get("FILING STATE") or "").strip(), "zip": zip5, "addr_norm": _normalize_addr(r.get("FILING ADDRESS 1") or ""), }) df = pd.DataFrame(rows).drop_duplicates(subset=["doc_num"]) print(f"[load] {len(df)} unique filings, across {df['entity_type'].nunique()} types") # write staging table df.to_sql("oh_business_filings", engine, if_exists="replace", index=False) with engine.begin() as conn: conn.execute(text("CREATE INDEX oh_biz_zip_idx ON oh_business_filings (zip);")) conn.execute(text("CREATE INDEX oh_biz_addr_idx ON oh_business_filings (addr_norm);")) conn.execute(text("CREATE INDEX oh_biz_city_idx ON oh_business_filings (city);")) # In-Python match: normalize BOTH sides identically, then merge. parcels_df = pd.read_sql( "SELECT id, adr, lat, lng FROM parcels WHERE lat IS NOT NULL", engine) parcels_df["addr_norm"] = parcels_df["adr"].fillna("").map(_normalize_addr) parcels_df = parcels_df.drop_duplicates(subset=["addr_norm"], keep="first") filings = df[df["zip"].isin(["43082","43081","43015","43074","43021","43054"])].copy() merged = filings.merge( parcels_df.rename(columns={"id":"parcel_id","adr":"parcel_addr"}), on="addr_norm", how="inner", suffixes=("","_p"), ) print(f"[match] {len(merged)} businesses matched to parcel (of {len(filings)} Delaware-area filings)") if len(merged) > 0: gdf = gpd.GeoDataFrame( merged.drop(columns=["addr_norm"]), geometry=[Point(r["lng"], r["lat"]) for _, r in merged.iterrows()], crs="EPSG:4326", ) with engine.begin() as conn: conn.execute(text("DROP TABLE IF EXISTS oh_business_points;")) gdf.to_postgis("oh_business_points", engine, if_exists="replace", index=False) with engine.begin() as conn: conn.execute(text( "CREATE INDEX oh_biz_pt_geom_idx ON oh_business_points USING GIST (geometry);")) (n,) = conn.execute(text("SELECT count(*) FROM oh_business_points")).one() print(f"[postgis] oh_business_points: {n}") def main(): _fetch_all() _load() if __name__ == "__main__": main()