"""Oracle NetSuite REST record API — outbound identity sync. Pushes Helix's HRIS-of-truth employee master data into NetSuite's REST record API (``/services/rest/record/v1/employee``). Backs the single ``netsuite`` sync-target kind. Unlike the SCIM / 汇联易 runners this is single-phase (employees only — NetSuite subsidiaries/departments are NetSuite-managed reference data, not something Helix projects). Key vendor facts (NetSuite 2024.1+ REST record API + TBA): * **Auth = Token-Based Authentication (TBA), OAuth 1.0a, HMAC-SHA256.** There is NO token-exchange step — the four static credentials (Consumer Key/Secret + Token ID/Secret) sign EVERY request. The ``realm`` is the account id (uppercase). Host is derived from the account id: ``{account_id_lower_dashed}.suitetalk.api.netsuite.com`` (e.g. ``7429932`` → ``7429932.suitetalk.api.netsuite.com``; a sandbox ``1234567_SB1`` → ``1234567-sb1.suitetalk.api.netsuite.com``). * **Idempotency = external id.** Helix sets each employee's NetSuite ``externalId`` to the HRIS 工号 (``P01431``-style) and addresses the record by ``employee/eid:{externalId}``. ``PUT eid:`` is NetSuite's documented UPSERT — create-if-absent, update-if-present — so the runner needs no lookup-first round-trip. A successful upsert returns ``204 No Content`` + a ``Location`` header carrying the internal id. * **离职 = 停用, never delete.** NetSuite employees carry an ``isInactive`` boolean; deactivate = ``PATCH …{"isInactive": true}``. Hard delete is intentionally NOT exposed (``supports_delete=False``) — same offboarding contract as 汇联易. * **OneWorld requires ``subsidiary``** on employee create. The mapping rules render ``subsidiary.id`` from ``company_code`` via a value_map; for accounts/tests without that map filled, the runner injects a ``runner_options.netsuite_default_subsidiary_id`` fallback so a single-person test push doesn't 400 on a missing required field. Wire layout: * :class:`NetSuiteClient` — per-request OAuth1 TBA signer + authed HTTP with bounded 429/5xx retries (mirrors ``HuilianyiClient`` conventions minus the token cache, which TBA doesn't need). * :class:`NetSuiteRunner` — employee upsert (PUT eid:) + deactivate (PATCH isInactive). External key = HRIS 工号. """ from __future__ import annotations import asyncio import base64 import hashlib import hmac import json import logging import re import time from typing import Any from urllib.parse import quote, unquote import httpx from app.domains.identity_sync.runners.base import ( DeptOutcome, DeptSyncRunner, PersonOutcome, SyncRunner, ) logger = logging.getLogger(__name__) # NetSuite account ids are alphanumeric + ``_`` (sandbox suffix) + ``-``. # Validate before building the host so a hostile account_id string can't # inject a different host into the derived suitetalk URL. _ACCOUNT_ID_RE = re.compile(r"^[A-Za-z0-9_-]+$") # REST record API base path. The record type is appended by the runner. _REST_RECORD_BASE = "/services/rest/record/v1" # NetSuite REST Web Services rejects requests when the token owner's UI # preference is not the company's default language. The production account # explicitly requires ``zh``; pinning the request language makes the # integration independent of that mutable user preference. _COMPANY_CONTENT_LANGUAGE = "zh" class NetSuiteError(RuntimeError): """Wraps an HTTP/transport failure from the NetSuite REST API.""" def __init__( self, message: str, *, status_code: int | None = None, body_excerpt: str | None = None, ): super().__init__(message) self.status_code = status_code self.body_excerpt = body_excerpt def _pencode(value: Any) -> str: """RFC 3986 percent-encoding (OAuth 1.0a flavour). ``quote`` always leaves the unreserved set (A-Z a-z 0-9 ``-._~``) untouched; ``safe=""`` forces everything else — including ``/`` and ``:`` — to be encoded, which is exactly what the OAuth signature base string and Authorization header require. """ return quote(str(value), safe="") def redact_credentials(text: str, *secrets: str | None) -> str: """Strip our credential material from vendor-echoed text before it lands in an error message / ledger row. NetSuite 4xx bodies don't usually echo the Authorization header, but a verbose validation message or proxy could — defense in depth.""" out = text or "" for sec in secrets: if sec and len(sec) >= 6: out = out.replace(sec, "***") return out class NetSuiteClient: """Per-request OAuth 1.0a (TBA, HMAC-SHA256) signer + authed HTTP. No token cache — TBA signs each request from the four static credentials. ``host`` is derived from ``account_id`` and is always ``*.suitetalk.api.netsuite.com`` (the account id is format-validated so it cannot inject a different host). """ def __init__( self, *, account_id: str, consumer_key: str, consumer_secret: str, token_id: str, token_secret: str, verify_ssl: bool = True, ca_bundle_path: str | None = None, timeout_seconds: int = 30, rate_limit_per_minute: int = 60, ): account_id = (account_id or "").strip() if not account_id: raise ValueError("netsuite client requires account_id") if not _ACCOUNT_ID_RE.match(account_id): raise ValueError( "netsuite account_id must be alphanumeric " "(letters/digits/_/-) — got an unexpected shape" ) if not (consumer_key and consumer_secret and token_id and token_secret): raise ValueError( "netsuite client requires consumer_key + consumer_secret " "+ token_id + token_secret" ) self._account_id = account_id # Realm is the account id uppercased; host is the account id # lowercased with ``_`` → ``-`` (NetSuite's documented URL form). self._realm = account_id.upper() host_slug = account_id.lower().replace("_", "-") self._host = f"{host_slug}.suitetalk.api.netsuite.com" self._base_url = f"https://{self._host}" self._consumer_key = consumer_key self._consumer_secret = consumer_secret self._token_id = token_id self._token_secret = token_secret # trust_env=False / follow_redirects=False — same hardening as the # other identity-sync clients: no ambient proxy can siphon the # signed request, no redirect can replay it cross-host. from app.domains.identity_sync.runners.tls import build_identity_sync_tls_verify tls_verify = build_identity_sync_tls_verify( base_url=self._base_url, verify_ssl=verify_ssl, ca_bundle_path=ca_bundle_path, ) self._client = httpx.AsyncClient( timeout=timeout_seconds, verify=tls_verify, follow_redirects=False, trust_env=False, ) self._min_interval = 60.0 / max(1, rate_limit_per_minute) self._rl_lock = asyncio.Lock() self._last_call_at: float = 0.0 # Circuit breaker — a dead host / revoked token otherwise costs # every person in a sweep the full retry budget of timeouts. self._consecutive_failures = 0 self._breaker_threshold = 5 @property def host(self) -> str: return self._host def _redact(self, text: str) -> str: return redact_credentials( text, self._consumer_secret, self._token_secret, ) def _check_breaker(self) -> None: if self._consecutive_failures >= self._breaker_threshold: raise NetSuiteError( f"circuit open: {self._consecutive_failures} consecutive " "failures this sweep — host unreachable or token revoked; " "fix and re-run." ) # ── OAuth 1.0a TBA signing ──────────────────────────────────────── def _auth_header( self, method: str, url_no_query: str, query: dict[str, str] | None, ) -> str: """Build the ``Authorization: OAuth …`` header for one request. ``url_no_query`` is the absolute URL without the query string; ``query`` carries any query params (included in the signature base string per the OAuth spec). The JSON body is NOT part of the signature (only form-encoded bodies are). """ oauth = { "oauth_consumer_key": self._consumer_key, "oauth_token": self._token_id, "oauth_signature_method": "HMAC-SHA256", "oauth_timestamp": str(int(time.time())), # 32 hex chars of nonce — unique per request. time_ns + a # process-monotonic counter avoids Math.random-style # collisions without needing a CSPRNG import here. "oauth_nonce": hashlib.sha256( f"{time.time_ns()}:{self._last_call_at}:{url_no_query}".encode() ).hexdigest()[:32], "oauth_version": "1.0", } all_params = {**(query or {}), **oauth} # Param string: sort by encoded key, then encoded value. param_pairs = sorted( (_pencode(k), _pencode(v)) for k, v in all_params.items() ) param_string = "&".join(f"{k}={v}" for k, v in param_pairs) base_string = "&".join([ method.upper(), _pencode(url_no_query), _pencode(param_string), ]) signing_key = ( f"{_pencode(self._consumer_secret)}&{_pencode(self._token_secret)}" ) signature = base64.b64encode( hmac.new( signing_key.encode("utf-8"), base_string.encode("utf-8"), hashlib.sha256, ).digest() ).decode("ascii") header_params = { **oauth, "oauth_signature": signature, } # ``realm`` is unquoted-key, quoted-value and comes first by # convention. All values percent-encoded + double-quoted. parts = [f'realm="{_pencode(self._realm)}"'] parts += [ f'{_pencode(k)}="{_pencode(v)}"' for k, v in sorted(header_params.items()) ] return "OAuth " + ", ".join(parts) # ── Rate limit ──────────────────────────────────────────────────── async def _rate_limit_wait(self) -> None: async with self._rl_lock: now = asyncio.get_event_loop().time() wait = self._min_interval - (now - self._last_call_at) if wait > 0: await asyncio.sleep(wait) self._last_call_at = asyncio.get_event_loop().time() # ── Authed request with bounded retries ─────────────────────────── async def request( self, method: str, path: str, *, json_body: dict[str, Any] | None = None, params: dict[str, str] | None = None, retry_on_429: int = 2, ) -> tuple[int, Any, str, str | None]: """Issue one signed request. Returns ``(status_code, body_json_or_None, raw_text, location_header)``. Retries 429 / 5xx with exponential backoff up to ``retry_on_429`` times. A 401/403 is a hard auth/permission failure (TBA has no refresh) — returned immediately. """ self._check_breaker() url = f"{self._base_url}{path}" attempt = 0 while True: await self._rate_limit_wait() auth = self._auth_header(method, url, params) headers = { "Authorization": auth, "Accept": "application/json", "Content-Language": _COMPANY_CONTENT_LANGUAGE, } if json_body is not None: headers["Content-Type"] = "application/json" try: resp = await self._client.request( method.upper(), url, params=params, json=json_body, headers=headers, ) except httpx.HTTPError as e: self._consecutive_failures += 1 raise NetSuiteError( f"transport error: {type(e).__name__}: {e!s:.200}" ) from e status = resp.status_code # Retry transient throttle / server errors. Honor ``Retry-After`` # when the vendor sends it (429s especially) — backing off for the # server-requested interval instead of a blind exponential clears # the throttle faster and avoids hammering (parity with the # Huilianyi client). Cap so a pathological header can't stall the # sweep; fall back to exponential when absent/unparseable. if status == 429 or 500 <= status < 600: if attempt < retry_on_429: attempt += 1 ra = resp.headers.get("Retry-After") if ra and str(ra).strip().isdigit(): await asyncio.sleep(min(int(ra), 30)) else: await asyncio.sleep(min(2 ** attempt, 8)) continue body: Any = None text = resp.text or "" if text: try: body = resp.json() except ValueError: body = None location = resp.headers.get("Location") if 200 <= status < 300: self._consecutive_failures = 0 elif status in (401, 403) or status == 429 or 500 <= status < 600: # The breaker exists to catch a DEAD HOST / REVOKED TOKEN, so # only genuinely infrastructure/auth failures may arm it: # • 401 / 403 → auth/permission hard failure (TBA has no # refresh; the whole run is doomed). # • 429 / 5xx that reached here → the retry budget above was # already exhausted, i.e. the host is throttling # or erroring persistently. self._consecutive_failures += 1 else: # Everything else non-2xx is a per-person BUSINESS outcome — # NOT a transport/auth failure — so it MUST NOT arm the # breaker (aligns with HuilianyiClient, which only counts # transport + retry-exhausted 429/5xx): # • 404 → record not adopted yet (update-only PATCH) or # already absent on deprovision; settled as # ``excluded``/``skipped`` by the caller. # • admin-record lock → policy skip (``excluded``). # • 400 USER_ERROR / validation / 409 conflict → one # person's bad data; the caller records a per-person # ``failed``. Counting these tripped the breaker at # the 5th data error and bricked the whole remaining # population with a misleading "host unreachable / # token revoked" (#1758). # Leave the counter untouched (don't reset either, so a real # run of transport/5xx failures interspersed with business # errors still accumulates toward the threshold). pass return status, body, text, location async def suiteql( self, q: str, *, page_size: int = 1000, max_rows: int = 20000, ) -> list[dict[str, Any]]: """Run a SuiteQL query, paginating, and return all rows. Requires the role's ``SuiteAnalytics Workbook`` permission (the upsert-only record permission is NOT enough — verified 2026-06-25). Used by the orchestrator to build the ``工号 → NetSuite internal id`` map for reference-field resolution (the role can't enumerate via the record collection API). Adds ``Prefer: transient`` so the result isn't persisted server-side. """ items: list[dict[str, Any]] = [] offset = 0 while len(items) < max_rows: self._check_breaker() await self._rate_limit_wait() url = f"{self._base_url}/services/rest/query/v1/suiteql" params = {"limit": str(page_size), "offset": str(offset)} headers = { "Authorization": self._auth_header("POST", url, params), "Accept": "application/json", "Content-Type": "application/json", "Prefer": "transient", } try: resp = await self._client.request( "POST", url, params=params, json={"q": q}, headers=headers, ) except httpx.HTTPError as e: self._consecutive_failures += 1 raise NetSuiteError( f"suiteql transport error: {type(e).__name__}: {e!s:.200}" ) from e if resp.status_code != 200: self._consecutive_failures += 1 raise NetSuiteError( self._redact( f"suiteql HTTP {resp.status_code}: " f"{_vendor_message(_safe_json(resp.text), resp.text)}" ), status_code=resp.status_code, ) self._consecutive_failures = 0 body = _safe_json(resp.text) or {} page = body.get("items") or [] items.extend(page) if not body.get("hasMore"): break offset += page_size return items async def aclose(self) -> None: await self._client.aclose() def _safe_json(text: str) -> Any: """Parse JSON text, returning None on empty/invalid input.""" if not text: return None try: return json.loads(text) except (ValueError, TypeError): return None def _vendor_message(body: Any, raw: str) -> str: """Pull a human-readable error out of a NetSuite error envelope. NetSuite REST errors look like:: {"type": ".../error", "title": "...", "status": 400, "o:errorDetails": [{"detail": "...", "o:errorCode": "..."}]} """ if isinstance(body, dict): details = body.get("o:errorDetails") if isinstance(details, list) and details: first = details[0] if isinstance(first, dict): msg = first.get("detail") or first.get("o:errorCode") if msg: return str(msg) title = body.get("title") or body.get("message") if title: return str(title) return (raw or "")[:300] def _error_code(body: Any) -> str | None: if isinstance(body, dict): details = body.get("o:errorDetails") if isinstance(details, list) and details and isinstance(details[0], dict): code = details[0].get("o:errorCode") if code: return str(code) return None # NetSuite refuses ANY non-admin role (the integration's TBA role is # deliberately non-admin — see the driver docstring) read/write access to # an Administrator-flagged employee record, with a 400 ``USER_ERROR``: # zh: "出于安全方面的原因，只有管理员才能够编辑管理员记录。" # en: "For security reasons, only an administrator can edit an # administrator record." # These are the handful of NetSuite IT-admin employees. By decision # (DECISIONS.md, 2026-06-24) they are EXCLUDED from Helix's outbound sync # and maintained manually, so this rejection is a *policy skip* # (``status='excluded'``), NOT a red ``failed`` — a full sweep must not # choke on the admins. Markers are matched locale-tolerantly on the vendor # detail message; ``o:errorCode`` is ``USER_ERROR`` (shared with many # other user errors) so the message text is the discriminator. _ADMIN_RECORD_LOCK_MARKERS = ( "管理员记录", # zh: "…编辑管理员记录" "administrator record", # en: "…edit an administrator record" ) def _is_admin_record_lock(body: Any, raw: str) -> bool: """True when the error is NetSuite's admin-record edit/read lock.""" msg = _vendor_message(body, raw).lower() return any(marker.lower() in msg for marker in _ADMIN_RECORD_LOCK_MARKERS) # NetSuite locks an employee's ``subsidiary`` once the employee carries # posted transactions (expense reports, payroll, journals…), with a 400 # ``USER_ERROR``: # zh: "您不能更改此实体的子公司，原因是该实体有一到多个事务处理。" # en: "You can't change the subsidiary of this entity because it has one # or more transactions." # This is NetSuite protecting financial integrity — for an established # employee the subsidiary (legal entity on the books) is effectively # immutable. Because ``subsidiary`` rides in the same PATCH as the identity # fields, this 400 would otherwise fail the WHOLE update (name / title / # department never apply, and the person fails every sweep). So on this # specific lock the runner drops ``subsidiary`` and retries once: the vendor # keeps its (financially-correct, locked) subsidiary and the rest of the # golden record still syncs. Net effect: subsidiary changes only ever land # for transaction-less employees (genuine new hires / nothing posted yet), # which is exactly NetSuite's own rule. _SUBSIDIARY_LOCKED_MARKERS = ( "更改此实体的子公司", # zh "change the subsidiary", # en ) def _is_subsidiary_locked(body: Any, raw: str) -> bool: """True when NetSuite refuses a subsidiary change (has transactions).""" msg = _vendor_message(body, raw).lower() return any(marker.lower() in msg for marker in _SUBSIDIARY_LOCKED_MARKERS) # NetSuite computes an employee's ENTITY NAME (their login identity) from the # name fields and enforces it UNIQUE among external-access employees, 400ing: # zh: "已有一位享有外部访问权限的员工正在使用此实体名称。所有享有外部访问 # 权限的员工都必须有一个唯一的实体名称作登录之用。" # en: "…entity name … already … external access … unique …" # This fires when Helix pushes ``firstName`` on the UPDATE of an employee who # collides with a same-name twin record (finance keeps some people as multiple # subsidiary records — see the create-skip rationale below). Because firstName # rides in the same PATCH as the rest of the golden record, the collision fails # the WHOLE update every sweep. Like the subsidiary lock: drop the offending # ``firstName`` slot and retry once so the rest of the record syncs, then # BLOCK (not fail) with an actionable note — the duplicate login is a manual # NetSuite cleanup, not something retrying can fix. _ENTITY_NAME_IN_USE_MARKERS = ( "实体名称", # zh: "…正在使用此实体名称" "entity name", # en ) def _is_entity_name_collision(body: Any, raw: str) -> bool: """True when NetSuite rejects a name change as a duplicate entity name (login identity) already used by another external-access employee.""" msg = _vendor_message(body, raw).lower() return any(m.lower() in msg for m in _ENTITY_NAME_IN_USE_MARKERS) def _internal_id_from_location(location: str | None) -> str | None: """Parse the trailing internal id out of a ``Location`` header. NetSuite returns ``Location: https://…/employee/12345`` on a successful upsert. Returns the last path segment when it's a bare integer id, else ``None``. """ if not location: return None tail = location.rstrip("/").rsplit("/", 1)[-1] return tail if tail.isdigit() else None class NetSuiteRunner(SyncRunner): """Push Helix employees into NetSuite via the REST record API. Addresses each person by external id (``employee/eid:{工号}``), so the operation is idempotent without a lookup. **Update-only by default**: the sync PATCHes the existing record (PATCH never creates and MERGES, so omitted fields — name / subsidiary / supervisor / department / ``custentity_swc_*`` — are preserved, never clobbered). A 404 (no record carries that external id) is returned as ``excluded`` ("not adopted"), NOT a create — this is the guard against mass-duplicating the existing PingCAP population, which lives in NetSuite under internal ids with an EMPTY standard externalId (工号 is in ``custentity_swc_work_num``). Opt into creating genuine new hires with ``create_missing=True`` only after the population is adopted (externalId backfilled), so a 404 then reliably means "new hire". Deactivate flips ``isInactive=true`` (离职 = 停用; hard delete not exposed). See DECISIONS.md 2026-06-24. """ # 离职 = 停用. NetSuite hard-delete of an employee with transaction # history is rejected by the vendor anyway, so the offboarding # contract is deactivate-only (same as 汇联易). supports_deactivate = True supports_delete = False # Payload key carrying the external id (HRIS 工号). Pulled OUT of the # body and used in the ``eid:`` URL on upsert. _EXTERNAL_ID_FIELD = "externalId" def __init__( self, *, client: NetSuiteClient, record_type: str = "employee", default_subsidiary_id: str | None = None, create_missing: bool = False, skip_create_work_nums: frozenset[str] | None = None, ): self._client = client # Record type segment of the REST path. Defaults to ``employee``; # kept configurable so a future vendor/customer record sync can # reuse this runner without a new class. self._record_type = (record_type or "employee").strip() or "employee" # OneWorld fallback — injected as ``subsidiary.id`` only when the # rendered payload didn't already carry a subsidiary (i.e. the # company_code value_map missed). Lets a single-person test push # succeed before the full subsidiary map is filled. self._default_subsidiary_id = ( str(default_subsidiary_id).strip() if default_subsidiary_id not in (None, "") else None ) # Update-only safety (default). The sync addresses each person by # ``eid:{工号}`` and UPDATES via PATCH. PATCH never creates — so a # person whose external id is absent in NetSuite (404) is returned # as ``excluded`` ("not adopted"), NOT created. This is the guard # against mass-duplication: the existing PingCAP population lives # in NetSuite under internal ids with an EMPTY standard externalId # (工号 sits in custentity_swc_work_num), so a blind ``PUT eid:`` # would create a parallel duplicate for every un-adopted person. # Set ``create_missing=True`` only AFTER the population is adopted # (externalId backfilled on the existing records), so a 404 then # genuinely means "new hire" and a create is safe. NetSuite's PUT # and PATCH both MERGE (verified 2026-06-24: omitted fields — # name / subsidiary / supervisor / department / custentity_swc_* — # are preserved, not cleared), so an update only ever writes the # non-empty mapped fields and never wipes SWC/Authing-era data. self._create_missing = bool(create_missing) # Per-person CREATE skip-list (工号), mirroring the Feishu runner's # ``skip_provision_person_ids``. Even with ``create_missing`` on, a # 404 for one of these external ids is returned as ``excluded`` # instead of PUT-creating. Protects records that must never be # auto-created: NS Administrator employees (a blind PUT would spawn a # duplicate, non-admin record beside them), a person finance keeps as # multiple subsidiary records, and test / junk-email rows. The # orchestrator resolves person UUIDs → 工号 before constructing the # runner. Update of an already-adopted record is unaffected — this # only gates the create path. self._skip_create_work_nums = frozenset(skip_create_work_nums or ()) def _record_path(self, addressing: str) -> str: return f"{_REST_RECORD_BASE}/{self._record_type}/{addressing}" @staticmethod def _work_num_from_ref(ref: str) -> str | None: """Extract the work number (工号) from an ``eid:{工号}`` external ref. Returns None for a bare internal-id ref (nothing to resolve — that id already 404'd so it's genuinely gone).""" if ref and ref.startswith("eid:"): return unquote(ref[len("eid:"):]).strip() or None return None async def _resolve_internal_id_by_work_num(self, work_num: str) -> str | None: """Look up the NetSuite internal id of the employee whose ``custentity_swc_work_num`` equals ``work_num`` (the offboarding zero-trust re-verify path). Best-effort: SuiteQL needs the SuiteAnalytics permission; any failure returns None so the caller falls back to the idempotent already-gone outcome rather than a hard error that re-fires every sweep. Escapes the value against SuiteQL string-literal injection.""" safe = work_num.replace("'", "''") try: rows = await self._client.suiteql( "SELECT id FROM employee " f"WHERE custentity_swc_work_num = '{safe}'", page_size=5, max_rows=5, ) except Exception: # noqa: BLE001 — re-verify is best-effort logger.info( "netsuite offboard re-verify SuiteQL for work_num=%s failed " "— treating as no live record (idempotent gone)", work_num, ) return None for r in rows: rid = r.get("id") if isinstance(r, dict) else None if rid: return str(rid) return None def describe(self) -> str: return f"NetSuite REST {self._record_type} @ {self._client.host}" async def upsert( self, *, person_email: str, payload: dict[str, Any], prior_external_id: str | None, ) -> PersonOutcome: body = dict(payload) external_id = str(body.pop(self._EXTERNAL_ID_FIELD, "") or "").strip() if not external_id: # externalId is the idempotency key. A person who reaches # here without one (no employee_no / mapping rule misfire) # would otherwise create a duplicate every sweep — fail loud. return PersonOutcome( status="failed", error=( "missing externalId in payload — map a rule with " "target_field 'externalId' from {employee_id}; refusing " "to push without a stable external key." ), ) # The eid: form is how NetSuite addresses this record by external # id (path segment for PUT/PATCH/GET). Reuse it for the upsert path # AND every failure outcome: if a failed external_id is later # persisted and handed to deactivate, a bare employee number would # be misread as an internal id, so it must carry the eid: prefix. # Percent-encode the external id in the path segment so a value # carrying ``/`` ``?`` ``#`` etc. can't break out of the record # path. (工号 is ^P\d+$-gated so this is a no-op for employees, but # it's defense-in-depth and keeps both runners consistent.) The # literal ``eid:`` prefix is NetSuite path syntax and stays raw. eid_ref = f"eid:{quote(external_id, safe='')}" path = self._record_path(eid_ref) # ── UPDATE-ONLY default: PATCH the existing record ────────────── # PATCH never creates (404 if absent), and MERGES (omitted fields # are preserved — name / subsidiary / supervisor / department / # custentity_swc_* stay intact). So a sync only writes the non-empty # mapped fields the golden record owns and never clobbers # SWC/Authing-era data. We do NOT inject a default subsidiary on the # PATCH path: the existing record already has the right subsidiary, # and overwriting it from a half-filled value_map would be the kind # of damage update-only mode exists to prevent. try: status, resp_body, raw, location = await self._client.request( "PATCH", path, json_body=body, ) except NetSuiteError as e: return PersonOutcome( status="failed", external_id=eid_ref, error=self._client._redact(str(e)), detail={"status_code": e.status_code}, ) if 200 <= status < 300: internal_id = _internal_id_from_location(location) stored = internal_id or eid_ref return PersonOutcome( status="success", external_id=stored, detail={"status_code": status, "op": "update"}, ) # NS Administrator-flagged record → excluded by policy, not failed. # The TBA role legitimately can't touch admin records; these IT # admins are maintained manually (DECISIONS.md 2026-06-24). if _is_admin_record_lock(resp_body, raw): return PersonOutcome( status="excluded", external_id=eid_ref, detail={ "status_code": status, "error_code": _error_code(resp_body), "reason": "netsuite_admin_record_locked", "note": ( "NetSuite refuses non-admin TBA role on an " "Administrator record; excluded by policy — " "maintain this employee manually in NetSuite." ), }, ) # NS locks subsidiary once the employee has transactions. Drop the # subsidiary field and retry once so the rest of the golden record # still syncs; the vendor keeps its (locked, financially-correct) # subsidiary. See _is_subsidiary_locked for the full rationale. if _is_subsidiary_locked(resp_body, raw) and "subsidiary" in body: retry_body = {k: v for k, v in body.items() if k != "subsidiary"} try: r_status, r_body, r_raw, r_loc = await self._client.request( "PATCH", path, json_body=retry_body, ) except NetSuiteError as e: return PersonOutcome( status="failed", external_id=eid_ref, error=self._client._redact(str(e)), detail={"status_code": e.status_code}, ) if 200 <= r_status < 300: internal_id = _internal_id_from_location(r_loc) return PersonOutcome( status="success", external_id=internal_id or eid_ref, detail={ "status_code": r_status, "op": "update", "subsidiary_preserved": True, "note": ( "NetSuite rejected the subsidiary change " "(employee has transactions — subsidiary is " "locked); kept the vendor's subsidiary and " "synced the remaining fields." ), }, ) # Retry without subsidiary still failed → report that failure. return PersonOutcome( status="failed", external_id=eid_ref, error=self._client._redact( f"HTTP {r_status} {_vendor_message(r_body, r_raw)}" ), detail={ "status_code": r_status, "error_code": _error_code(r_body), "note": "retry without subsidiary also failed", }, ) # NS entity-name (login identity) collision: Helix's ``firstName`` # would make this employee's computed entity name clash with another # external-access employee's (a same-name twin record). Drop firstName # and retry once so the rest of the golden record still syncs, then # BLOCK — the duplicate login is a manual NetSuite dedup, not something # retrying fixes; ``blocked`` surfaces it as actionable backlog without # reddening the run every sweep. (See _is_entity_name_collision.) if _is_entity_name_collision(resp_body, raw) and "firstName" in body: retry_body = {k: v for k, v in body.items() if k != "firstName"} try: r_status, r_body, r_raw, _r_loc = await self._client.request( "PATCH", path, json_body=retry_body, ) except NetSuiteError as e: return PersonOutcome( status="failed", external_id=eid_ref, error=self._client._redact(str(e)), detail={"status_code": e.status_code}, ) if 200 <= r_status < 300: return PersonOutcome( status="blocked", external_id=eid_ref, error=( "NetSuite rejected the name (entity name / login " "collides with another external-access employee — a " "duplicate/twin record); synced the rest of the record " "without the name. Resolve the duplicate login in " "NetSuite, then the name syncs." ), detail={ "status_code": r_status, "op": "update", "blocked_reason": "netsuite_entity_name_collision", "fields": ["firstName"], }, ) return PersonOutcome( status="failed", external_id=eid_ref, error=self._client._redact( f"HTTP {r_status} {_vendor_message(r_body, r_raw)}" ), detail={"status_code": r_status, "note": "retry without firstName also failed"}, ) # 404 = no NetSuite record carries this external id yet. In the # default update-only mode this is NOT a create — it means the # person hasn't been adopted (their existing NS record still has an # empty externalId) or is a genuine new hire. Either way we refuse # to PUT-create here, because for an un-adopted existing employee a # create would duplicate them. Return ``excluded`` with a clear # next-step. Enable ``create_missing`` only after the population is # adopted, when a 404 reliably means "new hire". if status == 404 and not self._create_missing: return PersonOutcome( status="excluded", external_id=eid_ref, detail={ "status_code": status, "error_code": _error_code(resp_body), "reason": "netsuite_not_adopted", "note": ( "no NetSuite record with this external id — adopt " "the existing employee (backfill externalId on " "their record) or enable create_missing once the " "population is adopted. Refusing to create to avoid " "duplicating an un-adopted existing employee." ), }, ) if status == 404 and self._create_missing: # Per-person create skip-list: never PUT-create these external # ids even with create_missing on (admins would get a duplicate # non-admin record; finance-multi-subsidiary people; test/junk). if external_id in self._skip_create_work_nums: return PersonOutcome( status="excluded", external_id=eid_ref, detail={ "status_code": status, "reason": "netsuite_skip_provision", "note": ( "on the create skip-list " "(runner_options.skip_provision_person_ids) — " "not auto-created; maintain manually if needed." ), }, ) # Opt-in create path for genuine new hires (population already # adopted). PUT eid: = NetSuite create-by-external-id. Email + # subsidiary are required on create, so inject the default # subsidiary when the mapping didn't render one. if not body.get("email"): return PersonOutcome( status="failed", external_id=eid_ref, error=( "missing email — NetSuite employee create requires " "an email; check the mapping rule for 'email'." ), ) # Subsidiary preflight: NetSuite create REQUIRES a subsidiary. If # the company_code value_map didn't render one and no default is # configured, DON'T burn a PUT that will 400 "请输入值:子公司" # every sweep — BLOCK with the exact fix (add the company_code → # subsidiary mapping). Auto-retries once the mapping is added. if not body.get("subsidiary"): if self._default_subsidiary_id: body["subsidiary"] = {"id": self._default_subsidiary_id} else: return PersonOutcome( status="blocked", external_id=eid_ref, error=( "cannot create NetSuite employee: no subsidiary " "resolved for this person's company_code (not in " "the target's subsidiary value_map, and no default " "configured). Add the company_code → subsidiary " "mapping, then it syncs." ), detail={"blocked_reason": "netsuite_no_subsidiary", "fields": ["subsidiary"]}, ) # firstName preflight: NetSuite create REQUIRES a name ("请输入 # 名称"). The rendered payload can miss it two ways: (a) the # person's display_name is genuinely empty, or (b) this person is # Helix-LINKED so the orchestrator rendered the UPDATE shape — # which omits ``create_only`` rules like firstName (#1949) — but # the NetSuite record behind the stored link is GONE (404) and we # are about to CREATE from an update-shaped body. Either way a # PUT would 400 every sweep; BLOCK with the exact fix instead. if not str(body.get("firstName") or "").strip(): return PersonOutcome( status="blocked", external_id=eid_ref, error=( "cannot create NetSuite employee: no firstName in the " "rendered payload. If this person is already linked, " "the stored NetSuite record is gone (404) and the " "UPDATE shape omits create-only name fields — clear " "the stored link so a fresh CREATE renders them (or " "restore the NetSuite record). Otherwise fill the " "person's display_name in HRIS." ), detail={ "blocked_reason": "netsuite_create_missing_first_name", "fields": ["firstName"], }, ) # NetSuite create REQUIRES a non-empty lastName (400 otherwise). # Helix's golden record has a single ``display_name`` (no separate # surname), and firstName carries the full display name (#1949) so # foreign hires get a name. Inject a create-only lastName from the # last whitespace token of the display name (a plausible surname # for western names) with a minimal "." fallback for single-token / # CJK names — create-only so an UPDATE never writes/clobbers it. if not body.get("lastName"): _fn = str(body.get("firstName") or "").strip() _parts = _fn.split() body["lastName"] = _parts[-1] if len(_parts) > 1 else (_fn or ".") try: cs, cb, craw, cloc = await self._client.request( "PUT", path, json_body=body, ) except NetSuiteError as e: return PersonOutcome( status="failed", external_id=eid_ref, error=self._client._redact(str(e)), detail={"status_code": e.status_code}, ) if 200 <= cs < 300: stored = _internal_id_from_location(cloc) or eid_ref return PersonOutcome( status="success", external_id=stored, detail={"status_code": cs, "op": "create", "location": cloc}, ) return PersonOutcome( status="failed", external_id=eid_ref, error=self._client._redact( f"HTTP {cs} {_vendor_message(cb, craw)}" ), detail={"status_code": cs, "error_code": _error_code(cb)}, ) return PersonOutcome( status="failed", external_id=eid_ref, error=self._client._redact( f"HTTP {status} {_vendor_message(resp_body, raw)}" ), detail={ "status_code": status, "error_code": _error_code(resp_body), }, ) async def deactivate( self, *, person_email: str, prior_external_id: str | None, ) -> PersonOutcome: if not prior_external_id: # Never synced to NetSuite → nothing to deactivate. Treat as # a no-op success so the offboarding blocker closes cleanly # (the goal — "inactive on the vendor" — is already met). return PersonOutcome( status="deactivated", external_id=None, detail={"note": "no prior external id; nothing to deactivate"}, ) # ``prior_external_id`` is either a bare internal id or the # ``eid:{工号}`` form — both are valid path segments. path = self._record_path(prior_external_id) try: status, resp_body, raw, _location = await self._client.request( "PATCH", path, json_body={"isInactive": True}, ) except NetSuiteError as e: return PersonOutcome( status="failed", external_id=prior_external_id, error=self._client._redact(str(e)), detail={"status_code": e.status_code}, ) if 200 <= status < 300: return PersonOutcome( status="deactivated", external_id=prior_external_id, detail={"status_code": status}, ) if status == 404: # ZERO-TRUST re-verify (2026-07 offboarding review): a 404 on an # ``eid:{工号}`` ref does NOT prove the leaver is gone — the # majority of NS employee records are Authing-era, keyed by # INTERNAL id with an EMPTY externalId, so ``PATCH eid:P0586`` # 404s while the person's REAL record (under an internal id, work # number in ``custentity_swc_work_num``) stays active. Claiming # "deactivated" off that 404 leaves a departed employee's NS # login live with a green dashboard. So: resolve the work number # to an internal id via SuiteQL and deactivate THAT; only claim # idempotent success when NS genuinely has no record for the 工号. work_num = self._work_num_from_ref(prior_external_id) if work_num: internal_id = await self._resolve_internal_id_by_work_num(work_num) # Re-PATCH only when SuiteQL resolved a record addressed # DIFFERENTLY from the ref we just 404'd on (``prior_external_id``, # the ``eid:{work_num}`` form). ``internal_id`` is a bare numeric # NS id, so this differs from the eid ref — avoids re-hitting the # identical addressing while still deactivating the real record. # (The prior ``f"eid:{work_num}" != internal_id`` compared the eid # ref to the numeric id — always true, a dead guard.) if internal_id and internal_id != prior_external_id: logger.warning( "netsuite deactivate: eid:%s 404'd but work_num " "resolves to live internal id %s — deactivating that " "(unadopted-record zero-trust re-verify)", work_num, internal_id, ) try: st2, _rb2, raw2, _ = await self._client.request( "PATCH", self._record_path(internal_id), json_body={"isInactive": True}, ) except NetSuiteError as e: return PersonOutcome( status="failed", external_id=internal_id, error=self._client._redact(str(e)), ) if 200 <= st2 < 300 or st2 == 404: return PersonOutcome( status="deactivated", external_id=internal_id, detail={"status_code": st2, "note": "deactivated via work_num re-verify"}, ) return PersonOutcome( status="failed", external_id=internal_id, error=f"PATCH {internal_id} (deactivate) → {st2}: {raw2[:200]}", ) # No work_num resolvable, or NS has no record for it → truly # absent → idempotent success. return PersonOutcome( status="deactivated", external_id=prior_external_id, detail={"status_code": status, "note": "already absent (404)"}, ) # NS Administrator record → the non-admin TBA role can't deactivate # it either. Mirror ``upsert``: excluded by policy (maintain the IT # admin manually), NOT a hard failure — otherwise an admin leaver's # deactivate would redden the sweep and bump retry_count every run # forever (the lock is permanent for this role). if _is_admin_record_lock(resp_body, raw): return PersonOutcome( status="excluded", external_id=prior_external_id, detail={ "status_code": status, "error_code": _error_code(resp_body), "reason": "netsuite_admin_record_locked", "note": ( "NetSuite refuses non-admin TBA role on an " "Administrator record; deactivate this IT admin " "manually in NetSuite." ), }, ) return PersonOutcome( status="failed", external_id=prior_external_id, error=self._client._redact( f"HTTP {status} {_vendor_message(resp_body, raw)}" ), detail={ "status_code": status, "error_code": _error_code(resp_body), }, ) async def aclose(self) -> None: await self._client.aclose() # NetSuite REST sets a *reference* field (the dept's ``parent``, which # points at another department record) by an object carrying either the # internal id (``{"id": "123"}``) or the external id # (``{"externalId": "DEPT_ROOT"}``). Helix keys every dept by its HRIS # org-node code (= ``externalId``) and pushes parents before children in # topological order, so we address the parent by ``externalId`` — no # internal-id round-trip needed and the parent we reference was just # upserted under that same external id. CAVEAT (verify against the live # API before enabling — the dept target is gated off by default): if a # given NetSuite account/version rejects the ``{"externalId": ...}`` form # for a reference field and demands the internal id, the fallback is to # resolve the parent's internal id first (from the prior upsert's stored # external_id when it's a bare integer, else a GET ``department/eid:{code}``) # and send ``{"id": ""}`` instead. We implement the externalId # form here because it keeps the runner lookup-free and matches NetSuite's # documented reference-by-externalId behaviour. class NetSuiteDepartmentRunner(DeptSyncRunner): """Phase A — push Helix's HRIS org tree into NetSuite departments. Mirrors :class:`NetSuiteRunner` (same client, same eid: addressing, same redaction + error-shape helpers) but targets the ``department`` record type instead of ``employee``. Idempotency key = the HRIS org-node code, set as the NetSuite ``externalId`` and addressed by ``department/eid:{dept_code}`` (PUT = upsert). Out-of-scope depts are soft-deleted (``isinactive=true``); hard delete is not exposed — NetSuite departments carry historical references and the offboarding contract is deactivate-only (same as the employee runner / 汇联易). Departments are NetSuite-managed reference data today (Authing-era); this runner is the long-term cutover that makes Helix the master. It adopts the EXISTING vendor records in place by their External ID, so no parallel SF-coded tree is created — see ``docs/architecture/DECISIONS.md``. """ # 失效 = 停用. NetSuite departments with posted transaction history # cannot be hard-deleted; soft-delete (isinactive) is the contract. supports_soft_delete = True # Payload key carrying the external id (HRIS dept code). Reused as the # ``eid:`` path segment, matching ``NetSuiteRunner._EXTERNAL_ID_FIELD``. _EXTERNAL_ID_FIELD = "externalId" def __init__( self, *, client: NetSuiteClient, record_type: str = "department", create_missing: bool = False, ): self._client = client self._record_type = ( (record_type or "department").strip() or "department" ) # Update-only safety, identical to the employee runner: existing # NetSuite departments live under internal ids with an empty # externalId (SWC/Authing-era reference data), so a blind # ``PUT eid:{code}`` would create a PARALLEL duplicate tree. Default # OFF → PATCH-update only; a 404 returns ``skipped`` ("not adopted") # rather than creating. Enable ``create_missing=True`` only after # the NS department tree is adopted (externalId backfilled on the # existing dept records). PATCH/PUT both merge (omitted fields # preserved), so an update never wipes a dept's other attributes. self._create_missing = bool(create_missing) def _record_path(self, addressing: str) -> str: return f"{_REST_RECORD_BASE}/{self._record_type}/{addressing}" def describe(self) -> str: return f"NetSuite REST {self._record_type} @ {self._client.host}" async def upsert_dept( self, *, dept_code: str, dept_name: str, parent_dept_code: str | None, leader_email: str | None, dept_order: int | None, prior_external_id: str | None, ) -> DeptOutcome: dept_code = (dept_code or "").strip() if not dept_code: return DeptOutcome( status="failed", external_id=prior_external_id, error=( "missing dept_code — refusing to push a NetSuite " "department without a stable external key." ), ) # Percent-encode the dept code in the path (org-node codes are # free-form HRIS strings, NOT ^P\d+$-gated) so a code with ``/`` / # ``?`` can't escape the record path. ``eid:`` prefix stays raw; the # body ``externalId`` is a JSON value (httpx-encoded) so it keeps # the raw code. eid_ref = f"eid:{quote(dept_code, safe='')}" # The UPDATE body OMITS ``name`` → PATCH-merge preserves the vendor's # existing dept name. SWC writes names as ``"{编码} {名称}"`` and Helix # must NOT strip that code prefix on adopted depts (decided # 2026-06-25); a NEW dept (the create path below) DOES get the clean # ``name``. ``isInactive`` is camelCase to match the REST record body # schema — the SuiteQL column is lowercase ``isinactive``, but the # record API field is ``isInactive`` (same as the employee runner); # the wrong case is silently ignored by NetSuite, which would make # soft-delete a no-op. body: dict[str, Any] = { "isInactive": False, self._EXTERNAL_ID_FIELD: dept_code, } # Reference field: address the parent by its External ID (the # parent was pushed just before this child in topological order). if parent_dept_code: body["parent"] = {"externalId": parent_dept_code} # UPDATE-ONLY default: PATCH the existing dept (merge; never create). path = self._record_path(eid_ref) try: status, resp_body, raw, location = await self._client.request( "PATCH", path, json_body=body, ) except NetSuiteError as e: return DeptOutcome( status="failed", external_id=eid_ref, error=self._client._redact(str(e)), detail={"status_code": e.status_code}, ) if 200 <= status < 300: internal_id = _internal_id_from_location(location) stored = internal_id or eid_ref return DeptOutcome( status="success", external_id=stored, detail={"status_code": status, "op": "update"}, ) # 404 = no dept carries this external id yet. Update-only mode does # NOT create (would build a parallel tree beside the un-adopted # Authing-era depts) — return ``skipped`` ("not adopted"). Enable # ``create_missing`` only after the NS dept tree is adopted. if status == 404 and not self._create_missing: return DeptOutcome( status="skipped", external_id=eid_ref, detail={ "status_code": status, "reason": "netsuite_dept_not_adopted", "note": ( "no NetSuite department with this external id — " "adopt the existing dept (backfill externalId) or " "enable create_missing once the tree is adopted. " "Refusing to create to avoid a parallel dept tree." ), }, ) if status == 404 and self._create_missing: # Opt-in create path (tree already adopted). PUT eid: = create. # A NEW dept DOES get the clean name (the update path omits it # to preserve the vendor's code-prefixed name). try: cs, cb, craw, cloc = await self._client.request( "PUT", path, json_body={**body, "name": dept_name or dept_code}, ) except NetSuiteError as e: return DeptOutcome( status="failed", external_id=eid_ref, error=self._client._redact(str(e)), detail={"status_code": e.status_code}, ) if 200 <= cs < 300: stored = _internal_id_from_location(cloc) or eid_ref return DeptOutcome( status="success", external_id=stored, detail={"status_code": cs, "op": "create", "location": cloc}, ) return DeptOutcome( status="failed", external_id=eid_ref, error=self._client._redact( f"HTTP {cs} {_vendor_message(cb, craw)}" ), detail={"status_code": cs, "error_code": _error_code(cb)}, ) return DeptOutcome( status="failed", external_id=eid_ref, error=self._client._redact( f"HTTP {status} {_vendor_message(resp_body, raw)}" ), detail={ "status_code": status, "error_code": _error_code(resp_body), }, ) async def soft_delete( self, *, dept_code: str, prior_external_id: str | None, ) -> DeptOutcome: # Address by the stored external id (bare internal id or eid: form # — already path-safe as stored) when we have one; otherwise fall # back to the eid: form derived from the dept code, percent-encoding # the (free-form) code so it can't break out of the path. addressing = prior_external_id or f"eid:{quote(dept_code, safe='')}" path = self._record_path(addressing) try: status, resp_body, raw, _location = await self._client.request( "PATCH", path, json_body={"isInactive": True}, ) except NetSuiteError as e: return DeptOutcome( status="failed", external_id=prior_external_id, error=self._client._redact(str(e)), detail={"status_code": e.status_code}, ) if 200 <= status < 300: return DeptOutcome( status="soft_deleted", external_id=prior_external_id, detail={"status_code": status}, ) if status == 404: # Already gone on the vendor side → idempotent success. return DeptOutcome( status="soft_deleted", external_id=prior_external_id, detail={"status_code": status, "note": "already absent (404)"}, ) return DeptOutcome( status="failed", external_id=prior_external_id, error=self._client._redact( f"HTTP {status} {_vendor_message(resp_body, raw)}" ), detail={ "status_code": status, "error_code": _error_code(resp_body), }, ) async def aclose(self) -> None: await self._client.aclose()