"""Feishu (Lark) Contact API driver — INBOUND partner roster. The third Feishu driver, and the only one that points at somebody else's tenant. The other two talk to OUR tenant: * ``feishu_chat_approval`` — submits approvals, sends cards. * ``feishu_contact`` (``identity_sync_feishu``) — PUSHES our HRIS truth into our own Feishu directory. This one PULLS a partner company's roster out of THEIR tenant. Concrete case: CBG. Their staff carry ``@pingcap.cn`` addresses in their own Lark tenant while holding ``@pingcap.com`` Google accounts, vendor seats and GitHub handles in ours — so they surface in every "we can't match this to an employee" bucket we have, and until now somebody classified them by hand (58 rows in ``external_account_registry`` at the time this shipped, 8 more Google orphans still untriaged). Two consequences follow from "somebody else's tenant" and both are load-bearing: 1. **Read-only, structurally.** The driver subclasses ``PartnerDirectoryDriver``, whose interface has no write verb at all. We are a guest in that directory; there must be no code path from here that mutates it. 2. **Their custom-attribute ids are theirs.** CBG's "GitHub ID" field is ``C-7356820809171140610``; ours is ``C-6934211695879389211``. Same display name, different tenant, different id namespace — which is also the check that proved these are two separate tenants rather than one app on ours. So the driver resolves the attribute **by display name** (``github_id_attr``, default ``GitHub ID``) against ``/contact/v3/custom_attrs`` at run time, and only falls back to a literally-configured id. Hard-coding the id would break the day CBG recreates the field. Scopes required on the partner's app. This list was established empirically against ``cli_aa1cca9909a3dbb4`` on 2026-09-01 by granting them one at a time and re-reading a real user, because **Feishu does not tell you which one is missing** (see below). Every one of them is load bearing: ===================================== ========================== Scope What stops being returned ===================================== ========================== ``contact:contact.base:readonly`` everything (the API itself 403s) ``contact:department.base:readonly`` department ``name`` ``contact:department.organize:readonly`` ``parent_department_id`` → dept path ``contact:user.base:readonly`` ``name`` / ``en_name`` ``contact:user.email:readonly`` ``email`` ``contact:user.employee:readonly`` ``enterprise_email``, ``custom_attrs`` (**the GitHub ID**), ``job_title``, ``employee_no``, ``status`` ``contact:user.department:readonly`` ``department_ids`` ===================================== ========================== The console no longer offers the old ``contact:contact:readonly_as_app`` umbrella — searching for it returns nothing — so the granular list is the only route. **The scope list is only half of it.** Feishu also has a *data* scope ("通讯录权限范围") set in the admin console, and the two fail differently. A missing data scope answers ``40004 no dept authority``; a missing API scope answers ``code: 0`` with the field simply ABSENT from the response body. The first probe of this app returned a perfectly successful ``GET /contact/v3/users/{open_id}`` carrying only ``open_id`` and ``union_id`` — no name, no email, no custom attrs. Nothing errored. Two grants later it returned name but still no email, again with ``code: 0``. That is why :meth:`test_connection` asserts on *fields present in a real user record* rather than on HTTP status: a green connection test that only checked status codes would have declared this app healthy at every one of those stages. What the first complete pull found (2026-09-01, CBG): 75 departments, 381 people, all in service, every department readable. 380 of them carry an ``@pingcap.cn`` address. 236 end up with a usable GitHub handle — see :func:`_clean_github_login` for why that is not the same as the 357 who have something in the field. """ from __future__ import annotations import asyncio import logging import re from typing import Any from app.approval.feishu_client import FeishuApprovalClient from app.core.exceptions import AppException from app.integrations.driver_base import ( PartnerDirectoryDriver, PartnerDirectoryMemberRecord, PartnerDirectorySnapshot, ) from app.integrations.registry import register_driver from app.integrations.schemas import ( IntegrationFieldInfo, IntegrationTestResult, ) logger = logging.getLogger(__name__) # Feishu's page cap for both endpoints we walk. _PAGE_SIZE = 50 # Departments are read concurrently; a partner tenant is a few hundred # people, so this is politeness rather than throughput. Feishu's contact # read endpoints are rate-limited at 50 QPS tenant-wide and we share the # token with nothing else, but a burst of 100 parallel calls is still a # good way to get throttled for no gain. _DEPT_CONCURRENCY = 4 # ``40004`` = "no dept authority": the app's *data* scope excludes this # department. On a whole-tenant sync that means an incomplete read, not # an error — some partners deliberately withhold a subtree. _ERR_NO_DEPT_AUTHORITY = 40004 _DEFAULT_GITHUB_ATTR = "GitHub ID" # GitHub's own rule for a login: alphanumerics and single inner hyphens, # 1–39 chars. Used as a REJECT filter, not a guess — anything that fails # it cannot be a handle, whatever it is. _GITHUB_LOGIN_RE = re.compile(r"^[A-Za-z0-9](?:[A-Za-z0-9]|-(?=[A-Za-z0-9])){0,38}$") # A profile URL pasted into the field. ``github.cn`` is in here because a # real CBG row has it — somebody typo'd the TLD and the account behind it # is perfectly real. _GITHUB_URL_RE = re.compile( r"^https?://(?:www\.)?github\.(?:com|cn)/([^/?#\s]+)/?$", re.IGNORECASE, ) def _clean_github_login(raw: str) -> str: """A usable GitHub handle out of a free-text field, or "". This field is typed by hand into the partner's directory and the 2026-09-01 full pull says so plainly — of 381 people, 357 had something in it and only 230 of those passed GitHub's own login syntax as typed: * 118 wrote the literal 「无」("none"), * 6 pasted the whole profile URL (one of them ``github.cn``), * 3 pasted a QQ email address, * 1 prefixed it with ``@``. Dropping everything non-conforming would throw away the ones that are real accounts in the wrong shape; keeping everything would put 118 rows into the index under the key 「无」and render a fake GitHub tag next to 118 names. So: unwrap the two known wrappers, then apply GitHub's own syntax rule as the gate — 230 usable as typed, **236** after unwrapping. """ value = (raw or "").strip() if not value: return "" url_match = _GITHUB_URL_RE.match(value) if url_match: value = url_match.group(1) value = value.lstrip("@").strip() if not _GITHUB_LOGIN_RE.match(value): return "" return value _CONFIG_SCHEMA = [ IntegrationFieldInfo( key="app_id", label="App ID", is_secret=False, required=True, placeholder="cli_xxxxxxxxxxxxxxxx", hint=( "合作方飞书/Lark 自建应用的 App ID —— 注意这是**对方租户**的应用," "不是我们自己的那个。需要对方开通 contact:contact:readonly_as_app" "(或 contact:department.base:readonly + contact:user.base:readonly " "+ contact:user.email:readonly + contact:user.id:readonly)、" "contact:custom_attr:readonly,并把「通讯录权限范围」放开到全部成员。" ), ), IntegrationFieldInfo( key="app_secret", label="App Secret", is_secret=True, required=True, hint="App Secret,仅在轮换时重新粘贴。", ), IntegrationFieldInfo( key="partner_email_domain", label="合作方邮箱域", is_secret=False, required=True, placeholder="pingcap.cn", hint=( "对方员工在自己租户里用的邮箱域(CBG 是 pingcap.cn)。" "打标时用「邮箱名跨域匹配」—— 对方的 zhang.san@<这个域> 认定为" "我们这边的 zhang.san@<本司域>。填错会让匹配整体失效,不会误标。" ), ), IntegrationFieldInfo( key="local_email_domain", label="本司邮箱域", is_secret=False, required=True, placeholder="pingcap.com", hint="我们这边的邮箱域,跨域匹配的另一端。", ), IntegrationFieldInfo( key="github_id_attr", label="GitHub ID 自定义字段名", is_secret=False, required=False, placeholder=_DEFAULT_GITHUB_ATTR, hint=( "对方通讯录里存 GitHub 账号的自定义字段**显示名**(默认 " f"「{_DEFAULT_GITHUB_ATTR}」)。按名字解析而不是写死 id —— " "自定义字段 id 是每个租户各自一套的。留空用默认值。" ), ), ] def _first_i18n(entry: dict[str, Any]) -> str: """Display name of a custom-attr definition. Feishu returns ``i18n_name`` as a list of ``{locale, value}``; the ``default`` locale is the one an admin actually typed. Some tenants ship empty strings for the localised variants (CBG's 云学堂 fields do), so an "any non-empty value" scan would pick a blank. """ items = entry.get("i18n_name") if not isinstance(items, list): return "" by_locale = { str(i.get("locale") or ""): str(i.get("value") or "") for i in items if isinstance(i, dict) } for locale in ("default", "zh_cn", "en_us"): if by_locale.get(locale): return by_locale[locale] return next((v for v in by_locale.values() if v), "") def _attr_text(attr: dict[str, Any]) -> str: """Scalar text out of one ``custom_attrs`` entry. TEXT attributes carry ``value.text``; HREF carries ``value.text`` + ``value.url``; enums carry ``value.option_value``. Anything else (GENERIC_USER, PICTURE_ENUM) has no scalar reading and is skipped — returning ``str(value)`` there would put a dict repr into a column that the matcher compares against GitHub handles. """ value = attr.get("value") if not isinstance(value, dict): return "" for key in ("text", "option_value", "value"): raw = value.get(key) if isinstance(raw, str) and raw.strip(): return raw.strip() return "" @register_driver class FeishuPartnerDirectoryDriver(PartnerDirectoryDriver): """A partner company's Feishu/Lark directory, read-only.""" provider = "feishu" display_name = "合作方通讯录 (飞书/Lark)" description = ( "只读拉取**合作方**飞书租户的员工名册(姓名 / 企业邮箱 / GitHub ID " "自定义字段 / 部门 / 工号 / 在职状态),用于把我们这边认不出人的" "账号(Google 孤儿账号、各厂商 external 坐席、GitHub 上没关联到人的 " "handle)标记成合作方员工。不写对方通讯录 —— 接口层就没有写方法。" "需要对方应用开通 contact:contact:readonly_as_app(或等价的细粒度读" "权限)+ contact:custom_attr:readonly,并把「通讯录权限范围」放开。" ) # One row per partner company. A second partner is a second row, not # a second driver. supports_multi_instance = True config_schema = _CONFIG_SCHEMA # ── config accessors ─────────────────────────────────────────── @property def partner_email_domain(self) -> str: return str(self.config.get("partner_email_domain") or "").strip().lower().lstrip("@") @property def local_email_domain(self) -> str: return str(self.config.get("local_email_domain") or "").strip().lower().lstrip("@") @property def github_id_attr(self) -> str: return str( self.config.get("github_id_attr") or _DEFAULT_GITHUB_ATTR, ).strip() def _client(self) -> FeishuApprovalClient: # ``FeishuApprovalClient`` is a misnomer inherited from the # approval driver — it is a thin Open-Platform wrapper (auth + # arbitrary /contact/v3 calls) and its tenant_access_token cache # is keyed by app_id, so the partner's token can never collide # with our own tenant's. return FeishuApprovalClient( app_id=str(self.config.get("app_id") or ""), app_secret=str(self.config.get("app_secret") or ""), ) # ── custom attribute resolution ──────────────────────────────── async def _resolve_github_attr_id( self, client: FeishuApprovalClient, ) -> str | None: """Partner-tenant id of the "GitHub ID" custom attribute. Resolved by display name every run. An admin who typed a literal ``C-…`` id into ``github_id_attr`` gets it used verbatim — an escape hatch for a tenant with two identically-named fields. """ configured = self.github_id_attr if configured.startswith("C-"): return configured try: attrs = await client.list_custom_attrs() except AppException: logger.warning( "partner_directory[%s]: cannot read custom_attrs — " "GitHub ID will be missing from this pull", self.integration_id, ) return None wanted = configured.casefold() for attr in attrs: if _first_i18n(attr).strip().casefold() == wanted: return str(attr.get("id") or "") or None logger.warning( "partner_directory[%s]: no custom attribute named %r in the " "partner tenant (found: %s)", self.integration_id, configured, [_first_i18n(a) for a in attrs], ) return None # ── roster pull ──────────────────────────────────────────────── async def list_members(self) -> PartnerDirectorySnapshot: client = self._client() github_attr_id = await self._resolve_github_attr_id(client) snapshot = PartnerDirectorySnapshot() if github_attr_id is None: snapshot.notes.append( f"未能在对方通讯录里解析到自定义字段「{self.github_id_attr}」" f"—— 本次同步不带 GitHub ID", ) departments = await self._list_departments(client, snapshot) dept_names = { d["open_department_id"]: d["name"] for d in departments if d.get("open_department_id") } dept_parent = { d["open_department_id"]: d.get("parent_department_id") or "" for d in departments if d.get("open_department_id") } # Department "0" is the tenant root: people who belong to no # department at all live there and are invisible to a walk over # the department list. It takes the numeric id type, not # open_department_id (which rejects "0" outright). targets: list[tuple[str, str]] = [("department_id", "0")] targets += [ ("open_department_id", d["open_department_id"]) for d in departments if d.get("open_department_id") ] by_external_id: dict[str, PartnerDirectoryMemberRecord] = {} sem = asyncio.Semaphore(_DEPT_CONCURRENCY) async def _one(id_type: str, dept_id: str) -> list[dict[str, Any]]: async with sem: return await self._list_users_in_department( client, id_type, dept_id, snapshot, ) results = await asyncio.gather( *(_one(t, d) for t, d in targets), return_exceptions=True, ) for outcome in results: if isinstance(outcome, BaseException): # A department we could not read at all. Recorded as # incompleteness, never swallowed — an incomplete # snapshot must not be allowed to retire rows. snapshot.complete = False snapshot.departments_unreadable += 1 logger.warning( "partner_directory[%s]: department read failed: %s", self.integration_id, outcome, ) continue for raw in outcome: record = self._to_record(raw, github_attr_id, dept_names, dept_parent) if record is None: continue # A person in three departments comes back three times. # Keep the first and merge the department lists so the # stored path reflects every posting. existing = by_external_id.get(record.external_id) if existing is None: by_external_id[record.external_id] = record else: for did in record.department_ids: if did not in existing.department_ids: existing.department_ids.append(did) snapshot.members = list(by_external_id.values()) return snapshot async def _list_departments( self, client: FeishuApprovalClient, snapshot: PartnerDirectorySnapshot, ) -> list[dict[str, Any]]: """Every department under the root, flattened. ``fetch_child=true`` makes Feishu return the whole subtree in one paginated walk instead of a recursive descent. """ items: list[dict[str, Any]] = [] page_token: str | None = None while True: qs = ( "parent_department_id=0&fetch_child=true" f"&page_size={_PAGE_SIZE}" "&department_id_type=open_department_id" ) if page_token: qs = f"{qs}&page_token={page_token}" try: data = await client.request("GET", f"/contact/v3/departments?{qs}") except AppException as exc: # The partner withheld the department tree. Everything # still reachable via the root pseudo-department is kept, # but the snapshot is explicitly incomplete. snapshot.complete = False snapshot.notes.append(f"部门列表读取失败:{exc.message}") logger.warning( "partner_directory[%s]: department list failed: %s", self.integration_id, exc.message, ) return items for entry in data.get("items") or []: if isinstance(entry, dict): status = entry.get("status") if isinstance(status, dict) and status.get("is_deleted"): continue items.append(entry) if not data.get("has_more"): break page_token = data.get("page_token") or None if not page_token: break snapshot.departments_scanned = len(items) return items async def _list_users_in_department( self, client: FeishuApprovalClient, id_type: str, dept_id: str, snapshot: PartnerDirectorySnapshot, ) -> list[dict[str, Any]]: items: list[dict[str, Any]] = [] page_token: str | None = None while True: qs = ( f"department_id={dept_id}" f"&department_id_type={id_type}" "&user_id_type=open_id" f"&page_size={_PAGE_SIZE}" ) if page_token: qs = f"{qs}&page_token={page_token}" try: data = await client.request( "GET", f"/contact/v3/users/find_by_department?{qs}", ) except AppException as exc: from app.approval.feishu_client import parse_feishu_error_code if parse_feishu_error_code(exc) == _ERR_NO_DEPT_AUTHORITY: # Deliberately withheld subtree — expected on a # partner tenant, still an incomplete read. snapshot.complete = False snapshot.departments_unreadable += 1 return items raise for entry in data.get("items") or []: if isinstance(entry, dict): items.append(entry) if not data.get("has_more"): break page_token = data.get("page_token") or None if not page_token: break return items def _to_record( self, raw: dict[str, Any], github_attr_id: str | None, dept_names: dict[str, str], dept_parent: dict[str, str], ) -> PartnerDirectoryMemberRecord | None: open_id = str(raw.get("open_id") or "").strip() if not open_id: return None status = raw.get("status") status = status if isinstance(status, dict) else {} # ``is_activated`` is "has logged in at least once", NOT # employment — a new hire who has not opened Lark yet is # is_activated=false and very much on the payroll. Employment is # is_resigned / is_exited; treating activation as employment # would drop every partner hire from their first day. departed = bool(status.get("is_resigned")) or bool(status.get("is_exited")) custom: dict[str, str] = {} github_login: str | None = None for attr in raw.get("custom_attrs") or []: if not isinstance(attr, dict): continue text = _attr_text(attr) if not text: continue attr_id = str(attr.get("id") or "") # The raw value is kept in ``custom_attrs`` even when the # cleaner rejects it — an admin looking at 「无」in the roster # can tell "nobody filled this in" apart from "we failed to # parse what they filled in". custom[attr_id] = text if github_attr_id and attr_id == github_attr_id: github_login = _clean_github_login(text) or None dept_ids = [ str(d) for d in (raw.get("department_ids") or []) if str(d).strip() ] return PartnerDirectoryMemberRecord( external_id=open_id, union_id=str(raw.get("union_id") or "") or None, partner_user_id=str(raw.get("user_id") or "") or None, employee_no=str(raw.get("employee_no") or "") or None, name=str(raw.get("name") or "") or None, en_name=str(raw.get("en_name") or "") or None, email=str(raw.get("enterprise_email") or "") or None, secondary_email=str(raw.get("email") or "") or None, job_title=str(raw.get("job_title") or "") or None, department_ids=dept_ids, department_path=_dept_path(dept_ids, dept_names, dept_parent), github_login=github_login, is_active=not departed, is_suspended=bool(status.get("is_frozen")), status_raw={k: bool(v) for k, v in status.items()}, custom_attrs=custom, ) # ── health ───────────────────────────────────────────────────── async def test_connection(self) -> IntegrationTestResult: """Probe every scope the sync depends on, and assert on FIELDS. Feishu answers a missing API scope with ``code: 0`` and the field simply absent, so a status-code-only check reports a healthy connection for an app that cannot read a single name. Each probe below therefore states what it expects to see in the body. """ client = self._client() probes: dict[str, dict[str, Any]] = {} def _mark(name: str, ok: bool, **extra: Any) -> None: probes[name] = {"ok": ok, **extra} try: await client.get_tenant_access_token() _mark("credentials", True) except AppException as exc: return IntegrationTestResult( ok=False, message=f"凭据无效:{exc.message}", detail={"probes": {"credentials": {"ok": False, "error": exc.message}}}, ) # 1. contact scope — what the app is allowed to see at all. scoped_departments = 0 scoped_users = 0 try: data = await client.request("GET", "/contact/v3/scopes?page_size=50") scoped_departments = len(data.get("department_ids") or []) scoped_users = len(data.get("user_ids") or []) _mark( "contact_scope", True, departments=scoped_departments, users=scoped_users, ) except AppException as exc: _mark("contact_scope", False, error=exc.message) # 2. departments — the data scope. This is the probe that # catches "通讯录权限范围 was never opened up", which no # amount of API-scope granting fixes. try: await client.request( "GET", "/contact/v3/departments?parent_department_id=0" "&page_size=1&department_id_type=open_department_id", ) _mark("departments", True) except AppException as exc: _mark("departments", False, error=exc.message) # 3. custom attrs — is the GitHub ID field resolvable by name? github_attr_id: str | None = None try: attrs = await client.list_custom_attrs() names = [_first_i18n(a) for a in attrs] wanted = self.github_id_attr.casefold() github_attr_id = next( ( str(a.get("id")) for a in attrs if _first_i18n(a).strip().casefold() == wanted ), None, ) _mark( "custom_attrs", True, available=names, github_attr=self.github_id_attr, github_attr_id=github_attr_id, ) except AppException as exc: _mark("custom_attrs", False, error=exc.message) # 4. a REAL user record — the only probe that can tell a granted # field scope from a silently-dropped one. sample_fields: list[str] = [] try: data = await client.request( "GET", "/contact/v3/users/find_by_department?department_id=0" "&department_id_type=department_id&user_id_type=open_id" "&page_size=1", ) items = [i for i in (data.get("items") or []) if isinstance(i, dict)] if items: sample_fields = sorted(items[0].keys()) _mark( "user_fields", bool({"name", "enterprise_email"} & set(sample_fields)), fields=sample_fields, sampled=len(items), ) except AppException as exc: _mark("user_fields", False, error=exc.message) missing: list[str] = [] if not probes.get("departments", {}).get("ok"): missing.append( "部门读取失败 —— 多半是「通讯录权限范围」没放开(飞书报 40004)," "加 API 权限解决不了,要在管理后台把可见范围改成全部成员" ) user_probe = probes.get("user_fields", {}) if not user_probe.get("ok"): got = user_probe.get("fields") missing.append( "读不到用户的姓名/企业邮箱 —— 缺 contact:user.base:readonly / " "contact:user.email:readonly。注意飞书**不报错**,只是把字段" f"从 response 里去掉了(本次实际拿到的字段:{got or '无'})" ) if github_attr_id is None and probes.get("custom_attrs", {}).get("ok"): missing.append( f"对方通讯录里没有名为「{self.github_id_attr}」的自定义字段 —— " "GitHub 打标会整体失效" ) if not self.partner_email_domain or not self.local_email_domain: missing.append("合作方邮箱域 / 本司邮箱域未填 —— 跨域邮箱名匹配无法进行") ok = not missing return IntegrationTestResult( ok=ok, message=( f"连接正常;授权范围:{scoped_departments} 个部门 / " f"{scoped_users} 个成员" if ok else ";".join(missing) ), detail={ "probes": probes, "github_attr_id": github_attr_id, "partner_email_domain": self.partner_email_domain, "local_email_domain": self.local_email_domain, }, ) def _dept_path( dept_ids: list[str], dept_names: dict[str, str], dept_parent: dict[str, str], ) -> list[str]: """Outermost-first name chain for the person's FIRST department. One chain, not one per posting: the path exists so a human reading the roster can place somebody, and three interleaved chains place nobody. Guarded against a cycle in the parent map (a partner tenant mid-reorg has produced them) by bounding the walk. """ if not dept_ids: return [] chain: list[str] = [] seen: set[str] = set() current = dept_ids[0] while current and current in dept_names and current not in seen: seen.add(current) chain.append(dept_names[current]) current = dept_parent.get(current) or "" return list(reversed(chain))