#!/usr/bin/env python3
"""
Recompute the figures in "The county does not call it Charlotte, and other
things the record will not tell you" from the public bulk files.

    https://glvtl.com/field-study/mecklenburg-commercial-record/

By Chris Klebl, Sterling Digital Partners. Free to reuse with attribution.

Standard library only. No install step.

    python3 count_mecklenburg.py                       # download both tables, then count
    python3 count_mecklenburg.py cama.zip sales.zip    # count files you already have

Sources, both free and both a direct download -- no login, no fee, no request form:

  Mecklenburg County, NC, Open Mapping bulk data
  https://maps.mecklenburgcountync.gov/opendata/
    Cama_Table.zip          https://maps.mecklenburgcountync.gov/opendata/Cama_Table.zip
    Parcel_Sales_Table.zip  https://maps.mecklenburgcountync.gov/opendata/Parcel_Sales_Table.zip
  Catalog of all published datasets:
  https://maps.mecklenburgcountync.gov/opendata/data.json

The published figures came from the files dated 2026-09-15 19:57:57 GMT
(Cama_Table.zip) and 2026-09-14 16:35:02 GMT (Parcel_Sales_Table.zip), downloaded
and counted on 2026-09-20. Both tables are republished on the county's own
schedule, so a LATER FILE WILL NOT REPRODUCE THESE FIGURES EXACTLY. That is
expected. What should reproduce is the method: same universe, same columns, same
arithmetic. If a number moves, the date moved with it.

NOTE ON THE SERVER: maps.mecklenburgcountync.gov returns 403 to the default
Python-urllib User-Agent. This script sets one. Still standard library only.

Aggregate counts only. This script never reads, stores or prints an owner name,
a mailing address, a grantor, a grantee or a legal description. The columns that
carry them are named in SUPPRESSED below and are skipped by read_rows().
"""

import csv, io, sys, zipfile, statistics, collections, math
import urllib.request

BASE = "https://maps.mecklenburgcountync.gov/opendata/"
CAMA_URL = BASE + "Cama_Table.zip"
SALES_URL = BASE + "Parcel_Sales_Table.zip"
CAMA_MEMBER = "Cama_Table.csv"
SALES_MEMBER = "Parcel_Sales_Table.csv"

# Columns carrying personal or conveyance-party data. Dropped on read; never counted.
SUPPRESSED = {
    "ownrlstnme", "ownrfrstnme", "ownr2lstnme", "ownr2frstnme", "mailaddr1",
    "mailaddr2", "mailcity", "zipcode", "state", "legaldesc", "grantor",
    "grantee", "full_owner_name", "txt_legaldesc",
}

# Mecklenburg land use codes. The leading letter is the class: C commercial,
# I industrial, O office, R residential, A multi-family, and numeric codes for
# institutional, agricultural and non-taxable land. Commercial real estate here
# means C, I or O -- the direct analogue of DOR 010-049 in studies No. 01 and 02,
# and like those it deliberately EXCLUDES multi-family (A).
CRE_CLASSES = ("C", "I", "O")

# A blank validity code is the county's "Qualified". Every other code is a
# reason the assessor did not treat the transfer as an open-market sale.
# 2024 is study No. 01's window. 2025 is study No. 02's, counted alongside so
# the comparison between the three counties cannot be a window artifact.
SALES_FROM_YEAR = 2024
COMPARISON_YEAR = 2025

# Two parcels sharing a house number and street name are only a genuine
# collision if they are actually somewhere else. Study No. 02 counted every
# address that mapped to more than one parcel and mostly counted condominium
# units, which is an artifact of ownership structure rather than a trap. The
# county publishes a coordinate for every parcel, so here the distance is
# measured instead: a mile apart is not a stacked unit and is not arguable.
COLLISION_MILES = 1.0

# Largest group for which the exact pairwise maximum is computed. The biggest
# key in the September 2026 file holds 408 parcels, so nothing is skipped.
MAX_POINTS = 1000

# Direction words the county writes in front of a street name, when it writes
# one at all. The point of finding 04 is that it usually does not.
DIRECTIONS = {"N", "S", "E", "W", "NE", "NW", "SE", "SW",
              "NORTH", "SOUTH", "EAST", "WEST"}

csv.field_size_limit(1 << 24)
UA = {"User-Agent": "count_mecklenburg/1.0 (+https://glvtl.com/field-study/)"}


def num(s):
    s = (s or "").replace(",", "").replace("$", "").strip()
    try:
        return float(s)
    except ValueError:
        return None


def sale_year(s):
    """'6/4/2007 0:00:00.000' -> 2007."""
    s = (s or "").strip()
    if "/" not in s:
        return None
    tail = s.split("/")[-1].split()[0]
    return int(tail) if tail.isdigit() else None


def cre_class(lusecode):
    c = (lusecode or "").strip().upper()
    return c[:1] if c[:1] in CRE_CLASSES else None


def street_key(streetnumber, streetname):
    """(house number, street name with any leading direction word removed).

    Mecklenburg's own file mostly stores the street WITHOUT its directional
    prefix -- North Tryon Street and South Tryon Street are both 'TRYON ST' --
    and where it does store one it is inconsistent ('W' and 'WEST' both occur).
    Stripping the prefix is therefore not the test being applied to the county;
    it is the state the file is already in, normalised so the two spellings do
    not count as different streets.
    """
    num_ = (streetnumber or "").strip().upper()
    toks = " ".join((streetname or "").upper().split()).split()
    if not num_ or not toks:
        return None, ""
    direction = ""
    if len(toks) > 1 and toks[0] in DIRECTIONS:
        direction, toks = toks[0], toks[1:]
    if not toks:
        return None, ""
    return (num_, " ".join(toks)), direction


def miles(a, b):
    """Great-circle distance between two (lat, lon) pairs."""
    (la1, lo1), (la2, lo2) = a, b
    p = math.pi / 180.0
    h = (0.5 - math.cos((la2 - la1) * p) / 2
         + math.cos(la1 * p) * math.cos(la2 * p) * (1 - math.cos((lo2 - lo1) * p)) / 2)
    return 7917.5 * math.asin(math.sqrt(max(0.0, min(1.0, h))))


def spread(points):
    """Largest distance in miles between any two points. Small sets only."""
    worst = 0.0
    for i in range(len(points)):
        for j in range(i + 1, len(points)):
            worst = max(worst, miles(points[i], points[j]))
    return worst


def fetch(url, path=None):
    if path:
        return zipfile.ZipFile(path)
    sys.stderr.write("downloading %s ...\n" % url)
    with urllib.request.urlopen(urllib.request.Request(url, headers=UA), timeout=900) as r:
        sys.stderr.write("  Last-Modified: %s\n" % r.headers.get("Last-Modified", "not stated"))
        blob = r.read()
    return zipfile.ZipFile(io.BytesIO(blob))


def read_rows(zf, member):
    """Stream one CSV member, dropping every suppressed column before it is seen."""
    with zf.open(member) as fh:
        reader = csv.DictReader(io.TextIOWrapper(fh, encoding="utf-8-sig",
                                                 errors="replace", newline=""))
        for row in reader:
            for col in SUPPRESSED:
                row.pop(col, None)
            yield row


def main(argv):
    paths = [a for a in argv if not a.startswith("--")]
    cama = fetch(CAMA_URL, paths[0] if len(paths) > 0 else None)
    sales = fetch(SALES_URL, paths[1] if len(paths) > 1 else None)

    # ---------- pass 1: the parcel table ----------
    rows = 0
    parcels = set()                      # every distinct parcelid
    cre_parcels = set()                  # distinct commercial/industrial/office
    cre_class_of = {}
    mun = collections.Counter()          # taxing municipality, commercial only
    mun_all = collections.Counter()      # taxing municipality, every parcel
    fire_differs = 0
    market = {}                          # parcelid -> total market value
    eq_market = lt_market = gt_market = has_market = 0
    keyed = collections.defaultdict(set)         # street key -> parcelids
    key_points = collections.defaultdict(dict)   # street key -> parcelid -> (lat, lon)
    key_dirs = collections.defaultdict(set)      # street key -> direction spellings seen
    key_muns = collections.defaultdict(set)      # street key -> municipalities seen
    cre_keys = collections.defaultdict(set)

    for r in read_rows(cama, CAMA_MEMBER):
        rows += 1
        pid = (r.get("parcelid") or "").strip()
        if not pid:
            continue
        first_time = pid not in parcels
        parcels.add(pid)

        cls = cre_class(r.get("lusecode"))
        if cls:
            cre_parcels.add(pid)
            cre_class_of[pid] = cls

        m = (r.get("taxmundist") or "").strip().upper()
        if first_time and m:
            mun_all[m] += 1
        f = (r.get("taxfiredist") or "").strip().upper()
        if first_time and m and f and f not in ("NA", "") and m not in ("NA", ""):
            if f.replace("CITY OF ", "").replace("TOWN OF ", "") != m:
                fire_differs += 1

        tv, tm = num(r.get("totalvalue")), num(r.get("totmarkval"))
        if tm and tm > 0:
            if pid not in market:
                market[pid] = tm
                has_market += 1
                if tv is not None:
                    if tv == tm:
                        eq_market += 1
                    elif tv < tm:
                        lt_market += 1
                    else:
                        gt_market += 1

        key, direction = street_key(r.get("streetnumber"), r.get("streetname"))
        if key:
            keyed[key].add(pid)
            key_dirs[key].add(direction)
            if m:
                key_muns[key].add(m)
            lat, lon = num(r.get("xcoord")), num(r.get("ycoord"))
            if lat and lon and -90 <= lat <= 90 and -180 <= lon <= 180:
                key_points[key][pid] = (lat, lon)
            if cls:
                cre_keys[key].add(pid)

    # second, cheap pass for the commercial municipality split (one row per parcel)
    seen = set()
    for r in read_rows(cama, CAMA_MEMBER):
        pid = (r.get("parcelid") or "").strip()
        if pid in cre_parcels and pid not in seen:
            seen.add(pid)
            mun[(r.get("taxmundist") or "").strip().upper() or "(blank)"] += 1

    # ---------- pass 2: the sales table ----------
    sold = not_qualified = zero_price = blank_price = 0
    sold_cmp = not_qualified_cmp = 0
    by_code = collections.Counter()
    qualified_prices = []       # (parcelid, price) for qualified CRE sales, recent
    for r in read_rows(sales, SALES_MEMBER):
        pid = (r.get("parcelid") or "").strip()
        if pid not in cre_parcels:
            continue
        y = sale_year(r.get("saledate"))
        if y is None or y < SALES_FROM_YEAR:
            continue
        sold += 1
        code = (r.get("salesvalidity") or "").strip()
        label = (r.get("naldesc") or "").strip()
        by_code[(code or "(qualified)", label or "Qualified")] += 1
        price = num(r.get("saleprice"))
        if code:
            not_qualified += 1
        if y >= COMPARISON_YEAR:
            sold_cmp += 1
            if code:
                not_qualified_cmp += 1
        if price is None:
            blank_price += 1
        elif price == 0:
            zero_price += 1
        if not code and price and price > 0 and y >= 2025:
            qualified_prices.append((pid, price))

    # sale price against the county's market value, qualified recent CRE sales
    ratios = []
    above = below = 0
    for pid, price in qualified_prices:
        mv = market.get(pid)
        if mv and mv > 0:
            ratios.append(price / mv)
            if price > mv:
                above += 1
            else:
                below += 1
    ratios.sort()

    # ---------- address collisions ----------
    multi = {k: v for k, v in keyed.items() if len(v) > 1}
    far, skipped = {}, 0
    for k in multi:
        pts = list(key_points[k].values())
        if len(pts) < 2:
            continue
        if len(pts) > MAX_POINTS:
            skipped += 1
            continue
        if spread(pts) >= COLLISION_MILES:
            far[k] = multi[k]
    cre_far = {k: v for k, v in far.items() if cre_keys.get(k)}
    cross_town = {k: v for k, v in far.items() if len(key_muns[k] - {"NA", ""}) > 1}

    # ---------- report ----------
    pct = lambda n, d: "%.1f%%" % (100.0 * n / d) if d else "n/a"
    line = lambda k, v: print("%-52s %s" % (k, v))

    print("=" * 78)
    print("MECKLENBURG COUNTY, NC COMMERCIAL PROPERTY RECORD - COUNTS")
    print("=" * 78)
    line("rows in %s" % CAMA_MEMBER, f"{rows:,}")
    line("distinct parcels", f"{len(parcels):,}")
    line("commercial / industrial / office (use code C, I, O)",
         f"{len(cre_parcels):,}  ({pct(len(cre_parcels), len(parcels))})")

    print("\n-- Finding 01: sales qualification, %d onward --" % SALES_FROM_YEAR)
    line("recorded sales on commercial parcels", f"{sold:,}")
    line("  not qualified (any validity code)", f"{not_qualified:,}  ({pct(not_qualified, sold)})")
    line("  price recorded as exactly $0", f"{zero_price:,}  ({pct(zero_price, sold)})")
    line("  price field empty", f"{blank_price:,}  ({pct(blank_price, sold)})")
    line("  carrying no usable price at all", f"{zero_price + blank_price:,}  ({pct(zero_price + blank_price, sold)})")
    print("  top codes:")
    for (code, label), n in by_code.most_common(8):
        print("    %-12s %6d  %-6s %s" % (code, n, pct(n, sold), label[:46]))
    print("  same count on study No. 02's window (%d onward), so the" % COMPARISON_YEAR)
    print("  comparison between counties is not an artifact of the dates:")
    line("    recorded sales %d onward" % COMPARISON_YEAR, f"{sold_cmp:,}")
    line("    not qualified", f"{not_qualified_cmp:,}  ({pct(not_qualified_cmp, sold_cmp)})")

    print("\n-- Finding 02: which government the parcel answers to --")
    line("commercial parcels placed", f"{sum(mun.values()):,}")
    for m, n in mun.most_common(12):
        line("  " + m.title(), f"{n:,}  ({pct(n, sum(mun.values()))})")
    line("all parcels, City of Charlotte", f"{mun_all.get('CHARLOTTE', 0):,}  ({pct(mun_all.get('CHARLOTTE', 0), sum(mun_all.values()))})")
    line("all parcels, unincorporated", f"{mun_all.get('MECKLENBURG COUNTY-UNINCORPORATED', 0):,}")

    print("\n-- Finding 03: assessed against market, and against the price paid --")
    line("parcels with a market value", f"{has_market:,}")
    line("  total value EQUALS market value", f"{eq_market:,}  ({pct(eq_market, has_market)})")
    line("  total value BELOW market value", f"{lt_market:,}  ({pct(lt_market, has_market)})")
    line("  total value ABOVE market value", f"{gt_market:,}")
    line("qualified commercial sales 2025+ with a market value", f"{len(ratios):,}")
    if ratios:
        q = lambda f: ratios[int(f * (len(ratios) - 1))]
        line("  median sale price / county market value", "%.2f" % statistics.median(ratios))
        line("  quartiles (p25 / p75)", "%.2f / %.2f" % (q(.25), q(.75)))
        line("  deciles  (p10 / p90)", "%.2f / %.2f" % (q(.10), q(.90)))
        line("  sold ABOVE the county's market value", f"{above:,}  ({pct(above, len(ratios))})")
        line("  sold at or below", f"{below:,}  ({pct(below, len(ratios))})")

    print("\n-- Finding 04: address collisions --")
    print("   key = (house number, street name with any leading direction removed)")
    line("distinct address keys", f"{len(keyed):,}")
    line("keys holding 2+ parcels", f"{len(multi):,}  ({pct(len(multi), len(keyed))})")
    line("  ... at least %.0f mile apart on the ground" % COLLISION_MILES,
         f"{len(far):,}")
    line("  parcels behind those keys", f"{sum(len(v) for v in far.values()):,}")
    line("  keys spanning 2+ municipalities", f"{len(cross_town):,}")
    line("  keys including a commercial parcel", f"{len(cre_far):,}")
    line("  commercial parcels involved",
         f"{sum(len(cre_keys[k]) for k in cre_far):,}")
    line("  keys too large to measure (>%d parcels)" % MAX_POINTS, f"{skipped:,}")
    spelled = sum(1 for v in key_dirs.values() if any(d for d in v))
    mixed = sum(1 for v in key_dirs.values() if len([d for d in v if d]) > 1)
    line("keys where the file DOES carry a direction", f"{spelled:,}  ({pct(spelled, len(keyed))})")
    line("  ... spelled two different ways (W and WEST)", f"{mixed:,}")

    print("\nAggregate counts only. No owner, mailing, grantor, grantee or")
    print("legal-description field is read or printed by this script.")


if __name__ == "__main__":
    main(sys.argv[1:])