"""Feishu-sourced HRBP (HR Business Partner) resolver. The Feishu contact directory carries an enterprise-defined custom attribute named "HRBP" whose value points at another Feishu user. The attribute is a user-pointing custom attr; Feishu models that as either ``USER`` (``value.user.user_id``) or ``GENERIC_USER`` (``value.generic_user.id``) — the PingCAP tenant uses ``GENERIC_USER``. HRIS now also carries the HRBP relationship directly (``IdentityPerson.hris_hrbp_person_id``, SF ``EmpJobRelationships`` type 2448), and :func:`resolve_hrbp_for_person` prefers that SF-authoritative column when present, falling back to the Feishu walk only when it's unset/stale. This module turns ``leaver_person_id`` into ``hrbp_person_id`` by walking three Feishu API calls: 1. ``list_custom_attrs()`` once per process (≈1h cached) to find the enterprise-specific ``custom_attrs.id`` whose ``i18n_name`` matches "HRBP" / "HR Business Partner". 2. ``get_user_with_custom_attrs(leaver_feishu_user_id)`` to read the leaver's HRBP custom-attr value (a Feishu user_id pointing at the HRBP). 3. ``get_user_with_custom_attrs(hrbp_feishu_user_id)`` to read the HRBP's email, which we match against ``identity_persons.work_email`` to land an :class:`IdentityPerson` row. The resolved id is written back to ``identity_persons.hrbp_person_id`` so the next dispatch is a single PK lookup. Cache invalidation is implicit — the next call after an HR re-org overwrites the column. Used by ``leaver_blocked.sla_service`` to target SLA URGENT/BREACH notifications at the leaver's actual HRBP rather than the entire HR admin pool. When this resolver returns ``None`` (Feishu has no HRBP configured for the leaver, or the email doesn't match a Helix person), the SLA dispatcher falls back to the it_admin pool and writes an audit row tagged ``hrbp_resolve_failed`` so the gap is visible to ops. Errors do NOT raise — every failure path returns ``None`` plus a structured reason string so the SLA dispatcher can audit "why we fell back" without the SLA tick blowing up. The full audit trail lives in ``audit_logs`` keyed by ``leaver_person_id``. """ from __future__ import annotations import logging import time from dataclasses import dataclass from typing import Any from uuid import UUID from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from app.approval.feishu_client import FeishuApprovalClient from app.core.exceptions import AppException from app.identity.person_models import IdentityPerson from app.integrations.compat import get_feishu_config logger = logging.getLogger(__name__) # Names we accept as "this attr is the HRBP". Feishu admins spell it # differently across tenants; we match case-insensitively against the # i18n_name array. ``in`` over startswith so "HR Business Partner" # matches without false-positives like "HR Director". _HRBP_NAMES: tuple[str, ...] = ( "hrbp", "hr business partner", "hr bp", ) # Cache the resolved (app_id → custom_attrs.id) mapping for one hour. # The HRBP attr id is created once when the tenant configures custom # attributes — it does not change at runtime. An hour TTL is generous # enough that re-discovery on a worker restart is cheap, and short # enough that a tenant who renames the attr sees the change within an # hour without bouncing the pod. _HRBP_ATTR_ID_CACHE: dict[str, tuple[str | None, float]] = {} _HRBP_ATTR_ID_TTL_SECONDS = 60 * 60 @dataclass(slots=True, frozen=True) class HrbpResolution: """Outcome of :func:`resolve_hrbp_for_person`. ``person_id`` set → resolution succeeded; the SLA dispatcher should deliver to that person AND record the resolution path in audit. ``person_id`` None + ``reason`` populated → fall back to it_admin pool; the reason ends up in the audit detail so ops can see why a given SLA tick missed the targeted route (typically: leaver has no Feishu account, no HRBP attr in Feishu, or HRBP email doesn't match any Helix person). """ person_id: UUID | None reason: str | None hrbp_email: str | None = None async def resolve_hrbp_for_person( db: AsyncSession, person: IdentityPerson, ) -> HrbpResolution: """Resolve and (on success) persist the HRBP for one person. Read path: 0. If ``person.hris_hrbp_person_id`` is set AND that row exists + is active, return it. This column is SF-authoritative (populated by the SuccessFactors sync from ``EmpJobRelationships`` type 2448) and is the preferred source — it avoids the Feishu round-trip entirely. ``hris_hrbp_person_id`` and the Feishu-cached ``hrbp_person_id`` are DISTINCT columns; we read the SF one first and never overwrite it from Feishu. 1. Else if ``person.hrbp_person_id`` (Feishu cache) is set AND the cached row still exists + is active, return it (single SELECT, no Feishu round-trip). 2. Otherwise, walk Feishu (custom_attrs → leaver user → HRBP user) to land an email and match it to ``identity_persons``. 3. On success, write back ``person.hrbp_person_id`` so the next call short-circuits at step 1. Never raises. Network / API failures land as ``HrbpResolution`` with ``person_id=None`` and a reason describing what happened. """ # Step 0 — SF-authoritative HRBP wins. The SuccessFactors sync # populates ``hris_hrbp_person_id`` directly from the HRIS work # relationship (type 2448), so when it's present we trust it over # the lazily-Feishu-resolved column and skip the Feishu hops. if person.hris_hrbp_person_id is not None: sf_hrbp = await db.get(IdentityPerson, person.hris_hrbp_person_id) if ( sf_hrbp is not None and sf_hrbp.deleted_at is None and sf_hrbp.employment_status == "active" ): return HrbpResolution( person_id=sf_hrbp.id, reason=None, hrbp_email=sf_hrbp.work_email, ) # SF-resolved HRBP no longer valid (deleted / inactive) — fall # through to the Feishu-cached column / live re-resolution. if person.hrbp_person_id is not None: cached = await db.get(IdentityPerson, person.hrbp_person_id) if ( cached is not None and cached.deleted_at is None and cached.employment_status == "active" ): return HrbpResolution( person_id=cached.id, reason=None, hrbp_email=cached.work_email, ) # Cached HRBP no longer valid (deleted / inactive) — fall through # to live re-resolution and overwrite the stale column. if not person.work_email: return HrbpResolution( person_id=None, reason=f"leaver person {person.id} has no work_email", ) try: cfg = await get_feishu_config(session=db) except AppException as e: return HrbpResolution( person_id=None, reason=f"feishu integration not configured: {e.message}", ) client = FeishuApprovalClient( app_id=str(cfg.get("app_id") or ""), app_secret=str(cfg.get("app_secret") or ""), ) attr_id = await _get_hrbp_attr_id(client) if attr_id is None: return HrbpResolution( person_id=None, reason="no Feishu custom_attr named HRBP found in tenant", ) try: leaver_user_id = await client.resolve_user_id_by_email(person.work_email) except AppException as e: return HrbpResolution( person_id=None, reason=f"leaver Feishu user_id resolve failed: {e.message}", ) except Exception as e: # noqa: BLE001 — Round-49: honour "never raises" # Network-layer errors (httpx.ConnectError, ReadTimeout, # JSONDecodeError, asyncio.TimeoutError) used to propagate # past the AppException catch and break the SLA-escalation # cron loop for every subsequent leaver. The function # docstring guarantees we never raise — extend the catch so # a Feishu DNS hiccup falls through to the it_admin pool # instead of aborting the tick. logger.warning( "feishu_hrbp: resolve_user_id_by_email network error " "for %s: %s", person.work_email, e, ) return HrbpResolution( person_id=None, reason=f"leaver Feishu user_id resolve network error: {e!r}", ) try: leaver_user = await client.get_user_with_custom_attrs(leaver_user_id) except AppException as e: return HrbpResolution( person_id=None, reason=f"leaver Feishu user fetch failed: {e.message}", ) except Exception as e: # noqa: BLE001 — Round-49: honour "never raises" logger.warning( "feishu_hrbp: get_user_with_custom_attrs network error " "for leaver %s: %s", leaver_user_id, e, ) return HrbpResolution( person_id=None, reason=f"leaver Feishu user fetch network error: {e!r}", ) hrbp_feishu_user_id = _extract_hrbp_user_id(leaver_user, attr_id) if hrbp_feishu_user_id is None: return HrbpResolution( person_id=None, reason=( f"leaver {person.work_email} has no HRBP value in Feishu " f"custom_attrs (attr_id={attr_id})" ), ) try: hrbp_user = await client.get_user_with_custom_attrs(hrbp_feishu_user_id) except AppException as e: return HrbpResolution( person_id=None, reason=f"HRBP Feishu user fetch failed: {e.message}", ) except Exception as e: # noqa: BLE001 — Round-49: honour "never raises" logger.warning( "feishu_hrbp: get_user_with_custom_attrs network error " "for hrbp %s: %s", hrbp_feishu_user_id, e, ) return HrbpResolution( person_id=None, reason=f"HRBP Feishu user fetch network error: {e!r}", ) hrbp_email = str(hrbp_user.get("email") or "").strip().lower() if not hrbp_email: return HrbpResolution( person_id=None, reason=( f"HRBP Feishu user {hrbp_feishu_user_id} has no email — " f"cannot match to identity_persons" ), ) hrbp_person = await db.scalar( select(IdentityPerson).where( IdentityPerson.work_email == hrbp_email, IdentityPerson.deleted_at.is_(None), ) ) if hrbp_person is None: return HrbpResolution( person_id=None, reason=( f"HRBP email {hrbp_email} does not match any active Helix " f"person — onboarding/de-provisioning out of sync?" ), hrbp_email=hrbp_email, ) if hrbp_person.employment_status != "active": return HrbpResolution( person_id=None, reason=( f"HRBP {hrbp_email} matched person {hrbp_person.id} but " f"employment_status={hrbp_person.employment_status!r} (not active)" ), hrbp_email=hrbp_email, ) # Cache hit — write back so the next escalation tick is a single PK. person.hrbp_person_id = hrbp_person.id await db.flush() return HrbpResolution( person_id=hrbp_person.id, reason=None, hrbp_email=hrbp_email, ) # ── Internals ──────────────────────────────────────────────── async def _get_hrbp_attr_id(client: FeishuApprovalClient) -> str | None: """Return the Feishu custom-attr id whose i18n_name is "HRBP". Caches per app_id with a 1h TTL. Returns ``None`` when the tenant has no such attribute defined — caller falls back to it_admin pool. Never raises; network errors invalidate the cache and re-attempt on the next call. """ now = time.time() cached = _HRBP_ATTR_ID_CACHE.get(client.app_id) if cached and now < cached[1]: return cached[0] try: items = await client.list_custom_attrs() except AppException as e: logger.warning( "feishu_hrbp: list_custom_attrs failed for app_id=%s: %s", client.app_id, e.message, ) return None except Exception as e: # noqa: BLE001 — network flakiness logger.warning( "feishu_hrbp: list_custom_attrs network error for app_id=%s: %s", client.app_id, e, ) return None for attr in items: if not isinstance(attr, dict): continue names = attr.get("i18n_name") or [] for name in names: if not isinstance(name, dict): continue value = str(name.get("value") or "").strip().lower() if value in _HRBP_NAMES: attr_id = str(attr.get("id") or "").strip() if attr_id: _HRBP_ATTR_ID_CACHE[client.app_id] = ( attr_id, now + _HRBP_ATTR_ID_TTL_SECONDS, ) return attr_id # Cache the negative result so we don't hammer Feishu when the # tenant has not configured HRBP. Same TTL — operator wires it # up, an hour later we re-discover. _HRBP_ATTR_ID_CACHE[client.app_id] = ( None, now + _HRBP_ATTR_ID_TTL_SECONDS, ) return None def _extract_hrbp_user_id( user: dict[str, Any], attr_id: str, ) -> str | None: """Pull the Feishu user_id pointed to by the HRBP custom_attr. Feishu has two user-pointing custom-attr schema types, and the value envelope differs between them: * ``USER`` → ``value.user.user_id`` * ``GENERIC_USER`` → ``value.generic_user.id`` The PingCAP tenant's HRBP attribute is ``GENERIC_USER`` (verified live 2026-06-16 against the prod Feishu tenant), e.g.:: {"id": "C-7143156434032312324", "type": "GENERIC_USER", "value": {"generic_user": {"id": "", "type": 1}}} The original implementation only read ``value.user.user_id`` and so returned ``None`` for every PingCAP user — silently routing every leaver-SLA HRBP escalation to the it_admin fallback pool. We now accept BOTH shapes so the resolver is robust to either schema type. Defensive against missing nodes — every layer is checked because Feishu's response schema is intentionally permissive (a tenant can rename / retype an attr, and we'd see partial dicts). Returns ``None`` when neither shape yields a user id. """ custom_attrs = user.get("custom_attrs") or [] for attr in custom_attrs: if not isinstance(attr, dict): continue if attr.get("id") != attr_id: continue value = attr.get("value") if not isinstance(value, dict): return None # USER shape: value.user.user_id user_node = value.get("user") if isinstance(user_node, dict): hrbp_user_id = str(user_node.get("user_id") or "").strip() if hrbp_user_id: return hrbp_user_id # GENERIC_USER shape: value.generic_user.id (PingCAP tenant) generic_node = value.get("generic_user") if isinstance(generic_node, dict): hrbp_user_id = str(generic_node.get("id") or "").strip() if hrbp_user_id: return hrbp_user_id return None return None __all__ = [ "HrbpResolution", "resolve_hrbp_for_person", ]