"""
Split Apex certificates of occupancy into detached / townhome / apartment
by joining the Town's CO addresses to Wake County assessor parcels.

  pip install pandas requests
  python3 co_by_type.py apex_co.csv

Writes co_by_type.csv (per-FY table) and co_joined.csv (row-level, for checking).
"""
import sys, re, json, requests, pandas as pd

PARCELS = "https://maps.wake.gov/arcgis/rest/services/Property/Parcels/FeatureServer/0/query"

def norm(a):
    a = str(a).upper().strip()
    a = re.sub(r"[.,#]", " ", a)
    a = re.sub(r"\s+(APT|UNIT|STE|LOT)\s+\S+$", "", a)
    rep = {" DRIVE": " DR", " ROAD": " RD", " STREET": " ST", " LANE": " LN", " COURT": " CT",
           " AVENUE": " AVE", " BOULEVARD": " BLVD", " CIRCLE": " CIR", " PLACE": " PL", " WAY": " WAY",
           " TRAIL": " TRL", " PARKWAY": " PKWY", " ALLEY": " ALY", " TERRACE": " TER", " LOOP": " LOOP",
           " RUN": " RUN", " PATH": " PATH", " ROW": " ROW", " GLEN": " GLN", " POINT": " PT", " COVE": " CV"}
    for k, v in rep.items():
        a = re.sub(k + r"\b", v, a)
    return re.sub(r"\s+", " ", a).strip()

def fetch_parcels():
    addr_f = "SITE_ADDRESS"
    cls_candidates = ["DESIGN_STYLE_DECODE", "TYPE_USE_DECODE", "LAND_CLASS_DECODE"]
    fields = ",".join([addr_f] + cls_candidates + ["PIN_NUM", "REID", "PLANNING_JURISDICTION", "CITY_DECODE", "YEAR_BUILT", "TOTUNITS"])
    where = "PLANNING_JURISDICTION = 'AP' OR UPPER(CITY_DECODE) LIKE 'APEX%'"
    out, offset = [], 0
    while True:
        r = requests.get(PARCELS, params={"where": where, "outFields": fields, "returnGeometry": "false",
                                           "resultOffset": offset, "resultRecordCount": 2000, "f": "json"}, timeout=120).json()
        if "error" in r:
            print("server error:", r["error"]); break
        feats = r.get("features", [])
        out += [f["attributes"] for f in feats]
        print("fetched", len(out), end="\r")
        if len(feats) < 2000: break
        offset += 2000
    df = pd.DataFrame(out)
    print()
    print("DESIGN_STYLE values:", df["DESIGN_STYLE_DECODE"].value_counts().head(12).to_dict())
    print("TYPE_USE values:", df["TYPE_USE_DECODE"].value_counts().head(12).to_dict())
    df["addr_n"] = df[addr_f].map(norm)
    return df, addr_f, cls_candidates

def classify(row, cls_cols):
    text = " ".join(str(row.get(c, "")) for c in cls_cols).upper()
    if "TOWNH" in text or "TOWNHOUSE" in text or "TWNHS" in text: return "Townhome"
    if "CONDO" in text: return "Condo"
    if "APART" in text or "MULTI" in text: return "Apartment"
    if "SINGLE" in text or "SFR" in text or "RESIDENTIAL" in text or "CONVENTIONAL" in text: return "Detached"
    return "Unclassified"

def main(co_path):
    co = pd.read_csv(co_path)
    co["addr_n"] = co["address"].map(norm)
    parcels, addr_f, cls_cols = fetch_parcels()
    parcels = parcels.drop_duplicates("addr_n")
    j = co.merge(parcels[["addr_n"] + cls_cols], on="addr_n", how="left")
    j["class"] = j.apply(lambda r: classify(r, cls_cols), axis=1)
    # Town's own Multi-Family tag wins for apartment certificates
    j.loc[j["type"].astype(str).str.contains("Multi", case=False, na=False), "class"] = "Apartment"
    j.to_csv("co_joined.csv", index=False)
    matched = (j[cls_cols[0]].notna()).mean() if cls_cols else 0
    print(f"\naddress match rate: {matched:.1%}")
    t = j.groupby(["fy_field", "class"])["units"].sum().unstack(fill_value=0)
    t["Total"] = t.sum(axis=1)
    print(t)
    t.to_csv("co_by_type.csv")
    print("\nunclassified sample (check these by hand):")
    print(j[j["class"] == "Unclassified"][["fy_field", "address", "units"]].head(15).to_string())

if __name__ == "__main__":
    main(sys.argv[1] if len(sys.argv) > 1 else "apex_co.csv")