#!/usr/bin/env python3 """ MA.csv -> public_directory.people_addresses importer SAFE DEFAULT: Without --commit, this script performs a dry run only. Designed for: Python 3.9+ MariaDB mysql-connector-python Production table: public_directory.people_addresses """ import argparse import csv import datetime import hashlib import json import os import re import sys from pathlib import Path from typing import Dict, Optional, Tuple try: import mysql.connector except ImportError: print("ERROR: mysql-connector-python is not installed.") print("Use the existing Public Directory API virtual environment.") sys.exit(1) PROJECT_ROOT = Path( "/home/mostincr/maillists.mostincrediblevalue.com/data/public-directory" ) REPORT_DIR = PROJECT_ROOT / "import-reports" DEFAULT_ENV_FILE = Path("/etc/public-directory-api.env") MASTER_TABLE = "people_addresses" STAGING_TABLE = "import_ma_staging" TARGET_COLUMNS = [ "first_name", "middle_name", "last_name", "full_name", "title", "street_number", "street", "unit", "city", "state", "zip_code", "county", "country", ] # --------------------------------------------------------- # Utility functions # --------------------------------------------------------- def clean_text(value: Optional[str]) -> Optional[str]: if value is None: return None value = str(value).strip() if not value: return None # Normalize whitespace only. value = re.sub(r"\s+", " ", value) return value.upper() def fit(value: Optional[str], max_length: int) -> Optional[str]: """ Never silently truncate production data. Return None here only when input is None. Length validation happens separately. """ if value is None: return None return value def normalize_zip(value: Optional[str]) -> Optional[str]: """ Preserve Massachusetts leading-zero ZIP codes. Handles: 02038 02038-1234 Also repairs values that a spreadsheet may have rendered as: 2038 2038.0 """ if value is None: return None z = str(value).strip() if not z: return None # Excel-like numeric representation. if re.fullmatch(r"\d+\.0", z): z = z[:-2] z = re.sub(r"\s+", "", z) # Massachusetts input is expected to use US ZIP codes. if re.fullmatch(r"\d{1,5}", z): return z.zfill(5) if re.fullmatch(r"\d{5}-\d{4}", z): return z if re.fullmatch(r"\d{9}", z): return z[:5] + "-" + z[5:] return clean_text(z) def parse_address( full_address: Optional[str], ) -> Tuple[Optional[str], Optional[str], Optional[str], Optional[str]]: """ Returns: street_number, street, unit, error Conservative parser. We keep the original street text as much as possible while separating common unit designators. """ address = clean_text(full_address) if not address: return None, None, None, "missing address" # Extract unit from end of address. # # Examples: # 12 MAIN ST APT 3 # 12 MAIN ST UNIT 3 # 12 MAIN ST STE 200 # 12 MAIN ST #3 unit = None unit_pattern = re.compile( r"\s+(APT(?:ARTMENT)?|UNIT|STE|SUITE|#)\s*([A-Z0-9\-]+)\s*$", re.IGNORECASE, ) m = unit_pattern.search(address) if m: designator = m.group(1).upper() number = m.group(2).upper() if designator == "APARTMENT": designator = "APT" elif designator == "SUITE": designator = "STE" if designator == "#": unit = "#" + number else: unit = designator + " " + number address = address[:m.start()].strip() # Street number may include: # 12 # 12A # 12-14 m = re.match( r"^([0-9]+(?:-[0-9]+)?[A-Z]?)\s+(.+)$", address, re.IGNORECASE, ) if not m: return None, address, unit, "unable to safely parse street number" street_number = clean_text(m.group(1)) street = clean_text(m.group(2)) if not street: return street_number, None, unit, "missing street name" return street_number, street, unit, None def derive_middle_name( first_name: Optional[str], last_name: Optional[str], full_name: Optional[str], ) -> Optional[str]: first_name = clean_text(first_name) last_name = clean_text(last_name) full_name = clean_text(full_name) if not first_name or not last_name or not full_name: return None parts = full_name.split() first_parts = first_name.split() last_parts = last_name.split() if len(parts) <= len(first_parts) + len(last_parts): return None # Require exact known first name at beginning. if parts[:len(first_parts)] != first_parts: return None # Require exact known last name at end. if parts[-len(last_parts):] != last_parts: return None middle = parts[len(first_parts):len(parts) - len(last_parts)] if not middle: return None suffixes = {"JR", "SR", "II", "III", "IV", "V"} if any(x.rstrip(".") in suffixes for x in middle): return None return " ".join(middle) def make_fingerprint(row: Dict[str, Optional[str]]) -> str: """ Stable duplicate fingerprint. Same person + same address => same fingerprint. Different people at same address remain separate. Same person at different address remains separate. """ fields = [ row.get("full_name") or "", row.get("street_number") or "", row.get("street") or "", row.get("unit") or "", row.get("city") or "", row.get("state") or "", row.get("zip_code") or "", ] canonical = "\x1f".join( re.sub(r"\s+", " ", str(v).strip().upper()) for v in fields ) return hashlib.sha256( canonical.encode("utf-8") ).hexdigest() # --------------------------------------------------------- # Environment # --------------------------------------------------------- def load_env_file(path: Path) -> Dict[str, str]: env = {} if not path.exists(): raise RuntimeError("Environment file not found: {}".format(path)) with path.open("r", encoding="utf-8") as f: for raw_line in f: line = raw_line.strip() if not line or line.startswith("#"): continue if "=" not in line: continue key, value = line.split("=", 1) key = key.strip() value = value.strip() # Remove matching outer quotes. if ( len(value) >= 2 and value[0] == value[-1] and value[0] in ("'", '"') ): value = value[1:-1] env[key] = value return env def connect_database(env: Dict[str, str]): required = [ "PD_DB_HOST", "PD_DB_PORT", "PD_DB_USER", "PD_DB_PASSWORD", "PD_DB_NAME", ] missing = [x for x in required if not env.get(x)] if missing: raise RuntimeError( "Missing database environment variables: {}".format( ", ".join(missing) ) ) return mysql.connector.connect( host=env["PD_DB_HOST"], port=int(env["PD_DB_PORT"]), user=env["PD_DB_USER"], password=env["PD_DB_PASSWORD"], database=env["PD_DB_NAME"], autocommit=False, charset="utf8mb4", use_unicode=True, ) # --------------------------------------------------------- # CSV validation / transformation # --------------------------------------------------------- REQUIRED_SOURCE_COLUMNS = { "buyer_first_name", "buyer_last_name", "buyer_full_name", "full_address", "city", "state", "zip", "county", } COLUMN_LIMITS = { "first_name": 40, "middle_name": 40, "last_name": 60, "full_name": 150, "title": 20, "street_number": 20, "street": 100, "unit": 30, "city": 100, "state": 10, "zip_code": 20, "county": 100, "country": 20, } def validate_lengths(row: Dict[str, Optional[str]]) -> Optional[str]: problems = [] for column, limit in COLUMN_LIMITS.items(): value = row.get(column) if value is not None and len(value) > limit: problems.append( "{} length {} exceeds {}".format( column, len(value), limit, ) ) if problems: return "; ".join(problems) return None def transform_source_row( source: Dict[str, str] ) -> Tuple[Dict[str, Optional[str]], Optional[str]]: first_name = clean_text(source.get("buyer_first_name")) last_name = clean_text(source.get("buyer_last_name")) full_name = clean_text(source.get("buyer_full_name")) if not full_name: components = [ x for x in [first_name, last_name] if x ] full_name = " ".join(components) if components else None middle_name = derive_middle_name( first_name, last_name, full_name, ) street_number, street, unit, address_error = parse_address( source.get("full_address") ) city = clean_text(source.get("city")) state = clean_text(source.get("state")) zip_code = normalize_zip(source.get("zip")) county = clean_text(source.get("county")) row = { "first_name": first_name, "middle_name": middle_name, "last_name": last_name, "full_name": full_name, "title": None, "street_number": street_number, "street": street, "unit": unit, "city": city, "state": state, "zip_code": zip_code, "county": county, "country": "USA", } errors = [] if not first_name: errors.append("missing first name") if not last_name: errors.append("missing last name") if not full_name: errors.append("missing full name") if not source.get("full_address", "").strip(): errors.append("missing address") if address_error: errors.append(address_error) if not city: errors.append("missing city") if not state: errors.append("missing state") if not zip_code: errors.append("missing ZIP") elif not re.fullmatch(r"\d{5}(?:-\d{4})?", zip_code): errors.append("malformed ZIP") length_error = validate_lengths(row) if length_error: errors.append(length_error) if errors: return row, "; ".join(dict.fromkeys(errors)) return row, None # --------------------------------------------------------- # Database schema / staging # --------------------------------------------------------- def verify_master_schema(cursor): cursor.execute( """ SELECT COLUMN_NAME, COLUMN_TYPE FROM information_schema.COLUMNS WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = %s ORDER BY ORDINAL_POSITION """, (MASTER_TABLE,), ) actual = {name: ctype for name, ctype in cursor.fetchall()} required = {"id"} | set(TARGET_COLUMNS) missing = required - set(actual) if missing: raise RuntimeError( "Master table is missing required columns: {}".format( ", ".join(sorted(missing)) ) ) def create_staging_table(cursor): cursor.execute( "DROP TEMPORARY TABLE IF EXISTS {}".format(STAGING_TABLE) ) cursor.execute( """ CREATE TEMPORARY TABLE import_ma_staging ( source_row BIGINT UNSIGNED NOT NULL, fingerprint CHAR(64) NOT NULL, first_name VARCHAR(40) NULL, middle_name VARCHAR(40) NULL, last_name VARCHAR(60) NULL, full_name VARCHAR(150) NULL, title VARCHAR(20) NULL, street_number VARCHAR(20) NULL, street VARCHAR(100) NULL, unit VARCHAR(30) NULL, city VARCHAR(100) NULL, state VARCHAR(10) NULL, zip_code VARCHAR(20) NULL, county VARCHAR(100) NULL, country VARCHAR(20) NULL, KEY idx_stage_fingerprint (fingerprint), KEY idx_stage_zip (zip_code) street, full_name ) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci """ ) def insert_stage_batch(cursor, batch): sql = """ INSERT INTO import_ma_staging ( source_row, fingerprint, first_name, middle_name, last_name, full_name, title, street_number, street, unit, city, state, zip_code, county, country ) VALUES ( %s,%s,%s,%s,%s, %s,%s,%s,%s,%s, %s,%s,%s,%s,%s ) """ cursor.executemany(sql, batch) # --------------------------------------------------------- # Duplicate analysis # --------------------------------------------------------- def create_unique_stage_table(cursor): cursor.execute( "DROP TEMPORARY TABLE IF EXISTS import_ma_unique" ) cursor.execute( """ CREATE TEMPORARY TABLE import_ma_unique AS SELECT s.* FROM import_ma_staging s INNER JOIN ( SELECT fingerprint, MIN(source_row) AS keep_row FROM import_ma_staging GROUP BY fingerprint ) d ON d.fingerprint = s.fingerprint AND d.keep_row = s.source_row """ ) cursor.execute( """ ALTER TABLE import_ma_unique ADD KEY idx_unique_zip (zip_code) street, full_name ) """ ) def master_duplicate_condition() -> str: """ We deliberately compare actual normalized fields instead of relying solely on a hash that does not exist in the master table. """ return """ m.full_name = s.full_name AND COALESCE(m.street_number, '') = COALESCE(s.street_number, '') AND COALESCE(m.street, '') = COALESCE(s.street, '') AND COALESCE(m.unit, '') = COALESCE(s.unit, '') AND COALESCE(m.city, '') = COALESCE(s.city, '') AND COALESCE(m.state, '') = COALESCE(s.state, '') AND COALESCE(m.zip_code, '') = COALESCE(s.zip_code, '') """ def analyze_duplicates(cursor): cursor.execute( "SELECT COUNT(*) FROM import_ma_staging" ) valid_stage_rows = cursor.fetchone()[0] cursor.execute( """ SELECT COALESCE(SUM(cnt - 1), 0) FROM ( SELECT COUNT(*) AS cnt FROM import_ma_staging GROUP BY fingerprint HAVING COUNT(*) > 1 ) x """ ) source_duplicates = int(cursor.fetchone()[0] or 0) cursor.execute( "SELECT COUNT(*) FROM import_ma_unique" ) unique_source_rows = cursor.fetchone()[0] duplicate_sql = """ SELECT COUNT(*) FROM import_ma_unique s WHERE EXISTS ( SELECT 1 FROM people_addresses m WHERE m.zip_code = s.zip_code AND m.full_name = s.full_name AND {} LIMIT 1 ) """.format(master_duplicate_condition()) cursor.execute(duplicate_sql) existing_duplicates = cursor.fetchone()[0] candidate_new = unique_source_rows - existing_duplicates return { "valid_stage_rows": valid_stage_rows, "source_duplicates": source_duplicates, "unique_source_rows": unique_source_rows, "existing_duplicates": existing_duplicates, "candidate_new": candidate_new, } # --------------------------------------------------------- # Commit # --------------------------------------------------------- def insert_new_rows(cursor): sql = """ INSERT INTO people_addresses ( first_name, middle_name, last_name, full_name, title, street_number, street, unit, city, state, zip_code, county, country ) SELECT s.first_name, s.middle_name, s.last_name, s.full_name, s.title, s.street_number, s.street, s.unit, s.city, s.state, s.zip_code, s.county, s.country FROM import_ma_unique s WHERE NOT EXISTS ( SELECT 1 FROM people_addresses m WHERE m.zip_code = s.zip_code AND m.full_name = s.full_name AND {} LIMIT 1 ) """.format(master_duplicate_condition()) cursor.execute(sql) return cursor.rowcount # --------------------------------------------------------- # Main # --------------------------------------------------------- def main(): parser = argparse.ArgumentParser( description="Safely merge MA.csv into Public Directory" ) mode = parser.add_mutually_exclusive_group() mode.add_argument( "--dry-run", action="store_true", help="Analyze only. This is the default.", ) mode.add_argument( "--commit", action="store_true", help="Actually insert new records.", ) parser.add_argument( "--input", required=True, help="Path to MA.csv", ) parser.add_argument( "--env-file", default=str(DEFAULT_ENV_FILE), help="Database environment file", ) parser.add_argument( "--batch-size", type=int, default=5000, ) args = parser.parse_args() commit_mode = bool(args.commit) input_path = Path(args.input) if not input_path.exists(): print("ERROR: input file not found: {}".format(input_path)) return 1 REPORT_DIR.mkdir( parents=True, exist_ok=True, ) timestamp = datetime.datetime.now().strftime( "%Y%m%d-%H%M%S" ) rejected_path = REPORT_DIR / ( "ma-import-rejected-{}.csv".format(timestamp) ) report_path = REPORT_DIR / ( "ma-import-report-{}.json".format(timestamp) ) env = load_env_file(Path(args.env_file)) print("=" * 72) print("PUBLIC DIRECTORY — MA CSV IMPORT") print("=" * 72) print("Input: {}".format(input_path)) print( "Mode: {}".format( "COMMIT" if commit_mode else "DRY RUN" ) ) print("Database: {}".format(env.get("PD_DB_NAME"))) print("Master table: {}".format(MASTER_TABLE)) print() conn = connect_database(env) cursor = conn.cursor() try: verify_master_schema(cursor) cursor.execute( "SELECT COUNT(*) FROM people_addresses" ) before_count = cursor.fetchone()[0] print( "Master records before analysis: {:,}".format( before_count ) ) create_staging_table(cursor) counters = { "source_rows": 0, "valid_rows": 0, "rejected_rows": 0, "missing_first_name": 0, "missing_last_name": 0, "missing_address": 0, "missing_city": 0, "missing_state": 0, "missing_zip": 0, "malformed_zip": 0, "address_parse_failure": 0, } rejected_fields = [ "source_row", "reason", "buyer_first_name", "buyer_last_name", "buyer_full_name", "full_address", "city", "state", "zip", "county", ] with rejected_path.open( "w", newline="", encoding="utf-8", ) as rejected_file: rejected_writer = csv.DictWriter( rejected_file, fieldnames=rejected_fields, ) rejected_writer.writeheader() with input_path.open( "r", newline="", encoding="utf-8-sig", errors="replace", ) as csv_file: reader = csv.DictReader(csv_file) if not reader.fieldnames: raise RuntimeError( "CSV has no header row" ) headers = { x.strip() for x in reader.fieldnames if x } missing_headers = ( REQUIRED_SOURCE_COLUMNS - headers ) if missing_headers: raise RuntimeError( "CSV is missing required columns: {}".format( ", ".join( sorted(missing_headers) ) ) ) batch = [] for source_row, source in enumerate( reader, start=2, ): counters["source_rows"] += 1 transformed, error = transform_source_row( source ) if error: counters["rejected_rows"] += 1 e = error.lower() if "missing first name" in e: counters["missing_first_name"] += 1 if "missing last name" in e: counters["missing_last_name"] += 1 if "missing address" in e: counters["missing_address"] += 1 if "missing city" in e: counters["missing_city"] += 1 if "missing state" in e: counters["missing_state"] += 1 if "missing zip" in e: counters["missing_zip"] += 1 if "malformed zip" in e: counters["malformed_zip"] += 1 if ( "unable to safely parse street number" in e ): counters[ "address_parse_failure" ] += 1 rejected_writer.writerow( { "source_row": source_row, "reason": error, "buyer_first_name": source.get( "buyer_first_name", "", ), "buyer_last_name": source.get( "buyer_last_name", "", ), "buyer_full_name": source.get( "buyer_full_name", "", ), "full_address": source.get( "full_address", "", ), "city": source.get("city", ""), "state": source.get("state", ""), "zip": source.get("zip", ""), "county": source.get("county", ""), } ) continue counters["valid_rows"] += 1 fingerprint = make_fingerprint( transformed ) batch.append( ( source_row, fingerprint, transformed["first_name"], transformed["middle_name"], transformed["last_name"], transformed["full_name"], transformed["title"], transformed["street_number"], transformed["street"], transformed["unit"], transformed["city"], transformed["state"], transformed["zip_code"], transformed["county"], transformed["country"], ) ) if len(batch) >= args.batch_size: insert_stage_batch( cursor, batch, ) batch = [] if ( counters["source_rows"] % 10000 == 0 ): print( "Processed={:,} Valid={:,} " "Rejected={:,}".format( counters["source_rows"], counters["valid_rows"], counters["rejected_rows"], ) ) if batch: insert_stage_batch( cursor, batch, ) create_unique_stage_table(cursor) duplicate_stats = analyze_duplicates(cursor) report = { "timestamp": timestamp, "mode": ( "commit" if commit_mode else "dry-run" ), "input_file": str(input_path), "database": env.get("PD_DB_NAME"), "master_table": MASTER_TABLE, "before_count": before_count, **counters, **duplicate_stats, } print() print("=" * 72) print("ANALYSIS") print("=" * 72) print( "Source rows: {:,}".format( counters["source_rows"] ) ) print( "Valid rows: {:,}".format( counters["valid_rows"] ) ) print( "Rejected rows: {:,}".format( counters["rejected_rows"] ) ) print( "Duplicates inside CSV: {:,}".format( duplicate_stats["source_duplicates"] ) ) print( "Already in master: {:,}".format( duplicate_stats["existing_duplicates"] ) ) print( "Candidate new records: {:,}".format( duplicate_stats["candidate_new"] ) ) if not commit_mode: print() print("DRY RUN COMPLETE.") print( "No records were added to people_addresses." ) conn.rollback() report["inserted_rows"] = 0 report["after_count"] = before_count report["status"] = "dry-run-complete" else: print() print( "COMMIT MODE: inserting candidate records..." ) inserted_rows = insert_new_rows(cursor) conn.commit() cursor.execute( "SELECT COUNT(*) FROM people_addresses" ) after_count = cursor.fetchone()[0] expected_difference = inserted_rows actual_difference = after_count - before_count report["inserted_rows"] = inserted_rows report["after_count"] = after_count report["expected_difference"] = ( expected_difference ) report["actual_difference"] = ( actual_difference ) if actual_difference != expected_difference: report["status"] = ( "COMMITTED-WITH-RECONCILIATION-ERROR" ) print() print("WARNING: COUNT RECONCILIATION FAILED.") else: report["status"] = "committed" print() print( "Inserted: {:,}".format( inserted_rows ) ) print( "Master before: {:,}".format( before_count ) ) print( "Master after: {:,}".format( after_count ) ) cursor.execute( "ANALYZE TABLE people_addresses" ) cursor.fetchall() with report_path.open( "w", encoding="utf-8", ) as f: json.dump( report, f, indent=2, sort_keys=True, ) print() print( "Rejected/review CSV:" ) print(rejected_path) print() print( "Import report:" ) print(report_path) return 0 except Exception as exc: conn.rollback() print() print("IMPORT FAILED:") print(str(exc)) return 1 finally: cursor.close() conn.close() if __name__ == "__main__": sys.exit(main()) #!/usr/bin/env python3 """ MA.csv -> public_directory.people_addresses importer SAFE DEFAULT: Without --commit, this script performs a dry run only. Designed for: Python 3.9+ MariaDB mysql-connector-python """ import argparse import csv import datetime import hashlib import json import re import sys from pathlib import Path from typing import Dict, Optional, Tuple try: import mysql.connector except ImportError: print("ERROR: mysql-connector-python is not installed.") print("Use the existing Public Directory API virtual environment.") sys.exit(1) PROJECT_ROOT = Path( "/home/mostincr/maillists.mostincrediblevalue.com/data/public-directory" ) REPORT_DIR = PROJECT_ROOT / "import-reports" DEFAULT_ENV_FILE = Path("/etc/public-directory-api.env") MASTER_TABLE = "people_addresses" STAGING_TABLE = "import_ma_staging" UNIQUE_TABLE = "import_ma_unique" TARGET_COLUMNS = [ "first_name", "middle_name", "last_name", "full_name", "title", "street_number", "street", "unit", "city", "state", "zip_code", "county", "country", ] COLUMN_LIMITS = { "first_name": 40, "middle_name": 40, "last_name": 60, "full_name": 150, "title": 20, "street_number": 20, "street": 100, "unit": 30, "city": 100, "state": 10, "zip_code": 20, "county": 100, "country": 20, } REQUIRED_SOURCE_COLUMNS = { "buyer_first_name", "buyer_last_name", "buyer_full_name", "full_address", "city", "state", "zip", "county", } # ========================================================= # TEXT NORMALIZATION # ========================================================= def clean_text(value: Optional[str]) -> Optional[str]: if value is None: return None value = str(value).strip() if not value: return None value = re.sub(r"\s+", " ", value) return value.upper() def normalize_zip(value: Optional[str]) -> Optional[str]: """ Preserve leading-zero ZIP codes. Examples: 02038 -> 02038 2038 -> 02038 2038.0 -> 02038 020381234 -> 02038-1234 02038-1234 -> 02038-1234 """ if value is None: return None z = str(value).strip() if not z: return None if re.fullmatch(r"\d+\.0", z): z = z[:-2] z = re.sub(r"\s+", "", z) if re.fullmatch(r"\d{1,5}", z): return z.zfill(5) if re.fullmatch(r"\d{9}", z): return z[:5] + "-" + z[5:] if re.fullmatch(r"\d{5}-\d{4}", z): return z return clean_text(z) # ========================================================= # ADDRESS PARSING # ========================================================= def parse_address( full_address: Optional[str], ) -> Tuple[ Optional[str], Optional[str], Optional[str], Optional[str], ]: """ Returns: street_number street unit parsing_error This parser is intentionally conservative. """ address = clean_text(full_address) if not address: return None, None, None, "missing address" unit = None # Examples: # # 12 MAIN ST APT 3 # 12 MAIN ST APARTMENT 3 # 12 MAIN ST UNIT 3 # 12 MAIN ST STE 200 # 12 MAIN ST SUITE 200 # 12 MAIN ST #3 unit_pattern = re.compile( r"\s+(APT(?:ARTMENT)?|UNIT|STE|SUITE|#)\s*" r"([A-Z0-9\-]+)\s*$", re.IGNORECASE, ) unit_match = unit_pattern.search(address) if unit_match: designator = unit_match.group(1).upper() unit_number = unit_match.group(2).upper() if designator == "APARTMENT": designator = "APT" if designator == "SUITE": designator = "STE" if designator == "#": unit = "#" + unit_number else: unit = "{} {}".format( designator, unit_number, ) address = address[:unit_match.start()].strip() # Handles: # # 12 MAIN ST # 12A MAIN ST # 12-14 MAIN ST street_match = re.match( r"^([0-9]+(?:-[0-9]+)?[A-Z]?)\s+(.+)$", address, re.IGNORECASE, ) if not street_match: return ( None, address, unit, "unable to safely parse street number", ) street_number = clean_text( street_match.group(1) ) street = clean_text( street_match.group(2) ) if not street: return ( street_number, None, unit, "missing street name", ) return ( street_number, street, unit, None, ) # ========================================================= # NAME HANDLING # ========================================================= def derive_middle_name( first_name: Optional[str], last_name: Optional[str], full_name: Optional[str], ) -> Optional[str]: first_name = clean_text(first_name) last_name = clean_text(last_name) full_name = clean_text(full_name) if not first_name: return None if not last_name: return None if not full_name: return None parts = full_name.split() first_parts = first_name.split() last_parts = last_name.split() if len(parts) <= ( len(first_parts) + len(last_parts) ): return None if parts[:len(first_parts)] != first_parts: return None if parts[-len(last_parts):] != last_parts: return None middle = parts[ len(first_parts): len(parts) - len(last_parts) ] if not middle: return None suffixes = { "JR", "SR", "II", "III", "IV", "V", } for item in middle: if item.rstrip(".") in suffixes: return None return " ".join(middle) # ========================================================= # DUPLICATE FINGERPRINT # ========================================================= def make_fingerprint( row: Dict[str, Optional[str]] ) -> str: """ Fingerprint represents: same person + same address Different people at the same address remain separate. Same person at different addresses remains separate. """ fields = [ row.get("full_name") or "", row.get("street_number") or "", row.get("street") or "", row.get("unit") or "", row.get("city") or "", row.get("state") or "", row.get("zip_code") or "", ] canonical = "\x1f".join( re.sub( r"\s+", " ", str(value).strip().upper(), ) for value in fields ) return hashlib.sha256( canonical.encode("utf-8") ).hexdigest() # ========================================================= # ENVIRONMENT FILE # ========================================================= def load_env_file(path: Path) -> Dict[str, str]: env = {} if not path.exists(): raise RuntimeError( "Environment file not found: {}".format( path ) ) with path.open( "r", encoding="utf-8", ) as file_handle: for raw_line in file_handle: line = raw_line.strip() if not line: continue if line.startswith("#"): continue if "=" not in line: continue key, value = line.split( "=", 1, ) key = key.strip() value = value.strip() if ( len(value) >= 2 and value[0] == value[-1] and value[0] in ("'", '"') ): value = value[1:-1] env[key] = value return env # ========================================================= # DATABASE CONNECTION # ========================================================= def connect_database( env: Dict[str, str] ): required = [ "PD_DB_HOST", "PD_DB_PORT", "PD_DB_USER", "PD_DB_PASSWORD", "PD_DB_NAME", ] missing = [ key for key in required if not env.get(key) ] if missing: raise RuntimeError( "Missing database environment variables: {}".format( ", ".join(missing) ) ) return mysql.connector.connect( host=env["PD_DB_HOST"], port=int(env["PD_DB_PORT"]), user=env["PD_DB_USER"], password=env["PD_DB_PASSWORD"], database=env["PD_DB_NAME"], autocommit=False, charset="utf8mb4", use_unicode=True, ) # ========================================================= # FIELD LENGTH VALIDATION # ========================================================= def validate_lengths( row: Dict[str, Optional[str]] ) -> Optional[str]: problems = [] for column, limit in COLUMN_LIMITS.items(): value = row.get(column) if ( value is not None and len(value) > limit ): problems.append( "{} length {} exceeds {}".format( column, len(value), limit, ) ) if problems: return "; ".join(problems) return None # ========================================================= # SOURCE ROW TRANSFORMATION # ========================================================= def transform_source_row( source: Dict[str, str] ) -> Tuple[ Dict[str, Optional[str]], Optional[str], ]: first_name = clean_text( source.get("buyer_first_name") ) last_name = clean_text( source.get("buyer_last_name") ) full_name = clean_text( source.get("buyer_full_name") ) if not full_name: components = [ value for value in [ first_name, last_name, ] if value ] if components: full_name = " ".join(components) middle_name = derive_middle_name( first_name, last_name, full_name, ) ( street_number, street, unit, address_error, ) = parse_address( source.get("full_address") ) city = clean_text( source.get("city") ) state = clean_text( source.get("state") ) zip_code = normalize_zip( source.get("zip") ) county = clean_text( source.get("county") ) row = { "first_name": first_name, "middle_name": middle_name, "last_name": last_name, "full_name": full_name, "title": None, "street_number": street_number, "street": street, "unit": unit, "city": city, "state": state, "zip_code": zip_code, "county": county, "country": "USA", } errors = [] if not first_name: errors.append( "missing first name" ) if not last_name: errors.append( "missing last name" ) if not full_name: errors.append( "missing full name" ) raw_address = source.get( "full_address", "", ) if not str(raw_address).strip(): errors.append( "missing address" ) if address_error: errors.append( address_error ) if not city: errors.append( "missing city" ) if not state: errors.append( "missing state" ) if not zip_code: errors.append( "missing ZIP" ) elif not re.fullmatch( r"\d{5}(?:-\d{4})?", zip_code, ): errors.append( "malformed ZIP" ) length_error = validate_lengths( row ) if length_error: errors.append( length_error ) if errors: # Remove duplicate error messages errors = list( dict.fromkeys(errors) ) return ( row, "; ".join(errors), ) return row, None # ========================================================= # MASTER SCHEMA CHECK # ========================================================= def verify_master_schema(cursor): cursor.execute( """ SELECT COLUMN_NAME, COLUMN_TYPE FROM information_schema.COLUMNS WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = %s ORDER BY ORDINAL_POSITION """, (MASTER_TABLE,), ) actual = { name: column_type for name, column_type in cursor.fetchall() } required = ( {"id"} | set(TARGET_COLUMNS) ) missing = ( required - set(actual) ) if missing: raise RuntimeError( "Master table is missing required columns: {}".format( ", ".join( sorted(missing) ) ) ) # ========================================================= # TEMPORARY STAGING TABLE # ========================================================= def create_staging_table(cursor): cursor.execute( """ DROP TEMPORARY TABLE IF EXISTS import_ma_staging """ ) cursor.execute( """ CREATE TEMPORARY TABLE import_ma_staging ( source_row BIGINT UNSIGNED NOT NULL, fingerprint CHAR(64) NOT NULL, first_name VARCHAR(40) NULL, middle_name VARCHAR(40) NULL, last_name VARCHAR(60) NULL, full_name VARCHAR(150) NULL, title VARCHAR(20) NULL, street_number VARCHAR(20) NULL, street VARCHAR(100) NULL, unit VARCHAR(30) NULL, city VARCHAR(100) NULL, state VARCHAR(10) NULL, zip_code VARCHAR(20) NULL, county VARCHAR(100) NULL, country VARCHAR(20) NULL, KEY idx_stage_fingerprint ( fingerprint ), KEY idx_stage_zip (zip_code) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci """ ) # ========================================================= # INSERT STAGING BATCH # ========================================================= def insert_stage_batch( cursor, batch, ): sql = """ INSERT INTO import_ma_staging ( source_row, fingerprint, first_name, middle_name, last_name, full_name, title, street_number, street, unit, city, state, zip_code, county, country ) VALUES ( %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s ) """ cursor.executemany( sql, batch, ) # ========================================================= # CREATE DEDUPLICATED STAGING TABLE # ========================================================= def create_unique_stage_table(cursor): cursor.execute( """ DROP TEMPORARY TABLE IF EXISTS import_ma_unique """ ) cursor.execute( """ CREATE TEMPORARY TABLE import_ma_unique AS SELECT s.* FROM import_ma_staging s INNER JOIN ( SELECT fingerprint, MIN(source_row) AS keep_row FROM import_ma_staging GROUP BY fingerprint ) duplicates ON duplicates.fingerprint = s.fingerprint AND duplicates.keep_row = s.source_row """ ) # IMPORTANT: # # Do not create the original very large composite # UTF8MB4 index here. It can exceed MariaDB's # maximum key length. # # ZIP + a prefix of full_name is sufficient to # narrow duplicate candidates before comparing # the complete address. cursor.execute( """ ALTER TABLE import_ma_unique ADD KEY idx_unique_zip (zip_code) """ ) # ========================================================= # MASTER DUPLICATE CONDITION # ========================================================= def master_duplicate_condition() -> str: return """ m.full_name = s.full_name AND COALESCE( m.street_number, '' ) = COALESCE( s.street_number, '' ) AND COALESCE( m.street, '' ) = COALESCE( s.street, '' ) AND COALESCE( m.unit, '' ) = COALESCE( s.unit, '' ) AND COALESCE( m.city, '' ) = COALESCE( s.city, '' ) AND COALESCE( m.state, '' ) = COALESCE( s.state, '' ) AND COALESCE( m.zip_code, '' ) = COALESCE( s.zip_code, '' ) """ # ========================================================= # DUPLICATE ANALYSIS # ========================================================= def analyze_duplicates(cursor): cursor.execute( """ SELECT COUNT(*) FROM import_ma_staging """ ) valid_stage_rows = ( cursor.fetchone()[0] ) cursor.execute( """ SELECT COALESCE( SUM(cnt - 1), 0 ) FROM ( SELECT COUNT(*) AS cnt FROM import_ma_staging GROUP BY fingerprint HAVING COUNT(*) > 1 ) duplicate_groups """ ) source_duplicates = int( cursor.fetchone()[0] or 0 ) cursor.execute( """ SELECT COUNT(*) FROM import_ma_unique """ ) unique_source_rows = ( cursor.fetchone()[0] ) duplicate_sql = """ SELECT COUNT(*) FROM import_ma_unique s WHERE EXISTS ( SELECT 1 FROM people_addresses m WHERE m.zip_code = s.zip_code AND m.full_name = s.full_name AND {} LIMIT 1 ) """.format( master_duplicate_condition() ) cursor.execute( duplicate_sql ) existing_duplicates = ( cursor.fetchone()[0] ) candidate_new = ( unique_source_rows - existing_duplicates ) return { "valid_stage_rows": valid_stage_rows, "source_duplicates": source_duplicates, "unique_source_rows": unique_source_rows, "existing_duplicates": existing_duplicates, "candidate_new": candidate_new, } # ========================================================= # PRODUCTION INSERT # ========================================================= def insert_new_rows(cursor): sql = """ INSERT INTO people_addresses ( first_name, middle_name, last_name, full_name, title, street_number, street, unit, city, state, zip_code, county, country ) SELECT s.first_name, s.middle_name, s.last_name, s.full_name, s.title, s.street_number, s.street, s.unit, s.city, s.state, s.zip_code, s.county, s.country FROM import_ma_unique s WHERE NOT EXISTS ( SELECT 1 FROM people_addresses m WHERE m.zip_code = s.zip_code AND m.full_name = s.full_name AND {} LIMIT 1 ) """.format( master_duplicate_condition() ) cursor.execute(sql) return cursor.rowcount # ========================================================= # MAIN # ========================================================= def main(): parser = argparse.ArgumentParser( description=( "Safely merge MA.csv " "into Public Directory" ) ) mode = ( parser.add_mutually_exclusive_group() ) mode.add_argument( "--dry-run", action="store_true", help=( "Analyze only. " "This is the default." ), ) mode.add_argument( "--commit", action="store_true", help=( "Actually insert new records." ), ) parser.add_argument( "--input", required=True, help="Path to MA.csv", ) parser.add_argument( "--env-file", default=str( DEFAULT_ENV_FILE ), help=( "Database environment file" ), ) parser.add_argument( "--batch-size", type=int, default=5000, ) args = parser.parse_args() commit_mode = bool( args.commit ) input_path = Path( args.input ) if not input_path.exists(): print( "ERROR: input file not found: {}".format( input_path ) ) return 1 REPORT_DIR.mkdir( parents=True, exist_ok=True, ) timestamp = ( datetime.datetime.now().strftime( "%Y%m%d-%H%M%S" ) ) rejected_path = ( REPORT_DIR / "ma-import-rejected-{}.csv".format( timestamp ) ) report_path = ( REPORT_DIR / "ma-import-report-{}.json".format( timestamp ) ) env = load_env_file( Path(args.env_file) ) print("=" * 72) print( "PUBLIC DIRECTORY — MA CSV IMPORT" ) print("=" * 72) print( "Input: {}".format( input_path ) ) print( "Mode: {}".format( "COMMIT" if commit_mode else "DRY RUN" ) ) print( "Database: {}".format( env.get("PD_DB_NAME") ) ) print( "Master table: {}".format( MASTER_TABLE ) ) print() conn = connect_database( env ) cursor = conn.cursor() try: # --------------------------------------------- # Verify production schema # --------------------------------------------- verify_master_schema( cursor ) cursor.execute( """ SELECT COUNT(*) FROM people_addresses """ ) before_count = ( cursor.fetchone()[0] ) print( "Master records before analysis: {:,}".format( before_count ) ) # --------------------------------------------- # Temporary staging # --------------------------------------------- create_staging_table( cursor ) counters = { "source_rows": 0, "valid_rows": 0, "rejected_rows": 0, "missing_first_name": 0, "missing_last_name": 0, "missing_address": 0, "missing_city": 0, "missing_state": 0, "missing_zip": 0, "malformed_zip": 0, "address_parse_failure": 0, } rejected_fields = [ "source_row", "reason", "buyer_first_name", "buyer_last_name", "buyer_full_name", "full_address", "city", "state", "zip", "county", ] # --------------------------------------------- # Parse CSV # --------------------------------------------- with rejected_path.open( "w", newline="", encoding="utf-8", ) as rejected_file: rejected_writer = ( csv.DictWriter( rejected_file, fieldnames=rejected_fields, ) ) rejected_writer.writeheader() with input_path.open( "r", newline="", encoding="utf-8-sig", errors="replace", ) as csv_file: reader = csv.DictReader( csv_file ) if not reader.fieldnames: raise RuntimeError( "CSV has no header row" ) headers = { header.strip() for header in reader.fieldnames if header } missing_headers = ( REQUIRED_SOURCE_COLUMNS - headers ) if missing_headers: raise RuntimeError( "CSV is missing required columns: {}".format( ", ".join( sorted( missing_headers ) ) ) ) batch = [] for ( source_row, source, ) in enumerate( reader, start=2, ): counters[ "source_rows" ] += 1 ( transformed, error, ) = transform_source_row( source ) # --------------------------------- # Rejected record # --------------------------------- if error: counters[ "rejected_rows" ] += 1 lower_error = ( error.lower() ) if ( "missing first name" in lower_error ): counters[ "missing_first_name" ] += 1 if ( "missing last name" in lower_error ): counters[ "missing_last_name" ] += 1 if ( "missing address" in lower_error ): counters[ "missing_address" ] += 1 if ( "missing city" in lower_error ): counters[ "missing_city" ] += 1 if ( "missing state" in lower_error ): counters[ "missing_state" ] += 1 if ( "missing zip" in lower_error ): counters[ "missing_zip" ] += 1 if ( "malformed zip" in lower_error ): counters[ "malformed_zip" ] += 1 if ( "unable to safely parse street number" in lower_error ): counters[ "address_parse_failure" ] += 1 rejected_writer.writerow( { "source_row": source_row, "reason": error, "buyer_first_name": source.get( "buyer_first_name", "", ), "buyer_last_name": source.get( "buyer_last_name", "", ), "buyer_full_name": source.get( "buyer_full_name", "", ), "full_address": source.get( "full_address", "", ), "city": source.get( "city", "", ), "state": source.get( "state", "", ), "zip": source.get( "zip", "", ), "county": source.get( "county", "", ), } ) continue # --------------------------------- # Valid record # --------------------------------- counters[ "valid_rows" ] += 1 fingerprint = ( make_fingerprint( transformed ) ) batch.append( ( source_row, fingerprint, transformed[ "first_name" ], transformed[ "middle_name" ], transformed[ "last_name" ], transformed[ "full_name" ], transformed[ "title" ], transformed[ "street_number" ], transformed[ "street" ], transformed[ "unit" ], transformed[ "city" ], transformed[ "state" ], transformed[ "zip_code" ], transformed[ "county" ], transformed[ "country" ], ) ) if ( len(batch) >= args.batch_size ): insert_stage_batch( cursor, batch, ) batch = [] if ( counters[ "source_rows" ] % 10000 == 0 ): print( "Processed={:,} " "Valid={:,} " "Rejected={:,}".format( counters[ "source_rows" ], counters[ "valid_rows" ], counters[ "rejected_rows" ], ) ) if batch: insert_stage_batch( cursor, batch, ) # --------------------------------------------- # Deduplicate incoming CSV # --------------------------------------------- create_unique_stage_table( cursor ) # --------------------------------------------- # Compare with production master # --------------------------------------------- duplicate_stats = ( analyze_duplicates( cursor ) ) report = { "timestamp": timestamp, "mode": ( "commit" if commit_mode else "dry-run" ), "input_file": str(input_path), "database": env.get( "PD_DB_NAME" ), "master_table": MASTER_TABLE, "before_count": before_count, **counters, **duplicate_stats, } # --------------------------------------------- # Display analysis # --------------------------------------------- print() print("=" * 72) print("ANALYSIS") print("=" * 72) print( "Source rows: {:,}".format( counters[ "source_rows" ] ) ) print( "Valid rows: {:,}".format( counters[ "valid_rows" ] ) ) print( "Rejected rows: {:,}".format( counters[ "rejected_rows" ] ) ) print( "Duplicates inside CSV: {:,}".format( duplicate_stats[ "source_duplicates" ] ) ) print( "Already in master: {:,}".format( duplicate_stats[ "existing_duplicates" ] ) ) print( "Candidate new records: {:,}".format( duplicate_stats[ "candidate_new" ] ) ) # --------------------------------------------- # DRY RUN # --------------------------------------------- if not commit_mode: print() print( "DRY RUN COMPLETE." ) print( "No records were added " "to people_addresses." ) conn.rollback() report[ "inserted_rows" ] = 0 report[ "after_count" ] = before_count report[ "status" ] = "dry-run-complete" # --------------------------------------------- # COMMIT MODE # --------------------------------------------- else: print() print( "COMMIT MODE: " "inserting candidate records..." ) inserted_rows = ( insert_new_rows( cursor ) ) conn.commit() cursor.execute( """ SELECT COUNT(*) FROM people_addresses """ ) after_count = ( cursor.fetchone()[0] ) expected_difference = ( inserted_rows ) actual_difference = ( after_count - before_count ) report[ "inserted_rows" ] = inserted_rows report[ "after_count" ] = after_count report[ "expected_difference" ] = expected_difference report[ "actual_difference" ] = actual_difference if ( actual_difference != expected_difference ): report[ "status" ] = ( "COMMITTED-WITH-" "RECONCILIATION-ERROR" ) print() print( "WARNING: " "COUNT RECONCILIATION FAILED." ) else: report[ "status" ] = "committed" print() print( "Inserted: {:,}".format( inserted_rows ) ) print( "Master before: {:,}".format( before_count ) ) print( "Master after: {:,}".format( after_count ) ) # Update optimizer statistics. # This does NOT rebuild all indexes. cursor.execute( """ ANALYZE TABLE people_addresses """ ) cursor.fetchall() # --------------------------------------------- # Write JSON report # --------------------------------------------- with report_path.open( "w", encoding="utf-8", ) as report_file: json.dump( report, report_file, indent=2, sort_keys=True, ) print() print( "Rejected/review CSV:" ) print( rejected_path ) print() print( "Import report:" ) print( report_path ) return 0 except Exception as exc: conn.rollback() print() print( "IMPORT FAILED:" ) print( str(exc) ) return 1 finally: cursor.close() conn.close() if __name__ == "__main__": sys.exit( main() )