"""汇联易 (Huilianyi) master-data runners — outbound identity sync. Pushes Helix's HRIS-of-truth employee + org-tree data into 汇联易's open API (``/api/open/...``), replacing the legacy 汇联易→Authing daily full-pull. Backs the single ``huilianyi`` sync-target kind — internally a two-phase sweep mirroring ``feishu_contact`` (Phase A reconciles departments, Phase B upserts users) but admin-facing as ONE target, ONE switch, ONE cron. Key vendor facts (verified against opendocs.huilianyi.com, 2026-06-10): * ``employeeID`` (工号, P01431-style) is the tenant-unique employee key — every employee endpoint addresses by it. ``email`` is ALSO tenant-unique, so a stale email on the vendor side makes an update collide; the reconciliation export must be checked before cutover. * Departments are keyed by ``custDeptNumber`` (部门编码) — unlike Feishu there is NO vendor-assigned opaque id to adopt; Helix's HRIS dept code IS the vendor key, so the dept phase needs no name-adoption catalog. * 离职 = 停用 (disable), never delete: ``DELETE /api/open/user/delete/v2/{employeeID}/``. A separate 恢复 endpoint exists but is intentionally NOT automated — restoring a resigned account is an HR decision, surfaced as a failed outcome with instructions instead. * Mutation responses carry ``{"errorCode": "0000", "message": ...}`` on success; error codes are documented per endpoint (120207 = 员工 不存在, 120107 = 部门编码已存在, 120110 = 部门编码不存在, …). * AccessToken (verified against the live tenant's 开放接口权限 catalog + opendocs implement guide, 2026-06-10): ``POST {gateway}/oauth/token`` with ``Authorization: Basic base64(权限ID:权限密钥)`` and form body ``grant_type=client_credentials&scope=write``; response carries top-level ``access_token`` + ``expires_in`` (~86400s). Credentials are minted per-permission in 汇联易后台 → 系统对接 → 开放接口权限, which also gates every endpoint individually — the Helix credential needs 登录 + 员工 + 部门 module endpoints enabled. Wire layout: * :class:`HuilianyiClient` — token cache + authed HTTP with bounded 429/5xx retries (mirrors ``ScimRunner`` conventions). * :class:`HuilianyiRunner` — Phase B (users): lookup-first by employeeID → PUT update / POST create; deactivate = 离职 V2. * :class:`HuilianyiDepartmentRunner` — Phase A (departments): POST/PUT by ``custDeptNumber``; soft_delete = 失效部门 V2. """ from __future__ import annotations import asyncio import base64 import time from typing import Any import httpx from app.core.safe_logging import get_safe_logger from app.domains.identity_sync.runners.base import ( DeptOutcome, DeptSyncRunner, PersonOutcome, SyncRunner, ) logger = get_safe_logger(__name__) # ── Vendor error codes (opendocs.huilianyi.com, per-endpoint docs) ─── # 汇联易 部门角色 (department role) positionCodes for PingCAP's tenant, # read live from a real dept's departmentPositionDTOList (2026-06-16): # 6103=部门负责人, 6105=HRBP, 6102=Budget Owner. Pushed via the # create/update-V2 ``departmentPositionList`` ({positionCode, employeeId}). # NOTE: 部门负责人 is set via the 6103 ROLE here, NOT the top-level # ``managerEmployeeId`` field — verified that real depts carry the leader # in departmentPositionList (managerEmployeeNumber was null), so the role is # the contract 汇联易's UI/Authing actually used. DEPT_ROLE_LEADER = "6103" # 部门负责人 DEPT_ROLE_HRBP = "6105" # HRBP DEPT_ROLE_BUDGET_OWNER = "6102" # Budget Owner OK_CODE = "0000" USER_NOT_FOUND = "120207" # 员工不存在 # 主岗直属领导工号 on 创建员工V2 / 更新员工V2. The vendor's own spelling # carries a typo ("Empoyee") — verified verbatim in the live opendocs field # tables 2026-08-27. The correctly spelled ``directManagerEmployeeId`` is a # DIFFERENT field that only exists inside ``userJobsDtos`` (per-position # leader); at top level the vendor would ignore it silently, which is exactly # how 入职时间 stayed blank for months (entryDate/entryTime). Never "correct" # this constant. _MANAGER_FIELD = "directManagerEmpoyeeId" # 手机号 slot: the number and its country travel together (the country # decides validation, and a country without a number is meaningless). _MOBILE_FIELD = "mobile" _PHONE_COUNTRY_FIELD = "countryCode" _PHONE_FIELDS = (_MOBILE_FIELD, _PHONE_COUNTRY_FIELD) # Vendor-DOCUMENTED phone rejections (创建/更新员工V2 错误码表, 2026-08-27). # Unlike 直属领导 — for which the vendor documents NO code, hence the blind # probe below — these are known, so the phone slot degrades on the CODE and # never on a guessed message marker: # 120206 手机号不合法 (format failed CN validation) # 120211 手机号已被占用 (tenant-unique collision) # 12002008 手机号已被占用,该手机号{0}已被工号为{1}的人员绑定 # (same collision, and the message NAMES the holder's 工号 — # keep the vendor text on the ledger row, it is the fix) # 120223 国家编码验证规则不存在 (countryCode outside the vendor's table) _PHONE_REJECT_CODES = ("120206", "120211", "12002008", "120223") DEPT_ALREADY_EXISTS = ("120101", "120107") # 部门已存在 / 部门编码已存在 # Dept-level "not found" — the vendor uses TWO distinct codes and they # are NOT interchangeable (2026-06-16 cutover incident): a department # PUT/DELETE against a missing custDeptNumber returns 120103 (部门不存在), # while 120110 (部门编码不存在) is the USER-create error when a person's # custDeptNumber is unknown. The dept runner only ever sees 120103; 120110 # was mis-set here and silently broke the PUT→POST recreate fallback (a # vanished dept was logged as a hard failure instead of being recreated). # Accept both defensively so neither path regresses. DEPT_NOT_FOUND = ("120103", "120110") # 部门不存在 / 部门编码不存在 # 失效部门 is rejected by THREE distinct vendor codes and all mean "this # dept still has dependents, disable it later" — surface them as the same # actionable human task (never cascade). 120102 = 存在子部门或员工; 120125 # = 部门存在启用的子部门 (a still-enabled child dept, verified live 2026-07-05 # on custDeptNumber 20250304); 120124 = 部门或者子部门存在员工,无法禁用 (the # dept — or a descendant — still has in-post employees). Each was in turn # UNMAPPED and fell through to the generic error path, so the sweep reported # a bare "errorCode=1201xx" with no "go clean up the vendor child/员工" # instruction and re-failed loudly every night. 120124 was captured live in # the #1954 ledger x9 across 2026-06/07 (e.g. custDeptNumbers 2-1-19, # 20240501-1) — see the dry-run/ledger provenance note in the module tests. DEPT_HAS_CHILDREN = ("120102", "120124", "120125") # 子部门/员工未清,无法失效 # 汇联易 employee ``status`` values (增量查询员工 doc). _STATUS_RESIGNED = "1003" # Substrings that mean "this create collided with an existing row" — # used as a POST→PUT fallback signal when the lookup raced a parallel # writer (e.g. HR creating the employee by hand mid-sweep). Matched # against the vendor ``message`` because the duplicate-employee error # code isn't documented; codes that ARE documented take priority. _DUPLICATE_MARKERS = ("已存在", "already exist", "duplicate") # Substrings that mean "this row is already in 离职 state" — used to # make 离职V2 idempotent when HR already resigned the person by hand # in the 汇联易 console (today's manual process keeps running through # the cutover parallel window). The goal of deactivate is "inactive on # the vendor"; already-resigned achieves it. # # Message-marker based (not code based) DELIBERATELY: a read of the full # #1954 ledger (90-day retention, 2026-07) shows 离职V2 is idempotent by # return — 35 clean deactivations, ZERO vendor rejections; the only 离职 # failures on record are the "no employeeID recorded" guard below, never a # business errorCode. So 汇联易 either returns 0000 on an already-resigned # row or answers with a 已离职 message the markers already catch. There is # no distinct already-resigned error code to whitelist (unlike the dept # DEPT_HAS_CHILDREN codes, which the ledger DID surface). If a future sweep # ever lands a bare errorCode on a 离职 attempt, capture it from the ledger # and add it here — do NOT guess one blind. _ALREADY_RESIGNED_MARKERS = ("已离职", "已经离职", "已停用", "resigned") class HuilianyiError(RuntimeError): """Wraps an HTTP/transport failure from the 汇联易 gateway.""" 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 _error_code(body: Any) -> str | None: if isinstance(body, dict): code = body.get("errorCode") if code is not None: return str(code) return None def _is_ok(status: int, body: Any) -> bool: """Vendor success = HTTP 2xx AND errorCode 0000 (when present). Some read endpoints return bare data without an errorCode wrapper; those count as OK on HTTP status alone. """ if status not in (200, 201, 204): return False code = _error_code(body) return code is None or code == OK_CODE def _vendor_message(body: Any, raw: str) -> str: if isinstance(body, dict): msg = body.get("message") if msg: return str(msg) return (raw or "")[:200] def redact_credentials(text: str, *secrets: str | None) -> str: """Strip known credential material from vendor-echoed text before it can reach an error message / API response / log line. Opaque 汇联易 tokens don't match the platform's generic JWT/base64 scrub patterns, so explicit replacement of the values we hold is the only reliable redaction.""" out = text or "" for sec in secrets: if sec and len(sec) >= 6: out = out.replace(sec, "***") return out def _looks_duplicate(body: Any, raw: str) -> bool: haystack = _vendor_message(body, raw).lower() return any(m in haystack for m in _DUPLICATE_MARKERS) def _looks_already_resigned(body: Any, raw: str) -> bool: haystack = _vendor_message(body, raw).lower() return any(m in haystack for m in _ALREADY_RESIGNED_MARKERS) class HuilianyiClient: """Token-cached HTTP client for the 汇联易 open API. The access token is cached for the client's lifetime (one sweep) and refreshed 60s before expiry or on the first 401. Defaults match the verified contract (``POST {gateway}/oauth/token``, Basic 权限ID:权限密钥, ``grant_type=client_credentials&scope=write``); ``token_path`` / ``token_auth_style`` stay configurable for private-deploy tenants whose gateway differs. """ def __init__( self, *, base_url: str, client_id: str, client_secret: str, token_path: str = "/oauth/token", token_auth_style: str = "basic", verify_ssl: bool = True, ca_bundle_path: str | None = None, timeout_seconds: int = 30, rate_limit_per_minute: int = 60, transport: httpx.AsyncBaseTransport | None = None, ): if not base_url: raise ValueError("huilianyi client requires base_url") # SSRF guard lives at the ASYNC boundaries that construct this # client (driver.test_connection + service._build_huilianyi_client) # via assert_safe_external_url_async — the sync variant does a # blocking DNS resolve, which must not run inside __init__ on # the event loop. if not client_id or not client_secret: raise ValueError("huilianyi client requires client_id + client_secret") # Config fields are free text on the integration form — normalise # before validating so "Basic" / "form " don't hard-fail a sweep. token_auth_style = (token_auth_style or "basic").strip().lower() token_path = (token_path or "/oauth/token").strip() if token_auth_style not in ("form", "basic"): raise ValueError( f"unsupported token_auth_style {token_auth_style!r}; " "use 'form' or 'basic'" ) self._base_url = base_url.rstrip("/") self._client_id = client_id self._client_secret = client_secret self._token_path = "/" + token_path.strip("/") self._token_auth_style = token_auth_style # trust_env=False / follow_redirects=False — same hardening # rationale as ScimRunner: no ambient proxy can siphon the # token, 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, transport=transport, ) self._min_interval = 60.0 / max(1, rate_limit_per_minute) self._rl_lock = asyncio.Lock() self._last_call_at: float = 0.0 self._token: str | None = None self._token_expires_at: float = 0.0 # Circuit breaker — counts CLIENT-LEVEL failures (token mint # failures + requests that exhausted their retry budget). A dead # gateway / revoked 权限密钥 otherwise costs every person in a # sweep ~minutes of timeouts each (the per-person rate-limit # wait + 3 retried 30s attempts). Once open, every further call # fails in microseconds; per-person failures still land on # state rows and self-heal on the next sweep after the # credential/gateway is fixed (fresh client per sweep). self._consecutive_failures = 0 self._breaker_threshold = 5 def _redact(self, text: str) -> str: """Strip our credential material from vendor-echoed text before it lands in an error message / ledger row — the generic scrub can't match opaque 汇联易 token shapes.""" return redact_credentials(text, self._client_secret, self._token) def _check_breaker(self) -> None: if self._consecutive_failures >= self._breaker_threshold: raise HuilianyiError( f"circuit open: {self._consecutive_failures} consecutive " "client-level failures this sweep — gateway down or " "credential revoked; fix and re-run." ) # ── Token ──────────────────────────────────────────────────────── async def get_access_token(self, *, force_refresh: bool = False) -> str: self._check_breaker() if ( not force_refresh and self._token and time.monotonic() < self._token_expires_at - 60 ): return self._token url = f"{self._base_url}{self._token_path}" data = {"grant_type": "client_credentials", "scope": "write"} headers: dict[str, str] = {} if self._token_auth_style == "basic": creds = f"{self._client_id}:{self._client_secret}".encode() headers["Authorization"] = ( f"Basic {base64.b64encode(creds).decode('ascii')}" ) else: data["client_id"] = self._client_id data["client_secret"] = self._client_secret try: resp = await self._client.post(url, data=data, headers=headers) except httpx.HTTPError as e: self._consecutive_failures += 1 raise HuilianyiError( f"token endpoint transport error: {type(e).__name__}: {e!s:.200}" ) from e body: dict[str, Any] | None = None try: body = resp.json() except ValueError: body = None if resp.status_code != 200 or not isinstance(body, dict): self._consecutive_failures += 1 raise HuilianyiError( self._redact( f"token endpoint {self._token_path} → " f"{resp.status_code}: {resp.text[:300]}" ), status_code=resp.status_code, body_excerpt=self._redact(resp.text[:500]), ) # Common shapes: top-level OAuth body, or wrapped under "data". token = body.get("access_token") expires_in = body.get("expires_in") if not token and isinstance(body.get("data"), dict): token = body["data"].get("access_token") expires_in = body["data"].get("expires_in") if not token: self._consecutive_failures += 1 raise HuilianyiError( self._redact( "token endpoint returned no access_token: " f"{resp.text[:300]}" ), status_code=resp.status_code, body_excerpt=self._redact(resp.text[:500]), ) try: ttl = float(expires_in) if expires_in is not None else 3600.0 except (TypeError, ValueError): ttl = 3600.0 self._token = str(token) self._token_expires_at = time.monotonic() + ttl self._consecutive_failures = 0 return self._token # ── Authed request with bounded retries ────────────────────────── 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() async def request( self, method: str, path: str, *, json_body: dict[str, Any] | None = None, params: dict[str, Any] | None = None, retry_on_429: int = 2, ) -> tuple[int, dict[str, Any] | list[Any] | None, str]: """Returns (status_code, body_json_or_None, raw_text). Retries 429/5xx with backoff; refreshes the token once on 401. Raises :class:`HuilianyiError` on transport failure or after the retry budget is exhausted on 429/5xx. """ self._check_breaker() url = f"{self._base_url}/{path.lstrip('/')}" attempt = 0 refreshed_token = False while True: await self._rate_limit_wait() token = await self.get_access_token() try: resp = await self._client.request( method, url, headers={ "Authorization": f"Bearer {token}", "Accept": "application/json", }, json=json_body, params=params, ) except httpx.HTTPError as e: # Non-idempotent methods (POST create / update) must NOT be # blind-replayed on a transport error that could have landed # server-side (read timeout / write error / protocol error / # pool timeout) — a replay would DUPLICATE the create (汇联易 # create is not idempotent: employeeID + email are tenant- # unique). Only CONNECT-phase errors (request never reached # the server) are safe to retry for a mutation; anything else # surfaces immediately so the caller's lookup-first path # (GET detail → update) converges on the next sweep without a # duplicate. GET (idempotent) retries any transport error. connect_phase = isinstance( e, (httpx.ConnectError, httpx.ConnectTimeout), ) retryable = method.upper() == "GET" or connect_phase attempt += 1 if attempt > retry_on_429 or not retryable: self._consecutive_failures += 1 raise HuilianyiError( f"transport error: {type(e).__name__}: {e!s:.200}" ) from e await asyncio.sleep(2 ** attempt) continue text = resp.text body: dict[str, Any] | list[Any] | None = None if text: try: parsed = resp.json() # Mutations answer {"errorCode": ...}; the 增量查询 # read endpoints answer a BARE ARRAY — keep both. body = parsed if isinstance(parsed, (dict, list)) else None except ValueError: body = None if resp.status_code == 401 and not refreshed_token: # Token expired server-side before our local clock said # so — refresh once and replay. refreshed_token = True await self.get_access_token(force_refresh=True) continue if resp.status_code in (429, 500, 502, 503, 504): attempt += 1 if attempt > retry_on_429: self._consecutive_failures += 1 raise HuilianyiError( self._redact( f"{method} {path} → {resp.status_code} after " f"{retry_on_429 + 1} attempts: {text[:300]}" ), status_code=resp.status_code, body_excerpt=self._redact(text[:500]), ) ra = resp.headers.get("Retry-After") if ra and ra.isdigit(): await asyncio.sleep(min(int(ra), 30)) else: await asyncio.sleep(2 ** attempt) continue # A completed HTTP exchange (any status) means the gateway # and credential are alive — reset the breaker. self._consecutive_failures = 0 return resp.status_code, body, text async def aclose(self) -> None: await self._client.aclose() # ── Read-only full-roster fetch (dry-run / reconciliation) ────────── # 增量查询员工/部门 take a REQUIRED date window + 1-based pages of ≤100. # Verified against the live tenant (2026-06-11): page=0 and page=1 BOTH # return the first page, page=2 is the second page — i.e. the server is # 1-indexed and clamps 0 up to 1. Iterating from 0 made page 1 look # like a full page of duplicates, tripping the fresh-row guard and # silently truncating every roster to exactly _PAGE_SIZE rows. Always # start at page=1. Rows are still deduped by their natural key as a # belt-and-braces guard against servers that ignore the page param. # "Full pull" = a window from 2000-01-01 to tomorrow. _FULL_WINDOW_START = "2000-01-01 00:00:00" _PAGE_SIZE = 100 _MAX_PAGES = 500 # 50k rows — far above any real tenant; runaway guard def _full_window_end() -> str: from datetime import datetime, timedelta return (datetime.now() + timedelta(days=1)).strftime("%Y-%m-%d %H:%M:%S") async def _fetch_all_pages( client: HuilianyiClient, path: str, *, base_params: dict[str, str], dedup_key: str, ) -> list[dict[str, Any]]: """Page through a bare-array read endpoint until a short page. Raises :class:`HuilianyiError` on a non-2xx page or an errorCode- wrapped body — a PARTIAL roster is worse than no roster for a reconciliation plan (missing rows read as "safe to create"). """ out: list[dict[str, Any]] = [] seen: set[str] = set() # 1-based: the server treats page=0 as an alias of page=1 (verified # live) — starting at 0 double-reads the first page and the # fresh-row guard then misreads the duplicate as end-of-data. for page in range(1, _MAX_PAGES + 1): params = dict(base_params) params["page"] = str(page) params["size"] = str(_PAGE_SIZE) status, body, raw = await client.request("GET", path, params=params) # STRICT list contract: anything else — an errorCode wrapper, # an HTML login page from a misrouted gateway (parses to # None), an empty 200 body — raises instead of reading as an # empty roster. An empty roster tells the reconciliation # "everything is safe to create", which is the most dangerous # possible misreading. if status != 200 or not isinstance(body, list): raise HuilianyiError( f"GET {path} page={page} → {status} (non-list body) " f"errorCode={_error_code(body)}: {_vendor_message(body, raw)}", status_code=status, body_excerpt=raw[:500], ) rows = body fresh = 0 for r in rows: if not isinstance(r, dict): continue key = str(r.get(dedup_key) or "") if not key or key in seen: continue seen.add(key) out.append(r) fresh += 1 if len(rows) < _PAGE_SIZE: break if fresh == 0: # A full page of nothing-new = a server that ignores the # page param (we start at 1, so the 0/1-alias quirk can no # longer produce this); stop rather than spin. break else: # Loop exhausted _MAX_PAGES of FULL pages — the roster is # bigger than the runaway guard. A silently truncated roster # has the same missing-rows-read-as-safe-to-create hazard as # an empty one: refuse. raise HuilianyiError( f"GET {path}: roster exceeds {_MAX_PAGES * _PAGE_SIZE} rows " "(_MAX_PAGES guard) — refusing a truncated reconciliation; " "raise _MAX_PAGES if the tenant is genuinely this large." ) return out async def fetch_all_employees( client: HuilianyiClient, *, include_leavers: bool = True, ) -> list[dict[str, Any]]: """Full vendor employee roster (工号-keyed dedup).""" return await _fetch_all_pages( client, "/api/open/user", base_params={ "startDate": _FULL_WINDOW_START, "endDate": _full_window_end(), "containLeavedUser": "true" if include_leavers else "false", }, dedup_key="employeeID", ) async def fetch_all_departments( client: HuilianyiClient, ) -> list[dict[str, Any]]: """Full vendor dept list (custDeptNumber-keyed dedup). Vendor dept ``status``: 101 启用 / 102 禁用 / 103 删除. """ return await _fetch_all_pages( client, "/api/open/department", base_params={ "startDate": _FULL_WINDOW_START, "endDate": _full_window_end(), }, dedup_key="custDeptNumber", ) class HuilianyiRunner(SyncRunner): """Phase B — employee upsert / 离职 against the 汇联易 open API. The rendered ``payload`` is the vendor body verbatim (flat keys: ``employeeID`` / ``fullName`` / ``email`` / ``custDeptNumber`` / ``companyCode`` / …) — the mapping engine owns field placement. ``external_id`` convention: the vendor-unique ``employeeID`` (工号). Every endpoint addresses by it, so storing it makes deactivate a single direct call with no lookup; the vendor's internal userOID is deliberately NOT stored (it appears in no mutation path). """ # 离职恢复 exists on the vendor side but is an HR decision — # surface, don't automate. Hard delete doesn't exist at all. supports_delete = False def __init__( self, *, client: HuilianyiClient, dept_code_map: dict[str, str] | None = None, ): self._client = client # In-place recode map (Helix SF dept code → 汇联易 custDeptNumber) — # same map the dept runner uses, so a user lands on the EXISTING # vendor dept rather than the SF-coded parallel. Empty = passthrough. self._code_map = dept_code_map or {} async def upsert( self, *, person_email: str, payload: dict[str, Any], prior_external_id: str | None, ) -> PersonOutcome: employee_id = str(payload.get("employeeID") or "").strip() if not employee_id: return PersonOutcome( status="failed", error=( f"rendered payload for {person_email} has no " "employeeID — check the mapping rule that targets " "the employeeID field." ), ) # In-place recode: translate the rendered SF dept code to the # existing 汇联易 custDeptNumber so the user attaches to the legacy # dept (no SF-coded parallel). No-op when the map is empty or the # code isn't mapped (new dept → keeps its SF code). if self._code_map: raw_dept = payload.get("custDeptNumber") mapped = self._code_map.get(str(raw_dept)) if raw_dept else None if mapped and mapped != raw_dept: payload = {**payload, "custDeptNumber": mapped} # ``entryTime`` (入职时间) — the WRITE field name on both 创建员工V2 # and 更新员工V2, documented as 格式:yyyy-MM-dd. (The 增量查询员工 READ # endpoint answers with ``entryDate`` instead; mapping to the read # name is what kept 入职时间 blank on every Helix-pushed employee # until 2026-08-27 — the vendor ignores unknown keys silently.) # The mapping engine renders {hire_date} as "YYYY-MM-DD", which is # already the documented shape; anything unparseable is dropped # rather than failing the whole person over one optional attribute. entry = payload.get("entryTime") if isinstance(entry, str) and entry: from datetime import datetime as _dt ds = entry.strip() try: _dt.strptime(ds[:10], "%Y-%m-%d") parseable = True except ValueError: parseable = False if parseable: # Normalise to the documented date-only shape (a padded # "… 00:00:00" tail is accepted by the vendor's DateTime # parser but the docs' own samples are date-only). payload = {**payload, "entryTime": ds[:10]} else: # A non-ISO value (value_map typo, locale-formatted # date) would make the vendor reject the WHOLE person # — drop the one optional attribute instead. payload = { k: v for k, v in payload.items() if k != "entryTime" } logger.info( "huilianyi: unparseable entryTime %r dropped for %s", entry, employee_id, ) # ── 手机号 pre-wire guards ──────────────────────────────────── # The vendor's ``mobile`` must be a bare national number: no # dialing prefix, digits only. HRIS is clean today (prod: 411 # numbers, 0 carrying a prefix) but a single bad value must cost # ONE field, never the person — and must never be "fixed" by # guessing, because 手机号 is tenant-unique: welding a dialing code # onto a national number can land on a DIFFERENT real employee. mob = str(payload.get(_MOBILE_FIELD) or "").strip() if mob: drop_reason = None if mob.startswith("+") or mob.startswith("00"): # ``phone_digits`` deliberately preserves the prefix so we # can refuse here instead of silently splitting it. drop_reason = "carries an international dialing prefix" elif not mob.isdigit(): drop_reason = "not digits-only after normalisation" if drop_reason: payload = { k: v for k, v in payload.items() if k not in _PHONE_FIELDS } logger.info( "huilianyi: %s mobile dropped (%s)", employee_id, drop_reason, ) mob = "" cc = str(payload.get(_PHONE_COUNTRY_FIELD) or "").strip() if cc and (len(cc) != 2 or not cc.isalpha()): # An unresolvable/odd country code would 120223 the whole # person; the number alone still lands (the vendor then treats # it as 中国大陆 and skips validation). payload = { k: v for k, v in payload.items() if k != _PHONE_COUNTRY_FIELD } logger.info( "huilianyi: %s countryCode %r dropped (not ISO-2)", employee_id, cc, ) elif cc and not mob: # A country with no number is meaningless and can only earn a # validation error. payload = { k: v for k, v in payload.items() if k != _PHONE_COUNTRY_FIELD } # 主岗直属领导 self-reference guard. A person can't be their own # 直属领导 — the org head's manager pointer is either NULL (field # already absent) or, in bad HRIS data, themselves. Sending it would # either be rejected or create a self-loop in 汇联易's approval # routing. Drop the one attribute; the rest of the person still syncs. if str(payload.get(_MANAGER_FIELD) or "").strip() == employee_id: payload = { k: v for k, v in payload.items() if k != _MANAGER_FIELD } logger.info( "huilianyi: %s is its own 直属领导 in the rendered payload; " "dropped %s", employee_id, _MANAGER_FIELD, ) # Required-field guard. The mapping engine DROPS a field whose # rule renders nothing (value_map miss with no default), and an # ABSENT field on the vendor PUT means "leave unchanged" — so a # missing companyCode/custDeptNumber would otherwise silently # misfile (or silently keep) the employee under the wrong legal # entity / department with a green run status. Fail loudly with # the fix instead. if not str(payload.get("companyCode") or "").strip(): # Non-transient, data/config-dependent (a new legal entity with no # value_map row) → BLOCK, not failed: persists cheaply (no vendor # re-call), keeps the run green, auto-retries once the mapping row # is added — instead of reddening every sweep until config catches # up (PR2 taxonomy). return PersonOutcome( status="blocked", external_id=prior_external_id or employee_id, error=( f"{employee_id} ({person_email}) 渲染后缺 companyCode —— " "company_code 不在 value_map 里(新法人主体?)。在同步目标的 " "companyCode 映射规则 value_map 中补一行 Helix company_code " "→ 汇联易公司编码 后重跑。" ), detail={"blocked_reason": "huilianyi_company_code_unmapped", "fields": ["companyCode"]}, ) if not str(payload.get("custDeptNumber") or "").strip(): return PersonOutcome( status="blocked", external_id=prior_external_id or employee_id, error=( f"{employee_id} ({person_email}) 渲染后缺 custDeptNumber" "(无主部门或部门编码为空)——检查该员工的 HRIS 部门归属。" ), detail={"blocked_reason": "huilianyi_dept_number_missing", "fields": ["custDeptNumber"]}, ) # Lookup-first: 汇联易 create is NOT idempotent (employeeID + # email are tenant-unique) and the tenant pre-dates Helix — # most employees already exist from the Authing-pull era, so # the steady state is "update". try: status, body, raw = await self._client.request( "GET", f"/api/open/user/detail/{employee_id}", ) except HuilianyiError as e: return PersonOutcome( status="failed", external_id=prior_external_id, error=f"GET user detail {employee_id} failed: {e!s:.300}", ) exists = False if _is_ok(status, body) and isinstance(body, dict): exists = True raw_status = body.get("status") data = body.get("data") if raw_status is None and isinstance(data, dict): raw_status = data.get("status") vendor_status = str(raw_status or "") if vendor_status == _STATUS_RESIGNED: # The vendor row is in 离职 state. Restoring is an HR # decision (离职员工恢复 endpoint) — do NOT silently # rehire on the expense system. Surface with the exact # next step instead. return PersonOutcome( status="blocked", external_id=employee_id, error=( f"{employee_id} ({person_email}) 在汇联易侧是离职" "状态;如确需恢复,请在汇联易后台执行「离职员工" "恢复」后重跑同步。" ), detail={"blocked_reason": "huilianyi_vendor_resigned"}, ) elif status == 404 or _error_code(body) == USER_NOT_FOUND: exists = False elif not _is_ok(status, body): return PersonOutcome( status="failed", external_id=prior_external_id, error=( f"GET user {employee_id} → {status} " f"errorCode={_error_code(body)}: " f"{_vendor_message(body, raw)}" ), ) if exists: return await self._update(employee_id, person_email, payload) return await self._create(employee_id, person_email, payload) async def _create( self, employee_id: str, person_email: str, payload: dict[str, Any], *, allow_manager_retry: bool = True, allow_phone_retry: bool = True, ) -> PersonOutcome: try: status, body, raw = await self._client.request( "POST", "/api/open/user/create/v2", json_body=payload, ) except HuilianyiError as e: return PersonOutcome( status="failed", external_id=employee_id, error=f"POST create {employee_id} failed: {e!s:.300}", ) if _is_ok(status, body): # ``op`` tells the ledger the REAL vendor operation. Without it the # orchestrator falls back to ``was_new`` (= "no payload hash on the # state row"), which a mapping-config change sets to NULL for the # WHOLE population — so a config rollout labelled all 407 people # 「已创建」 while every call was in fact a PUT update (2026-08-27). return PersonOutcome( status="success", external_id=employee_id, detail={"op": "create"}, ) if _looks_duplicate(body, raw): # Raced a parallel writer (HR console / another flow) — # converge via update instead of failing the sweep row. logger.info( "huilianyi create %s collided with an existing row; " "falling back to update", employee_id, ) return await self._update(employee_id, person_email, payload) hint = "" if payload.get("custDeptNumber"): # Hire-day edge: a person can land in a dept created in HRIS # the same day — the single-person event push runs WITHOUT # Phase A, so the dept may not exist on the vendor yet. The # nightly full sweep (Phase A before Phase B) self-heals. hint = ( "(若部门为当日新建,等下一次全量同步先建部门后自愈)" ) original_error = ( f"POST create {employee_id} → {status} " f"errorCode={_error_code(body)}: {_vendor_message(body, raw)}" f"{hint}" ) return await self._degrade_or_fail( employee_id, person_email, payload, original_error, code=_error_code(body), create=True, allow_manager_retry=allow_manager_retry, allow_phone_retry=allow_phone_retry, ) async def _update( self, employee_id: str, person_email: str, payload: dict[str, Any], *, allow_manager_retry: bool = True, allow_phone_retry: bool = True, ) -> PersonOutcome: try: status, body, raw = await self._client.request( "PUT", "/api/open/user/v2", json_body=payload, ) except HuilianyiError as e: return PersonOutcome( status="failed", external_id=employee_id, error=f"PUT update {employee_id} failed: {e!s:.300}", ) if _is_ok(status, body): return PersonOutcome( status="success", external_id=employee_id, detail={"op": "update"}, ) if _error_code(body) == USER_NOT_FOUND: # Deleted behind our back between lookup and update — # recreate (mirrors ScimRunner's PUT-404 → POST fallback). try: status2, body2, raw2 = await self._client.request( "POST", "/api/open/user/create/v2", json_body=payload, ) except HuilianyiError as e: return PersonOutcome( status="failed", external_id=employee_id, error=f"PUT 120207 then POST failed: {e!s:.300}", ) if _is_ok(status2, body2): return PersonOutcome( status="success", external_id=employee_id, detail={"op": "create"}, ) return PersonOutcome( status="failed", external_id=employee_id, error=( f"PUT 120207 then POST → {status2} " f"errorCode={_error_code(body2)}: " f"{_vendor_message(body2, raw2)}" ), ) original_error = ( f"PUT update {employee_id} → {status} " f"errorCode={_error_code(body)}: {_vendor_message(body, raw)}" ) return await self._degrade_or_fail( employee_id, person_email, payload, original_error, code=_error_code(body), create=False, allow_manager_retry=allow_manager_retry, allow_phone_retry=allow_phone_retry, ) async def _degrade_or_fail( self, employee_id: str, person_email: str, payload: dict[str, Any], original_error: str, *, code: str | None, create: bool, allow_manager_retry: bool, allow_phone_retry: bool, ) -> PersonOutcome: """Decide whether ONE optional field can be sacrificed to land the rest. Two different mechanisms, picked by what the vendor actually tells us: * **手机号 — by ERROR CODE.** The phone rejections are documented (``_PHONE_REJECT_CODES``), so the slot is dropped only on those codes. No message-marker guessing. * **主岗直属领导 — by PROBE.** The vendor documents no code for an unusable 直属领导, so the retry itself is the experiment (see :meth:`_retry_without_manager`). Order matters: the phone check runs first because its evidence is exact. Each mechanism fires at most once per person (the flags are threaded into the retry), so the worst case is 3 vendor calls and the recursion cannot loop. Required fields are NEVER sacrificed — companyCode / custDeptNumber are guarded earlier and would return ``blocked``, not silently ship a reduced payload. """ if ( allow_phone_retry and code in _PHONE_REJECT_CODES and any(payload.get(f) for f in _PHONE_FIELDS) ): return await self._retry_without_phone( employee_id, person_email, payload, original_error, create=create, allow_manager_retry=allow_manager_retry, ) if allow_manager_retry and payload.get(_MANAGER_FIELD): return await self._retry_without_manager( employee_id, person_email, payload, original_error, create=create, allow_phone_retry=allow_phone_retry, ) return PersonOutcome( status="failed", external_id=employee_id, error=original_error, ) async def _retry_without_phone( self, employee_id: str, person_email: str, payload: dict[str, Any], original_error: str, *, create: bool, allow_manager_retry: bool, ) -> PersonOutcome: """Re-attempt once with the 手机号 slot removed. Fires only on a DOCUMENTED phone rejection. The two that matter in practice are both uniqueness: 汇联易's 手机号 is tenant-unique, so a number that HR already bound to someone else (or an ex-colleague's row) can never be applied by us — and 12002008's message names the holder's 工号, which is exactly what the admin needs. Dropping the number keeps name / dept / company / 入职时间 / 直属领导 flowing; the slot is re-attempted on the next sweep (the orchestrator clears ``last_verified_at`` off ``slot_outcomes``), so it self-heals the moment the conflict is resolved on either side. """ stripped = { k: v for k, v in payload.items() if k not in _PHONE_FIELDS } logger.info( "huilianyi: %s rejected on the phone slot (%s); retrying without it", employee_id, original_error[:160], ) if create: outcome = await self._create( employee_id, person_email, stripped, allow_manager_retry=allow_manager_retry, allow_phone_retry=False, ) else: outcome = await self._update( employee_id, person_email, stripped, allow_manager_retry=allow_manager_retry, allow_phone_retry=False, ) if outcome.status != "success": # Something else is also wrong — surface the FIRST error, which # is the phone one, plus whatever the retry said. return PersonOutcome( status="failed", external_id=employee_id, error=original_error, ) note = ( f"partial: 手机号未应用(汇联易侧手机号唯一/校验拒绝),其余字段已同步;" f"厂商原话:{original_error}" ) logger.warning("huilianyi: %s", note) return PersonOutcome( status="success", external_id=employee_id, error=note, detail={**(outcome.detail or {})}, slot_outcomes=[ *(outcome.slot_outcomes or []), { "slot": _MOBILE_FIELD, "status": "dropped", "detail": "mobile rejected by vendor", "vendor_error": original_error[:300], }, ], ) async def _retry_without_manager( self, employee_id: str, person_email: str, payload: dict[str, Any], original_error: str, *, create: bool, allow_phone_retry: bool = True, ) -> PersonOutcome: """Re-attempt one rejected upsert with 主岗直属领导 stripped. The vendor documents NO error code for an unusable ``directManagerEmpoyeeId`` (the create/update 错误码 tables cover 工号/邮箱/手机/部门/公司 only), so there is no marker to match on. Rather than guess one, the failure path itself is the probe: retry once without the manager field and let the vendor answer. If it now succeeds, the manager reference was the sole cause — the person's real attributes (dept, company, name, email, 入职时间) land instead of the whole row going red over ONE optional attribute, and the dropped field is recorded on the outcome so the admin can see it. If it fails again, the manager was never the problem: return the ORIGINAL error verbatim, so no diagnosis is lost. Safe to replay: this path is only reached on a DEFINITIVE vendor rejection (a parsed non-0000 body), never on a transport error where a POST might have landed server-side — that hazard is handled in ``HuilianyiClient.request`` and must stay that way. """ stripped = {k: v for k, v in payload.items() if k != _MANAGER_FIELD} logger.info( "huilianyi: %s rejected with %s=%r; retrying once without it", employee_id, _MANAGER_FIELD, payload.get(_MANAGER_FIELD), ) if create: outcome = await self._create( employee_id, person_email, stripped, allow_manager_retry=False, allow_phone_retry=allow_phone_retry, ) else: outcome = await self._update( employee_id, person_email, stripped, allow_manager_retry=False, allow_phone_retry=allow_phone_retry, ) if outcome.status != "success": # Not the manager's fault — surface the first, richer error. return PersonOutcome( status="failed", external_id=employee_id, error=original_error, ) note = ( f"partial: 主岗直属领导 ({_MANAGER_FIELD}=" f"{payload.get(_MANAGER_FIELD)}) 被汇联易拒绝,其余字段已同步" "(汇联易侧保留原有直属领导);原始报错:" f"{original_error}" ) logger.warning("huilianyi: %s", note) # Slot-degraded success (PR2 resilient-writes contract): the row # landed but ONE slot didn't, so the orchestrator clears # ``last_verified_at`` and re-attempts the slot next sweep instead of # hash-skipping the degradation for the whole drift window. return PersonOutcome( status="success", external_id=employee_id, error=note, detail={"op": "create" if create else "update"}, slot_outcomes=[ *(outcome.slot_outcomes or []), { "slot": _MANAGER_FIELD, "status": "dropped", "detail": "direct manager rejected by vendor", "manager_employee_no": payload.get(_MANAGER_FIELD), }, ], ) async def deactivate( self, *, person_email: str, prior_external_id: str | None, ) -> PersonOutcome: if not prior_external_id: # No employeeID on the state row = this person was never # successfully pushed by Helix. There is no email-keyed # disable endpoint, so guessing is impossible — surface # instead of silently claiming success against a vendor row # that might exist from the Authing era. return PersonOutcome( status="failed", error=( f"no employeeID recorded for {person_email}; run a " "full sweep first (or set identity_sync_state." "external_id to the 工号) so 离职 can address the " "vendor row." ), ) # Trailing slash per the vendor doc's literal path shape # (``/api/open/user/delete/v2/{employeeId}/``). Empty body = # leavingDate defaults to now on the vendor side. try: status, body, raw = await self._client.request( "DELETE", f"/api/open/user/delete/v2/{prior_external_id}/", json_body={}, ) except HuilianyiError as e: return PersonOutcome( status="failed", external_id=prior_external_id, error=f"离职 {prior_external_id} failed: {e!s:.300}", ) if _is_ok(status, body): return PersonOutcome( status="deactivated", external_id=prior_external_id, ) if status == 404 or _error_code(body) == USER_NOT_FOUND: # Already gone — goal (not active on the vendor) achieved. return PersonOutcome(status="deactivated", external_id=None) if _looks_already_resigned(body, raw): # HR already resigned this person by hand in the 汇联易 # console — idempotent success, keep the vendor key. return PersonOutcome( status="deactivated", external_id=prior_external_id, ) return PersonOutcome( status="failed", external_id=prior_external_id, error=( f"离职 {prior_external_id} → {status} " f"errorCode={_error_code(body)}: {_vendor_message(body, raw)}" ), ) def describe(self) -> str: return f"汇联易 {self._client._base_url}" async def aclose(self) -> None: await self._client.aclose() class HuilianyiDepartmentRunner(DeptSyncRunner): """Phase A — department reconciliation against 汇联易. 汇联易 keys departments by ``custDeptNumber``, so Helix's HRIS dept code IS the vendor key — no name-adoption catalog (the Feishu od-XXX problem doesn't exist here). ``external_id`` stores the vendor's returned ``oid`` when present (informational), falling back to the dept code; all addressing is by code regardless. ``parent_dept_code`` is passed through as ``parentCustDeptNumber``. The orchestrator feeds depts in topological order and None's out parents that fell outside the sweep scope, so a parent reference always points at a dept that was just pushed (or omitted → vendor treats as top-level). """ def __init__( self, *, client: HuilianyiClient, dept_code_map: dict[str, str] | None = None, ): self._client = client # In-place recode map: Helix SF org-node code → existing 汇联易 # ``custDeptNumber``. When set, the runner addresses the EXISTING # vendor dept (UPDATE in place) instead of creating an SF-coded # parallel — the cutover from the Authing/北森-era code tree. # Empty (default) = pass codes through unchanged (current behaviour). self._code_map = dept_code_map or {} def _vendor_code(self, helix_code: str | None) -> str | None: if helix_code is None: return None return self._code_map.get(helix_code, helix_code) 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, leader_employee_no: str | None = None, cost_center_label: str | None = None, hrbp_employee_no: str | None = None, budget_owner_employee_no: str | None = None, ) -> DeptOutcome: # Translate Helix SF codes → vendor codes (in-place recode). Both the # dept's own code and its parent map; everything below talks to the # vendor in vendor-code space. dept_code = self._vendor_code(dept_code) parent_dept_code = self._vendor_code(parent_dept_code) body: dict[str, Any] = { "custDeptNumber": dept_code, "departmentName": dept_name or dept_code, } if parent_dept_code: body["parentCustDeptNumber"] = parent_dept_code # 部门负责人 / HRBP / Budget Owner → ``departmentPositionList`` roles # ({positionCode, employeeId}). Each 工号 is derived by the dept phase # from the HRIS golden record (leader = org head; HRBP/Budget Owner = # member-shared SF 工作关系). Only roles with a resolved 工号 are # included. CAVEAT (verify before enabling — target is enabled=0): # this sends a PARTIAL ``departmentPositionList`` when only a subset # of the three roles resolves. Safe ONLY if 汇联易 MERGES the list # per-position; if update-V2 treats this list field as a full REPLACE # (common for list fields), a sweep that resolves only the leader # would WIPE a previously-set HRBP/Budget-Owner. The vendor's # list-merge semantics are NOT verifiable from code and must be # confirmed against the live API before this sync is turned on; if # replace-semantics, always send all three (reading current vendor # state to preserve unresolved roles) or use a per-role patch. # ``managerEmployeeId`` is intentionally NOT used # — real depts carry the leader via the 6103 ROLE (managerEmployeeId # was null), so the role list is the contract the vendor actually # honours. ``leader_email`` / ``dept_order`` stay unused. # ORDERING NOTE: Phase A (depts) precedes Phase B (users), so on a # FIRST sweep a role 工号 can reference a person 汇联易 hasn't received # yet → the vendor may 120202 (员工不存在); surfaced (not silent) and # self-heals next sweep once Phase B lands the person. In steady state # every referenced person already exists. (These kwargs are # huilianyi-specific; the base DeptSyncRunner signature omits them.) positions: list[dict[str, str]] = [] if leader_employee_no: positions.append( {"positionCode": DEPT_ROLE_LEADER, "employeeId": leader_employee_no}, ) if hrbp_employee_no: positions.append( {"positionCode": DEPT_ROLE_HRBP, "employeeId": hrbp_employee_no}, ) if budget_owner_employee_no: positions.append( {"positionCode": DEPT_ROLE_BUDGET_OWNER, "employeeId": budget_owner_employee_no}, ) if positions: body["departmentPositionList"] = positions # 成本中心 — pushed as a customFormValues extension field (fieldCode # CC), the FULL vendor label ("C0906 BO-Business Operation") resolved # by the dept phase from HUILIANYI_COST_CENTER_LABEL_MAP; an unmapped # code arrives as None → field omitted → vendor keeps its value. if cost_center_label: body["customFormValues"] = [ {"fieldCode": "CC", "value": cost_center_label}, ] if prior_external_id: outcome = await self._put(dept_code, body) if outcome is not None: return outcome # PUT said the code doesn't exist — vendor row vanished; # fall through to POST recreate. try: status, resp, raw = await self._client.request( "POST", "/api/open/department/v2", json_body=body, ) except HuilianyiError as e: return DeptOutcome( status="failed", external_id=prior_external_id, error=f"POST dept {dept_code} failed: {e!s:.300}", ) if _is_ok(status, resp): oid = (resp or {}).get("oid") return DeptOutcome( status="success", external_id=str(oid) if oid else dept_code, ) if _error_code(resp) in DEPT_ALREADY_EXISTS: outcome = await self._put(dept_code, body) if outcome is not None: return outcome return DeptOutcome( status="failed", external_id=prior_external_id, error=( f"dept {dept_code}: POST says exists but PUT says " "not found — vendor state inconsistent, re-run the " "sweep." ), ) return DeptOutcome( status="failed", external_id=prior_external_id, error=( f"POST dept {dept_code} → {status} " f"errorCode={_error_code(resp)}: {_vendor_message(resp, raw)}" ), ) async def _put( self, dept_code: str, body: dict[str, Any], ) -> DeptOutcome | None: """PUT update; returns None when the vendor says the dept code doesn't exist (caller falls back to POST).""" try: status, resp, raw = await self._client.request( "PUT", "/api/open/department/v2", json_body=body, ) except HuilianyiError as e: return DeptOutcome( status="failed", error=f"PUT dept {dept_code} failed: {e!s:.300}", ) if _is_ok(status, resp): oid = (resp or {}).get("oid") return DeptOutcome( status="success", external_id=str(oid) if oid else dept_code, ) if _error_code(resp) in DEPT_NOT_FOUND: return None return DeptOutcome( status="failed", error=( f"PUT dept {dept_code} → {status} " f"errorCode={_error_code(resp)}: {_vendor_message(resp, raw)}" ), ) async def soft_delete( self, *, dept_code: str, prior_external_id: str | None, ) -> DeptOutcome: # Disable the vendor-coded dept (in-place recode space). dept_code = self._vendor_code(dept_code) try: status, resp, raw = await self._client.request( "DELETE", f"/api/open/department/delete/v2/{dept_code}", params={"disableChildren": "false"}, ) except HuilianyiError as e: return DeptOutcome( status="failed", external_id=prior_external_id, error=f"失效 dept {dept_code} failed: {e!s:.300}", ) if _is_ok(status, resp): return DeptOutcome( status="soft_deleted", external_id=prior_external_id, ) if status == 404 or _error_code(resp) in DEPT_NOT_FOUND: # Already gone — idempotent success. return DeptOutcome(status="soft_deleted", external_id=None) if _error_code(resp) in DEPT_HAS_CHILDREN: # The dept still has enabled sub-depts / active employees, so # NOT retiring it is the CORRECT outcome, not a failure — a dept # can't be soft-deleted while it holds people. ``skipped`` (green) # + an actionable note, not ``failed`` (which downgrades the whole # dept phase to partial every sweep until a human migrates the # members). It retires automatically once its members/children # move (their moves change the tree; the next sweep re-attempts). return DeptOutcome( status="skipped", external_id=prior_external_id, error=( f"dept {dept_code} 在汇联易侧仍有启用的子部门或在职员工," "暂不失效(非错误);先在汇联易后台迁走/停用其子部门与人员," "下次同步会自动失效该部门。" ), ) return DeptOutcome( status="failed", external_id=prior_external_id, error=( f"失效 dept {dept_code} → {status} " f"errorCode={_error_code(resp)}: {_vendor_message(resp, raw)}" ), ) def describe(self) -> str: return f"汇联易 departments {self._client._base_url}" async def aclose(self) -> None: await self._client.aclose()