"""Feishu (Lark) Contact API driver — outbound identity sync. Pushes Helix's HRIS-of-truth employee + org-tree data into Feishu's ``/contact/v3`` endpoints, so the Feishu directory stays consistent with SuccessFactors via Helix. Backs the single ``feishu_contact`` sync-target kind — internally a two-phase sweep (Phase A reconciles departments, Phase B upserts users + assigns them) but admin-facing as ONE target, ONE switch, ONE cron. Why a separate driver from ``feishu_chat_approval`` even though both talk to the same Feishu open-platform tenant: * Identity sync needs different OAuth scopes (``contact:user.base:write`` + ``contact:department.base:write``) than the approval driver (``approval:*``). Splitting the kind makes the scope diff obvious in the integrations UI — the description spells out exactly which scopes to grant on the Feishu app. * One integration row carries the entire credential surface for a Feishu tenant — admin doesn't fill credentials twice. ``app_id`` / ``app_secret`` are the same as the approval app's. * The driver registers under its own ``kind`` so the credential form + ``test_connection`` are bespoke (the SCIM ``identity_sync_endpoint`` config form with its ``base_url`` + bearer-token fields is the wrong shape for Feishu). Wire-protocol work lives in ``app.domains.identity_sync.runners.feishu_contact`` (user phase) and ``app.domains.identity_sync.runners.feishu_department`` (the internal helper invoked during Phase A — NOT a registered sync kind itself). The driver is catalog + ``test_connection`` only; runners read ``app_id`` / ``app_secret`` off the resolved integration row when constructed. To create the Feishu sync target in the Identity Sync Center, pick this integration row + the ``feishu_contact`` preset. """ from __future__ import annotations import logging from typing import Any from app.approval.feishu_client import FeishuApprovalClient from app.core.exceptions import AppException from app.integrations.driver_base import IntegrationDriver from app.integrations.registry import register_driver from app.integrations.schemas import ( IntegrationFieldInfo, IntegrationTestResult, ) logger = logging.getLogger(__name__) _CONFIG_SCHEMA = [ IntegrationFieldInfo( key="app_id", label="App ID", is_secret=False, required=True, placeholder="cli_xxxxxxxxxxxxxxxx", hint=( "飞书开放平台 > 凭证与基础信息 > App ID。可与「飞书审批」连接器复用同一个" "自建应用,但请确认应用已开通 contact:user.base:write + " "contact:department:base:write + contact:user.email:write 权限。" ), ), IntegrationFieldInfo( key="app_secret", label="App Secret", is_secret=True, required=True, hint="App Secret,仅在轮换时重新粘贴。", ), ] @register_driver class FeishuContactDriver(IntegrationDriver): """Feishu open-platform Contact API — outbound identity sync.""" kind = "identity_sync_feishu" provider = "feishu" display_name = "飞书 (Lark) 通讯录" description = ( "飞书开放平台 /contact/v3 API。支持 feishu_contact 同步目标:内部分两个阶段" "(部门同步 → 用户同步),管理员视角是一个目标一个开关。需要在飞书自建应用" "上开通 contact:user.base:write、contact:user.email:write、" "contact:department.base:write、contact:user.employee_id:readonly、" "contact:custom_attr:readonly 等权限。可与审批连接器复用同一个 app。" ) supports_multi_instance = False config_schema = _CONFIG_SCHEMA def _client(self) -> FeishuApprovalClient: # FeishuApprovalClient is a misnomer — it's a thin wrapper # around the Open Platform auth + /contact/v3 endpoints. Reused # here to share the tenant_access_token cache (class-level dict # keyed by app_id) so identity sync + approval flows on the # same app don't burn separate tokens. Identical app_id = # identical token = single quota cost. return FeishuApprovalClient( app_id=str(self.config.get("app_id") or ""), app_secret=str(self.config.get("app_secret") or ""), ) async def test_connection(self) -> IntegrationTestResult: """Per-scope health probe. All calls are READ-ONLY (no mutation). Probes: 1. ``POST /auth/v3/tenant_access_token/internal`` — credentials valid. 2. ``GET /contact/v3/scopes`` — contact:contact.scope:readonly 3. ``GET /contact/v3/departments?parent_department_id=0&page_size=1`` — contact:department.base:readonly 4. ``GET /contact/v3/custom_attrs?page_size=50`` — contact:custom_attr:readonly (+ surfaces real schema names so admins can cross-reference with mapping_rules) 5. ``POST /contact/v3/users/batch_get_id`` with no emails (dry probe) — verifies the user read endpoint is reachable. Result shape: ok=True when (1) + (2) + (3) all succeed. (4) + (5) failures downgrade to warnings in detail but the connection is still usable for the dept sync target. The user-sync target needs (4) + (5) AND additional write scopes which we don't probe here (a real test fires when the admin clicks Sync Now). `detail` carries: * ``probes``: per-stage {ok, error?} so admin sees exactly which scope is missing. * ``custom_attrs``: list of {id, name} for configured schemas — used by the admin UI to render "your mapping rules reference these schemas, here's what Feishu has". """ client = self._client() probes: dict[str, dict[str, Any]] = {} # ── 1. tenant_access_token ────────────────────────────────── try: await client.get_tenant_access_token() probes["tenant_access_token"] = {"ok": True} except AppException as e: return IntegrationTestResult( ok=False, message=f"凭证无效: {e.code}: {e.message}", detail={"stage": "tenant_access_token"}, ) except Exception as e: # noqa: BLE001 return IntegrationTestResult( ok=False, message=f"凭证无效: {type(e).__name__}: {e}", detail={"stage": "tenant_access_token"}, ) # ── 2. contact:contact.scope:readonly ─────────────────────── try: await client.request( "GET", "/contact/v3/scopes?user_id_type=user_id&page_size=1", ) probes["scopes_read"] = {"ok": True} except AppException as e: probes["scopes_read"] = { "ok": False, "error": f"{e.code}: {e.message}", "hint": ( "缺少 contact:contact.scope:readonly。请到 飞书开放平台 → " "我的应用 → 权限管理,开通后等 1-2 分钟生效。" ), } except Exception as e: # noqa: BLE001 probes["scopes_read"] = {"ok": False, "error": f"{type(e).__name__}: {e}"} # ── 3. contact:department.base:readonly ───────────────────── try: await client.request( "GET", "/contact/v3/departments?parent_department_id=0" "&department_id_type=open_department_id&page_size=1", ) probes["department_read"] = {"ok": True} except AppException as e: probes["department_read"] = { "ok": False, "error": f"{e.code}: {e.message}", "hint": ( "缺少 contact:department.base:readonly。这个权限是部门同步" "的硬前置 —— 没有它部门 sync 完全无法工作。" ), } except Exception as e: # noqa: BLE001 probes["department_read"] = {"ok": False, "error": f"{type(e).__name__}: {e}"} # ── 4. contact:custom_attr:readonly ───────────────────────── custom_attr_summary: list[dict[str, Any]] = [] try: attrs_resp = await client.request( "GET", "/contact/v3/custom_attrs?page_size=50", ) items = (attrs_resp or {}).get("items") or [] for it in items: if not isinstance(it, dict): continue sid = it.get("id") attr_type = it.get("type") i18n = it.get("i18n_name") or [] # Pick the default-locale name for the admin UI. name = None if isinstance(i18n, list): for loc in i18n: if isinstance(loc, dict) and loc.get("locale") == "default": name = loc.get("value") break if not name: for loc in i18n: if isinstance(loc, dict): val = loc.get("value") if isinstance(val, str) and val: name = val break if sid: custom_attr_summary.append({ "id": sid, "type": attr_type, "name": name, }) probes["custom_attr_read"] = { "ok": True, "count": len(custom_attr_summary), } except AppException as e: probes["custom_attr_read"] = { "ok": False, "error": f"{e.code}: {e.message}", "hint": ( "缺少 contact:custom_attr:readonly。用户同步的 custom_attrs " "字段(PinYin / Company Entity / Location 等)无法自动适配," "推送会按裸字段名走 — 飞书侧 schema 名称必须严格匹配。" ), } except Exception as e: # noqa: BLE001 probes["custom_attr_read"] = { "ok": False, "error": f"{type(e).__name__}: {e}", } # ── 5. user read endpoint reachability ────────────────────── # POST with empty emails — Feishu returns code 0 + empty list, # confirms the user read scope without enumerating any user. try: await client.request( "POST", "/contact/v3/users/batch_get_id?user_id_type=user_id", json_body={"emails": []}, ) probes["user_read"] = {"ok": True} except AppException as e: probes["user_read"] = { "ok": False, "error": f"{e.code}: {e.message}", "hint": ( "缺少 contact:user.base:readonly 或 contact:user.email:readonly。" "用户同步无法找到飞书侧已存在的账号,会全部当成新人 POST 创建 " "→ 创建一堆重复账号。这个权限必须开。" ), } except Exception as e: # noqa: BLE001 probes["user_read"] = {"ok": False, "error": f"{type(e).__name__}: {e}"} # ── Aggregate ──────────────────────────────────────────────── hard_required = {"scopes_read", "department_read", "user_read"} missing = [k for k in hard_required if not probes[k]["ok"]] if missing: return IntegrationTestResult( ok=False, message=( f"飞书权限不全,缺少必备 scope: {missing}。" "前往 飞书开放平台 → 我的应用 → 权限管理 一一对应开通。" ), detail={ "probes": probes, "custom_attrs": custom_attr_summary, }, ) # All hard-required green; custom_attr_read is soft. soft_ok = probes["custom_attr_read"].get("ok") is True msg = ( "OK · tenant_access_token 获取成功 · " f"custom_attrs configured: {len(custom_attr_summary)}" ) if not soft_ok: msg += " · ⚠ custom_attr_read 缺,custom_attrs 同步效果受限" return IntegrationTestResult( ok=True, message=msg, detail={ "probes": probes, "custom_attrs": custom_attr_summary, }, )