#!/usr/bin/env python3 """Compare the expected state with a destination export, checking only owned fields. Usage: python3 scripts/reconcile_readback.py --source EXPECTED.csv --dest DESTINATION.csv --key SOURCE_KEY[:DEST_KEY] [--map FIELD_MAP.csv --flow FLOW_ID | --fields a,b,c] [--baseline DEST_BEFORE.csv] [--dest-id DEST_RECORD_ID_COLUMN] [--expected-absent KEYS.csv] [--membership current] [--dest-case-insensitive] [--tolerance 0.005] [--show 20] [--hide-values] [--json] [--json-out FILE] EXPECTED.csv is the state the source says the destination should hold, using the destination's field names (IDs and the flow's fields only). DESTINATION.csv is an export taken after the run. Owned fields are the field-map rows for --flow whose writer is this_flow; --fields lists them instead, or narrows them to a subset when given with --map and --flow (types still come from the map). Every owned field must be a column in both exports: a missing one exits 2 instead of comparing less than it claims. HubSpot UI exports head columns with property labels, so rename them to the field map's internal names first. --baseline is a destination export from before the run; it shows columns this flow does not own that changed anyway. Categories: R1 MISSING FAIL key expected, not in the destination R2 DUPLICATE FAIL key more than once in the destination (or in the source) R3 MISMATCH FAIL an owned field differs after normalisation R4 NULL_OVERWRITE FAIL the source has a value, the destination is blank R5 KEY_MANGLED FAIL key matches only after stripping leading zeros or folding case R6 ORPHAN WARN destination key with no source row (FAIL with --membership current) R7 NOT_REMOVED FAIL key listed in --expected-absent still present R8 NON_OWNED_CHANGED WARN with --baseline: a column this flow does not own changed R9 CASE_COLLISION WARN two source keys differ only by case (FAIL with --dest-case-insensitive) R10 MIXED_SF_ID WARN Salesforce-style keys mix 15- and 18-character forms Normalisation: trim; blank and missing are equal; number, integer and currency fields compare numerically (1200 equals 1200.00) within --tolerance; booleans accept true, false, yes, no, 1, 0; id, enum, string, date and datetime fields compare as exact text. Without a field map, two values compare numerically only when both are plain numbers without leading zeros. Personal data: keys are printed so records can be found, so a key, --dest-id or --expected-absent value that looks like an email address exits 2; key on a stable ID. Values are printed truncated to 40 characters, except values of columns the field map marks pii yes (any flow) and values that look like email addresses, which print as (hidden). --hide-values hides every value, also in --json output. Exit 1 on any FAIL, 2 on bad input, 0 otherwise. A source export with no keyed row is bad input unless --expected-absent lists keys, so empty files never pass. Reads local files only and changes nothing. """ import argparse import csv import json import os import re import sys from collections import OrderedDict, defaultdict CATEGORIES = OrderedDict([ ("R1", "MISSING"), ("R2", "DUPLICATE"), ("R3", "MISMATCH"), ("R4", "NULL_OVERWRITE"), ("R5", "KEY_MANGLED"), ("R6", "ORPHAN"), ("R7", "NOT_REMOVED"), ("R8", "NON_OWNED_CHANGED"), ("R9", "CASE_COLLISION"), ("R10", "MIXED_SF_ID"), ]) NUMERIC_TYPES = {"number", "integer", "currency"} TEXT_TYPES = {"id", "enum", "string", "date", "datetime"} TRUE, FALSE = {"true", "yes", "1"}, {"false", "no", "0"} PLAIN_NUMBER = re.compile(r"^-?(0|[1-9]\d*)(\.\d+)?$") EMAIL_LIKE = re.compile(r"[^@\s]+@[^@\s]+\.[A-Za-z]{2,}") SF_ID = re.compile(r"^[A-Za-z0-9]{15}([A-Za-z0-9]{3})?$") class InputError(Exception): pass def read_csv(path, label): if not os.path.isfile(path): raise InputError(f"{label} file not found: {os.path.basename(path)}") try: with open(path, encoding="utf-8-sig", newline="") as handle: reader = csv.reader(handle) header = [h.strip() for h in next(reader, [])] if not any(header): raise InputError(f"{label} {os.path.basename(path)} has no header row") rows = [] for raw in reader: if any(cell.strip() for cell in raw): rows.append({name: (raw[i].strip() if i < len(raw) else "") for i, name in enumerate(header)}) return header, rows except (UnicodeDecodeError, csv.Error): raise InputError(f"{label} {os.path.basename(path)} is not UTF-8 CSV text") def find_column(header, name, label): lookup = {h.lower(): h for h in header} if name.lower() not in lookup: raise InputError(f"column '{name}' not found in {label}") return lookup[name.lower()] def strip_zeros(key): return key.lstrip("0") or "0" def sf18(id15): """Standard 15 to 18 character Salesforce ID conversion (case-safe suffix).""" alphabet = "ABCDEFGHIJKLMNOPQRSTUVWXYZ012345" suffix = "" for i in range(3): chunk = id15[i * 5:(i + 1) * 5] suffix += alphabet[sum(1 << j for j, c in enumerate(chunk) if "A" <= c <= "Z")] return id15 + suffix def compare(expected, found, ftype, tolerance): """Return None when equal, else NULL_OVERWRITE or MISMATCH.""" if expected == "" and found == "": return None if found == "": return "NULL_OVERWRITE" if expected == "": return "MISMATCH" ftype = (ftype or "").lower() if ftype == "boolean" or (not ftype and expected.lower() in TRUE | FALSE and found.lower() in TRUE | FALSE and not (PLAIN_NUMBER.match(expected) and PLAIN_NUMBER.match(found))): e, f = expected.lower(), found.lower() if e in TRUE | FALSE and f in TRUE | FALSE: return None if (e in TRUE) == (f in TRUE) else "MISMATCH" return None if e == f else "MISMATCH" numeric = ftype in NUMERIC_TYPES or (not ftype and PLAIN_NUMBER.match(expected) and PLAIN_NUMBER.match(found)) if numeric and ftype not in TEXT_TYPES: try: return None if abs(float(expected) - float(found)) <= tolerance else "MISMATCH" except ValueError: pass return None if expected == found else "MISMATCH" def short(value, hide): if hide or EMAIL_LIKE.search(value): return "(hidden)" return '"' + (value if len(value) <= 40 else value[:37] + "...") + '"' def read_map(args): """Return (owned fields of --flow as name -> type, lowercased names marked pii yes on any row).""" header, rows = read_csv(args.map, "Field map") fid, field, writer = (find_column(header, c, "the field map") for c in ("flow_id", "destination_field", "writer")) lookup = {h.lower(): h for h in header} ftype, pii = lookup.get("type"), lookup.get("pii") owned, private = OrderedDict(), set() for row in rows: name = row[field] if pii and name and row[pii].lower() == "yes": private.add(name.lower()) if args.flow and row[fid] == args.flow and row[writer].lower() == "this_flow": owned.setdefault(name, row[ftype] if ftype else "") return owned, private def owned_fields(args): """Return (owned field name -> declared type, '' when unknown; lowercased pii field names).""" owned, private = read_map(args) if args.map else (OrderedDict(), set()) if args.map and args.flow and not owned: raise InputError(f"no field-map rows with flow_id '{args.flow}' and writer this_flow") if args.fields: listed = [n.strip() for n in args.fields.split(",") if n.strip()] if args.map and args.flow: known = {k.lower(): (k, v) for k, v in owned.items()} stray = [n for n in listed if n.lower() not in known] if stray: raise InputError(f"--fields names {', '.join(stray)}, which the field map does not list as written by " f"flow {args.flow}") owned = OrderedDict(known[n.lower()] for n in listed) else: owned = OrderedDict((n, "") for n in listed) if not owned: raise InputError("no owned fields to compare") return owned, private def parse_args(argv): p = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) p.add_argument("--source", required=True, help="expected state CSV") p.add_argument("--dest", required=True, help="destination export CSV") p.add_argument("--key", required=True, help="SOURCE_KEY or SOURCE_KEY:DEST_KEY") p.add_argument("--map", help="field map CSV (with --flow); its pii column also hides values") p.add_argument("--flow", help="flow_id in the field map") p.add_argument("--fields", help="comma-separated owned fields, instead of --map and --flow, or a subset with them") p.add_argument("--baseline", help="destination export taken before the run") p.add_argument("--dest-id", help="destination record ID column, used in messages and to join the baseline") p.add_argument("--expected-absent", help="one-column CSV of keys that must not be in the destination") p.add_argument("--membership", choices=["current", "add_only"], help="current: an orphan should have been removed") p.add_argument("--dest-case-insensitive", action="store_true", help="the destination compares keys ignoring case") p.add_argument("--tolerance", type=float, default=0.0, help="absolute tolerance for numeric fields (default 0)") p.add_argument("--show", type=int, default=20, help="examples printed per category (default 20)") p.add_argument("--hide-values", action="store_true", help="never print field values (pii yes columns and " "email-like values are always hidden)") p.add_argument("--json", action="store_true", help="print findings as JSON") p.add_argument("--json-out", help="also write the JSON result to this file") args = p.parse_args(argv) if not args.fields and not (args.map and args.flow): p.error("pass --map and --flow, or --fields") return args def main(argv=None): args = parse_args(argv) findings, not_checked = [], [] def add(rule, level, key, message, field=""): findings.append({"rule": rule, "category": CATEGORIES[rule], "level": level, "key": key, "field": field, "message": message}) try: src_header, src_rows = read_csv(args.source, "Source") dst_header, dst_rows = read_csv(args.dest, "Destination") src_key_name, _, dst_key_name = args.key.partition(":") src_key = find_column(src_header, src_key_name, "the source") dst_key = find_column(dst_header, dst_key_name or src_key_name, "the destination") dst_id = find_column(dst_header, args.dest_id, "the destination") if args.dest_id else None fields, private = owned_fields(args) src_lookup = {h.lower(): h for h in src_header} dst_lookup = {h.lower(): h for h in dst_header} gaps = OrderedDict() for name in fields: lacking = [label for label, lookup in (("source", src_lookup), ("destination", dst_lookup)) if name.lower() not in lookup] if lacking: where = "in neither export" if len(lacking) == 2 else f"not in the {lacking[0]} export" gaps.setdefault(where, []).append(name) if gaps: where = "; ".join(f"{', '.join(names)} {label}" for label, names in gaps.items()) raise InputError(f"owned field(s) {where}. Rename the columns to the field map's internal names, or pass " "--fields with the subset you mean to compare") absent_keys = [] if args.expected_absent: if not os.path.isfile(args.expected_absent): raise InputError(f"file not found: {os.path.basename(args.expected_absent)}") try: with open(args.expected_absent, encoding="utf-8-sig", newline="") as handle: cells = [r[0].strip() for r in csv.reader(handle) if r and r[0].strip()] except (UnicodeDecodeError, csv.Error): raise InputError(f"{os.path.basename(args.expected_absent)} is not UTF-8 CSV text") if cells and cells[0].lower() in {src_key_name.lower(), (dst_key_name or src_key_name).lower(), "key"}: cells = cells[1:] absent_keys = cells key_values = [r[src_key] for r in src_rows] + [r[dst_key] for r in dst_rows] + absent_keys + \ ([r[dst_id] for r in dst_rows] if dst_id else []) if any(EMAIL_LIKE.search(v) for v in key_values): raise InputError("a key, --dest-id or --expected-absent value looks like an email address; findings print " "keys, so key on a stable ID (source or record ID) instead") if args.json_out: inputs = [args.source, args.dest, args.map, args.baseline, args.expected_absent] if any(p and os.path.abspath(p) == os.path.abspath(args.json_out) for p in inputs): raise InputError("--json-out would overwrite an input file") except InputError as exc: print(f"error: {exc}", file=sys.stderr) return 2 compared = fields def ref(row): return f" ({dst_id} {row[dst_id]})" if dst_id and row.get(dst_id) else "" src_index, dst_index = OrderedDict(), OrderedDict() blank = [0, 0] for row in src_rows: if row[src_key]: src_index.setdefault(row[src_key], []).append(row) else: blank[0] += 1 for row in dst_rows: if row[dst_key]: dst_index.setdefault(row[dst_key], []).append(row) else: blank[1] += 1 if any(blank): not_checked.append(f"rows with a blank key skipped: {blank[0]} in the source, {blank[1]} in the destination") if not src_index and not absent_keys: print("error: the source export has no row with a key, so there is nothing to compare", file=sys.stderr) return 2 if not src_index: not_checked.append("R1 to R5, R9 and R10: the source export has no row with a key; only the --expected-absent keys were read") for key, rows in src_index.items(): if len(rows) > 1: add("R2", "FAIL", key, f"appears {len(rows)} times in the source; the expected state is not unique") for key, rows in dst_index.items(): if len(rows) > 1: ids = ", ".join(r[dst_id] for r in rows if dst_id and r.get(dst_id)) if dst_id else "" add("R2", "FAIL", key, f"{len(rows)} destination records" + (f" ({dst_id} {ids})" if ids else "")) absent_set = set(absent_keys) missing = [k for k in src_index if k not in dst_index] orphans = [k for k in dst_index if k not in src_index and k not in absent_set] pairs = [(k, k) for k in src_index if k in dst_index] repaired = {} for norm in (strip_zeros, str.casefold, lambda k: strip_zeros(k).casefold()): by_norm = defaultdict(list) for d in orphans: by_norm[norm(d)].append(d) for s in list(missing): hits = [d for d in by_norm.get(norm(s), []) if d in orphans] if hits: d = hits[0] missing.remove(s) orphans.remove(d) repaired[s] = d pairs.append((s, d)) for s, d in repaired.items(): how = "leading zeros stripped" if strip_zeros(s) == strip_zeros(d) and s.casefold() != d.casefold() else \ "case changed" if s.casefold() == d.casefold() else "leading zeros stripped and case changed" add("R5", "FAIL", s, f"destination stores the key as \"{d}\" ({how}){ref(dst_index[d][0])}; " "fix the key, then match exactly") for key in missing: add("R1", "FAIL", key, "expected in the destination, not found") clean = 0 for s, d in pairs: expected = src_index[s][0] record_clean = True for drow in dst_index[d]: for name, ftype in compared.items(): e, f = expected[src_lookup[name.lower()]], drow[dst_lookup[name.lower()]] outcome = compare(e, f, ftype, args.tolerance) hide = args.hide_values or name.lower() in private if outcome == "NULL_OVERWRITE": add("R4", "FAIL", s, f"expected {short(e, hide)}, destination blank{ref(drow)}", name) elif outcome == "MISMATCH": add("R3", "FAIL", s, f"expected {short(e, hide)}, found {short(f, hide)}{ref(drow)}", name) if outcome: record_clean = False clean += record_clean for key in orphans: level = "FAIL" if args.membership == "current" else "WARN" reason = "it should have been removed under a current-membership contract" if level == "FAIL" else \ "confirm whether another writer created it or a removal was missed" add("R6", level, key, f"in the destination with no source row{ref(dst_index[key][0])}; {reason}") if args.expected_absent: dst_norm = {strip_zeros(k).casefold(): k for k in dst_index} for key in absent_keys: if key in src_index: add("R7", "WARN", key, "listed in both the source and --expected-absent; fix the inputs") elif key in dst_index: add("R7", "FAIL", key, f"still present{ref(dst_index[key][0])}; the removal did not reach the destination") elif strip_zeros(key).casefold() in dst_norm: d = dst_norm[strip_zeros(key).casefold()] add("R7", "FAIL", key, f"still present as \"{d}\" (mangled key){ref(dst_index[d][0])}") else: not_checked.append("R7 NOT_REMOVED: pass --expected-absent with the keys that must be gone") if args.baseline: try: base_header, base_rows = read_csv(args.baseline, "Baseline") join = dst_id or dst_key base_join = find_column(base_header, join, "the baseline") except InputError as exc: print(f"error: {exc}", file=sys.stderr) return 2 base_index = {} for row in base_rows: base_index.setdefault(row[base_join], row) base_lookup = {h.lower(): h for h in base_header} skip = {n.lower() for n in fields} | {dst_key.lower(), join.lower()} columns = [h for h in dst_header if h.lower() in base_lookup and h.lower() not in skip] for drow in dst_rows: before = base_index.get(drow.get(join, "")) if not before: continue for col in columns: b, a = before[base_lookup[col.lower()]], drow[col] if compare(b, a, "", 0.0) and compare(a, b, "", 0.0): hide = args.hide_values or col.lower() in private add("R8", "WARN", f"{join} {drow[join]}", f"{short(b, hide)} before, {short(a, hide)} now; " "this flow does not own it, so confirm which writer changed it", col) else: not_checked.append("R8 NON_OWNED_CHANGED: pass --baseline with a destination export from before the run") by_fold = defaultdict(list) for key in src_index: by_fold[key.casefold()].append(key) for group in by_fold.values(): if len(group) > 1: level = "FAIL" if args.dest_case_insensitive else "WARN" add("R9", level, group[0], f"source keys {', '.join(group)} differ only by case" + ("; a case-insensitive destination merges them" if level == "FAIL" else "; check the destination's comparison rule")) sf_keys = [k for k in list(src_index) + list(dst_index) if SF_ID.match(k) and re.search(r"\d", k) and re.search(r"[A-Za-z]", k)] lengths = {len(k) for k in sf_keys} if {15, 18} <= lengths: sample = next(k for k in sf_keys if len(k) == 15) add("R10", "WARN", sample, f"{sum(1 for k in sf_keys if len(k) == 15)} keys across both files have 15 characters and " f"{sum(1 for k in sf_keys if len(k) == 18)} have 18; convert to 18 before joining " f"({sample} becomes {sf18(sample)})") order = list(CATEGORIES) findings.sort(key=lambda f: order.index(f["rule"])) # stable within a category checked = [r for r in order if not (r == "R7" and not args.expected_absent) and not (r == "R8" and not args.baseline) and (src_index or r in ("R6", "R7", "R8"))] passes = [{"rule": r, "category": CATEGORIES[r], "level": "PASS", "message": "none found"} for r in checked if not any(f["rule"] == r for f in findings)] counts = {lvl: sum(1 for f in findings if f["level"] == lvl) for lvl in ("FAIL", "WARN")} counts["PASS"] = len(passes) summary = f"{counts['FAIL']} FAIL, {counts['WARN']} WARN, {counts['PASS']} PASS" if not_checked: summary += f", {len(not_checked)} NOT CHECKED" summary += ". Nothing was changed." tail = ("Not checked by this script: fields this flow does not own (beyond --baseline), downstream actions " "(messages, tasks, enrolments), values outside the two exports, and whether the source export was complete.") per_rule = OrderedDict((CATEGORIES[r], sum(1 for f in findings if f["rule"] == r)) for r in order) hidden = [n for n in compared if args.hide_values or n.lower() in private] result = {"source_rows": len(src_rows), "destination_rows": len(dst_rows), "owned_fields_compared": list(compared), "owned_fields_values_hidden": hidden, "matched_exact": len(pairs) - len(repaired), "matched_after_key_repair": len(repaired), "matched_clean": clean, "counts": per_rule, "findings": findings, "passes": passes, "not_checked": not_checked, "summary": counts} if args.json_out: with open(args.json_out, "w", encoding="utf-8") as handle: json.dump(result, handle, indent=2) if args.json: print(json.dumps(result, indent=2)) else: origin = f"field map flow {args.flow}" + (", narrowed by --fields" if args.fields else "") if args.map and args.flow \ else "--fields" print(f"Read-back: {len(src_rows)} source rows ({os.path.basename(args.source)}), {len(dst_rows)} destination rows " f"({os.path.basename(args.dest)}), key {src_key}" + (f":{dst_key}" if dst_key != src_key else "")) print(f"Owned fields compared ({len(compared)}, from {origin}): {', '.join(compared) or 'none'}") if hidden: print(f"Values not printed for: {', '.join(hidden)}") print(f"Matched: {len(pairs) - len(repaired)} exact, {len(repaired)} after key repair; " f"{clean} with every owned field equal") print("Counts: " + ", ".join(f"{name} {n}" for name, n in per_rule.items())) shown = defaultdict(int) for f in findings: shown[f["rule"]] += 1 if shown[f["rule"]] <= args.show: field = f" {f['field']}" if f["field"] else "" print(f"{f['level']} {f['rule']} {f['category']} {f['key']}{field}: {f['message']}") for rule, n in shown.items(): if n > args.show: print(f" ... {n - args.show} more {CATEGORIES[rule]} findings (raise --show or use --json)") for p in passes: print(f"PASS {p['rule']} {p['category']}: {p['message']}") for item in not_checked: print(f"NOT CHECKED {item}") print(summary) print(tail) return 1 if counts["FAIL"] else 0 if __name__ == "__main__": sys.exit(main())