#!/usr/bin/env python3 from __future__ import annotations import argparse import base64 import json import urllib.parse import urllib.request from pathlib import Path from typing import Any DEFAULT_LOGIN_DOMAIN = "login.salesforce.com" DEFAULT_API_VERSION = "v63.0" DEFAULT_PACK_DIR = "packs/Salesforce-PingCAP-L2" DEFAULT_CONFIG = "~/.salesforce/readonly_config" DEFAULT_TARGETS_JSON = "configs/l2_pack_salesforce_describe_targets.json" def load_kv_config(path: Path) -> dict[str, str]: data: dict[str, str] = {} for raw in path.read_text(encoding="utf-8").splitlines(): line = raw.strip() if not line or line.startswith("#") or ":" not in line: continue key, value = line.split(":", 1) data[key.strip()] = value.strip() return data class SalesforceClient: def __init__(self, config: dict[str, str], api_version: str) -> None: self.login_domain = (config.get("tokenDomain") or DEFAULT_LOGIN_DOMAIN).replace("https://", "").replace( "http://", "" ).strip("/") self.client_id = config["clientId"] self.client_secret = config["clientSecret"] self.username = config["username"] self.password = config["password"] self.api_version = api_version self.instance_url: str | None = None self.access_token: str | None = None def login(self) -> None: token_url = f"https://{self.login_domain}/services/oauth2/token" payload = urllib.parse.urlencode( { "grant_type": "password", "client_id": self.client_id, "client_secret": self.client_secret, "username": self.username, "password": self.password, } ).encode("utf-8") req = urllib.request.Request(token_url, data=payload, method="POST") with urllib.request.urlopen(req, timeout=30) as resp: data = json.loads(resp.read().decode("utf-8")) self.instance_url = data["instance_url"] self.access_token = data["access_token"] def _headers(self) -> dict[str, str]: if not self.instance_url or not self.access_token: self.login() assert self.access_token is not None return {"Authorization": f"Bearer {self.access_token}"} def _api(self, path: str) -> str: if not self.instance_url or not self.access_token: self.login() assert self.instance_url is not None return f"{self.instance_url}/services/data/{self.api_version}{path}" def describe(self, object_name: str) -> dict[str, Any]: req = urllib.request.Request(self._api(f"/sobjects/{object_name}/describe"), headers=self._headers()) with urllib.request.urlopen(req, timeout=60) as resp: return json.loads(resp.read().decode("utf-8")) def pack_type_for_describe(field: dict[str, Any]) -> str: field_type = str(field.get("type") or "").strip().lower() type_map = { "id": "id", "string": "string", "textarea": "textarea", "phone": "phone", "email": "email", "url": "url", "boolean": "boolean", "int": "double", "double": "double", "currency": "currency", "percent": "percent", "date": "date", "datetime": "datetime", "picklist": "picklist", "multipicklist": "multipicklist", "reference": "reference", } return type_map.get(field_type, field_type or "string") def patchable_field_spec(field: dict[str, Any]) -> dict[str, Any]: spec: dict[str, Any] = { "apiName": field["name"], "type": pack_type_for_describe(field), "label": field.get("label") or field["name"], } if field.get("length") not in (None, 0) and spec["type"] == "string": spec["length"] = field["length"] if field.get("precision") not in (None, 0) and spec["type"] in {"double", "currency", "percent"}: spec["precision"] = field["precision"] if field.get("scale") not in (None, 0) and spec["type"] in {"double", "currency", "percent"}: spec["scale"] = field["scale"] if field.get("nillable") is False: spec["required"] = True controller_name = field.get("controllerName") if isinstance(controller_name, str) and controller_name.strip(): spec["controllingField"] = controller_name.strip() if spec["type"] == "reference": reference_to = [str(item).strip() for item in field.get("referenceTo", []) if str(item).strip()] if reference_to: spec["referenceTo"] = reference_to if len(reference_to) > 1: spec["isPolymorphic"] = True relationship_name = str(field.get("relationshipName") or "").strip() if relationship_name: spec["relationshipName"] = relationship_name if spec["type"] in {"picklist", "multipicklist"}: options = [] default_value: str | None = None for item in field.get("picklistValues", []): if not isinstance(item, dict) or item.get("active") is False: continue value = item.get("value") if value in (None, ""): continue label = item.get("label") or value options.append({"value": value, "label": label}) if item.get("defaultValue") is True: default_value = str(value) spec["options"] = options if default_value: spec["default"] = default_value return spec def decode_valid_for(valid_for: str, controller_count: int) -> list[bool]: raw = base64.b64decode(valid_for) bits: list[bool] = [] for byte in raw: for shift in range(8): bits.append(bool(byte & (0x80 >> shift))) return bits[:controller_count] def extract_picklist_dependency(field: dict[str, Any], describe_fields: dict[str, dict[str, Any]]) -> dict[str, list[str]] | None: controller_name = str(field.get("controllerName") or "").strip() if not controller_name: return None controller_field = describe_fields.get(controller_name) if not controller_field: return None controller_values = [ str(item.get("value")) for item in controller_field.get("picklistValues", []) if isinstance(item, dict) and item.get("active") is not False and item.get("value") not in (None, "") ] if not controller_values: return None dependency: dict[str, list[str]] = {value: [] for value in controller_values} for item in field.get("picklistValues", []): if not isinstance(item, dict) or item.get("active") is False: continue dependent_value = item.get("value") if dependent_value in (None, ""): continue valid_for = str(item.get("validFor") or "").strip() if not valid_for: continue bits = decode_valid_for(valid_for, len(controller_values)) for idx, enabled in enumerate(bits): if enabled: dependency[controller_values[idx]].append(str(dependent_value)) # Keep only real mappings. If every controller resolves to the full set, we still # emit the map because the downstream pack should mirror Salesforce explicitly. return dependency def merge_field(existing: dict[str, Any] | None, patched: dict[str, Any]) -> dict[str, Any]: if existing is None: return patched merged = json.loads(json.dumps(existing)) merged["type"] = patched["type"] merged["label"] = patched["label"] if patched["type"] != "reference": merged.pop("referenceTo", None) merged.pop("relationshipName", None) merged.pop("isPolymorphic", None) if patched["type"] not in {"picklist", "multipicklist"}: merged.pop("options", None) merged.pop("controllingField", None) merged.pop("default", None) if patched["type"] not in {"string"}: merged.pop("length", None) if patched["type"] not in {"double", "currency", "percent"}: merged.pop("precision", None) merged.pop("scale", None) if "required" in patched: merged["required"] = patched["required"] elif "required" in merged and patched["type"] not in {"picklist", "multipicklist"}: merged.pop("required", None) if "length" in patched: merged["length"] = patched["length"] if "precision" in patched: merged["precision"] = patched["precision"] if "scale" in patched: merged["scale"] = patched["scale"] if "controllingField" in patched: merged["controllingField"] = patched["controllingField"] else: merged.pop("controllingField", None) merged.pop("picklistDependency", None) if "referenceTo" in patched: merged["referenceTo"] = patched["referenceTo"] if "relationshipName" in patched: merged["relationshipName"] = patched["relationshipName"] if "isPolymorphic" in patched: merged["isPolymorphic"] = patched["isPolymorphic"] else: merged.pop("isPolymorphic", None) if "options" in patched: merged["options"] = patched["options"] if "picklistDependency" in patched: merged["picklistDependency"] = patched["picklistDependency"] elif patched["type"] in {"picklist", "multipicklist"}: merged.pop("picklistDependency", None) if "default" in patched: merged["default"] = patched["default"] elif patched["type"] in {"picklist", "multipicklist"}: merged.pop("default", None) return merged def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser(description="Patch the checked-in L2 pack using live Salesforce describe metadata.") 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="Optional JSON file mapping object API names to field API name arrays.", ) parser.add_argument("--dry-run", action="store_true", help="Report changes without writing files.") return parser.parse_args() def load_targets(path: str | None) -> dict[str, list[str]]: if not path: raise ValueError("targets JSON path is required") payload = json.loads(Path(path).expanduser().resolve().read_text(encoding="utf-8")) if not isinstance(payload, dict): raise ValueError("targets JSON must be an object") normalized: dict[str, list[str]] = {} for object_name, fields in payload.items(): if not isinstance(object_name, str) or not isinstance(fields, list): raise ValueError("targets JSON must map object names to arrays") normalized[object_name] = [str(item).strip() for item in fields if str(item).strip()] return normalized def main() -> int: args = parse_args() pack_dir = Path(args.pack_dir).expanduser().resolve() config = load_kv_config(Path(args.config).expanduser().resolve()) targets = load_targets(args.targets_json) client = SalesforceClient(config, api_version=args.api_version) changed_files: list[Path] = [] for object_name, field_names in targets.items(): object_path = pack_dir / "objects" / f"{object_name}.json" if not object_path.exists(): raise FileNotFoundError(f"missing object spec: {object_path}") payload = json.loads(object_path.read_text(encoding="utf-8")) if not isinstance(payload, dict) or not isinstance(payload.get("fields"), list): raise ValueError(f"unexpected object spec format: {object_path}") describe = client.describe(object_name) describe_fields = { str(field.get("name") or "").strip(): field for field in describe.get("fields", []) if isinstance(field, dict) and str(field.get("name") or "").strip() } fields = payload["fields"] field_index = { str(field.get("apiName") or "").strip(): index for index, field in enumerate(fields) if isinstance(field, dict) and str(field.get("apiName") or "").strip() } object_changed = False for field_name in field_names: describe_field = describe_fields.get(field_name) if not describe_field: continue patched = patchable_field_spec(describe_field) dependency = extract_picklist_dependency(describe_field, describe_fields) if dependency is not None: patched["picklistDependency"] = dependency index = field_index.get(field_name) existing = fields[index] if index is not None else None merged = merge_field(existing, patched) if index is None: fields.append(merged) field_index[field_name] = len(fields) - 1 object_changed = True continue if merged != existing: fields[index] = merged object_changed = True if object_changed: changed_files.append(object_path) if not args.dry_run: object_path.write_text(json.dumps(payload, indent=2) + "\n", encoding="utf-8") for path in changed_files: print(f"[patched] {path}") if not changed_files: print("[patched] no changes") return 0 if __name__ == "__main__": raise SystemExit(main())