"""Writes the files under --out and reads only --previous and an earlier calibration sample there. No network calls, and nothing is written outside --out. """ import csv import hashlib import json import os from collections import Counter, defaultdict from hygiene_core import ( FORMULA_START, InputError, OBJECTS, PERSONS, SCHEMA, STATUSES, idsort, is_placeholder, mask, mask_owner, pct, plural, rate, shape) from hygiene_duplicates import assoc_count, ukey # ---------- outputs ---------- CAL_HEADER = ["pair_id", "band", "object", "record_id_a", "record_id_b", "rule_id", "basis", "score", "reviewer_mark", "notes"] PREFIX = {"contacts": "CT", "companies": "CO", "deals": "DL", "leads": "LD"} FAMILY = "XFERDOLA" def safe(v): s = "" if v is None else str(v) return "'" + s if s[:1] in FORMULA_START else s def write_csv(path, header, rows): with open(path, "w", newline="", encoding="utf-8") as fh: w = csv.writer(fh, lineterminator="\n") w.writerow(header) for r in rows: w.writerow([safe(x) for x in r]) def key_of(obj, d, unmask): if obj in PERSONS: if d["email"]: return d["email"] if unmask else mask(d["email"], "email") return d["name"] if unmask else mask(d["name"], "name") if obj == "companies": return d["domain"] or d["name"] return "amount " + (f"{d['amount']:.0f}" if isinstance(d["amount"], float) else "blank") def sort_checks(checks): return sorted(checks, key=lambda c: (FAMILY.index(c["id"][0]), c["id"], OBJECTS.index(c["object"]) if c["object"] in OBJECTS else 9, c["field"])) def line(c, ids=5): label = f"{c['id']} {c['name']} {c['object']}" + (f".{c['field']}" if c["field"] else "") if c["numerator"] is not None and c["denominator"] is not None: body = f"{rate(c['numerator'], c['denominator'])} {c['population']}" elif c["numerator"] is not None: body = f"{c['numerator']} {c['population']}" else: body = "" out = f"{label}: {body}".rstrip(": ") if c["threshold"]: out += f". Threshold: {c['threshold']}" if c["detail"]: out += f". {c['detail']}" if c["record_ids"] and ids: more = len(c["record_ids"]) - ids out += ". Records: " + ", ".join(c["record_ids"][:ids]) + (f" (+{more} more)" if more > 0 else "") return out.replace("..", ".") def group_rows(dup, unmask): rows, ids = [], {} for obj in OBJECTS: if obj not in dup: continue for n, g in enumerate(dup[obj]["groups"], start=1): gid = f"{PREFIX[obj]}-{n:02d}" ids[(obj, n)] = gid for k, d in enumerate(g["members"]): role = "survivor-candidate (suggestion)" if k == 0 else "merge-candidate (suggestion)" extra = [d.get("name", ""), d.get("email", "")] if unmask else [] rows.append([gid, g["band"], obj, d["rid"], ";".join(g["rules"]), g["score"], "yes" if g["conflict"] else "no", role, g["why"] if k == 0 else "", d["created"].isoformat() if d["created"] else "", mask_owner(d["owner"], unmask), None, d["source"], key_of(obj, d, unmask), ukey(d)] + extra) return rows def write_outputs(ctx, res, dup, args, tables, rules_meta): os.makedirs(args.out, exist_ok=True) checks = sort_checks(res.checks) totals = Counter(c["status"] for c in checks) # duplicates.csv rows = group_rows(dup, args.unmask) for r in rows: d = ctx.rec(r[2], r[3]) r[11] = assoc_count(ctx, r[2], d) if d else 0 hdr = ["group_id", "band", "object", "record_id", "rule_ids", "best_score", "key_conflict", "suggested_role", "suggestion_reason", "created_at", "owner", "association_count", "source", "masked_key", "unique_key"] write_csv(os.path.join(args.out, "duplicates.csv"), hdr + (["name", "email"] if args.unmask else []), rows) prow, n = [], 0 for obj in OBJECTS: if obj not in dup: continue for l in dup[obj]["pairs"]: n += 1 a, b = dup[obj]["recs"][l["i"]], dup[obj]["recs"][l["j"]] reason = l["basis"] + ("; refused: the link would put different unique keys in one group" if l["refused"] else "; key conflict: different unique keys" if l["conflict"] else "") prow.append([f"PP-{n:03d}", "possible", obj, a["rid"], b["rid"], ";".join(l["rules"]), l["score"], "yes" if l["conflict"] else "no", reason]) for c in dup.get("_cross", []): n += 1 prow.append([f"PP-{n:03d}", "probable", "leads-contacts", c["lead"]["rid"], c["contact"]["rid"], "D07", 1.0, "no", "unconverted lead with a contact's email; convert the lead into the contact"]) write_csv(os.path.join(args.out, "possible-pairs.csv"), ["pair_id", "band", "object", "record_id_a", "record_id_b", "rule_id", "score", "key_conflict", "reason"], prow) write_csv(os.path.join(args.out, "findings.csv"), ["check_id", "status", "object", "record_id", "field", "detail"], sorted(res.findings, key=lambda f: (FAMILY.index(f[0][0]), f[0], f[2], idsort(f[3])))) crow = [] for name, t in tables.items(): rev = {i: f for f, i in t.col.items()} for i, h in enumerate(t.headers): vals = [r[i].strip() for r in t.rows if i < len(r)] filled = [v for v in vals if v] ph = sum(1 for v in filled if is_placeholder(v, rev.get(i, ""), set())) shapes = Counter(shape(v) for v in filled).most_common(3) crow.append([name, h, rev.get(i, "").replace("__ukey", "unique key"), pct(len(filled), len(vals)), pct(ph, len(filled)) if filled else "n/a", len(set(filled)), "; ".join(f"{s} ({c})" for s, c in shapes)]) write_csv(os.path.join(args.out, "columns.csv"), ["object", "column", "canonical_field", "fill_rate", "placeholder_rate", "distinct_count", "top_shape_patterns"], crow) unique = {} for obj, _ in ctx.tables(): f = {i for c in checks if c["object"] == obj and c["status"] == "FAIL" for i in c["record_ids"]} fw = {i for c in checks if c["object"] == obj and c["status"] in ("FAIL", "WARN") for i in c["record_ids"]} unique[obj] = {"fail": len(f), "fail_or_warn": len(fw)} excess = {} for obj in OBJECTS: if obj in dup: g = dup[obj]["groups"] inside = sum(len(x["members"]) for x in g) excess[obj] = {"certain_groups": sum(1 for x in g if x["band"] == "certain"), "probable_groups": sum(1 for x in g if x["band"] == "probable"), "records_in_groups": inside, "excess": inside - len(g), "population": len(dup[obj]["recs"]), "possible_pairs": len(dup[obj]["pairs"])} profile = { "schema": SCHEMA, "as_of": ctx.as_of.isoformat(), "crm": ctx.crm, "rules_version": rules_meta["version"], "rules_hash": rules_meta["hash"], "inputs": [{"object": k, "file": t.name, "rows": len(t.rows), "sha256": t.sha256, "encoding": t.encoding, "delimiter": t.delimiter} for k, t in tables.items()], "mapping": {k: {f: v[1] for f, v in t.mapinfo.items() if v[0] == "mapped"} for k, t in tables.items()}, "coverage": {c["object"]: ("FULL" if c["status"] == "PASS" else "PARTIAL" if c["status"] == "FAIL" else "UNCONFIRMED") for c in checks if c["id"] == "X08"}, "checks": [{k: c[k] for k in ("id", "object", "field", "status", "numerator", "denominator", "population", "threshold", "record_ids")} for c in checks], "duplicates": {"excess": excess, "groups": [ {"group_id": r[0], "band": r[1], "object": r[2], "record_id": r[3], "rule_ids": r[4], "created_at": r[9], "source": r[12], "unique_key": r[14]} for r in rows], "pairs": [{"pair_id": p[0], "object": p[2], "a": p[3], "b": p[4], "rule_id": p[5]} for p in prow]}, "record_ids": {obj: sorted((d["rid"] for d in ctx.info[obj] if d["rid"]), key=idsort) for obj, _ in ctx.tables()}, "created": {obj: {d["rid"]: (d["created"].isoformat() if d["created"] else "") for d in ctx.info[obj] if d["rid"]} for obj, _ in ctx.tables()}, "unique_keys": {obj: {d["rid"]: ukey(d) for d in ctx.info[obj] if d["rid"] and ukey(d)} for obj, t in ctx.tables() if "__ukey" in t.col}, "converted_leads": sorted((d["rid"] for d in ctx.info.get("leads", []) if d["converted"] and d["rid"]), key=idsort), "unique_affected": unique, "totals": dict(totals), } with open(os.path.join(args.out, "profile.json"), "w", encoding="utf-8") as fh: json.dump(profile, fh, indent=1, sort_keys=True) fh.write("\n") cal = write_calibration(ctx, dup, args) if args.calibration else None trend = write_trend(args, profile) write_summary(ctx, args, tables, checks, totals, excess, unique, rules_meta, trend, cal) return checks, totals def write_calibration(ctx, dup, args): """Up to N pairs per band per object in a stable order seeded by --as-of, so a rerun after tightening one rule keeps every pair it still finds. Marks already typed into the file in --out are carried over.""" path, marks = os.path.join(args.out, "calibration-sample.csv"), {} earlier = os.path.exists(path) if earlier: try: with open(path, newline="", encoding="utf-8") as fh: for r in csv.DictReader(fh): if (r.get("reviewer_mark") or "").strip() or (r.get("notes") or "").strip(): marks[(r.get("object"), r.get("record_id_a"), r.get("record_id_b"))] = (r.get("reviewer_mark") or "", r.get("notes") or "") except (OSError, csv.Error, UnicodeDecodeError): marks = {} rows, n = [], 0 for obj in OBJECTS: if obj not in dup: continue recs = dup[obj]["recs"] order = lambda l: hashlib.sha256(f"{ctx.as_of}|{obj}|{recs[l['i']]['rid']}|{recs[l['j']]['rid']}".encode()).hexdigest() for band in ("certain", "probable", "possible"): pool = [l for l in dup[obj]["links"] if l["band"] == band] for l in sorted(sorted(pool, key=order)[:args.calibration], key=lambda l: (l["i"], l["j"])): n += 1 a, b = recs[l["i"]]["rid"], recs[l["j"]]["rid"] mark, note = marks.get((obj, a, b), ("", "")) rows.append([f"CAL-{n:03d}", band, obj, a, b, ";".join(l["rules"]), l["basis"], l["score"], mark, note]) write_csv(path, CAL_HEADER, rows) return {"pairs": len(rows), "kept": sum(1 for r in rows if r[8] or r[9]), "earlier": earlier} def write_trend(args, cur): if not args.previous: with open(os.path.join(args.out, "trend.md"), "w", encoding="utf-8") as fh: fh.write("# CRM hygiene trend\n\n- No previous profile was given (--previous), so this run has no trend.\n") return None try: with open(args.previous, encoding="utf-8") as fh: prev = json.load(fh) except (OSError, ValueError) as e: raise InputError(f"cannot read --previous: {e}") out = ["# CRM hygiene trend", "", f"- Previous profile as of {prev.get('as_of')}, rules {prev.get('rules_version')} " f"({prev.get('rules_hash')}). Current as of {cur['as_of']}, rules {cur['rules_version']} ({cur['rules_hash']})."] if prev.get("rules_hash") != cur["rules_hash"]: out.append("- Rules changed between the two profiles: read every delta below as possibly a definition change.") pc = {(c["id"], c["object"], c.get("field", "")): c for c in prev.get("checks", [])} out += ["", "## Check deltas", ""] for c in cur["checks"]: p = pc.get((c["id"], c["object"], c["field"])) if not p or c["numerator"] is None: continue pids, cids = set(p.get("record_ids", [])), set(c["record_ids"]) if p["numerator"] == c["numerator"] and p["status"] == c["status"] and pids == cids: continue fmt = lambda x: f"{x['status']} {x['numerator']}" + (f" of {x['denominator']}" if x["denominator"] is not None else "") out.append(f"- {c['id']} {c['object']}{('.' + c['field']) if c['field'] else ''}: {fmt(p)} to {fmt(c)}.") old = set(prev.get("record_ids", {}).get(c["object"], [])) resolved, again = sorted(pids - cids, key=idsort), sorted((cids - pids) & old, key=idsort) fresh = sorted(cids - pids - old, key=idsort) if resolved: out.append(f" - Resolved since last profile: {len(resolved)} ({', '.join(resolved[:5])}).") if again: out.append(f" - New on records that existed last time (regressions or newly broken): {len(again)} ({', '.join(again[:5])}).") if fresh: out.append(f" - On records created since the previous profile: {len(fresh)} ({', '.join(fresh[:5])}).") out += ["", "## Records missing since the previous profile (merged or deleted)", ""] for obj, ids in sorted(prev.get("record_ids", {}).items()): gone = sorted(set(ids) - set(cur["record_ids"].get(obj, [])), key=idsort) out.append(f"- {obj}: {len(gone)}" + (f" ({', '.join(gone[:10])})" if gone else "")) out += ["", "## Duplicate groups with a member created since the previous as-of, by creation source", ""] since = defaultdict(lambda: defaultdict(set)) for g in cur["duplicates"]["groups"]: if g["created_at"] and g["created_at"] > prev.get("as_of", ""): since[g["object"]][g["source"] or "unknown source"].add(g["group_id"]) if not since: out.append("- None.") for obj, by in sorted(since.items()): for src, gids in sorted(by.items()): out.append(f"- {obj}, {src}: {plural(len(gids), 'group')} ({', '.join(sorted(gids))}).") prev_members = defaultdict(set) for g in prev.get("duplicates", {}).get("groups", []): prev_members[(g["object"], g["group_id"])].add(g["record_id"]) cur_members = defaultdict(set) for g in cur["duplicates"]["groups"]: cur_members[(g["object"], g["group_id"])].add(g["record_id"]) open_groups = [k for k, m in cur_members.items() if any(len(m & pm) > 1 for pm in prev_members.values())] out += ["", "## Unresolved review queue", "", f"- {len(open_groups)} current groups were already in the previous profile (open since {prev.get('as_of')} or earlier)."] with open(os.path.join(args.out, "trend.md"), "w", encoding="utf-8") as fh: fh.write("\n".join(out) + "\n") return len(open_groups) def coverage_line(name, t, checks): c = next((c for c in checks if c["id"] == "X08" and c["object"] == name), None) delim = {",": "comma", ";": "semicolon", "\t": "tab", "|": "pipe"}.get(t.delimiter, t.delimiter) base = f"{name}: {t.name}, {len(t.rows)} rows, {t.encoding}, {delim}" if not c: return base + "." if c["status"] == "PASS": return base + f". Coverage FULL ({len(t.rows)} of {c['denominator']})." if c["status"] == "FAIL": return base + f". Coverage PARTIAL ({len(t.rows)} rows, CRM count {c['denominator']})." return base + ". Coverage UNCONFIRMED (no expected count)." def write_summary(ctx, args, tables, checks, totals, excess, unique, rules_meta, trend, cal): L = ["# CRM hygiene profile", "", f"Read-only profile of CSV exports, as of {ctx.as_of.isoformat()}. Nothing was changed.", "", "## Baseline", "", f"- CRM preset: {ctx.crm}. Rules: version {rules_meta['version']}, hash {rules_meta['hash']}."] if rules_meta["unknown"]: L.append("- Unknown rules keys ignored: " + ", ".join(rules_meta["unknown"]) + ".") for name, t in tables.items(): L.append("- " + coverage_line(name, t, checks) + f" sha256 {t.sha256[:12]}.") conv = sum(1 for d in ctx.info.get("leads", []) if d["converted"]) if conv: L.append(f"- leads: {plural(conv, 'converted lead')} set aside as archived records; only the X checks, F07 and L10 " "read them, and they are never grouped as duplicates.") elif ctx.crm == "salesforce" and "leads" in tables and not tables["leads"].usable("is_converted"): L.append("- leads: no IsConverted values, so converted leads could not be set aside and can show as false duplicate, " "bounce or owner findings. Re-export with IsConverted and ConvertedContactId before trusting lead findings.") L += ["", "## Mapping", ""] for name, t in tables.items(): mapped = [f"{f} = \"{v[1]}\"" + (f" ({v[2]})" if v[2] else "") for f, v in sorted(t.mapinfo.items()) if v[0] == "mapped" and not f.startswith("__")] if "__ukey" in t.col: mapped.append(f"unique key = \"{t.headers[t.col['__ukey']]}\"") L.append(f"- {name} mapped: " + ("; ".join(mapped) if mapped else "none")) amb = [f"{f} ({v[2]})" for f, v in sorted(t.mapinfo.items()) if v[0] == "ambiguous"] if amb: L.append(f"- {name} ambiguous, fix with --map: " + "; ".join(amb)) if name in OBJECTS: un = [f for f, v in sorted(t.mapinfo.items()) if v[0] == "unmapped"] L.append(f"- {name} not mapped (checks that need them are NOT CHECKED): " + (", ".join(un) if un else "none")) L += ["", "## Scorecard", "", f"Totals: " + ", ".join(f"{totals.get(s, 0)} {s}" for s in STATUSES) + ".", ""] for s in STATUSES: rows = [c for c in checks if c["status"] == s] if rows: L += [f"### {s}", ""] + ["- " + line(c, ids=10 if s in ("FAIL", "WARN") else 0) for c in rows] + [""] L += ["## Duplicates", "", "Excess duplicate records = records in certain and probable groups minus the number of groups, " "over the object's records excluding test records. Possible pairs are never grouped or counted as excess.", ""] for obj, e in excess.items(): L.append(f"- {obj}: {plural(e['certain_groups'], 'certain group')} and {plural(e['probable_groups'], 'probable group')} " f"holding {plural(e['records_in_groups'], 'record')}, {plural(e['possible_pairs'], 'possible pair')}. " f"Excess {rate(e['excess'], e['population'])}.") L += ["", "## Unique affected records", "", "Counts are per object and never added across checks.", ""] for obj, u in unique.items(): L.append(f"- {obj}: {u['fail']} records with at least one FAIL; {u['fail_or_warn']} with a FAIL or WARN.") L += ["", "## Next steps for the reviewer", "", "- Integrity first: any FAIL in X01, X02, X03 or X05 means a clean re-export before anything else.", "- Review duplicates.csv by band. A person with CRM access marks calibration-sample.csv (same, different or unsure).", "- Trace one flagged record per FAIL line (at least 10) and 10 passing records in the CRM; at least 95% must match.", "- Decay sample (R04) is a manual step: 200 records from one population, first-party evidence only."] if cal is not None: L.append(f"- calibration-sample.csv holds {plural(cal['pairs'], 'pair')}" + (f"; {cal['kept']} marked pairs were kept from the earlier file in this folder." if cal["earlier"] else ".")) elif os.path.exists(os.path.join(args.out, "calibration-sample.csv")): L.append("- calibration-sample.csv in this folder is from an earlier run: --calibration 0 left it untouched.") if trend is not None: L.append(f"- trend.md compares with the previous profile; {trend} groups are still open from last time.") L += ["", "Nothing was changed."] with open(os.path.join(args.out, "summary.md"), "w", encoding="utf-8") as fh: fh.write("\n".join(L) + "\n")