#!/usr/bin/env python3 from __future__ import annotations import argparse import json from pathlib import Path from typing import Any from patch_l2_pack_from_salesforce_describe import ( DEFAULT_API_VERSION, DEFAULT_CONFIG, DEFAULT_PACK_DIR, DEFAULT_TARGETS_JSON, SalesforceClient, load_kv_config, load_targets, patchable_field_spec, ) def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser(description="Audit targeted L2 pack fields against live Salesforce describe.") parser.add_argument("--config", default=DEFAULT_CONFIG, help="Path to Salesforce readonly config.") parser.add_argument("--pack-dir", default=DEFAULT_PACK_DIR, help="Path to Salesforce-PingCAP-L2 pack.") parser.add_argument("--api-version", default=DEFAULT_API_VERSION, help="Salesforce API version.") parser.add_argument("--targets-json", default=DEFAULT_TARGETS_JSON, help="Targets JSON path.") parser.add_argument("--output-dir", default=".tmp/l2-pack-describe-audit", help="Directory for audit outputs.") return parser.parse_args() def normalized_summary(field: dict[str, Any] | None) -> dict[str, Any] | None: if field is None: return None summary: dict[str, Any] = { "apiName": field.get("apiName"), "type": field.get("type"), "label": field.get("label"), } if "required" in field: summary["required"] = field.get("required") if "controllingField" in field: summary["controllingField"] = field.get("controllingField") if "referenceTo" in field: summary["referenceTo"] = field.get("referenceTo") if "relationshipName" in field: summary["relationshipName"] = field.get("relationshipName") if "isPolymorphic" in field: summary["isPolymorphic"] = field.get("isPolymorphic") if "options" in field: summary["options"] = field.get("options") if "default" in field: summary["default"] = field.get("default") if "length" in field: summary["length"] = field.get("length") if "precision" in field: summary["precision"] = field.get("precision") if "scale" in field: summary["scale"] = field.get("scale") return summary def classify_diff(current: dict[str, Any] | None, expected: dict[str, Any] | None) -> list[str]: tags: list[str] = [] if current is None: return ["missing_field"] if expected is None: return ["missing_in_salesforce"] if current.get("type") != expected.get("type"): tags.append("type") if current.get("label") != expected.get("label"): tags.append("label") if current.get("required") != expected.get("required"): tags.append("required") if current.get("controllingField") != expected.get("controllingField"): tags.append("controllingField") if current.get("referenceTo") != expected.get("referenceTo"): tags.append("referenceTo") if current.get("relationshipName") != expected.get("relationshipName"): tags.append("relationshipName") if current.get("isPolymorphic") != expected.get("isPolymorphic"): tags.append("isPolymorphic") if current.get("default") != expected.get("default"): tags.append("default") if current.get("length") != expected.get("length"): tags.append("length") if current.get("precision") != expected.get("precision"): tags.append("precision") if current.get("scale") != expected.get("scale"): tags.append("scale") if current.get("options") != expected.get("options"): tags.append("options") return tags def render_markdown(report: dict[str, Any]) -> str: lines: list[str] = [] lines.append("# L2 Pack Describe Audit") lines.append("") lines.append(f"- Objects audited: `{report['objectCount']}`") lines.append(f"- Fields audited: `{report['fieldCount']}`") lines.append(f"- Fields matching live describe: `{report['matchingFieldCount']}`") lines.append(f"- Fields differing from live describe: `{report['diffFieldCount']}`") lines.append("") for object_name, fields in report["objects"].items(): lines.append(f"## {object_name}") lines.append("") for item in fields: status = "match" if not item["diffKinds"] else f"diff: {', '.join(item['diffKinds'])}" lines.append(f"- `{item['fieldApiName']}`: {status}") lines.append("") return "\n".join(lines).rstrip() + "\n" def main() -> int: args = parse_args() pack_dir = Path(args.pack_dir).expanduser().resolve() output_dir = Path(args.output_dir).expanduser().resolve() output_dir.mkdir(parents=True, exist_ok=True) config = load_kv_config(Path(args.config).expanduser().resolve()) targets = load_targets(args.targets_json) client = SalesforceClient(config, api_version=args.api_version) report: dict[str, Any] = { "objectCount": 0, "fieldCount": 0, "matchingFieldCount": 0, "diffFieldCount": 0, "objects": {}, } for object_name, field_names in targets.items(): object_path = pack_dir / "objects" / f"{object_name}.json" payload = json.loads(object_path.read_text(encoding="utf-8")) current_fields = { str(field.get("apiName") or "").strip(): normalized_summary(field) for field in payload.get("fields", []) if isinstance(field, dict) and str(field.get("apiName") or "").strip() } describe = client.describe(object_name) describe_fields = { str(field.get("name") or "").strip(): patchable_field_spec(field) for field in describe.get("fields", []) if isinstance(field, dict) and str(field.get("name") or "").strip() } object_rows: list[dict[str, Any]] = [] for field_name in field_names: current = current_fields.get(field_name) expected = normalized_summary(describe_fields.get(field_name)) diff_kinds = classify_diff(current, expected) row = { "fieldApiName": field_name, "diffKinds": diff_kinds, "current": current, "expected": expected, } object_rows.append(row) report["fieldCount"] += 1 if diff_kinds: report["diffFieldCount"] += 1 else: report["matchingFieldCount"] += 1 report["objects"][object_name] = object_rows report["objectCount"] += 1 json_path = output_dir / "l2_pack_describe_audit.json" md_path = output_dir / "l2_pack_describe_audit.md" json_path.write_text(json.dumps(report, indent=2) + "\n", encoding="utf-8") md_path.write_text(render_markdown(report), encoding="utf-8") print(f"[ok] json={json_path}") print(f"[ok] markdown={md_path}") print(f"[ok] fields={report['fieldCount']} diffs={report['diffFieldCount']}") return 0 if __name__ == "__main__": raise SystemExit(main())