#!/usr/bin/env python3 """ Business list cleanup — NYC Legally Operating Businesses. Takes a raw export and produces three files: clean.csv rows that passed every rule, normalised and deduplicated flagged.csv rows a human needs to look at, each with the reason why report.json counts for every rule that fired Guiding rule: never invent data. If a value can't be repaired with certainty, it is left empty and the row is flagged with the reason. A blank cell with a note is worth more to a client than a confident guess. Standard library only — no install step on the client's machine. Usage: python3 scripts/clean_businesses.py data/raw/in.csv data/clean """ import csv import json import re import sys from collections import Counter, defaultdict from pathlib import Path # Genuine acronyms — these stay upper-cased when a name is rebuilt from ALL CAPS. # Note that Inc./Corp./Co./Ltd. are deliberately NOT here: they are abbreviations, # not acronyms, and a client reading "ACME CORP." next to "Acme Bakery" sees a # half-finished job. ALWAYS_UPPER = { "LLC", "L.L.C.", "LLP", "PLLC", "PC", "P.C.", "LP", "L.P.", "USA", "US", "NY", "NYC", "NJ", "DBA", "RC", "II", "III", "IV", "VI", "VII", "TV", "DJ", "BBQ", "ATM", "HVAC", "AC", "BP", "JFK", } LOWERCASE_PARTICLES = {"and", "of", "the", "for", "at", "in", "on", "de", "la", "van", "von"} # Dates outside this window are placeholders, not facts. MIN_PLAUSIBLE_YEAR = 1950 MAX_PLAUSIBLE_YEAR = 2050 def collapse_whitespace(value): """Trim, and collapse runs of internal whitespace to a single space.""" return re.sub(r"\s+", " ", value).strip() def _capitalise_token(token): """ Capitalise one token without the damage str.title() does. str.title() turns "ALICIA'S" into "Alicia'S" and "WAL-MART" into "Wal-Mart" only by accident. This splits on apostrophes and hyphens explicitly, and leaves a one-letter possessive suffix alone. """ for separator in ("-", "'", "\u2019"): if separator in token: parts = token.split(separator) rebuilt = [_capitalise_token(parts[0])] for part in parts[1:]: # "ALICIA'S" -> "Alicia's", but "O'BRIEN" -> "O'Brien" rebuilt.append(part.lower() if len(part) <= 1 else _capitalise_token(part)) return separator.join(rebuilt) if not token: return token # "MCMILLEN," -> "McMillen,". Applied to the leading run of letters so that # attached punctuation does not defeat it. Deliberately not applied to # "Mac", which would turn MACHINE into MacHine; Mc is unambiguous, Mac is not. letters = re.match(r"^([A-Za-z]+)(.*)$", token) if letters and len(letters.group(1)) > 3 and letters.group(1).upper().startswith("MC"): core = letters.group(1) return "Mc" + core[2].upper() + core[3:].lower() + letters.group(2) return token[0].upper() + token[1:].lower() def title_case_name(raw): """ Rebuild a readable name from an ALL CAPS source. Preserves acronyms, single initials ("J."), and ordinals ("5TH"), all of which naive title-casing mangles into "Llc", "5Th" or "Alicia'S". """ cleaned = collapse_whitespace(raw) if not cleaned: return "" out = [] for index, word in enumerate(cleaned.split(" ")): bare = word.strip(",.").upper() if bare in ALWAYS_UPPER: out.append(word.upper()) elif re.fullmatch(r"\d+(ST|ND|RD|TH)", bare): out.append(bare[:-2] + bare[-2:].lower()) elif re.fullmatch(r"[A-Z]\.?", word): out.append(word.upper()) # single initial elif index > 0 and word.lower() in LOWERCASE_PARTICLES: out.append(word.lower()) else: out.append(_capitalise_token(word)) return " ".join(out) def normalise_phone(raw): """ Return (formatted, extension, error). Accepts the eight formats observed in the source data. A US number is ten digits, optionally prefixed with a country code of 1. Anything else is returned as an error rather than trimmed into looking valid. """ value = collapse_whitespace(raw) if not value: return "", "", None extension = "" ext_match = re.search(r"(?:X|EXT\.?|#)\s*(\d+)$", value, re.IGNORECASE) if ext_match: extension = ext_match.group(1) value = value[: ext_match.start()] digits = re.sub(r"\D", "", value) if len(digits) == 11 and digits.startswith("1"): digits = digits[1:] if len(digits) != 10: return "", extension, f"phone has {len(digits)} digits, expected 10" if digits[0] in "01": return "", extension, "phone area code cannot start with 0 or 1" return f"({digits[:3]}) {digits[3:6]}-{digits[6:]}", extension, None def normalise_zip(raw): """ Return (zip5, zip4, error). Handles the three defects present in the source: ZIP+4 run together without a hyphen, hyphenated ZIP+4, and four-digit codes that lost a leading zero somewhere upstream in a spreadsheet. """ value = collapse_whitespace(raw) if not value: return "", "", None digits = re.sub(r"\D", "", value) if len(digits) == 4: return "0" + digits, "", None # restored a lost leading zero if len(digits) == 5: return digits, "", None if len(digits) == 9: return digits[:5], digits[5:], None return "", "", f"ZIP has {len(digits)} digits, expected 5 or 9" def normalise_date(raw): """ Return (iso_date, error). Strips the time component the export adds, and rejects placeholder dates that are outside a plausible range. """ value = collapse_whitespace(raw) if not value: return "", None match = re.match(r"(\d{4})-(\d{2})-(\d{2})", value) if not match: return "", f"unrecognised date format: {value!r}" year = int(match.group(1)) if year < MIN_PLAUSIBLE_YEAR or year > MAX_PLAUSIBLE_YEAR: return "", f"placeholder date: {match.group(0)}" return match.group(0), None def dedupe_key(row): """ Two rows are the same business when the name and street address match after normalisation. Licence number is deliberately not part of the key: one business renewing under a new licence is the exact duplicate we are trying to collapse. """ def squash(text): return re.sub(r"[^A-Z0-9]", "", text.upper()) address = squash(row["address_building"] + row["address_street_name"] + row["address_zip"]) return squash(row["business_name"]), address def completeness(row): """How many of the fields a client actually wants are populated.""" wanted = ("contact_phone", "address_building", "address_street_name", "address_zip", "address_state", "business_category") return sum(1 for field in wanted if row.get(field, "").strip()) def main(source_path, output_dir): output_dir = Path(output_dir) output_dir.mkdir(parents=True, exist_ok=True) with open(source_path, newline="", encoding="utf-8") as handle: raw_rows = list(csv.DictReader(handle)) stats = Counter() stats["rows_in"] = len(raw_rows) # ---- pass 1: normalise every row, recording what each rule repaired ---- processed = [] for row in raw_rows: issues = [] original_name = row["business_name"] name = title_case_name(original_name) if collapse_whitespace(original_name) != original_name: stats["whitespace_trimmed"] += 1 if original_name.isupper(): stats["names_recased"] += 1 phone, extension, phone_error = normalise_phone(row["contact_phone"]) if phone_error: stats["phones_invalid"] += 1 issues.append(phone_error) elif phone and re.sub(r"\D", "", row["contact_phone"]) != re.sub(r"\D", "", phone): stats["phones_reformatted"] += 1 elif phone and row["contact_phone"] != phone: stats["phones_reformatted"] += 1 if not row["contact_phone"].strip(): stats["phones_missing"] += 1 zip5, zip4, zip_error = normalise_zip(row["address_zip"]) if zip_error: stats["zips_invalid"] += 1 issues.append(zip_error) elif len(re.sub(r"\D", "", row["address_zip"])) == 4: stats["zips_leading_zero_restored"] += 1 elif zip4: stats["zips_plus4_split"] += 1 created, created_error = normalise_date(row["license_creation_date"]) expires, expires_error = normalise_date(row["lic_expir_dd"]) for error in (created_error, expires_error): if error: stats["dates_placeholder" if "placeholder" in error else "dates_unparsed"] += 1 issues.append(error) street = collapse_whitespace(row["address_street_name"]) if street != row["address_street_name"]: stats["whitespace_trimmed"] += 1 state = collapse_whitespace(row["address_state"]).upper() if not state: stats["states_missing"] += 1 issues.append("state is empty") processed.append({ "record": { "license_number": row["license_nbr"], "business_name": name, "trade_name": title_case_name(row["dba_trade_name"]), "category": collapse_whitespace(row["business_category"]), "status": collapse_whitespace(row["license_status"]), "phone": phone, "phone_ext": extension, "address": collapse_whitespace(f"{row['address_building']} {title_case_name(street)}"), "city": title_case_name(row["address_city"]), "state": state, "zip": zip5 + (f"-{zip4}" if zip4 else ""), "licensed_since": created, "expires": expires, }, "key": dedupe_key(row), "completeness": completeness(row), "issues": issues, }) # ---- pass 2: collapse duplicates, keeping the most complete record ---- groups = defaultdict(list) for item in processed: groups[item["key"]].append(item) kept, removed = [], [] for key, members in groups.items(): if len(members) == 1 or not key[1]: kept.extend(members) # no address: can't safely merge continue members.sort(key=lambda item: (item["completeness"], item["record"]["licensed_since"]), reverse=True) kept.append(members[0]) for loser in members[1:]: loser["issues"].append( f"duplicate of licence {members[0]['record']['license_number']}") removed.append(loser) stats["duplicates_removed"] = len(removed) stats["duplicate_groups"] = sum(1 for m in groups.values() if len(m) > 1 and m[0]["key"][1]) clean = [item for item in kept if not item["issues"]] flagged = [item for item in kept if item["issues"]] + removed stats["rows_clean"] = len(clean) stats["rows_flagged"] = len(flagged) # ---- write outputs ---- columns = list(clean[0]["record"].keys()) if clean else [] with open(output_dir / "businesses-clean.csv", "w", newline="", encoding="utf-8") as handle: writer = csv.DictWriter(handle, fieldnames=columns) writer.writeheader() for item in sorted(clean, key=lambda i: i["record"]["business_name"]): writer.writerow(item["record"]) with open(output_dir / "businesses-flagged.csv", "w", newline="", encoding="utf-8") as handle: writer = csv.DictWriter(handle, fieldnames=columns + ["review_reason"]) writer.writeheader() for item in sorted(flagged, key=lambda i: i["record"]["business_name"]): writer.writerow({**item["record"], "review_reason": "; ".join(item["issues"])}) report = { "source": str(source_path), "rules": dict(sorted(stats.items())), } (output_dir / "report.json").write_text(json.dumps(report, indent=2) + "\n") width = max(len(k) for k in stats) for name, count in sorted(stats.items()): print(f"{name.replace('_', ' '):<{width}} {count:>6,}") if __name__ == "__main__": if len(sys.argv) != 3: sys.exit(__doc__.strip().splitlines()[-1]) main(sys.argv[1], sys.argv[2])