#!/usr/bin/env python3 """Check a proposed CRM writeback against a field ownership contract. Usage: python3 check_writeback.py --contract CONTRACT.csv --current CRM_EXPORT.csv --proposed WRITEBACK.csv [--object contacts] [--key record_id] [--identity email[=email|hs_additional_emails][,domain...]] [--blank-mode ignore|send] [--clear-token __CLEAR__] [--out change-list.csv] [--show-values] [--json] CONTRACT.csv uses the columns of assets/field-ownership-contract.csv. CRM_EXPORT.csv is a fresh export keyed by record ID kept as text. It holds the update targets plus the result of a read-only lookup on each create row's email or domain, because duplicate creates are checked only against this file. It needs a column for every contract property the update rows fill or clear, and for every identity column in WRITEBACK.csv. A missing column stops the script (exit 2), because a blank cannot be told from a missing column. WRITEBACK.csv has action (update or create), the key column, writer, and one column per property, named with internal names (HubSpot) or API names (Salesforce). Row numbers are spreadsheet rows: the header is row 1. Cells: an empty cell means no value; the clear token means clear the property. --blank-mode send models writers that send empty cells as clears (Clay HubSpot Update object with Ignore blank values off, Apollo run_dedupe with empty values, Clay Salesforce text fields with Ignore blank values off, or a number field mapped with a null fallback). In send mode an empty cell counts as a clear only in a column that the same writer fills in at least one update row, because that column is in its field mapping. Mapping: a column that a writer fills but may not write (not in the contract, never written, writer not allowed) is reported once per writer and column. In send mode it is a FAIL when its blank cells would clear current values. Identity columns (--identity, comma-separated) are never written on update: a value that matches none of the record's values blocks the row as a likely wrong match. On create they are required and checked for duplicates. By default a writeback column is compared with the export column of the same name. Map it to several export columns with = and |, for example --identity "email=email|hs_additional_emails" (quote it: | is a shell pipe), so a create whose email is another record's additional email is caught. Email and domain cells in the export may hold a semicolon or comma list; each item is compared. Email columns compare ignoring case and surrounding spaces. Columns named domain or website also drop the scheme, a leading www., port, path, query and trailing dot or slash, so https://www.example.com/ matches example.com. Other columns compare as text after trimming spaces. Findings name the row, action, record ID, property and reason, never cell values, so contact data stays out of the transcript. --show-values adds the values to findings; use it on fictional or test data only. The --out change list keeps current and proposed values: it goes to the approver only, stays out of version control and is deleted once the change is applied or rejected. Decisions: fill, overwrite, create, candidate (value goes to the candidate property for review), no_change, needs_approval, blocked. Checks W18 and W19 (references/rules.md); decision rules in references/crm-writeback.md. Standard library only. Writes only --out. Nothing is sent to the CRM. Exit status: 0 no FAIL, 1 at least one FAIL, 2 bad input. """ import argparse import csv import json import re import sys from collections import Counter, OrderedDict WRITE_RULES = ("never", "fill_if_blank", "overwrite", "propose_only") ON_CREATE = ("allowed", "forbidden") BLANK_INPUT = ("ignore", "clear_with_approval") CONTRACT_COLUMNS = ("object", "property", "owner", "allowed_writers", "write_rule", "on_create", "blank_input", "candidate_property") DECISIONS = ("fill", "overwrite", "create", "candidate", "no_change", "needs_approval", "blocked") CHECKS = {"W18": "field-ownership", "W19": "create-update-identity"} SCIENTIFIC = re.compile(r"^\d+(\.\d+)?[eE][+-]?\d+$") TRAILING_ZERO = re.compile(r"^\d+\.0+$") SALESFORCE_ID = re.compile(r"^[A-Za-z0-9]{15}([A-Za-z0-9]{3})?$") SCHEME = re.compile(r"^[a-z][a-z0-9+.-]*://") LIST_SEPARATOR = re.compile(r"[;,]") IGNORED = {"email": "case and spaces ignored", "domain": "case, scheme, www, port and path ignored", "text": "surrounding spaces ignored"} SEVERITY_ORDER = {"FAIL": 0, "WARN": 1} SHOWN_CLEAR = "(clear)" class InputError(Exception): pass def text(value): return str(value).strip() if value is not None else "" def plural(count, word): return "%d %s%s" % (count, word, "" if count == 1 else "s") def identity_kind(column): lowered = column.lower() if "email" in lowered: return "email" if "domain" in lowered or "website" in lowered: return "domain" return "text" def norm_identity(kind, value): value = text(value) if kind == "email": return value.lower() if kind == "domain": value = SCHEME.sub("", value.lower()) value = re.split(r"[/?#]", value, maxsplit=1)[0] value = value.rsplit("@", 1)[-1].split(":", 1)[0].rstrip(".") return value[4:] if value.startswith("www.") else value return value def identity_values(kind, value): """Normalized values in one export cell; an email or domain cell may hold a list.""" parts = LIST_SEPARATOR.split(text(value)) if kind in ("email", "domain") else [value] return [v for v in (norm_identity(kind, p) for p in parts) if v] def parse_identity(value): """--identity email=email|hs_additional_emails,domain -> {writeback column: [export columns]}.""" identity = OrderedDict() for part in value.split(","): if not part.strip(): continue name, sep, targets = part.partition("=") name = name.strip() columns = [t.strip() for t in targets.split("|") if t.strip()] if sep else [name] if not name or not columns or name in identity: raise InputError("--identity takes writeback columns, each optionally mapped to export columns, for " "example \"email=email|hs_additional_emails,domain\"") identity[name] = columns if not identity: raise InputError("--identity needs at least one column, for example email") return identity def read_csv(path, label): try: with open(path, newline="", encoding="utf-8-sig") as handle: reader = csv.DictReader(handle) columns = [text(c) for c in (reader.fieldnames or [])] rows = [] for line, raw in enumerate(reader, 2): row = {text(k): (v if v is not None else "") for k, v in raw.items() if k is not None} row["_line"] = line rows.append(row) except (OSError, UnicodeDecodeError, csv.Error) as exc: raise InputError("cannot read %s %s as a UTF-8 CSV: %s" % (label, path, exc)) if not columns: raise InputError("%s %s has no header row" % (label, path)) return columns, rows def load_contract(path, obj): columns, rows = read_csv(path, "contract") missing = [c for c in CONTRACT_COLUMNS if c not in columns] if missing: raise InputError("contract is missing columns: %s" % ", ".join(missing)) contract = OrderedDict() for row in rows: if text(row.get("object")).lower() != obj.lower() or not text(row.get("property")): continue entry = { "writers": [w.strip() for w in text(row.get("allowed_writers")).split(";") if w.strip()], "write_rule": text(row.get("write_rule")).lower(), "on_create": text(row.get("on_create")).lower(), "blank_input": text(row.get("blank_input")).lower(), "candidate": text(row.get("candidate_property")), } where = "contract row %d (%s)" % (row["_line"], text(row["property"])) if entry["write_rule"] not in WRITE_RULES: raise InputError("%s: write_rule must be one of %s" % (where, ", ".join(WRITE_RULES))) if entry["on_create"] not in ON_CREATE: raise InputError("%s: on_create must be allowed or forbidden" % where) if entry["blank_input"] not in BLANK_INPUT: raise InputError("%s: blank_input must be ignore or clear_with_approval" % where) if entry["write_rule"] == "propose_only" and not entry["candidate"]: raise InputError("%s: propose_only needs a candidate_property" % where) if text(row["property"]) in contract: raise InputError("%s: the property is listed twice for %s" % (where, obj)) contract[text(row["property"])] = entry if not contract: raise InputError("the contract has no rows for object %s" % obj) return contract def main(): parser = argparse.ArgumentParser(description="Check a proposed CRM writeback against a field ownership contract.") parser.add_argument("--contract", required=True) parser.add_argument("--current", required=True) parser.add_argument("--proposed", required=True) parser.add_argument("--object", default="contacts") parser.add_argument("--key", default="record_id") parser.add_argument("--identity", default="email") parser.add_argument("--blank-mode", choices=("ignore", "send"), default="ignore") parser.add_argument("--clear-token", default="__CLEAR__") parser.add_argument("--out") parser.add_argument("--show-values", action="store_true") parser.add_argument("--json", action="store_true") args = parser.parse_args() try: identity = parse_identity(args.identity) contract = load_contract(args.contract, args.object) current_columns, current_rows = read_csv(args.current, "export") if args.key not in current_columns: raise InputError("the export has no %s column" % args.key) columns, rows = read_csv(args.proposed, "writeback") for need in ("action", args.key, "writer"): if need not in columns: raise InputError("the writeback has no %s column" % need) if not rows: raise InputError("the writeback %s has no rows, so there is nothing to check" % args.proposed) props = [c for c in columns if c not in ("action", args.key, "writer")] # A missing export column reads as blank: a conflict would look like a fill and a clear would vanish. touched = {p for r in rows if text(r.get("action")).lower() == "update" for p in props if text(r.get(p))} missing = [] for p in props: need = identity[p] if p in identity else ([p] if p in contract and p in touched else []) missing.extend(c for c in need if c not in current_columns and c not in missing) if missing: raise InputError("the export has no %s column%s; export every mapped property by internal name, or blank " "cannot be told from missing" % (", ".join(missing), "" if len(missing) == 1 else "s")) except InputError as exc: print("error: %s" % exc, file=sys.stderr) return 2 findings = [] changes = [] counts = Counter() blocked_rows = set() def add(severity, check, message, line=None): findings.append({"severity": severity, "check": check, "name": CHECKS[check], "row": line, "message": message}) def change(row, prop, current, proposed, decision, check, reason, target=""): changes.append(OrderedDict([ ("row", row["_line"]), ("action", row["_action"]), ("key", row["_key"] or "(new)"), ("property", prop), ("current", current), ("proposed", proposed), ("decision", decision), ("target_property", target), ("rule", check), ("reason", reason)])) def block_row(row, check, message): blocked_rows.add(row["_line"]) add("FAIL", check, message, row["_line"]) change(row, "", "", "", "blocked", check, message) def block_cell(row, prop, current, proposed, reason): counts["blocked"] += 1 label = "%s %s" % (row["_action"], row["_key"] or "(new)") if args.show_values: detail = '; proposed "%s" over "%s"' % (proposed, current) if current else '; proposed "%s"' % proposed elif reason.startswith("clear not allowed"): detail = "" else: detail = "; proposed a clear" if proposed == SHOWN_CLEAR else "; proposed a new value" add("FAIL", "W18", "row %d %s %s: %s%s" % (row["_line"], label, prop, reason, detail), row["_line"]) change(row, prop, current, proposed, "blocked", "W18", reason) current = OrderedDict() duplicates_in_export = [] for row in current_rows: key = text(row.get(args.key)) if not key: continue if key in current: duplicates_in_export.append(key) continue current[key] = row if duplicates_in_export: add("WARN", "W19", "the export lists record IDs more than once (%s); deduplicate the export before trusting " "the comparison" % ", ".join(sorted(set(duplicates_in_export)))) kinds = {name: identity_kind(name) for name in identity} # index[writeback column][normalized value] = {record ID: export column it was found in} index = {name: {} for name in identity} for key, row in current.items(): for name, targets in identity.items(): for column in targets: for value in identity_values(kinds[name], row.get(column)): index[name].setdefault(value, OrderedDict()).setdefault(key, column) salesforce_lengths = set() for row in rows: raw = row.get(args.key) or "" row["_key"] = raw.strip() row["_action"] = text(row.get("action")).lower() row["_writer"] = text(row.get("writer")) if row["_key"] and raw != row["_key"]: add("WARN", "W19", "row %d: %s %r has surrounding spaces; trimmed before matching" % (row["_line"], args.key, raw), row["_line"]) if row["_key"] and SALESFORCE_ID.match(row["_key"]) and not row["_key"].isdigit(): salesforce_lengths.add(len(row["_key"])) if len(salesforce_lengths) > 1: add("WARN", "W19", "the file mixes 15- and 18-character Salesforce IDs; convert all to 18 characters before matching") seen_creates = {} for row in rows: key, line = row["_key"], row["_line"] if row["_action"] not in ("update", "create"): block_row(row, "W19", "row %d: action %r is not update or create" % (line, row.get("action"))) continue if not row["_writer"]: block_row(row, "W18", "row %d: no writer named; every write needs an accountable writer" % line) continue if row["_action"] == "update": if not key: block_row(row, "W19", "row %d: update without a record ID; never fall through to create" % line) elif SCIENTIFIC.match(key) or TRAILING_ZERO.match(key): block_row(row, "W19", "row %d: record ID %s looks damaged by a spreadsheet; re-export with IDs as text" % (line, key)) elif key not in current: block_row(row, "W19", "row %d: update target %s is not in the export; never fall through to create" % (line, key)) else: for name, targets in identity.items(): value = text(row.get(name)) known = {v for column in targets for v in identity_values(kinds[name], current[key].get(column))} if value and norm_identity(kinds[name], value) not in known: if args.show_values: shown = "%s %s differs from the record's %s %s" % ( name, value, targets[0], text(current[key].get(targets[0])) or "(blank)") elif len(targets) == 1: shown = "%s differs from the record's value" % name else: shown = "%s matches none of the record's %s" % (name, ", ".join(targets)) block_row(row, "W19", "row %d: %s; the row may be matched to the wrong record" % (line, shown)) break continue if key: block_row(row, "W19", "row %d: create row carries record ID %s; use update or remove the ID" % (line, key)) continue values = [(name, norm_identity(kinds[name], row.get(name))) for name in identity] if not any(v for c, v in values): block_row(row, "W19", "row %d: create without identity evidence (%s empty)" % (line, ", ".join(identity))) continue problem = "" for name, value in values: shown = "%s %s" % (name, text(row.get(name))) if args.show_values else name if value and value in index[name]: found = ", ".join(rid if column == name else "%s by %s" % (rid, column) for rid, column in index[name][value].items()) problem = "row %d: duplicate create; %s matches existing record %s (%s)" % ( line, shown, found, IGNORED[kinds[name]]) break if value and (name, value) in seen_creates: problem = "row %d: duplicate create; %s repeats row %d" % (line, shown, seen_creates[(name, value)]) break if problem: block_row(row, "W19", problem) continue for column, value in values: if value: seen_creates[(column, value)] = line groups = OrderedDict() for row in rows: if row["_action"] == "update" and row["_line"] not in blocked_rows: groups.setdefault(row["_key"], []).append(row) skipped = set() for key, group in groups.items(): if len(group) < 2: continue lines = " and ".join(str(r["_line"]) for r in group) conflicts = [p for p in props if len({text(r.get(p)) for r in group if text(r.get(p))}) > 1] if conflicts: message = "rows %s update record %s with different values for %s; hold both until one value is chosen" % ( lines, key, ", ".join(conflicts)) add("FAIL", "W18", message, group[0]["_line"]) for r in group: blocked_rows.add(r["_line"]) change(r, "", "", "", "blocked", "W18", message) else: add("WARN", "W18", "rows %s repeat record %s with the same values; only row %d is evaluated" % (lines, key, group[0]["_line"]), group[0]["_line"]) skipped.update(r["_line"] for r in group[1:]) # A writer's field mapping is every property it fills in at least one update row. mapping = OrderedDict() for row in rows: if row["_action"] != "update" or not row["_writer"]: continue mapped = mapping.setdefault(row["_writer"], set()) for prop in props: if prop not in identity and text(row.get(prop)): mapped.add(prop) def refusal(entry, writer): if entry is None: return "not in the contract, so no writer owns it" if entry["write_rule"] == "never": return "never written by a writeback" if writer not in entry["writers"]: return "writer %s is not allowed" % writer return "" implicit = OrderedDict() for row in rows: if row["_line"] in blocked_rows or row["_line"] in skipped: continue writer = row["_writer"] label = "%s %s" % (row["_action"], row["_key"] or "(new)") if row["_action"] == "create": for prop in props: proposed = text(row.get(prop)) if not proposed or proposed == args.clear_token: continue entry = contract.get(prop) if entry is None: block_cell(row, prop, "", proposed, "not in the contract, so no writer owns it") elif writer not in entry["writers"]: block_cell(row, prop, "", proposed, "writer %s is not allowed" % writer) elif entry["on_create"] == "forbidden": block_cell(row, prop, "", proposed, "not set on create") else: counts["create"] += 1 change(row, prop, "", proposed, "create", "", "allowed on create") continue record = current[row["_key"]] for prop in props: if prop in identity: continue raw = text(row.get(prop)) implicit_clear = False if raw == args.clear_token: operation = "clear" elif raw == "": if args.blank_mode == "send" and prop in mapping.get(writer, set()): operation = "clear" implicit_clear = True else: continue else: operation = "set" existing = text(record.get(prop)) if operation == "clear" and not existing: continue shown = SHOWN_CLEAR if operation == "clear" else raw if operation == "set" and raw == existing: counts["no_change"] += 1 change(row, prop, existing, raw, "no_change", "", "same as current") continue entry = contract.get(prop) reason = refusal(entry, writer) if reason and implicit_clear: # Reported once per column by the mapping finding below. implicit.setdefault((writer, prop), []).append(row["_line"]) counts["blocked_by_mapping"] += 1 change(row, prop, existing, shown, "blocked", "W18", "blank would clear it; " + reason) elif reason: block_cell(row, prop, existing, shown, reason) elif operation == "clear": if entry["blank_input"] == "clear_with_approval": counts["needs_approval"] += 1 what = 'clears "%s"' % existing if args.show_values else "clears the current value" add("WARN", "W18", "row %d %s %s: %s; allowed only with approval" % (row["_line"], label, prop, what), row["_line"]) change(row, prop, existing, shown, "needs_approval", "W18", "clear needs approval") else: block_cell(row, prop, existing, shown, "clear not allowed (blank_input is ignore)") elif entry["write_rule"] == "propose_only" or (existing and entry["write_rule"] == "fill_if_blank"): if entry["candidate"]: counts["candidate"] += 1 if args.show_values: what = '"%s" differs from current "%s"' % (raw, existing) else: what = "differs from the current value" add("WARN", "W18", "row %d %s %s: %s; write to %s for review" % (row["_line"], label, prop, what, entry["candidate"]), row["_line"]) change(row, prop, existing, raw, "candidate", "W18", "differs from current value", entry["candidate"]) else: block_cell(row, prop, existing, raw, "fill_if_blank with a different current value and no " "candidate property") elif not existing: counts["fill"] += 1 change(row, prop, existing, raw, "fill", "", "current value is blank") else: counts["overwrite"] += 1 change(row, prop, existing, raw, "overwrite", "", "writer owns the property") for writer, mapped in mapping.items(): for prop in props: reason = refusal(contract.get(prop), writer) if prop in mapped else "" if not reason: continue lines = implicit.get((writer, prop), []) if lines: shown = ", ".join(str(n) for n in lines[:5]) + (" and %d more" % (len(lines) - 5) if len(lines) > 5 else "") add("FAIL", "W18", "mapping: writer %s maps %s (%s); with blank mode send its %d blank cells would " "clear current values (rows %s); remove it from the mapping" % (writer, prop, reason, len(lines), shown)) else: add("WARN", "W18", "mapping: writer %s maps %s (%s); remove it from that writer's field mapping" % (writer, prop, reason)) if args.out: try: with open(args.out, "w", newline="", encoding="utf-8") as handle: writer = csv.writer(handle) names = ["row", "action", "key", "property", "current", "proposed", "decision", "target_property", "rule", "reason"] writer.writerow(names) for item in sorted(changes, key=lambda c: c["row"]): writer.writerow([item[n] for n in names]) except OSError as exc: print("error: cannot write %s: %s" % (args.out, exc), file=sys.stderr) return 2 updates = sum(1 for r in rows if r["_action"] == "update") creates = sum(1 for r in rows if r["_action"] == "create") if creates and not current: add("WARN", "W19", "the export holds no records, so %s had no duplicate check; it must hold a lookup on each " "create row's email or domain" % plural(creates, "create row")) flagged = {f["check"] for f in findings} passes = [] if "W18" not in flagged: passes.append(("PASS", "W18", "%s, all by allowed writers on owned properties" % plural(counts["fill"] + counts["overwrite"] + counts["create"], "cell change"))) if "W19" not in flagged: if creates: passes.append(("PASS", "W19", "%s matched by record ID; %s with identity evidence and no duplicate among " "the %s" % (plural(updates, "update"), plural(creates, "create"), plural(len(current), "exported record")))) else: passes.append(("PASS", "W19", "%s matched by record ID; no creates" % plural(updates, "update"))) if creates: compared = "; ".join("%s (export column%s %s)" % (name, "" if len(t) == 1 else "s", ", ".join(t)) for name, t in identity.items()) passes.append(("INFO", "W19", "%s checked for duplicates only against the %s in the export, on %s; it must " "hold a lookup on each create row's email or domain" % (plural(creates, "create row"), plural(len(current), "record"), compared))) findings.sort(key=lambda f: (SEVERITY_ORDER[f["severity"]], 0 if f["row"] is None else 1, f["row"] or 0)) fails = sum(1 for f in findings if f["severity"] == "FAIL") warns = sum(1 for f in findings if f["severity"] == "WARN") blocked_row_count = len(blocked_rows) if args.json: print(json.dumps({ "object": args.object, "key": args.key, "identity": identity, "blank_mode": args.blank_mode, "rows": len(rows), "updates": updates, "creates": creates, "show_values": args.show_values, "decisions": {d: counts[d] for d in DECISIONS}, "blocked_by_mapping": counts["blocked_by_mapping"], "blocked_rows": blocked_row_count, "findings": findings, "passes": [{"severity": s, "check": c, "message": m} for s, c, m in passes], "fail": fails, "warn": warns, }, indent=2)) return 1 if fails else 0 print("Writeback check: %s against %s (object %s, key %s, blank mode %s)" % (args.proposed, args.current, args.object, args.key, args.blank_mode)) print("Rows: %d (%d update, %d create)" % (len(rows), updates, creates)) print("") print("Findings:") for f in findings: print("%s %s %s: %s" % (f["severity"], f["check"], f["name"], f["message"])) for severity, check, message in passes: print("%s %s %s: %s" % (severity, check, CHECKS[check], message)) print("") print("Decisions: fill %d, overwrite %d, create %d, candidate %d, no_change %d, needs_approval %d, blocked %d" % (counts["fill"], counts["overwrite"], counts["create"], counts["candidate"], counts["no_change"], counts["needs_approval"], counts["blocked"])) print("Blocked through a mapping finding: %d blank cells. Rows blocked whole: %d." % (counts["blocked_by_mapping"], blocked_row_count)) if args.out: print("Change list with current and proposed values, for the approver only: %s" % args.out) print("%d FAIL, %d WARN. Nothing was changed." % (fails, warns)) return 1 if fails else 0 if __name__ == "__main__": sys.exit(main())