#!/usr/bin/env python3 """Post-pack-apply reconcile for the APAC Sub-unit (PoD) feature — run via REST AFTER the staging deploy's `setup apply`, so the schema + migrated data already exist and REST writes rebuild the group-membership closure automatically. It re-applies the three non-pack pieces so they self-heal after every reset:true deploy (which wipes the branch): 1. TAGS — Account.Sub_Unit__c set by Account_ID__c (reviewed 270-row mapping) 2. FLS — FieldPermissions (read+edit) on Account.Sub_Unit__c for every profile-owned PermissionSet, so end users can see the field 3. SHARING — per PoD: a Public Group (member = the virtual-lead User) + a criteria SharingRule Sub_Unit__c == "" -> group (Read) NON-FATAL BY DESIGN: every phase is wrapped; any error is logged and the next phase still runs, and the process exits 0 unless --strict is passed. So a reconcile problem (a 504, a schema surprise) can NEVER fail the deploy the way the earlier apply-patch SQL bundle did. Idempotent: query-then-create/patch. Creds come from flags or, for the deploy, env: --base/$RECONCILE_BASE_URL (default $PUBLIC_BASE_URL), --username/$RECONCILE_ADMIN_USERNAME (default admin@example.invalid), --password/$RECONCILE_ADMIN_PASSWORD (default $NEXUS_ADMIN_PASSWORD). """ from __future__ import annotations import argparse import json import os import sys import time import urllib.error import urllib.parse import urllib.request from pathlib import Path from typing import Any FIELD = "Sub_Unit__c" OBJECT = "Account" ACCESS_LEVEL = "Read" # PoD picklist value -> virtual-lead full Name (validated active 2026-07-07). LEADS: dict[str, str] = { "AI": "Lux Li", "Web3": "Nicole Li", "Hospitality": "Andy Hsu", "Strategic Growth": "Zhan Jie", "IN": "Siddharth Taliyan", "ANZ": "Brian Niu", "Alliance Sales": "Xianming Dou", "CSM": "Xianming Dou", } DEFAULT_MAPPING = Path(__file__).with_name("apac_sub_unit_tags.json") def log(msg: str) -> None: print(f"[reconcile-apac-sub-unit] {msg}", flush=True) def slug(v: str) -> str: return "".join(c if c.isalnum() else "_" for c in v).strip("_") def esc(v: str) -> str: return v.replace("\\", "\\\\").replace("'", "\\'") class Client: def __init__(self, base: str, user: str, pw: str, cid: str, csec: str, host: str | None = None) -> None: self.base = base.rstrip("/") # Optional Host header: set when base is a 127.0.0.1: localhost URL so the backend # (which routes/validates by Host) still resolves the correct server, exactly as the # deploy's own localhost smoke + cache-refresh calls do. self.host = (host or "").strip() or None self._auth = dict(grant_type="password", client_id=cid, client_secret=csec, username=user, password=pw) self.token: str | None = None def login(self) -> None: data = urllib.parse.urlencode(self._auth).encode() headers = {"Host": self.host} if self.host else {} req = urllib.request.Request(f"{self.base}/services/oauth2/token", data=data, method="POST", headers=headers) with urllib.request.urlopen(req, timeout=30) as r: self.token = str(json.loads(r.read().decode()).get("access_token") or "").strip() if not self.token: raise RuntimeError("login returned no access_token") def _h(self, js: bool = False) -> dict[str, str]: h = {"Authorization": f"Bearer {self.token}"} if self.host: h["Host"] = self.host if js: h["Content-Type"] = "application/json" return h def query(self, soql: str) -> list[dict[str, Any]]: url = f"{self.base}/services/data/v63.0/query?" + urllib.parse.urlencode({"q": soql}) out: list[dict[str, Any]] = [] while True: with urllib.request.urlopen(urllib.request.Request(url, headers=self._h()), timeout=60) as r: body = json.loads(r.read().decode()) out.extend(body.get("records") or []) nxt = str(body.get("nextRecordsUrl") or "").strip() if not nxt: return out url = f"{self.base}{nxt}" def create(self, sobject: str, fields: dict[str, Any]) -> str: req = urllib.request.Request(f"{self.base}/services/data/v63.0/sobjects/{sobject}", data=json.dumps(fields).encode(), method="POST", headers=self._h(True)) with urllib.request.urlopen(req, timeout=60) as r: return str(json.loads(r.read().decode()).get("id") or "").strip() def patch_composite(self, records: list[dict[str, Any]]) -> list[dict[str, Any]]: req = urllib.request.Request(f"{self.base}/services/data/v63.0/composite/sobjects", data=json.dumps({"allOrNone": False, "records": records}).encode(), method="PATCH", headers=self._h(True)) with urllib.request.urlopen(req, timeout=120) as r: results = json.loads(r.read().decode()) return [x for x in results if not x.get("success", False)] def phase_tags(c: Client, mapping: Path) -> str: data = json.loads(mapping.read_text(encoding="utf-8")) want = {r["account_id"]: r["sub_unit"] for r in data.get("rows", []) if r.get("account_id") and r.get("sub_unit")} ids = list(want) state: dict[str, dict[str, Any]] = {} for i in range(0, len(ids), 100): chunk = ids[i:i + 100] inlist = ",".join(f"'{esc(x)}'" for x in chunk) for r in c.query(f"SELECT Id, Account_ID__c, {FIELD} FROM {OBJECT} WHERE Account_ID__c IN ({inlist})"): aid = str(r.get("Account_ID__c") or "").strip() if aid: state[aid] = {"Id": r.get("Id"), "cur": (r.get(FIELD) or r.get(FIELD.lower()))} to_patch = [{"attributes": {"type": OBJECT}, "id": state[a]["Id"], FIELD: want[a]} for a in ids if a in state and state[a]["cur"] != want[a]] applied = fail = 0 for i in range(0, len(to_patch), 5): try: fail += len(c.patch_composite(to_patch[i:i + 5])) applied += len(to_patch[i:i + 5]) except urllib.error.HTTPError: time.sleep(1) # transient (e.g. 504); leave for next reconcile return f"matched={len(state)}/{len(ids)} set={applied - fail} record_errors={fail}" def phase_fls(c: Client) -> str: permsets = c.query("SELECT Id FROM PermissionSet WHERE IsOwnedByProfile = true") existing = {str(r.get("ParentId") or "") for r in c.query(f"SELECT ParentId FROM FieldPermissions WHERE Field = '{OBJECT}.{FIELD}'")} created = 0 for ps in permsets: pid = ps.get("Id") if pid and pid not in existing: try: c.create("FieldPermissions", {"ParentId": pid, "SobjectType": OBJECT, "Field": f"{OBJECT}.{FIELD}", "PermissionsRead": True, "PermissionsEdit": True}) created += 1 except Exception as e: # noqa: BLE001 - non-fatal per permset log(f" FLS grant failed for permset {pid}: {str(e)[:120]}") return f"profile_permsets={len(permsets)} already={len(existing)} granted={created}" def phase_sharing(c: Client) -> str: roster = c.query("SELECT Id, Name, IsActive FROM User WHERE UserRole.Name LIKE 'APAC%'") by_name: dict[str, list[dict[str, Any]]] = {} for r in roster: by_name.setdefault(str(r.get("Name")), []).append(r) made = {"group": 0, "member": 0, "rule": 0, "opp_rule": 0, "skip": 0} for value, lead in LEADS.items(): active = [h for h in by_name.get(lead, []) if h.get("IsActive")] if len(active) != 1: made["skip"] += 1 log(f" skip {value}: lead '{lead}' not uniquely active ({len(active)})") continue uid = active[0]["Id"] dev = f"Sub_Unit_Lead_{slug(value)}" rname = f"Share_Account_SubUnit_{slug(value)}" grp = c.query(f"SELECT Id FROM Group WHERE DeveloperName = '{esc(dev)}'") gid = grp[0]["Id"] if grp else c.create("Group", {"Name": dev, "DeveloperName": dev, "Label": f"Sub-unit Lead — {value}", "Type": "Regular"}) if not grp: made["group"] += 1 if not c.query(f"SELECT Id FROM GroupMember WHERE GroupId = '{esc(gid)}' AND UserOrGroupId = '{esc(uid)}'"): c.create("GroupMember", {"GroupId": gid, "UserOrGroupId": uid}) made["member"] += 1 rule = c.query(f"SELECT Id FROM SharingRule WHERE ObjectApiName = '{OBJECT}' AND Name = '{esc(rname)}'") if not rule: c.create("SharingRule", {"ObjectApiName": OBJECT, "Name": rname, "Criteria": f'{FIELD} == "{value}"', "AccessLevel": ACCESS_LEVEL, "ShareWithGroupId": gid}) made["rule"] += 1 # Opportunity visibility: reuse the SAME PoD group; criteria keys on the # Opportunity.Sub_Unit__c formula field (= text(Account.Sub_Unit__c)), so a lead # reads every Opportunity whose account is tagged to their PoD, even non-owned ones. orname = f"Share_Opportunity_SubUnit_{slug(value)}" orule = c.query(f"SELECT Id FROM SharingRule WHERE ObjectApiName = 'Opportunity' AND Name = '{esc(orname)}'") if not orule: c.create("SharingRule", {"ObjectApiName": "Opportunity", "Name": orname, "Criteria": f'{FIELD} == "{value}"', "AccessLevel": ACCESS_LEVEL, "ShareWithGroupId": gid}) made["opp_rule"] += 1 return (f"groups+{made['group']} members+{made['member']} " f"account_rules+{made['rule']} opp_rules+{made['opp_rule']} skipped={made['skip']}") def main() -> int: p = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) p.add_argument("--base", default=os.environ.get("RECONCILE_BASE_URL") or os.environ.get("PUBLIC_BASE_URL", "")) p.add_argument("--username", default=os.environ.get("RECONCILE_ADMIN_USERNAME", "admin@example.invalid")) p.add_argument("--password", default=os.environ.get("RECONCILE_ADMIN_PASSWORD") or os.environ.get("NEXUS_ADMIN_PASSWORD", "")) p.add_argument("--client-id", default="nexuscrm-ui") p.add_argument("--client-secret", default="nexuscrm-ui-internal") p.add_argument("--host", default=os.environ.get("RECONCILE_HOST", ""), help="Optional Host header; set when --base is a 127.0.0.1: localhost URL.") p.add_argument("--mapping", default=str(DEFAULT_MAPPING)) p.add_argument("--strict", action="store_true", help="Exit non-zero on any phase error (default: never fail).") args = p.parse_args() if not args.base or not args.password: log("no base url / password -> nothing to do (skipping reconcile)") return 0 errors = 0 try: c = Client(args.base, args.username, args.password, args.client_id, args.client_secret, host=args.host) c.login() log(f"connected to {args.base} as {args.username}") except Exception as e: # noqa: BLE001 log(f"login failed -> skipping reconcile (non-fatal): {str(e)[:160]}") return 1 if args.strict else 0 for name, fn in (("tags", lambda: phase_tags(c, Path(args.mapping))), ("fls", lambda: phase_fls(c)), ("sharing", lambda: phase_sharing(c))): try: log(f"{name}: {fn()}") except Exception as e: # noqa: BLE001 - non-fatal per phase errors += 1 log(f"{name}: FAILED (non-fatal): {str(e)[:200]}") log(f"done (phase_errors={errors})") return 1 if (args.strict and errors) else 0 if __name__ == "__main__": sys.exit(main())