#!/usr/bin/env python3 """Read-only HubSpot property retirement audit. Ranks custom properties by how safe they are to retire, using fill rate, last written date and the sources (property history) that wrote values. Live mode needs env HUBSPOT_TOKEN (private app token with read scopes only). Mock mode (--mock sample/) swaps ONLY the transport; every other function is the same code path. This script never writes to HubSpot. It only issues GET requests and POST requests to the search endpoint (which is a read). """ import argparse import csv import difflib import json import os import re import sys import time import urllib.error import urllib.parse import urllib.request from collections import Counter from datetime import datetime, timezone DEFAULT_BASE_URL = "https://api.hubapi.com" LOW_FILL_PCT = 2.0 STALE_DAYS = 365 VERDICT_ORDER = {"RETIRE_CANDIDATE": 0, "REVIEW": 1, "KEEP": 2} RISKY_SOURCES = {"INTEGRATION", "API"} # The "last modified" property differs by object type. Check the current docs. LASTMOD_PROP = { "contacts": "lastmodifieddate", "companies": "hs_lastmodifieddate", "deals": "hs_lastmodifieddate", } GENERIC_TOKENS = {"date", "time", "at", "the", "of", "id", "name", "number", "status", "type", "is", "a", "an", "and", "for", "to"} DATE_TOKENS = {"date", "time"} # --------------------------------------------------------------- transports class LiveTransport: """The only code that talks to the network.""" def __init__(self, token, base_url=DEFAULT_BASE_URL, sleep=0.25, max_retries=5): self.token = token self.base_url = base_url.rstrip("/") self.sleep = sleep self.max_retries = max_retries def _send(self, method, path, params=None, body=None): url = self.base_url + path if params: url += "?" + urllib.parse.urlencode(params) data = json.dumps(body).encode() if body is not None else None headers = {"Authorization": "Bearer " + self.token, "Content-Type": "application/json"} for attempt in range(self.max_retries + 1): time.sleep(self.sleep) req = urllib.request.Request(url, data=data, headers=headers, method=method) try: with urllib.request.urlopen(req, timeout=60) as resp: return json.loads(resp.read().decode() or "{}") except urllib.error.HTTPError as e: if e.code in (429, 502, 503, 504) and attempt < self.max_retries: wait = e.headers.get("Retry-After") if e.headers else None try: wait = float(wait) except (TypeError, ValueError): wait = min(2 ** attempt, 30) time.sleep(wait) continue raise RuntimeError("HTTP %s from %s %s" % (e.code, method, path)) raise RuntimeError("retries exhausted for %s %s" % (method, path)) def http_get(self, path, params=None): return self._send("GET", path, params=params) def http_post(self, path, body): return self._send("POST", path, body=body) class MockTransport: """Answers the same calls from fixture files in sample/.""" def __init__(self, sample_dir): def load(name): with open(os.path.join(sample_dir, name), encoding="utf-8") as f: return json.load(f) self.properties = load("properties.json") self.totals = load("search_totals.json") self.history = load("history.json") self.calls = [] def _obj(self, object_type): if object_type not in self.totals: raise RuntimeError("sample/ only has fixtures for: " + ", ".join(self.totals)) return self.totals[object_type] def http_get(self, path, params=None): self.calls.append(("GET", path)) m = re.fullmatch(r"/crm/v3/properties/(\w+)", path) if m: self._obj(m.group(1)) return self.properties m = re.fullmatch(r"/crm/v3/objects/(\w+)/(\w+)", path) if m: self._obj(m.group(1)) rec_id = m.group(2) prop = (params or {}).get("propertiesWithHistory") entries = self.history.get(rec_id, {}).get(prop, []) return {"id": rec_id, "propertiesWithHistory": {prop: entries}} raise RuntimeError("mock: unhandled GET " + path) def http_post(self, path, body): self.calls.append(("POST", path)) m = re.fullmatch(r"/crm/v3/objects/(\w+)/search", path) if not m: raise RuntimeError("mock: unhandled POST " + path) obj = self._obj(m.group(1)) groups = body.get("filterGroups") or [] if not groups: return {"total": obj["total_records"], "results": []} prop = groups[0]["filters"][0]["propertyName"] total = obj["fill"].get(prop, 0) ids = obj["recent_ids"].get(prop, [])[: body.get("limit", 1)] if body.get("limit", 1) == 1 and not body.get("sorts"): ids = [] return {"total": total, "results": [{"id": i} for i in ids]} # ------------------------------------------------------------------ helpers def parse_ts(s): if not s: return None s = s.replace("Z", "+00:00") try: d = datetime.fromisoformat(s) except ValueError: return None if d.tzinfo is None: d = d.replace(tzinfo=timezone.utc) return d def tokens(label): return [t for t in re.split(r"[^a-z0-9]+", label.lower()) if t] # ------------------------------------------------------------- data fetching def list_custom_properties(api, object_type): data = api.http_get("/crm/v3/properties/" + object_type) return [p for p in data.get("results", []) if not p.get("hubspotDefined", False)] def count_total_records(api, object_type): data = api.http_post("/crm/v3/objects/%s/search" % object_type, {"filterGroups": [], "limit": 1}) return int(data.get("total", 0)) def has_property_body(name, limit, sort_prop=None): body = {"filterGroups": [{"filters": [{"propertyName": name, "operator": "HAS_PROPERTY"}]}], "limit": limit} if sort_prop: body["sorts"] = [{"propertyName": sort_prop, "direction": "DESCENDING"}] return body def fill_count(api, object_type, name): data = api.http_post("/crm/v3/objects/%s/search" % object_type, has_property_body(name, 1)) return int(data.get("total", 0)) def sample_ids(api, object_type, name, sample_size): data = api.http_post("/crm/v3/objects/%s/search" % object_type, has_property_body(name, sample_size, LASTMOD_PROP[object_type])) return [r["id"] for r in data.get("results", [])] def collect_history(api, object_type, name, ids): entries = [] for rec_id in ids: data = api.http_get("/crm/v3/objects/%s/%s" % (object_type, rec_id), {"propertiesWithHistory": name}) for e in (data.get("propertiesWithHistory") or {}).get(name, []): entries.append({"timestamp": e.get("timestamp"), "sourceType": e.get("sourceType") or "UNKNOWN", "sourceId": e.get("sourceId"), "sourceLabel": e.get("sourceLabel")}) return entries # ------------------------------------------------------------------ scoring def score(fill_pct, filled, last_written, sources, now): """Return (verdict, reasons). Rules are deliberately simple and listed in README.""" reasons = [] risky = {s: n for s, n in sources.items() if s in RISKY_SOURCES} days = (now - last_written).days if last_written else None stale = last_written is None or days > STALE_DAYS low_fill = fill_pct < LOW_FILL_PCT if filled > 0 and last_written is None: return "REVIEW", ["has values but no history was sampled, cannot judge"] if filled == 0: reasons.append("no records have a value") if low_fill and filled > 0: reasons.append("fill rate %.2f%% is under %.0f%%" % (fill_pct, LOW_FILL_PCT)) if stale and filled > 0: reasons.append("no write in sample for over %d days" % STALE_DAYS) if risky: reasons.append("sample shows writes from " + ", ".join( "%s x%d" % kv for kv in sorted(risky.items()))) if risky: return "REVIEW", reasons + ["find out which integration or API client writes here"] if stale: return "RETIRE_CANDIDATE", reasons if low_fill: return "REVIEW", reasons + ["low fill but written recently"] return "KEEP", ["fill %.1f%%, last write %d days ago" % (fill_pct, days)] def audit_property(api, object_type, prop, total_records, sample_size, now): name = prop["name"] filled = fill_count(api, object_type, name) ids = sample_ids(api, object_type, name, sample_size) if filled else [] entries = collect_history(api, object_type, name, ids) times = [t for t in (parse_ts(e["timestamp"]) for e in entries) if t] last_written = max(times) if times else None sources = Counter(e["sourceType"] for e in entries) source_ids = sorted({str(e["sourceId"]) for e in entries if e["sourceId"]}) source_labels = sorted({e["sourceLabel"] for e in entries if e["sourceLabel"]}) fill_pct = (100.0 * filled / total_records) if total_records else 0.0 verdict, reasons = score(fill_pct, filled, last_written, sources, now) return { "object": object_type, "name": name, "label": prop.get("label", name), "type": prop.get("type", ""), "group": prop.get("groupName", ""), "total_records": total_records, "filled": filled, "fill_rate_pct": round(fill_pct, 2), "last_written": last_written.date().isoformat() if last_written else "", "days_since_write": (now - last_written).days if last_written else "", "sample_records": len(ids), "write_sources": ";".join("%s:%d" % kv for kv in sorted(sources.items())), "distinct_source_ids": len(source_ids), "source_ids": "|".join(source_ids), "source_labels": "|".join(source_labels), "verdict": verdict, "reasons": "; ".join(reasons), } def sort_rows(rows): def key(r): lw = r["last_written"] or "0000-00-00" return (VERDICT_ORDER[r["verdict"]], r["fill_rate_pct"], lw, r["name"]) return sorted(rows, key=key) # --------------------------------------------------------------- duplicates def find_merge_pairs(props): pairs = [] for i in range(len(props)): for j in range(i + 1, len(props)): a, b = props[i], props[j] la, lb = a.get("label", a["name"]), b.get("label", b["name"]) ratio = difflib.SequenceMatcher(None, la.lower(), lb.lower()).ratio() ta, tb = set(tokens(la)), set(tokens(lb)) shared = ta & tb specific = shared - GENERIC_TOKENS why = [] if ratio > 0.7: why.append("label similarity %.2f" % ratio) if specific: why.append("shared word(s): " + ", ".join(sorted(specific))) elif (shared & DATE_TOKENS) and a.get("type") == b.get("type") == "date": why.append("both date fields sharing '%s'" % sorted(shared & DATE_TOKENS)[0]) if why: pairs.append({"a": a["name"], "a_label": la, "b": b["name"], "b_label": lb, "ratio": round(ratio, 2), "why": "; ".join(why)}) return sorted(pairs, key=lambda p: -p["ratio"]) # ------------------------------------------------------------------- output CSV_FIELDS = ["object", "name", "label", "type", "group", "total_records", "filled", "fill_rate_pct", "last_written", "days_since_write", "sample_records", "write_sources", "distinct_source_ids", "source_ids", "source_labels", "verdict", "reasons"] CHECKLIST = [ "Check the property's 'Used in' tab (workflows, lists, forms, reports) and the archive panel by hand. This script cannot see them.", "Export the values with record IDs first.", "Rename it with a DEPRECATED_ prefix.", "Hide it from forms.", "Wait one full reporting cycle.", "Then archive it (do not hard delete).", ] def write_csv(path, rows): with open(path, "w", newline="", encoding="utf-8") as f: w = csv.DictWriter(f, fieldnames=CSV_FIELDS) w.writeheader() w.writerows(rows) def write_md(path, rows, pairs, object_type, now, sample_size): counts = Counter(r["verdict"] for r in rows) L = ["# HubSpot property retirement audit: %s" % object_type, "", "Run date: %s. Sample size per property: %d records." % (now.date().isoformat(), sample_size), "", "Verdicts: " + ", ".join("%s %d" % (k, counts.get(k, 0)) for k in VERDICT_ORDER), "", "A verdict is a ranking hint from data, not permission to delete.", "Last written and write sources come from the property history of a sample of the most recently modified records that have a value, not every record.", "", "| Verdict | Property | Label | Fill | Last written | Write sources | Why |", "|---|---|---|---|---|---|---|"] for r in rows: L.append("| %s | `%s` | %s | %s/%s (%s%%) | %s | %s | %s |" % ( r["verdict"], r["name"], r["label"], r["filled"], r["total_records"], r["fill_rate_pct"], r["last_written"] or "n/a", r["write_sources"] or "none", r["reasons"])) L += ["", "## Likely duplicates (merge pairs)", ""] if pairs: L += ["| Property A | Property B | Why |", "|---|---|---|"] for p in pairs: L.append("| `%s` (%s) | `%s` (%s) | %s |" % (p["a"], p["a_label"], p["b"], p["b_label"], p["why"])) L += ["", "Overlapping names can still mean different events. Write down what each field represents before merging."] else: L.append("None found.") L += ["", "## Checklist before retiring anything", ""] todo = [r for r in rows if r["verdict"] != "KEEP"] if not todo: L.append("No RETIRE_CANDIDATE or REVIEW properties.") for r in todo: L.append("### `%s` (%s)" % (r["name"], r["verdict"])) if r["source_labels"] or r["source_ids"]: L.append("Seen writers: labels [%s], source ids [%s]." % ( r["source_labels"].replace("|", ", "), r["source_ids"].replace("|", ", "))) for i, step in enumerate(CHECKLIST, 1): L.append("%d. %s" % (i, step)) L.append("") with open(path, "w", encoding="utf-8") as f: f.write("\n".join(L) + "\n") # --------------------------------------------------------------------- main def run_audit(api, object_type, sample_size, now): props = list_custom_properties(api, object_type) total = count_total_records(api, object_type) rows = [audit_property(api, object_type, p, total, sample_size, now) for p in props] return sort_rows(rows), find_merge_pairs(props) def main(argv=None): ap = argparse.ArgumentParser(description="Read-only HubSpot property retirement audit") ap.add_argument("--object", required=True, choices=["contacts", "companies", "deals"]) ap.add_argument("--sample-size", type=int, default=20) ap.add_argument("--mock", metavar="DIR", help="run offline against fixtures in DIR") ap.add_argument("--out", default="out") ap.add_argument("--base-url", default=DEFAULT_BASE_URL) ap.add_argument("--sleep", type=float, default=0.25, help="seconds between live calls") ap.add_argument("--now", help="override today (YYYY-MM-DD), useful for repeatable tests") args = ap.parse_args(argv) now = parse_ts(args.now) if args.now else datetime.now(timezone.utc) if args.mock: api = MockTransport(args.mock) else: token = os.environ.get("HUBSPOT_TOKEN") if not token: sys.exit("Set HUBSPOT_TOKEN (private app token with read scopes) or use --mock sample/") api = LiveTransport(token, args.base_url, args.sleep) try: rows, pairs = run_audit(api, args.object, args.sample_size, now) except RuntimeError as e: sys.exit("error: %s" % e) os.makedirs(args.out, exist_ok=True) write_csv(os.path.join(args.out, "property_audit.csv"), rows) write_md(os.path.join(args.out, "property_audit.md"), rows, pairs, args.object, now, args.sample_size) for r in rows: print("%-17s %-24s fill %6.2f%% last %s" % ( r["verdict"], r["name"], r["fill_rate_pct"], r["last_written"] or "n/a")) print("merge pairs: %d" % len(pairs)) print("wrote %s and %s" % (os.path.join(args.out, "property_audit.csv"), os.path.join(args.out, "property_audit.md"))) if __name__ == "__main__": main()