#!/usr/bin/env python3 """Provision HubSpot sandbox properties declared by the PingCAP companion pack. The script is intentionally narrow and idempotent: - reads checked-in property manifests, not live exports; - refuses to mutate a non-SANDBOX HubSpot portal; - creates missing properties and verifies existing properties have compatible immutable shape. """ from __future__ import annotations import argparse import json import os import sys from pathlib import Path from urllib.error import HTTPError, URLError from urllib.parse import quote from urllib.request import Request, urlopen DEFAULT_API_BASE = "https://api.hubapi.com" DEFAULT_MANIFEST_ROOT = Path("packs/PingCAP-HubSpot-Companion/hubspot_properties") MUTABLE_PROPERTY_KEYS = ("label", "description", "groupName") class ProvisionError(Exception): pass def load_env_file(path: Path) -> dict[str, str]: values: dict[str, str] = {} if not path.is_file(): return values for raw_line in path.read_text(encoding="utf-8").splitlines(): line = raw_line.strip() if not line or line.startswith("#") or "=" not in line: continue key, value = line.split("=", 1) value = value.strip() if len(value) >= 2 and value[0] == value[-1] and value[0] in ("'", '"'): value = value[1:-1] values[key.strip()] = value return values def resolve_token(envv: dict[str, str]) -> str: for key in ("EXP_HUBSPOT_ACCESS_TOKEN", "HUBSPOT_STAGING_TOKEN", "HUBSPOT_TOKEN"): value = os.environ.get(key, "").strip() or envv.get(key, "").strip() if value: return value return "" def request_json( method: str, path: str, token: str, *, api_base: str, payload: dict | None = None, ok_statuses: tuple[int, ...] = (200,), ) -> tuple[int, dict]: data = json.dumps(payload).encode("utf-8") if payload is not None else None headers = {"Authorization": f"Bearer {token}"} if data is not None: headers["Content-Type"] = "application/json" req = Request(f"{api_base.rstrip('/')}{path}", data=data, headers=headers, method=method) try: with urlopen(req, timeout=30) as resp: body = resp.read().decode("utf-8") parsed = json.loads(body) if body.strip() else {} if resp.status not in ok_statuses: raise ProvisionError(f"{method} {path} returned HTTP {resp.status}") return resp.status, parsed except HTTPError as exc: body = exc.read().decode("utf-8", errors="replace") if exc.code in ok_statuses: return exc.code, json.loads(body) if body.strip() else {} raise ProvisionError(f"{method} {path} returned HTTP {exc.code}: {body[:500]}") from exc except URLError as exc: raise ProvisionError(f"{method} {path} failed: {exc}") from exc def ensure_sandbox(token: str, *, api_base: str, expected_portal_id: str = "") -> dict: _, body = request_json("GET", "/account-info/v3/details", token, api_base=api_base) account_type = str(body.get("accountType") or "").strip() portal_id = str(body.get("portalId") or "").strip() if account_type != "SANDBOX": raise ProvisionError(f"refusing to provision HubSpot properties: portal {portal_id} accountType={account_type}") if expected_portal_id and portal_id != expected_portal_id: raise ProvisionError( f"HubSpot portal {portal_id} does not match expected portal {expected_portal_id}" ) return {"portalId": portal_id, "accountType": account_type} def load_property_specs(manifest_root: Path) -> list[dict]: specs: list[dict] = [] if not manifest_root.is_dir(): raise ProvisionError(f"HubSpot property manifest root not found: {manifest_root}") for path in sorted(manifest_root.glob("*.json")): payload = json.loads(path.read_text(encoding="utf-8")) if not isinstance(payload, list): raise ProvisionError(f"HubSpot property manifest must be a list: {path}") for item in payload: if not isinstance(item, dict): raise ProvisionError(f"invalid property entry in {path}") item = dict(item) item["_manifestPath"] = str(path) for key in ("objectType", "name", "label", "type", "fieldType", "groupName"): if not str(item.get(key) or "").strip(): raise ProvisionError(f"{path} property entry missing {key}") specs.append(item) return specs def create_payload(spec: dict) -> dict: out = { "name": spec["name"], "label": spec["label"], "type": spec["type"], "fieldType": spec["fieldType"], "groupName": spec["groupName"], } if spec.get("description"): out["description"] = spec["description"] if spec.get("options"): out["options"] = spec["options"] return out def patch_payload(spec: dict) -> dict: return {key: spec[key] for key in MUTABLE_PROPERTY_KEYS if key in spec} def ensure_property(spec: dict, token: str, *, api_base: str) -> str: object_type = quote(str(spec["objectType"]), safe="") name = quote(str(spec["name"]), safe="") status, existing = request_json( "GET", f"/crm/v3/properties/{object_type}/{name}", token, api_base=api_base, ok_statuses=(200, 404), ) if status == 404: request_json( "POST", f"/crm/v3/properties/{object_type}", token, api_base=api_base, payload=create_payload(spec), ok_statuses=(200, 201), ) return "created" for key in ("type", "fieldType"): actual = str(existing.get(key) or "").strip() expected = str(spec[key]).strip() if actual != expected: raise ProvisionError( f"HubSpot property {spec['objectType']}/{spec['name']} has {key}={actual}, expected {expected}" ) updates = patch_payload(spec) changed = any(str(existing.get(key) or "") != str(value) for key, value in updates.items()) if changed: request_json( "PATCH", f"/crm/v3/properties/{object_type}/{name}", token, api_base=api_base, payload=updates, ok_statuses=(200,), ) return "updated" return "present" def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser(description="Provision PingCAP HubSpot sandbox properties") parser.add_argument("--env-file", help="runtime.env file to read EXP_HUBSPOT_ACCESS_TOKEN from") parser.add_argument("--manifest-root", default=str(DEFAULT_MANIFEST_ROOT)) parser.add_argument("--api-base", default=os.environ.get("HUBSPOT_API_BASE", DEFAULT_API_BASE)) parser.add_argument("--expected-portal-id", default=os.environ.get("HUBSPOT_EXPECTED_PORTAL_ID", "")) parser.add_argument("--allow-missing-token", action="store_true") return parser.parse_args() def main() -> int: args = parse_args() envv = load_env_file(Path(args.env_file).expanduser()) if args.env_file else {} token = resolve_token(envv) if not token: if args.allow_missing_token: print("hubspot property provisioning skipped: no HubSpot token") return 0 raise ProvisionError("missing HubSpot token") repo_root = Path(__file__).resolve().parents[1] manifest_root = Path(args.manifest_root) if not manifest_root.is_absolute(): manifest_root = repo_root / manifest_root portal = ensure_sandbox(token, api_base=args.api_base, expected_portal_id=args.expected_portal_id) specs = load_property_specs(manifest_root) results = [] for spec in specs: status = ensure_property(spec, token, api_base=args.api_base) results.append(f"{spec['objectType']}/{spec['name']}={status}") print( "hubspot property provisioning ok: " f"portal={portal['portalId']} accountType={portal['accountType']} " + ", ".join(results) ) return 0 if __name__ == "__main__": try: raise SystemExit(main()) except ProvisionError as exc: print(f"hubspot property provisioning failed: {exc}", file=sys.stderr) raise SystemExit(1)