Joining Property Records to Your Own Data by APN and County FIPS

import pandas as pd

mine = pd.read_csv("my_parcels.csv", dtype=str)       # everything as text

# Case 1: a five-digit code that lost its leading zero ("6037" -> "06037").
mine["county_fips"] = mine["county_fips"].str.strip().str.zfill(5)

# Case 2: separate state and county codes ("6" and "37").
mine["county_fips"] = (
    mine["state_fips"].str.strip().str.zfill(2)
    + mine["county_code"].str.strip().str.zfill(3)
)

# Either way, check the result is exactly five digits.
bad = mine[~mine["county_fips"].str.fullmatch(r"\d{5}", na=False)]
print(len(bad), "rows with an invalid county_fips")

# Case 3: only county and state names. Build a lookup from the Census
# Bureau's county reference file, national_county2020.txt from census.gov
# (pipe-delimited; columns STATE, STATEFP, COUNTYFP, COUNTYNAME, ...).
ref = pd.read_csv("national_county2020.txt", sep="|", dtype=str)
ref["county_fips"] = ref["STATEFP"] + ref["COUNTYFP"]

def county_key(state: pd.Series, county: pd.Series) -> pd.Series:
    # "Fulton County" and "Fulton" both become "GA|FULTON".
    name = (county.fillna("").str.upper()
            .str.replace(r"\b(COUNTY|PARISH|BOROUGH|CENSUS AREA|CITY AND BOROUGH|MUNICIPALITY)\b", "", regex=True)
            .str.replace(r"[^A-Z ]", "", regex=True).str.split().str.join(" "))
    return state.fillna("").str.strip().str.upper() + "|" + name

ref["key"] = county_key(ref["STATE"], ref["COUNTYNAME"])

def add_fips(df: pd.DataFrame, state_col: str, county_col: str) -> pd.DataFrame:
    out = df.assign(key=county_key(df[state_col], df[county_col]))
    out = out.merge(ref[["key", "county_fips"]], on="key", how="left",
                    validate="many_to_one").drop(columns="key")
    print(out["county_fips"].isna().sum(), "rows without a FIPS match; inspect them by name")
    return out

mine = add_fips(mine, "state", "county")        # your own column names

-- The same lookup in Postgres. county_ref is national_county2020.txt loaded
-- as a table (state, statefp, countyfp, countyname); properties is the export
-- loaded with its own header names, so they are quoted.
CREATE OR REPLACE FUNCTION county_key(st text, name text) RETURNS text
LANGUAGE sql IMMUTABLE AS $$
  SELECT upper(trim(coalesce(st, ''))) || '|' || trim(regexp_replace(regexp_replace(regexp_replace(
           upper(coalesce(name, '')),
           '\m(COUNTY|PARISH|BOROUGH|CENSUS AREA|CITY AND BOROUGH|MUNICIPALITY)\M', '', 'g'),
           '[^A-Z ]', '', 'g'), '\s+', ' ', 'g'))
$$;

ALTER TABLE properties ADD COLUMN county_fips char(5);

UPDATE properties p
SET county_fips = r.statefp || r.countyfp
FROM county_ref r
WHERE county_key(r.state, r.countyname) = county_key(p."State", p."County");

SELECT "State", "County", count(*) FROM properties
WHERE county_fips IS NULL GROUP BY 1, 2 ORDER BY 3 DESC;   -- expect few or none

-- One normalised form for both sides: upper case, every character that is
-- not a letter or digit removed. Keep the original column untouched.
CREATE OR REPLACE FUNCTION apn_key(apn text) RETURNS text
LANGUAGE sql IMMUTABLE AS $$
  SELECT nullif(upper(regexp_replace(coalesce(apn, ''), '[^A-Za-z0-9]', '', 'g')), '')
$$;

ALTER TABLE properties ADD COLUMN apn_norm text GENERATED ALWAYS AS (apn_key("APN")) STORED;
ALTER TABLE my_parcels ADD COLUMN apn_norm text GENERATED ALWAYS AS (apn_key(apn)) STORED;

CREATE INDEX ON properties (county_fips, apn_norm);
CREATE INDEX ON my_parcels (county_fips, apn_norm);

-- Did normalising merge two different parcels? Expect no rows.
SELECT county_fips, apn_norm, array_agg("APN") AS originals
FROM properties
GROUP BY county_fips, apn_norm
HAVING count(DISTINCT "APN") > 1;

def apn_key(s: pd.Series) -> pd.Series:
    return s.fillna("").str.upper().str.replace(r"[^A-Z0-9]", "", regex=True).replace("", pd.NA)

props = pd.read_csv("state.csv", dtype={"APN": str, "Property ID": str, "Zip": str,
                                        "Owner Mailing Zip": str})
props = add_fips(props, "State", "County")      # the export has names, not FIPS
props["apn_norm"] = apn_key(props["APN"])
mine["apn_norm"] = apn_key(mine["apn"])

joined = mine.merge(
    props, on=["county_fips", "apn_norm"], how="left",
    suffixes=("", "_prop"), indicator=True, validate="many_to_one",
)

SELECT
  m.county_fips,
  count(*)                                             AS my_rows,
  count(p."APN")                                       AS matched,
  round(100.0 * count(p."APN") / count(*), 1)          AS match_pct
FROM my_parcels m
LEFT JOIN properties p
  ON p.county_fips = m.county_fips AND p.apn_norm = m.apn_norm
GROUP BY m.county_fips
ORDER BY match_pct ASC, my_rows DESC;

-- Compare APN "shapes" for one county: digits become 9, letters become A.
-- Different dominant shapes on the two sides explain most low match rates.
WITH shapes AS (
  SELECT 'export' AS side,
         translate(upper("APN"), '0123456789ABCDEFGHIJKLMNOPQRSTUVWXYZ',
                                 '9999999999AAAAAAAAAAAAAAAAAAAAAAAAAA') AS shape
  FROM properties WHERE county_fips = '13121'
  UNION ALL
  SELECT 'mine', translate(upper(apn), '0123456789ABCDEFGHIJKLMNOPQRSTUVWXYZ',
                                      '9999999999AAAAAAAAAAAAAAAAAAAAAAAAAA')
  FROM my_parcels WHERE county_fips = '13121'
)
SELECT side, shape, count(*) AS n
FROM shapes
GROUP BY side, shape
ORDER BY side, n DESC;

-- If one side drops leading zeros in a section of the APN, pad that side's
-- normalised key to the export's length for the affected county only.
UPDATE my_parcels
SET apn_fixed = lpad(apn_norm, 10, '0')
WHERE county_fips = '13121' AND length(apn_norm) < 10;

import re

SUFFIX = {"STREET": "ST", "AVENUE": "AVE", "ROAD": "RD", "DRIVE": "DR",
          "LANE": "LN", "COURT": "CT", "BOULEVARD": "BLVD", "PLACE": "PL"}
DIRS = {"NORTH": "N", "SOUTH": "S", "EAST": "E", "WEST": "W"}

def addr_key(a: str) -> str | None:
    if not isinstance(a, str):
        return None
    a = re.sub(r"[^A-Z0-9 ]", " ", a.upper())
    a = re.sub(r"\b(APT|UNIT|STE|SUITE)\b.*$", "", a)       # drop unit designators
    words = [SUFFIX.get(w, DIRS.get(w, w)) for w in a.split()]
    if not words or not words[0].isdigit():
        return None                                           # need a house number
    return " ".join(words)

unmatched = joined[joined["_merge"] == "left_only"][mine.columns].copy()
unmatched["addr_key"] = unmatched["address"].map(addr_key)   # your street line
props["addr_key"] = props["Address"].map(addr_key)   # export: street line only

# Only accept an address match that is unique within the county.
candidates = props.dropna(subset=["addr_key"])
counts = candidates.groupby(["county_fips", "addr_key"]).size().rename("n")
unique = candidates.join(counts, on=["county_fips", "addr_key"])
unique = unique[unique["n"] == 1].drop(columns="n")

fallback = unmatched.merge(unique, on=["county_fips", "addr_key"], how="inner",
                           suffixes=("", "_prop"))
fallback["match_method"] = "address"
print(len(fallback), "of", len(unmatched), "unmatched rows recovered by address")

CREATE TABLE parcel_crosswalk (
  my_id          text        NOT NULL,          -- your own record id
  county_fips    char(5)     NOT NULL,          -- derived from State + County
  apn            text        NOT NULL,          -- "APN" as delivered in the export
  match_method   text        NOT NULL CHECK (match_method IN ('apn', 'apn_fixed', 'address', 'manual')),
  matched_month  date        NOT NULL,          -- export month the match was made against
  PRIMARY KEY (my_id)
);