"""Read-only full-field audit against prod TiDB. Per object: - every _System_Field row (on- and off-layout), with HelpText/Description coverage - fields displayed on Detail layouts (_System_PageLayout.Config sections) - fill rate per field split by cohort: migrated (15/18-char Salesforce Ids kept by the migration) vs native (36-char uuids created in NexusCRM); other Id shapes are counted as unknown_cohort and surfaced - twin-field divergence counters (only when the relevant object is probed) SELECT/SHOW/DESCRIBE only. Writes JSON per object to the output directory. Usage: probe_full_audit.py [OBJECT ...] [--out DIR] """ from __future__ import annotations import json import re import sys from typing import Any from common import atomic_write, connect, existing_columns, parse_args CHUNK = 40 # aggregate expressions per query # _System_Field metadata columns: (result key, candidate column names in # preference order). Optional columns select NULL when absent so the probe # runs on any schema generation (e.g. FormulaStorageMode vs FormulaMode). FIELD_META_COLS: list[tuple[str, list[str]]] = [ ("api_name", ["ApiName"]), ("label", ["Label"]), ("type", ["Type"]), ("physical", ["PhysicalName"]), ("required", ["Required"]), ("createable", ["Createable"]), ("updateable", ["Updateable"]), ("formula", ["Formula"]), ("formula_storage", ["FormulaStorageMode", "FormulaMode"]), ("rollup", ["RollupConfig"]), ("default", ["DefaultValue"]), ("customized", ["IsCustomized"]), ("owner_pack", ["OwnerPack"]), ("is_system", ["IsSystem"]), ] # Classification-bearing columns are required: absent createable/updateable # would silently classify every field as auto, absent formula/rollup would let # computed fields fall into fill-based retirement buckets, absent required # would blank the Required column everywhere. Truly informational columns # (physical name, storage mode, default, ownership) may degrade to NULL. REQUIRED_META = {"api_name", "label", "type", "createable", "updateable", "required", "formula", "rollup"} def walk_layout_fields(node: Any, acc: set[str]) -> None: """Collect field api names from a layout 'sections' subtree.""" if isinstance(node, dict): for k, v in node.items(): if k == "fields" and isinstance(v, list): acc.update(x for x in v if isinstance(x, str)) elif k in ("fieldApiName", "apiName", "field") and isinstance(v, str): acc.add(v) else: walk_layout_fields(v, acc) elif isinstance(node, list): for item in node: walk_layout_fields(item, acc) def main() -> None: objects, out_dir, _ = parse_args(sys.argv[1:]) conn = connect() out_dir.mkdir(parents=True, exist_ok=True) # Tombstone previous-run freshness markers FIRST: if this run crashes # mid-way, downstream consumers must hard-fail instead of assembling a # green report from two different runs. twins.json is included so a subset # run cannot inherit a prior run's SC-4 counters. for stale in ("manifest.json", "twins.json"): (out_dir / stale).unlink(missing_ok=True) probed_cols: dict[str, set[str]] = {} # objects actually probed -> physical columns try: with conn.cursor() as cur: sf_cols = existing_columns(cur, "_System_Field") if sf_cols is None: raise SystemExit("_System_Field table not found — wrong database?") help_cols = sorted(c for c in sf_cols if re.search(r"help|descri", c, re.I)) print(f"_System_Field semantic columns: {help_cols}") select_exprs = [] for key, candidates in FIELD_META_COLS: col = next((c for c in candidates if c in sf_cols), None) if col is None and key in REQUIRED_META: raise SystemExit(f"_System_Field lacks required column (any of {candidates})") select_exprs.append(f"f.`{col}`" if col else "NULL") select_exprs.extend(f"f.`{c}`" for c in help_cols) meta_select = ", ".join(select_exprs) n_meta = len(FIELD_META_COLS) skipped: list[str] = [] for obj in objects: cur.execute( "SELECT COUNT(*) FROM `_System_Object` WHERE ApiName=%s AND (IsDeleted=0 OR IsDeleted IS NULL)", (obj,) ) if cur.fetchone()[0] == 0: print(f"{obj}: not found in _System_Object, skipped") skipped.append(obj) continue result: dict[str, Any] = {"object": obj} cur.execute( f"SELECT {meta_select} FROM `_System_Field` f " "JOIN `_System_Object` o ON f.ObjectId=o.Id " "WHERE o.ApiName=%s AND (f.IsDeleted=0 OR f.IsDeleted IS NULL)", (obj,), ) fields = {} for r in cur.fetchall(): m = {key: r[i] for i, (key, _) in enumerate(FIELD_META_COLS)} helptexts = {c: (r[n_meta + i] or "").strip() for i, c in enumerate(help_cols)} fields[m["api_name"]] = { "label": m["label"], "type": m["type"], "physical": m["physical"] or m["api_name"], "required": bool(m["required"]), "createable": bool(m["createable"]), "updateable": bool(m["updateable"]), "is_formula": bool(m["formula"]), "formula_storage": m["formula_storage"], "is_rollup": m["rollup"] not in (None, "", "null", "{}"), "default": (m["default"] or "")[:60], "customized": bool(m["customized"]), "owner_pack": m["owner_pack"], "is_system": bool(m["is_system"]), "helptexts": helptexts, } result["field_count"] = len(fields) result["help_cols"] = help_cols result["help_filled"] = { c: sum(1 for f in fields.values() if f["helptexts"].get(c)) for c in help_cols } cur.execute( "SELECT LayoutName, Type, Config FROM `_System_PageLayout` " "WHERE ObjectApiName=%s AND (IsDeleted=0 OR IsDeleted IS NULL)", (obj,), ) layout_fields: dict[str, list[str]] = {} for lname, ltype, cfg in cur.fetchall(): cfg = json.loads(cfg) if isinstance(cfg, str) else (cfg or {}) acc: set[str] = set() walk_layout_fields(cfg.get("sections"), acc) layout_fields[f"{lname} [{ltype}]"] = sorted(a for a in acc if a in fields) result["layouts"] = {k: len(v) for k, v in layout_fields.items()} on_any_layout = set() for v in layout_fields.values(): on_any_layout.update(v) cur.execute(f"DESCRIBE `{obj}`") physical_cols = {r[0]: r[1] for r in cur.fetchall()} probed_cols[obj] = set(physical_cols) # Exclude soft-deleted rows from every data measurement on this # object (schema-gated: not every table carries IsDeleted). live = "WHERE ((IsDeleted=0 OR IsDeleted IS NULL) OR IsDeleted IS NULL)" if "IsDeleted" in physical_cols else "" # Cohorts: migrated = Salesforce Ids (15 or 18 chars), native = # NexusCRM uuids (36 chars). Anything else is counted as unknown # and surfaced rather than silently folded into a cohort. cur.execute( f"SELECT SUM(LENGTH(Id) IN (15,18)), SUM(LENGTH(Id)=36), COUNT(*) FROM `{obj}` {live}" ) n_mig, n_nat, n_all = cur.fetchone() n_mig, n_nat, n_all = int(n_mig or 0), int(n_nat or 0), int(n_all) unknown = n_all - n_mig - n_nat result["rows"] = { "migrated": n_mig, "native": n_nat, "unknown_cohort": unknown, "total": n_all, } if unknown: print(f"WARNING {obj}: {unknown} rows with Id length outside 15/18/36 excluded from cohorts") # Emptiness must branch on column type: comparing date/decimal # columns with <>'' falsely reports them dead. text_types = ("varchar", "text", "longtext", "mediumtext", "char") measurable = [] for api, f in fields.items(): col = f["physical"] if col not in physical_cols: f["fill_note"] = "no-physical-column" continue coltype = physical_cols[col].lower() if coltype.startswith("tinyint(1)"): expr_m = f"SUM(CASE WHEN LENGTH(Id) IN (15,18) AND `{col}`=1 THEN 1 ELSE 0 END)" expr_n = f"SUM(CASE WHEN LENGTH(Id)=36 AND `{col}`=1 THEN 1 ELSE 0 END)" f["bool"] = True elif any(coltype.startswith(t) for t in text_types): expr_m = f"SUM(CASE WHEN LENGTH(Id) IN (15,18) AND NULLIF(TRIM(`{col}`),'') IS NOT NULL THEN 1 ELSE 0 END)" expr_n = f"SUM(CASE WHEN LENGTH(Id)=36 AND NULLIF(TRIM(`{col}`),'') IS NOT NULL THEN 1 ELSE 0 END)" elif coltype.startswith("json"): expr_m = f"SUM(CASE WHEN LENGTH(Id) IN (15,18) AND `{col}` IS NOT NULL AND JSON_TYPE(`{col}`)<>'NULL' THEN 1 ELSE 0 END)" expr_n = f"SUM(CASE WHEN LENGTH(Id)=36 AND `{col}` IS NOT NULL AND JSON_TYPE(`{col}`)<>'NULL' THEN 1 ELSE 0 END)" else: expr_m = f"SUM(CASE WHEN LENGTH(Id) IN (15,18) AND `{col}` IS NOT NULL THEN 1 ELSE 0 END)" expr_n = f"SUM(CASE WHEN LENGTH(Id)=36 AND `{col}` IS NOT NULL THEN 1 ELSE 0 END)" measurable.append((api, expr_m, expr_n)) for i in range(0, len(measurable), CHUNK): chunk = measurable[i : i + CHUNK] exprs = ", ".join(f"{em}, {en}" for _, em, en in chunk) cur.execute(f"SELECT {exprs} FROM `{obj}` {live}") row = cur.fetchone() for j, (api, _, _) in enumerate(chunk): fields[api]["filled_migrated"] = int(row[2 * j] or 0) fields[api]["filled_native"] = int(row[2 * j + 1] or 0) for api, f in fields.items(): f["on_layout"] = api in on_any_layout result["fields"] = fields out = out_dir / f"{obj}.json" atomic_write(out, json.dumps(result, ensure_ascii=False, indent=1, default=str)) print( f"{obj}: {len(fields)} fields ({sum(1 for f in fields.values() if f['on_layout'])} on layout), " f"rows mig={result['rows']['migrated']} nat={result['rows']['native']}, " f"help_filled={result['help_filled']}" ) # Twin divergence counters (epic #4863 SC-4). Gated on objects actually # probed AND on column existence, so the audit keeps working after a # twin field is retired (it reports the absence instead of crashing). def has_cols(obj: str, *cols: str) -> bool: return obj in probed_cols and all(c in probed_cols[obj] for c in cols) twins = {} if "Opportunity" in probed_cols: key = "Opportunity.CloseDate_vs_Close_Date__c" if has_cols("Opportunity", "CloseDate", "Close_Date__c"): # Non-convergence includes one-sided population, not only # both-set-unequal rows. live = "WHERE ((IsDeleted=0 OR IsDeleted IS NULL) OR IsDeleted IS NULL)" if "IsDeleted" in probed_cols["Opportunity"] else "" cur.execute( "SELECT COUNT(*), " "SUM(CASE WHEN Close_Date__c IS NOT NULL AND CloseDate IS NOT NULL " "AND Close_Date__c<>CloseDate THEN 1 ELSE 0 END), " "SUM(CASE WHEN CloseDate IS NOT NULL AND Close_Date__c IS NULL THEN 1 ELSE 0 END), " "SUM(CASE WHEN Close_Date__c IS NOT NULL AND CloseDate IS NULL THEN 1 ELSE 0 END) " f"FROM `Opportunity` {live}" ) total, both_unequal, only_std, only_custom = cur.fetchone() twins[key] = { "total_rows": int(total), "diverged_both_set": int(both_unequal or 0), "only_CloseDate_set": int(only_std or 0), "only_Close_Date__c_set": int(only_custom or 0), "not_converged_total": int(both_unequal or 0) + int(only_std or 0) + int(only_custom or 0), } else: twins[key] = {"status": "column absent (twin retired?)"} if "Account" in probed_cols: key = "Account.AccountStatus__c_vs_Account_Status__c" if has_cols("Account", "AccountStatus__c", "Account_Status__c"): live = "WHERE ((IsDeleted=0 OR IsDeleted IS NULL) OR IsDeleted IS NULL)" if "IsDeleted" in probed_cols["Account"] else "" cur.execute( "SELECT SUM(NULLIF(TRIM(AccountStatus__c),'') IS NOT NULL), " f"SUM(NULLIF(TRIM(Account_Status__c),'') IS NOT NULL), COUNT(*) FROM `Account` {live}" ) a, b, t = cur.fetchone() # Unlike the Opportunity date pair, these two fields carry # INTENTIONALLY distinct semantics (sales stage vs tenant # status; owner-confirmed dual-active). Equality is not the # goal, so there is no per-row divergence check — the # counters track usage of each side. twins[key] = { "semantics": "intentionally-distinct (usage counters, not convergence)", "AccountStatus__c_filled": int(a or 0), "Account_Status__c_filled": int(b or 0), "total_rows": int(t), } else: twins[key] = {"status": "column absent (twin retired?)"} if twins: atomic_write(out_dir / "twins.json", json.dumps(twins, indent=1)) print("twins:", json.dumps(twins)) # Run manifest: records what THIS run actually probed, so downstream # consumers can distinguish fresh output from stale files left in the # output directory by earlier runs. atomic_write( out_dir / "manifest.json", json.dumps({"probed": sorted(probed_cols), "skipped": skipped}, indent=1), ) finally: conn.close() if __name__ == "__main__": main()