"""Feishu (Lark) Contact department runner — lookup-by-name + adopt. Critical design fact (verified against the live tenant 2026-05-28): * Feishu auto-generates ``open_department_id`` like ``od-581b2e24e340c72f0a7d49e1207fccc5`` — admins cannot pre-pick it on the API surface. So **Helix does NOT get to use its HRIS dept_code as Feishu open_department_id**. Doing so would create duplicate departments side-by-side with Authing-built originals. * PingCAP's existing 20 top-level Feishu departments include all 13 HRIS top-level BUs by exact-name match, plus 7 geographic-office / intern / contractor depts (Tokyo Office, SG Office, SV Office, MY Office, China Business Group, PingCAP Intern, Consultants). The 7 extras are admin-maintained outside Helix's HRIS authority and MUST be left untouched. Strategy: **adopt by stable id, then sibling-scoped name**: 1. ``prime_catalog()`` walks ``/contact/v3/departments`` (BFS over the whole tree) and builds THREE indices over every active dept: ``name → od`` (flat, for the office lookup + public ``lookup_external_id_by_name``), ``(parent_od, name) → od`` (sibling-scoped), and ``name → distinct-od count`` (uniqueness guard). Called once at runner build time by the orchestrator. 2. ``upsert_dept(dept_code, payload, prior_external_id)``: - If ``prior_external_id`` set (= we adopted previously) → PATCH by that ``od-XXX``. This is the stable-key path; renames stay in-place on the same dept (members + 部门群 preserved) and never re-create. - Else classify via ``match_existing`` (shared with the dry-run): * **sibling-scoped** ``(parent_od, name)`` match → PATCH. The parent is already adopted (orchestrator sweeps parents-first), so two same-named depts under different parents (e.g. two "Sales") resolve to DISTINCT ods instead of both aliasing to a single flat-name entry and thrashing one Feishu dept. * else a **globally-unique flat-name** match → PATCH. Covers a benign root-depth mismatch (HRIS BUs under root vs nested under a "PingCAP" dept on Feishu) where the resolved parent_od won't line up; a name seen exactly once tenant-wide is unambiguous. A match returns the ``od-XXX`` in the outcome so the orchestrator persists it to ``identity_sync_department_state.external_id``. - Else POST create. Feishu auto-assigns ``open_department_id``; we capture it from the response and return it. 3. ``soft_delete(dept_code, prior_external_id)`` PATCHes ``status.is_deleted=true``. Only Helix-adopted depts ever reach this path — the 7 geographic extras never enter Helix's state table, so they're not in scope. ``parent_department_id`` translation: the mapping engine renders the parent's HRIS dept code; the runner translates it through the SAME catalog index (or through the dept_code → external_id map the orchestrator threads in via ``known_parent_index``) to Feishu's ``od-XXX``. Roots use Feishu's reserved ``"0"`` sentinel. Leader translation: ``leader_user_id_email`` is resolved through ``FeishuApprovalClient.resolve_user_id_by_email``; the request adds ``?user_id_type=user_id`` so Feishu interprets the value as user_id (not open_id). Translation failure drops the slot rather than blocking the dept upsert. """ from __future__ import annotations import logging from typing import Any from urllib.parse import quote from app.approval.feishu_client import ( FeishuApprovalClient, ) from app.approval.feishu_client import ( parse_feishu_error_code as _parse_feishu_code, ) from app.core.exceptions import AppException from app.domains.identity_sync.runners.base import DeptOutcome, DeptSyncRunner logger = logging.getLogger(__name__) # Feishu dept-API codes meaning "already in target state / not found" # — treat as idempotent success on soft_delete; PATCH/POST will short- # circuit when the runner observes these. _DEPT_ABSENT_OR_DONE_CODES: frozenset[int] = frozenset({ 230002, # idempotent already-in-target-state (e.g. already deleted) 230009, # department not found 230018, # department_id invalid 99991663, # generic resource-not-found }) def _normalise_name(name: str | None) -> str: """Normalise dept name for catalog lookup — trim, NFKC, lower. HRIS may store ``"Engineering Group"`` while Feishu has the same string verbatim; defence-in-depth in case Authing introduced trailing whitespace or case variance. Lower-case is a safe one-way equivalence because we only USE this index for matching; the original name from the payload is what gets PATCHed. """ if not name: return "" return " ".join(name.split()).strip().lower() class FeishuDepartmentRunner(DeptSyncRunner): """Adopt-existing-then-create dept runner for Feishu. The runner holds these indices, populated at build time: * :attr:`_name_to_external` — ``normalised_name → od-XXX`` (flat, last-writer-wins on collisions). Drives the public ``lookup_external_id_by_name`` (office index) + the unique-name adoption fallback. * :attr:`_sibling_to_external` — ``(parent_od, normalised_name) → od-XXX``. The primary adoption key: disambiguates same-named depts under different parents. * :attr:`_name_counts` — ``normalised_name → distinct-od count``. Gates the flat-name fallback to tenant-wide-unique names only. * :attr:`_helix_dept_to_external` — ``HRIS dept_code → od-XXX`` from the persistent ``identity_sync_department_state`` table. Authoritative when present; used to resolve the parent dept's Feishu id when pushing children. Grows through the sweep as parents adopt/create. All can be empty safely — the runner falls back to POST-create on miss. """ supports_soft_delete = True def __init__( self, *, client: FeishuApprovalClient, name_to_external: dict[str, str] | None = None, sibling_to_external: dict[tuple[str, str], str] | None = None, helix_dept_to_external: dict[str, str] | None = None, label: str = "Feishu Contact (dept)", ) -> None: self._client = client # Defensive copies + lower-case normalisation. The index keys # always match `_normalise_name(name)`; values are raw od-XXX. self._name_to_external: dict[str, str] = { _normalise_name(k): v for k, v in (name_to_external or {}).items() if k and v } # Sibling-scoped index: ``(parent_od, normalised_name) → od-XXX``. # Disambiguates same-named depts under different parents (two # "Sales" under APAC vs NA), which the flat ``_name_to_external`` # aliases to one od (last-writer-wins) and would thrash. Built in # :meth:`prime_catalog`; injectable here for unit tests. self._sibling_to_external: dict[tuple[str, str], str] = dict( sibling_to_external or {}, ) # ``normalised_name → count of DISTINCT ods`` carrying that name # tenant-wide. Drives the "globally-unique flat-name" fallback in # :meth:`upsert_dept` (a name seen exactly once is unambiguous to # adopt even when the sibling key misses — e.g. HRIS root depth # differs from Feishu's). prime_catalog() recomputes this with the # real distinct-od counts; here we seed it from an injected # ``name_to_external`` (dict keys are unique → every name counts # once) so a runner built WITHOUT prime_catalog (unit tests; never # the orchestrator, which always primes) still adopts by name. self._name_counts: dict[str, int] = { n: 1 for n in self._name_to_external } self._helix_dept_to_external: dict[str, str] = dict( helix_dept_to_external or {} ) self._label = label async def prime_catalog(self) -> None: """Walk ``/contact/v3/departments`` BFS-style and populate :attr:`_name_to_external` with every active dept in the tenant. Called once at runner-build time. **Fail-loud** — a page-level fetch failure or an exhausted step/page guard RAISES to abort this round's Phase A, matching the contract of :func:`fetch_feishu_directory`. A PARTIAL catalog must never be treated as complete: an existing Feishu dept that wasn't indexed (because its subtree's page errored) would miss the adopt match in :meth:`upsert_dept` and get a DUPLICATE parallel dept POSTed under it, whose od is then permanently adopted (#1757). Aborting the sweep and retrying on a clean full catalog is the only safe input for adopt/create decisions. """ if self._name_to_external: return # BFS from root "0". Each level fetched via parent_department_id # = + page_size=50 (Feishu's max). queue: list[str] = ["0"] visited: set[str] = set() index: dict[str, str] = {} sibling_index: dict[tuple[str, str], str] = {} # normalised_name → set of distinct ods (for the uniqueness # guard on the flat-name fallback). name_ods: dict[str, set[str]] = {} # Per-collection page bound — a page_token cycle (has_more=true # forever with recycled tokens) is a protocol-level wedge. Mirrors # fetch_feishu_directory's ``max_pages`` fail-loud guard. max_pages = 200 # Hard upper bound — defensive against unbounded recursion # / cyclic tenant data. Real org trees are ≤ 6 levels deep; # 500 dept iterations covers any plausible tenant. steps_remaining = 500 while queue and steps_remaining > 0: steps_remaining -= 1 parent = queue.pop(0) if parent in visited: continue visited.add(parent) page_token: str | None = None for _page in range(max_pages): qs = ( f"parent_department_id={quote(parent, safe='')}" "&department_id_type=open_department_id" "&page_size=50" "&fetch_child=false" ) if page_token: qs += f"&page_token={quote(page_token, safe='')}" # Do NOT swallow a page fetch failure. A silently truncated # catalog reads as "these existing depts don't exist" and # triggers duplicate creates (#1757) — fail loud instead. data = await self._client.request( "GET", f"/contact/v3/departments?{qs}", ) items = (data or {}).get("items") or [] for d in items: if not isinstance(d, dict): continue # Skip soft-deleted depts so we don't try to PATCH them. status = (d.get("status") or {}) if status.get("is_deleted"): continue name = d.get("name") od = d.get("open_department_id") if isinstance(name, str) and isinstance(od, str) and od: norm = _normalise_name(name) index[norm] = od # ``parent`` is this level's parent od ("0" for # top-level) — exactly the key the upsert path # resolves via _resolve_parent_external. sibling_index[(parent, norm)] = od name_ods.setdefault(norm, set()).add(od) queue.append(od) if not (data or {}).get("has_more"): break page_token = (data or {}).get("page_token") if not page_token: break else: raise RuntimeError( f"feishu_dept catalog: dept paginator under {parent!r} " f"did not terminate within {max_pages} pages — refusing " "a possibly-cyclic / truncated catalog." ) if queue: # Step guard tripped with depts still unvisited: same fail-loud # contract as fetch_feishu_directory. A truncated catalog would # make existing depts look absent → duplicate creates. raise RuntimeError( "feishu_dept catalog: dept BFS exceeded the 500-step guard " f"with {len(queue)} dept(s) unvisited — refusing a truncated " "catalog for adopt/create decisions." ) self._name_to_external = index self._sibling_to_external = sibling_index self._name_counts = {k: len(v) for k, v in name_ods.items()} logger.info( "feishu_dept catalog primed: %d depts indexed " "(%d sibling keys, %d colliding names)", len(index), len(sibling_index), sum(1 for c in self._name_counts.values() if c > 1), ) def lookup_external_id_by_name(self, name: str | None) -> str | None: """Public lookup of ``name → od-XXX`` from the primed catalog. Callers outside the runner (specifically the orchestrator's ``_build_office_name_index``) use this so they don't need to touch :attr:`_name_to_external` directly. Returns None on miss. """ if not name: return None return self._name_to_external.get(_normalise_name(name)) def match_existing( self, *, dept_name: str | None, parent_od: str | None, ) -> tuple[str | None, str]: """Classify how an upsert would adopt ``dept_name`` under ``parent_od``: ``(od-XXX, match_kind)`` or ``(None, "create")``. Single source of truth for both the live upsert path (:meth:`upsert_dept`) and the read-only preview (``service.dry_run_feishu``) so the two can never drift: * ``sibling`` — exact ``(parent_od, name)`` hit. Disambiguates same-named depts under different parents. * ``unique-name`` — sibling miss but the name is unique tenant-wide; safe to adopt despite a parent-depth mismatch. * ``create`` — no safe match; a real run POSTs a new dept. ``parent_od`` defaults to the root sentinel when falsy. """ if not dept_name: return None, "create" norm = _normalise_name(dept_name) pod = parent_od or self._root_parent_sentinel() od = self._sibling_to_external.get((pod, norm)) if od: return od, "sibling" if self._name_counts.get(norm, 0) == 1: od = self._name_to_external.get(norm) if od: return od, "unique-name" return None, "create" # ── Helpers ────────────────────────────────────────────────────── @staticmethod def _root_parent_sentinel() -> str: return "0" async def _resolve_leader_user_id(self, email: str | None) -> str | None: """Email → Feishu user_id, narrowly. Only swallow the documented "leader not on Feishu yet" case (``FEISHU_USER_NOT_FOUND``). Auth / rate-limit / 5xx codes propagate so the orchestrator's Phase A can short-circuit the whole sweep with one clear error rather than silently dropping the leader slot for every dept under bad creds. """ if not email: return None try: return await self._client.resolve_user_id_by_email(email) except AppException as e: if e.code == "FEISHU_USER_NOT_FOUND": return None raise def _resolve_parent_external( self, parent_dept_code: str, *, omit_if_unresolved: bool = False, ) -> str | None: """Translate HRIS ``parent_dept_code`` → Feishu ``od-XXX``. Lookup order: 1. ``_helix_dept_to_external`` (state-table-derived index) — authoritative; carries the persistent adoption. 2. Empty parent code → root sentinel ``"0"`` (genuinely a top-level dept). 3. A NON-empty parent code that isn't mapped yet — the "parent not yet pushed / transiently unadopted" race. For a CREATE we still need a concrete parent, so fall back to root "0". But for an UPDATE (``omit_if_unresolved=True``) we return ``None`` so the caller OMITS ``parent_department_id`` — sending "0" here would RE-PARENT an already-correctly-placed dept to the tenant root, and the hash would then freeze it there (the next sweep, once the parent re-adopts, sees an unchanged HRIS ``parent_dept_code`` and skips). Omitting preserves Feishu's current parent; the dept converges when the parent resolves. """ if not parent_dept_code: return self._root_parent_sentinel() mapped = self._helix_dept_to_external.get(parent_dept_code) if mapped: return mapped return None if omit_if_unresolved else self._root_parent_sentinel() def _build_body( self, *, dept_code: str, dept_name: str, parent_dept_code: str | None, leader_user_id: str | None, dept_order: int | None, is_update: bool = False, ) -> dict[str, Any]: """Build the ``POST/PATCH /contact/v3/departments`` body. Direct projection from IdentityOrgNode fields to Feishu wire format — no mapping engine in this layer, because dept fields are admin-invariant (Feishu's slots are fixed; no per-tenant customization use case). ``is_update`` — when this body targets an already-adopted dept (PATCH), a set-but-unresolved parent OMITS the slot rather than re-parenting to root "0" (see :meth:`_resolve_parent_external`). """ body: dict[str, Any] = {"name": str(dept_name or "")} parent = self._resolve_parent_external( str(parent_dept_code or ""), omit_if_unresolved=is_update, ) if parent is not None: body["parent_department_id"] = parent # ``code`` slot — Helix HRIS dept_code carried into Feishu's # admin-supplied ``code`` field for cross-reference in the # Feishu admin console + future "match by code" paths. if dept_code: body["code"] = str(dept_code) if leader_user_id: body["leader_user_id"] = leader_user_id if dept_order is not None: try: body["order"] = str(int(dept_order)) except (TypeError, ValueError): pass return body # ── DeptSyncRunner implementation ──────────────────────────────── 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: if not dept_code: return DeptOutcome( status="failed", error="dept_code is empty — cannot push to Feishu", ) leader_user_id = await self._resolve_leader_user_id(leader_email) body = self._build_body( dept_code=dept_code, dept_name=dept_name, parent_dept_code=parent_dept_code, leader_user_id=leader_user_id, dept_order=dept_order, # An adopted dept (prior_external_id) is a PATCH — omit an # unresolved parent rather than re-parenting it to root "0". is_update=bool(prior_external_id), ) # 1) Adopted before → PATCH by external_id. if prior_external_id: outcome = await self._patch_dept( external_id=prior_external_id, body=body, ) if outcome.status != "failed": # Record the helix→external mapping for parent lookups # later in this sweep (covers re-parenting cases). self._helix_dept_to_external[dept_code] = prior_external_id return outcome # 2) Adopt an existing Feishu dept by name. # # (a) Sibling-scoped (parent_od, name): disambiguates # same-named depts under different parents — two "Sales" # (one under APAC, one under NA) get distinct ods instead # of both aliasing to the single flat-name entry and # thrashing one Feishu dept (re-parented back and forth) # while the other never gets its own. The parent is # already adopted (orchestrator sweeps parents-before- # children), so ``body["parent_department_id"]`` is its # real od — the same key prime_catalog indexed under. # # (b) Globally-unique flat-name fallback: when the sibling key # misses but the name occurs EXACTLY ONCE tenant-wide, # adopt it anyway. Covers a benign root-depth mismatch # (HRIS has BUs directly under root while Feishu nests them # under a "PingCAP" dept, so the resolved parent_od differs # from the indexed one) without risking a wrong adoption — # a unique name is unambiguous. A COLLIDING name with no # sibling hit is genuinely ambiguous, so it falls through # to create rather than guess. rendered_name = body.get("name") if rendered_name: parent_od = ( body.get("parent_department_id") or self._root_parent_sentinel() ) existing_od, match_kind = self.match_existing( dept_name=rendered_name, parent_od=parent_od, ) if existing_od: logger.info( "feishu_dept: adopting existing Feishu dept %r as %s " "(helix dept_code=%s, match=%s, parent_od=%s)", rendered_name, existing_od, dept_code, match_kind, parent_od, ) outcome = await self._patch_dept( external_id=existing_od, body=body, ) if outcome.status != "failed": self._helix_dept_to_external[dept_code] = existing_od return outcome # 3) Create. Feishu assigns open_department_id; capture it. try: resp = await self._client.request( "POST", "/contact/v3/departments?department_id_type=open_department_id", json_body=body, ) except AppException as e: num = _parse_feishu_code(e) if num in _DEPT_ABSENT_OR_DONE_CODES: # On a POST-create these "absent/not-found" codes # (230009 dept-not-found / 230018 dept_id-invalid / # 99991663 resource-not-found) do NOT mean "Feishu already # created it under us" — a genuine concurrent create would # surface as a name/code collision, not a not-found. Here # they almost always mean the ``parent_department_id`` in # the body is stale/absent (the parent's od was deleted on # the Feishu side outside Helix). Returning ``success`` with # ``external_id=None`` makes the orchestrator store the hash # against a NULL adoption → next Phase A hash-skips and the # dept is NEVER created, and every child falls back to root # '0' (#1738). Return ``failed`` (external_id stays None so # the orchestrator clears adoption) and let the next sweep # retry once the parent re-adopts. logger.warning( "feishu_dept: POST %r got absent/not-found code %s " "(likely stale parent_department_id); failing so the " "next sweep retries after the parent re-adopts", dept_code, num, ) return DeptOutcome( status="failed", external_id=None, error=( f"POST /departments → {e.code} (code={num}): " "parent_department_id absent/invalid on Feishu " "(stale od); retrying next sweep" ), ) return DeptOutcome( status="failed", error=f"POST /departments → {e.code}: {e.message}", ) new_od: str | None = None if isinstance(resp, dict): dept_obj = (resp.get("department") or {}) if isinstance(dept_obj, dict): new_od = dept_obj.get("open_department_id") if new_od: self._helix_dept_to_external[dept_code] = new_od # Keep the parent-resolution + exact-sibling indices hot so a # child created later this sweep resolves against this dept. # NOTE: deliberately do NOT bump ``_name_counts`` — it is the # pre-sweep catalog snapshot that gates the unique-name # fallback. Incrementing it here would make a SECOND same-named # create under a DIFFERENT parent later in the sweep see # count==1 and wrongly adopt THIS freshly-created dept via the # flat-name fallback instead of creating its own. if rendered_name: norm = _normalise_name(rendered_name) self._name_to_external[norm] = new_od parent_od = ( body.get("parent_department_id") or self._root_parent_sentinel() ) self._sibling_to_external[(parent_od, norm)] = new_od return DeptOutcome(status="success", external_id=new_od) async def _patch_dept( self, *, external_id: str, body: dict[str, Any], ) -> DeptOutcome: """PATCH ``/contact/v3/departments/{external_id}``. Feishu rejects the request when immutable id slots (``department_id``, ``open_department_id``, ``code`` — actually code IS mutable but only via a dedicated endpoint). Strip them before sending. Body must carry at least ``name`` OR one other mutable field, otherwise Feishu 400s — the caller's body always has ``name``. """ safe = quote(external_id, safe="") patch_body = dict(body) # Immutable on PATCH. patch_body.pop("department_id", None) patch_body.pop("open_department_id", None) try: await self._client.request( "PATCH", f"/contact/v3/departments/{safe}" "?department_id_type=open_department_id", json_body=patch_body, ) return DeptOutcome(status="success", external_id=external_id) except AppException as e: num = _parse_feishu_code(e) if num in _DEPT_ABSENT_OR_DONE_CODES: # Dept disappeared between adoption and patch — surface # as failed so the orchestrator drops the state row's # external_id and the next sweep will re-adopt or POST. return DeptOutcome( status="failed", external_id=None, error=( f"PATCH /departments/{external_id} → absent " f"({num}); will re-adopt next sweep" ), ) return DeptOutcome( status="failed", external_id=external_id, error=f"PATCH /departments/{external_id} → {e.code}: {e.message}", ) async def soft_delete( self, *, dept_code: str, prior_external_id: str | None, ) -> DeptOutcome: """PATCH ``status.is_deleted=true``. Hard delete intentionally not exposed — admins delete dead depts manually in Feishu. Only Helix-adopted depts ever reach this path. The 7 geographic-office / intern / contractor depts that exist outside HRIS's authority NEVER enter Helix's state table, so they're not in scope for this method (orchestrator never identifies them as "fell out of scope"). """ target_id = prior_external_id if not target_id: # No prior external_id and no name-lookup available for # disappeared depts — treat as already done. return DeptOutcome(status="soft_deleted", external_id=None) safe = quote(target_id, safe="") body = {"status": {"is_deleted": True}} try: await self._client.request( "PATCH", f"/contact/v3/departments/{safe}" "?department_id_type=open_department_id", json_body=body, ) return DeptOutcome(status="soft_deleted", external_id=target_id) except AppException as e: num = _parse_feishu_code(e) if num in _DEPT_ABSENT_OR_DONE_CODES: return DeptOutcome(status="soft_deleted", external_id=None) return DeptOutcome( status="failed", external_id=target_id, error=f"PATCH /departments/{target_id} (soft_delete) → {e.code}: {e.message}", ) def describe(self) -> str: return self._label