Enrich CRM Property Records Without Overwriting Your Team's Data

# prepare_lookup.py: one lookup row per CRM record (prefer a confirmed Zillow link)
import csv
with open("crm_export.csv", newline="", encoding="utf-8-sig") as f, \
     open("lookup.csv", "w", newline="", encoding="utf-8") as g:
    w = csv.writer(g)
    w.writerow(["id", "input"])
    for r in csv.DictReader(f):
        w.writerow([r["crm_record_id"].strip(), (r.get("zillow_url") or r["address"]).strip()])

"""collect_snapshot.py: look up a supplied list and save every full response.

Usage: python collect_snapshot.py shortlist.csv snapshot.json
shortlist.csv needs id and input columns (input = a Zillow link or a full address).
"""
import csv, json, os, sys
from datetime import datetime, timezone
import requests

BATCH = "https://api.propertiesapi.com/v1/zillow/batch"
CHUNK = 50  # the batch endpoint accepts 1 to 50 items

def load(path):
    with open(path, newline="", encoding="utf-8-sig") as f:
        rows = [{"id": r["id"].strip(), "input": r["input"].strip()} for r in csv.DictReader(f)]
    ids = [r["id"] for r in rows]
    if not rows or len(set(ids)) != len(ids) or not all(r["id"] and r["input"] for r in rows):
        raise SystemExit("every row needs a unique id and an input")
    return rows

def collect(rows):
    out = []
    with requests.Session() as s:
        s.headers["x-api-key"] = os.environ["PROPERTIESAPI_KEY"]  # server side only
        for start in range(0, len(rows), CHUNK):
            chunk = rows[start:start + CHUNK]
            r = s.post(BATCH, json={"items": [x["input"] for x in chunk]},
                       timeout=(10, 330), allow_redirects=False)
            if r.status_code != 200:  # auth, credits, rate limit: stop and keep the rest visible
                print(f"stopped at row {start}: HTTP {r.status_code} {r.text[:300]}", file=sys.stderr)
                out += [{**x, "ok": False, "data": None, "meta": None,
                         "error": {"code": "not_requested"}} for x in rows[start:]]
                break
            for item in r.json()["data"]:
                src = chunk[item["index"]]
                out.append({**src, "ok": item["ok"], "data": item.get("data"),
                            "meta": item.get("meta"), "error": item.get("error")})
    return out

if __name__ == "__main__":
    rows = collect(load(sys.argv[1]))
    snapshot = {"exported_at": datetime.now(timezone.utc).isoformat(), "rows": rows}
    with open(sys.argv[2], "w", encoding="utf-8") as f:
        json.dump(snapshot, f, indent=2)
    failed = [r for r in rows if not r["ok"]]
    print(f"saved {len(rows)} rows, {len(failed)} failed")

"""merge_for_crm.py: build crm_import.csv that only ever writes api_* columns.

Usage: python merge_for_crm.py crm_export.csv snapshot.json crm_import.csv
"""
import csv, json, sys

API_COLUMNS = {                      # CRM column -> response field
    "api_zpid": "zpid", "api_detail_url": "detailUrl", "api_address": "address",
    "api_unit": "unit", "api_home_status": "homeStatus", "api_price": "price",
    "api_currency": "currency", "api_zestimate": "zestimate",
    "api_rent_zestimate": "rentZestimate", "api_beds": "beds", "api_baths": "baths",
    "api_living_area": "livingArea", "api_year_built": "yearBuilt",
}
COMPARE = {"beds": "beds", "baths": "baths", "sqft": "livingArea", "year_built": "yearBuilt"}
OUT = ["crm_record_id", *API_COLUMNS, "api_fetched_at", "api_cached",
       "api_status", "api_error", "api_conflicts"]

def safe(v):  # stop spreadsheet formula injection in text cells
    return "'" + v if isinstance(v, str) and v[:1] in ("=", "+", "-", "@", "\t", "\r") else v

def number(v):
    try:
        return float(v)
    except (TypeError, ValueError):
        return None

def merge(team, result):
    row = {"crm_record_id": team["crm_record_id"]}
    if not result or not result["ok"]:
        row.update(api_status="failed", api_error=((result or {}).get("error") or {}).get("code", "missing"))
        return row                                   # api_* stay empty; nothing is cleared in the CRM
    d, meta = result["data"], result["meta"] or {}
    review = []
    if meta.get("warning"):
        review.append("identity warning")
    if not (d.get("zpid") or d.get("detailUrl")):
        review.append("no returned identity")
    team_unit = (team.get("unit") or "").strip().lstrip("#").lstrip("0").upper()
    api_unit = (d.get("unit") or "").lstrip("0").upper()
    if team_unit and team_unit != api_unit:
        review.append(f"unit differs: team {team_unit or '-'}, source {api_unit or '-'}")
    if review:                                       # do not write observations for an unconfirmed match
        row.update(api_status="needs_review", api_error="; ".join(review))
        return row
    row.update({col: d.get(field) for col, field in API_COLUMNS.items()})
    row.update(api_fetched_at=meta.get("fetched_at"), api_cached=meta.get("cached"), api_status="ok")
    conflicts = []
    for team_col, field in COMPARE.items():
        mine, theirs = number(team.get(team_col)), d.get(field)
        if mine is not None and theirs is not None and mine != theirs:
            conflicts.append(f"{team_col}: team {team[team_col]}, source {theirs}")
    row["api_conflicts"] = "; ".join(conflicts)       # reported, never resolved automatically
    return row

if __name__ == "__main__":
    crm_path, snap_path, out_path = sys.argv[1:4]
    with open(crm_path, newline="", encoding="utf-8-sig") as f:
        team_rows = list(csv.DictReader(f))
    results = {r["id"]: r for r in json.load(open(snap_path, encoding="utf-8"))["rows"]}
    rows = [merge(t, results.get(t["crm_record_id"].strip())) for t in team_rows]
    with open(out_path, "w", newline="", encoding="utf-8") as f:
        w = csv.DictWriter(f, fieldnames=OUT, extrasaction="ignore")
        w.writeheader()
        w.writerows({k: safe(v) for k, v in r.items()} for r in rows)
    counts = {s: sum(r.get("api_status") == s for r in rows) for s in ("ok", "needs_review", "failed")}
    print(counts)