"""Thin HTTP wrapper around the Feishu Open Platform approval v4 API. Credentials are **injected** via the constructor — the integration framework resolves them from the integration row and hands them off here. No module-level credential reads. Includes :meth:`ensure_subscribed` because Feishu's per-approval instance events only fire after the app explicitly subscribes to each ``approval_code``. The driver calls this on every ``create_instance`` and whenever it fetches a template, so new approval_codes added in the config center start firing callbacks on first use without any manual step. """ from __future__ import annotations import asyncio import hashlib import json import logging import re import time from datetime import UTC, datetime, timedelta from typing import Any from urllib.parse import quote from uuid import uuid4 import httpx from fastapi import UploadFile from app.core.credential_redaction import redact_credentials from app.core.exceptions import AppException from app.core.upload import read_upload_bounded logger = logging.getLogger(__name__) # Same-tenant corporate email suffixes that differ only by the .com/.com.cn # tail: Helix stores work_email as ``@pingcap.com`` while many Feishu # accounts are registered under ``@pingcap.com.cn``. Used to resolve a # person across the two when an exact-address lookup misses. _PINGCAP_EMAIL_SIBLING: dict[str, str] = { "pingcap.com": "pingcap.com.cn", "pingcap.com.cn": "pingcap.com", } # Cap concurrent ``GET /contact/v3/users/{id}`` calls so a wide # ``batch_get_user_names`` call (admin opens a modal for an approval # with 20 approvers) doesn't burst Feishu rate limits or stall the # event loop. 5 is conservative — Feishu's per-tenant QPS budget on # contact endpoints is in the tens. _USER_LOOKUP_CONCURRENCY = 5 # ``POST /contact/v3/users/batch_get_id`` caps the ``emails`` array at # 50. Overrunning it fails the WHOLE request with 99992402 and names # nothing about length (same class of trap as the 50-char ``uuid`` cap # on ``im/v1/messages``), so the chunking lives here at the wire # contract rather than in each caller. _BATCH_GET_ID_CHUNK = 50 # ``page_size`` ceiling on ``approval/v4/instances/query`` and # ``approval/v4/tasks/search``. Feishu rejects anything larger with a # field-validation error rather than clamping. _APPROVAL_SEARCH_MAX_PAGE_SIZE = 200 _FEISHU_APPROVAL_SUBSCRIPTION_EXISTS_CODE = 1390007 _MAX_APPROVAL_IMAGE_BYTES = 10 * 1024 * 1024 _MAX_APPROVAL_FILE_BYTES = 50 * 1024 * 1024 # ── Shared error-code parser ──────────────────────────────────────── # FeishuApprovalClient formats every non-zero API response into an # AppException whose ``message`` starts with ``"Feishu API error :"`` # (see ``_request`` below). Multiple callers — feishu_software's # ``_feishu_error_code``, the identity_sync feishu_contact / # feishu_department runners — re-parse the code to drive idempotent # short-circuits (already-frozen / already-deleted / not-found). # Anchored on the documented prefix so a numeric run inside the API's # localised ``msg`` body can't be mis-read as the API code. _FEISHU_ERROR_RE = re.compile(r"Feishu API error (\d{4,8}):") # ``POST /im/v1/messages`` caps its dedup key (``uuid``) at 50 characters and # rejects the WHOLE request with 99992402 "field validation failed" when it is # longer — before it even validates the receiver. A caller composing a readable # key from a prefix plus a row id overshoots easily (``assistant-proposal-`` + # a 36-char UUID is 55), and the failure is total and silent-looking: the send # raises, the feature degrades, and nothing in the message names the length. # Verified against the live API 2026-08-17: 55 chars → 99992402, 15 chars → # the expected receiver error. _FEISHU_MESSAGE_UUID_MAX = 50 def _message_dedup_field(idempotency_key: str | None) -> dict[str, str]: """Render the ``uuid`` dedup field, shortening over-long keys safely. Truncation is not an option: two keys sharing a 50-char prefix would collapse into one, and Feishu would silently drop the second message as a duplicate. A digest keeps the property that actually matters — the same logical key maps to the same wire value, different keys almost surely do not — while always fitting the field. """ key = (idempotency_key or "").strip() if not key: return {} if len(key) <= _FEISHU_MESSAGE_UUID_MAX: return {"uuid": key} return {"uuid": hashlib.sha256(key.encode("utf-8")).hexdigest()[:32]} def parse_feishu_error_code(exc: AppException) -> int | None: """Extract the numeric Feishu API error code from an AppException raised by :class:`FeishuApprovalClient`. Returns None when the message doesn't carry the canonical ``Feishu API error :`` prefix (i.e. the exception was raised from somewhere else, or the format has changed). Used by Helix runners + drivers to decide whether a given Feishu error is idempotent (e.g. 230002 already-frozen → success) vs a real failure. Single source of truth — runners should NOT duplicate this regex. """ m = _FEISHU_ERROR_RE.search(exc.message or "") if not m: return None # No try/except — the ``\d{4,8}`` capture group is regex-guaranteed # to be int-parseable. A failure here would be a real bug, not a # graceful-degradation case. return int(m.group(1)) # Feishu API numeric codes that indicate a SYSTEMIC failure — bad # credentials, expired tenant_access_token, scope missing, tenant-wide # rate limit. Identity-sync runners use ``is_systemic_feishu_failure`` # below (NOT this set directly) to decide whether to halt the whole # sweep with one alert vs burn rate limit recording N per-person # failures. The set is best-effort: Feishu's docs don't always publish # the exact code list, so the function ALSO short-circuits on # ``FEISHU_TOKEN_ERROR`` exception codes (which carry no numeric code # at all — raised before the request even goes out). # # Code references gleaned from: # - feishu_software.py error-code tables (PingCAP-observed in prod) # - Feishu open-platform docs published code list _FEISHU_SYSTEMIC_NUMERIC_CODES: frozenset[int] = frozenset({ 99991661, # invalid tenant_access_token (token rotation / wrong app) 99991671, # tenant-wide rate limit exceeded # 99991672 is NOT a rate limit — it is "Access denied, the app has not # been granted " (the comment here said "tenant-wide frequency # limit" until 2026-08-31, which is 99991671's meaning; the mislabel # cost a session a wrong root-cause read). It stays in the systemic set # because on a LOOKUP path a missing read scope means nothing in the # sweep can work, so halting once beats N identical per-person rows. # WRITE paths must classify it with ``is_feishu_permission_failure`` # FIRST and return a per-person ``blocked`` instead: a scope is granted # per API surface, so a missing Directory *create* scope must not halt # the Contact *update* path that is still working fine for everyone # else (2026-08-31 — the create surface was dead for 12 days while # updates ran green). 99991672, # access denied — app lacks the required scope 91402, # invalid app_id (modern endpoints) 91403, # invalid app_secret (modern endpoints) }) # Subset of the above that means "a human must grant a scope in the Feishu # admin console and PUBLISH A NEW APP VERSION" — not bad credentials, not a # transient blip. Retrying changes nothing until someone acts, so callers on # a write path turn this into ``blocked`` (with the vendor's own remediation # link, which 99991672 always carries) rather than ``failed``. _FEISHU_PERMISSION_NUMERIC_CODES: frozenset[int] = frozenset({ 99991672, }) def is_systemic_feishu_failure(exc: AppException) -> bool: """True when the exception indicates a sweep-halt-worthy condition. Caller (identity_sync user / dept runners) propagates these out of per-record loops so one alert fires instead of N noisy state rows written under the same bad credentials. Detects two shapes: 1. ``FEISHU_TOKEN_ERROR`` exception code — raised before any per-request call when ``get_tenant_access_token`` itself fails. Has no numeric payload, so the regex parser returns None for it; we detect via the exception code string. 2. ``FEISHU_API_ERROR`` exception code + numeric code in the documented systemic set. Per-record failures (user not found, dept not found, field validation failed, idempotent already-frozen, etc.) return False so the runner continues processing the rest of the sweep. """ if exc.code == "FEISHU_TOKEN_ERROR": return True return parse_feishu_error_code(exc) in _FEISHU_SYSTEMIC_NUMERIC_CODES def is_feishu_permission_failure(exc: AppException) -> bool: """True when Feishu refused because the APP LACKS A SCOPE (99991672). Distinct from :func:`is_systemic_feishu_failure` (bad credentials / rate limit) even though the code is in both sets: the remedy here is a human granting the scope at ``open.feishu.cn/app//auth`` **and publishing a new app version** — a scope that is merely "开通" but not released does not take effect. Nothing the sweep does changes the outcome, so a write path should record ``blocked`` (a real, actionable state that alerts once and stops reddening every run) rather than ``failed`` (which reads as "transient, will retry"). Scope is granted PER API SURFACE. A missing Directory create scope says nothing about the Contact update scope, which is why the caller must not escalate this to a sweep halt. """ if exc.code == "FEISHU_TOKEN_ERROR": return False return parse_feishu_error_code(exc) in _FEISHU_PERMISSION_NUMERIC_CODES # Backward-compat alias for callers that import the old name (the # identity_sync feishu_contact runner). The set itself is no longer # the authoritative check — :func:`is_systemic_feishu_failure` is. FEISHU_AUTH_FAILURE_CODES = _FEISHU_SYSTEMIC_NUMERIC_CODES class FeishuApprovalClient: """One instance per driver invocation (driver is per-request). Class-level token cache keyed by app_id so multiple requests in the same process don't hammer the tenant_access_token endpoint. The cache is invalidated when the app_id changes (admin rotated the app credentials in the integration config). """ _cached_tokens: dict[str, tuple[str, float]] = {} # (app_id, approval_code) pairs we've already POSTed /subscribe for. # Feishu's subscribe endpoint is idempotent; the cache just skips # the extra round-trip on every subsequent call in the same process. _subscribed: set[tuple[str, str]] = set() # (app_id, user_id_type, email) → resolved_id. user_id and open_id # are stable for the lifetime of a Feishu user account — we cache # for the lifetime of the process (no TTL) so the hot # "我的申请 → 创建申请 → 提交" path doesn't re-hit batch_get_id on # every submission. _user_id_cache: dict[tuple[str, str, str], str] = {} # (app_id, id_type, id_value) → display_name. Used by the admin # approval detail modal to resolve Feishu task_list approvers # (whose ``user_name`` is ALWAYS null in our tenant's responses) # back to real names. Process-lifetime cache — names change rarely # and worker / API processes restart on every deploy. Only positive # hits are cached; transient lookup failures (network blip, 404 # for a deleted user) are NOT cached so a subsequent retry can # succeed. Class-level so concurrent requests in the same process # share the lookup work. _user_name_cache: dict[tuple[str, str, str], str] = {} def __init__(self, *, app_id: str, app_secret: str) -> None: if not app_id or not app_secret: raise AppException( code="FEISHU_NOT_CONFIGURED", message=( "飞书 app_id / app_secret 未配置。请到 管理后台 > 连接器 " "里编辑飞书连接器并填写凭据。" ), status_code=503, ) self.app_id = app_id self.app_secret = app_secret self.base_url = "https://open.feishu.cn/open-apis" async def get_tenant_access_token(self) -> str: cached = self._cached_tokens.get(self.app_id) if cached and time.time() < cached[1] - 60: return cached[0] async with httpx.AsyncClient( timeout=20, trust_env=False, follow_redirects=False, ) as client: response = await client.post( f"{self.base_url}/auth/v3/tenant_access_token/internal", json={"app_id": self.app_id, "app_secret": self.app_secret}, ) payload = response.json() # 🔴 **不是每个飞书端点都用 ``data`` 信封。** v3 那批(``/bot/v3/info``) # 把结果放在自己的键下(``bot``),``code``/``msg`` 之外没有 ``data`` —— # 于是默认的 ``payload.get("data", {})`` 返回空字典,而**调用方拿到的是一个 # 语法上完全正常的空结果**:没有异常、没有非零 code,只是什么都没有。 # 2026-09-08 就是这样让机器人的身份永远解析不出来(见 ``get_bot_info``)。 if payload.get("code") != 0: raise AppException( code="FEISHU_TOKEN_ERROR", message=payload.get("msg", "Failed to get Feishu tenant access token"), status_code=502, ) token = payload["tenant_access_token"] expire_at = time.time() + int(payload.get("expire", 7200)) self._cached_tokens[self.app_id] = (token, expire_at) return token async def _request( self, method: str, path: str, *, json_body: dict[str, Any] | None = None, data: dict[str, Any] | None = None, files: dict[str, Any] | None = None, params: dict[str, Any] | None = None, timeout: httpx.Timeout | float | int = 30, benign_error_codes: set[int] | frozenset[int] | None = None, envelope_key: str = "data", ) -> dict[str, Any]: token = await self.get_tenant_access_token() headers = {"Authorization": f"Bearer {token}"} async with httpx.AsyncClient( timeout=timeout, trust_env=False, follow_redirects=False, ) as client: response = await client.request( method, f"{self.base_url}{path}", headers=headers, params=params, json=json_body, data=data, files=files, ) payload = response.json() if payload.get("code") != 0: api_code = payload.get("code") if ( isinstance(api_code, int) and benign_error_codes and api_code in benign_error_codes ): return payload.get(envelope_key, {}) # Feishu error payloads can echo request fields and private # directory attributes. Keep logs useful for correlation without # copying the response body or localized message into clear text. # Do not interpolate request paths or response-derived values here: # approval paths can contain private instance/person identifiers. # The sanitized AppException below remains the audited diagnostic. logger.warning("feishu: API returned a non-zero response") # Embed the numeric Feishu code into the message text so # downstream consumers can parse it without depending on # a structured-detail channel. The Feishu API returns # localised ``msg`` (zh-CN tenants get Chinese strings), # which means substring-matching on "not found" doesn't # work for non-EN tenants. The numeric code is stable # across locales — feishu_software.py's # ``_feishu_error_code`` regex extracts it from this # prefix to drive idempotent short-circuits (already- # frozen / already-in-group / already-absent codes). api_msg = redact_credentials( payload.get("msg", "Feishu API request failed"), max_len=300, ) # Coerce non-int code to "?" before formatting so the # downstream regex (``\b(\d{4,8})\b`` in # feishu_software._feishu_error_code) doesn't accidentally # pull a numeric run from the API msg body when the API # returned ``code: null``. Without this, an upstream payload # of {"code": null, "msg": "...230002..."} would format as # "Feishu API error None: ...230002..." and the regex would # treat 230002 as the API code → false "already-frozen" # success. (Round 4 security P2.) code_str = str(api_code) if isinstance(api_code, int) else "?" raise AppException( code="FEISHU_API_ERROR", message=f"Feishu API error {code_str}: {api_msg}", status_code=502, ) return payload.get(envelope_key, {}) # ── Public API surface ─ async def request( self, method: str, path: str, *, json_body: dict[str, Any] | None = None, ) -> dict[str, Any]: """Public wrapper around the internal ``_request``. Same semantics — preferred entry point for code outside this module so cross-module ``noqa: SLF001`` is no longer needed. """ return await self._request(method, path, json_body=json_body) # ── Approval API ─ async def get_approval_definition( self, approval_code: str, *, subscribe: bool = True, with_admin_id: bool = False, user_id_type: str = "user_id", ) -> dict[str, Any]: """Read one approval definition. ``subscribe=False`` for definitions Helix only *observes* (the approval-flow registry reads every flow the company runs, most of which are HR / finance / legal). Subscribing to those would point their instance events at our webhook: thousands of payloads a day that no handler wants, on flows we have no business receiving. Subscription stays opt-in for the flows the platform actually drives. ``with_admin_id=True`` adds ``approval_admin_ids`` — the definition's 流程管理员, the people the Feishu console shows as answering for it. **The field is absent from the response entirely unless the query param is sent**, which is worth stating plainly: the approval-flow registry spent its first release calling this method without it, read the missing key as 「飞书没有负责人这个概念」, wrote that conclusion into its own docstrings, and published 「还没人认领」 for 32 flows whose managers Feishu knew by name. Verified live against this tenant 2026-08-27: no param → keys are name/status/form/node_list/ viewers; ``?with_admin_id=true&user_id_type=user_id`` → ``approval_admin_ids: ["89322g24"]``. """ query: list[str] = [] if with_admin_id: query.append("with_admin_id=true") query.append(f"user_id_type={quote(user_id_type, safe='')}") suffix = f"?{'&'.join(query)}" if query else "" data = await self._request( "GET", f"/approval/v4/approvals/{approval_code}{suffix}", ) if subscribe: await self.ensure_subscribed(approval_code) return data async def batch_get_user_profiles( self, user_ids: list[str], ) -> tuple[dict[str, dict[str, Any]], set[str]]: """Resolve Feishu ``user_id`` values to contact profiles. Returns ``({user_id: {"name", "email", "resigned"}}, failed)``. **The second element is the point.** An id can be missing from the map for two very different reasons, and a caller that cannot tell them apart will do the wrong thing with one of them: * **Genuinely gone** — Feishu answers with an error code (``99992360``: the account was deleted; this tenant's approval admin lists carry one such id today). The register should show that flow as having lost an administrator. * **The lookup failed** — a timeout, a 5xx, a rate-limit. The person still exists. Treating this as "gone" would delete a live administrator's name off the all-staff page for a night and write a false 「管理员变更」 into the flow's trail. Observed for real while measuring this tenant: ``2b34ag3a`` resolved on one pass and errored on the next, which flipped a flow from 「有唯一管理员」 to 「一个人都没有」 between two runs minutes apart. ``failed`` holds only the second kind. Ids simply absent from the map are the first kind. Deliberately per-id rather than ``GET /contact/v3/users/batch``: that endpoint fails the **whole** batch with 99992360 when one id in it is stale, and stale ids are exactly what a register of other teams' approval admins is full of (measured against this tenant: 1 of 37 admin ids no longer resolves). ``email`` prefers ``enterprise_email`` — the address that matches ``identity_persons.work_email``; the personal ``email`` field is often blank or a legacy address. """ wanted: list[str] = [] for raw in user_ids: if not isinstance(raw, str): continue value = raw.strip() if value and value not in wanted: wanted.append(value) if not wanted: return {}, set() sem = asyncio.Semaphore(_USER_LOOKUP_CONCURRENCY) async def _bounded(uid: str) -> tuple[str, dict[str, Any] | None, bool]: """``(uid, user, transient_failure)``.""" async with sem: safe_id = quote(uid, safe="") try: data = await self._request( "GET", f"/contact/v3/users/{safe_id}?user_id_type=user_id", ) except AppException as exc: # Feishu answered. Whatever it said, the account does # not resolve right now and that is upstream's own # statement, not a hiccup on our side. logger.debug("feishu: admin lookup miss (%s): %s", uid, exc) return uid, None, False except Exception as exc: # noqa: BLE001 logger.warning( "feishu: admin lookup transient error (%s): %s", uid, exc, ) return uid, None, True user = (data or {}).get("user") return uid, (user if isinstance(user, dict) else None), False gathered = await asyncio.gather( *(_bounded(u) for u in wanted), return_exceptions=True, ) out: dict[str, dict[str, Any]] = {} failed: set[str] = set() answered: set[str] = set() for outcome in gathered: if isinstance(outcome, BaseException): # We do not know WHICH id blew up; it is picked up below # as "no answer at all", which is a failure, not a # deletion. logger.warning("feishu: admin lookup task crashed: %s", outcome) continue uid, user, transient = outcome answered.add(uid) if transient: failed.add(uid) continue if not user: continue name = user.get("name") or user.get("en_name") email = user.get("enterprise_email") or user.get("email") status = user.get("status") or {} out[uid] = { "name": name.strip() if isinstance(name, str) else None, "email": email.strip().lower() if isinstance(email, str) else None, # Feishu keeps a resigned person's account resolvable. # An approval admin who left is a governance hole, and # the register has to be able to say so rather than # publishing their name as the answer to 「找谁」. "resigned": bool( status.get("is_resigned") or status.get("is_frozen"), ), } # A crashed task leaves an id with no answer at all. # Absent-because-crashed is not absent-because-deleted. failed.update(u for u in wanted if u not in answered) return out, failed async def batch_resolve_user_ids_by_email( self, emails: list[str], ) -> dict[str, str]: """Resolve many work emails to Feishu ``user_id`` in one sweep. Returns ``{lower-cased input email: user_id}``, omitting misses — a person with no Feishu account is a fact to report, not an exception to raise in the middle of a 500-person loop. Chunked at :data:`_BATCH_GET_ID_CHUNK`, which is Feishu's cap on the ``emails`` array. The same corporate-domain sibling tolerance as :meth:`resolve_user_id_by_email` applies: Helix stores ``x@pingcap.com`` while the Feishu account is often registered under ``x@pingcap.com.cn``, and an exact-match-only lookup silently drops those people from every sweep. """ wanted: list[str] = [] for raw in emails: norm = (raw or "").strip().lower() if norm and norm not in wanted: wanted.append(norm) if not wanted: return {} # Query address → the input address(es) it can satisfy. query_map: dict[str, list[str]] = {} for email in wanted: local, _, domain = email.partition("@") query_map.setdefault(email, []).append(email) sibling = _PINGCAP_EMAIL_SIBLING.get(domain) if sibling and local: query_map.setdefault(f"{local}@{sibling}", []).append(email) queries = list(query_map) resolved: dict[str, str] = {} for start in range(0, len(queries), _BATCH_GET_ID_CHUNK): chunk = queries[start:start + _BATCH_GET_ID_CHUNK] data = await self._request( "POST", "/contact/v3/users/batch_get_id?user_id_type=user_id", json_body={"emails": chunk}, ) for user in data.get("user_list") or []: uid = user.get("user_id") echoed = (user.get("email") or "").strip().lower() if not uid or not echoed: continue for target in query_map.get(echoed, []): # First hit wins: the exact address is queried before # its sibling, so an account that exists under both # resolves to the one Helix actually stores. resolved.setdefault(target, uid) self._user_id_cache[(self.app_id, "user_id", echoed)] = uid return resolved async def query_instances_for_user( self, user_id: str, *, page_size: int = 200, max_pages: int = 30, start_time_ms: str | None = None, end_time_ms: str | None = None, ) -> tuple[list[dict[str, Any]], bool]: """Every approval instance this user is party to. ``POST /approval/v4/instances/query``. Returns ``(items, truncated)`` — ``truncated`` is true when Feishu still had more pages at ``max_pages``, so a caller can report the cut instead of presenting a partial sweep as complete. Two wire facts, both verified live against the tenant (2026-08-26) rather than read off the docs: * At least one of ``user_id`` / ``approval_code`` / ``instance_code`` / … must be present, or Feishu answers ``1390001 all required query criteria is empty``. A bare time window is not a valid query, which is why this sweep is per-person. * The ``group`` object on each item IS reliable here (every code reported exactly one group name across a 1196-row sample) — unlike the same field on ``tasks/search``, which returned several different group names for the same definition. Read groups from this endpoint only. """ size = max(5, min(int(page_size), _APPROVAL_SEARCH_MAX_PAGE_SIZE)) body: dict[str, Any] = {"user_id": user_id} if start_time_ms and end_time_ms: # Feishu rejects a half-open window ("must be set together") # and any span wider than 30 days. body["instance_start_time_from"] = start_time_ms body["instance_start_time_to"] = end_time_ms return await self._paged_approval_search( "/approval/v4/instances/query", body, list_key="instance_list", page_size=size, max_pages=max_pages, ) async def search_tasks_for_user( self, user_id: str, *, page_size: int = 200, max_pages: int = 30, ) -> tuple[list[dict[str, Any]], bool]: """Every approval task assigned to this user — i.e. what they actually decide. ``POST /approval/v4/tasks/search``. Used to infer who is responsible for a flow. **Ignore the ``group`` object on these items**: measured against ``instances/query`` on the same tenant, 13 of 25 definitions came back with more than one group name here, and 12 disagreed outright with the instance API. """ size = max(5, min(int(page_size), _APPROVAL_SEARCH_MAX_PAGE_SIZE)) return await self._paged_approval_search( "/approval/v4/tasks/search", {"user_id": user_id}, list_key="task_list", page_size=size, max_pages=max_pages, ) async def _paged_approval_search( self, path: str, body: dict[str, Any], *, list_key: str, page_size: int, max_pages: int, ) -> tuple[list[dict[str, Any]], bool]: items: list[dict[str, Any]] = [] page_token: str | None = None for _ in range(max(1, max_pages)): qs = f"user_id_type=user_id&page_size={page_size}" if page_token: qs = f"{qs}&page_token={page_token}" data = await self._request("POST", f"{path}?{qs}", json_body=body) page_items = data.get(list_key) or [] items.extend(i for i in page_items if isinstance(i, dict)) if not data.get("has_more"): return items, False page_token = data.get("page_token") or None if not page_token: return items, False return items, True async def ensure_subscribed(self, approval_code: str) -> None: """Register this app to receive instance events for a specific approval definition. Idempotent on Feishu's side + cached here. Failures are logged and swallowed so the main flow continues; the next ``create_instance`` / ``get_approval_definition`` call retries automatically. """ cache_key = (self.app_id, approval_code) if cache_key in type(self)._subscribed: return try: await self._request( "POST", f"/approval/v4/approvals/{approval_code}/subscribe", benign_error_codes={ _FEISHU_APPROVAL_SUBSCRIPTION_EXISTS_CODE, }, ) type(self)._subscribed.add(cache_key) logger.info("feishu: subscribed to approval %s events", approval_code) except AppException as e: if parse_feishu_error_code(e) == _FEISHU_APPROVAL_SUBSCRIPTION_EXISTS_CODE: type(self)._subscribed.add(cache_key) logger.info( "feishu: approval %s event subscription already exists", approval_code, ) return # Already-subscribed is not an error; other transients retry next call. logger.warning( "feishu: subscribe to approval %s failed (will retry): %s", approval_code, e, ) except Exception as e: # noqa: BLE001 logger.warning( "feishu: subscribe to approval %s failed (will retry): %s", approval_code, e, ) async def resolve_user_id_by_email(self, email: str) -> str: # Normalize to lowercase to match the case-insensitive UNIQUE index # on identity_persons.work_email. Feishu echoes back the exact case # we sent, so we compare on lowercase too. norm_email = (email or "").strip().lower() cache_key = (self.app_id, "user_id", norm_email) cached = self._user_id_cache.get(cache_key) if cached is not None: return cached # Domain-mismatch tolerance: Helix stores work_email as # ``x@pingcap.com`` but a person's Feishu account is often registered # under ``x@pingcap.com.cn`` (and vice-versa). An exact-email lookup # then silently misses — and for an HRBP/leader reference that means # the slot is left unset and the Feishu field never updates # (2026-07-06: an HRBP push was omitted because the HRBP resolved # only under .com.cn). Query BOTH the given address and its corporate # sibling, then accept a hit by LOCAL-PART within the pingcap # corporate domains — never by local-part alone (that would cross # tenants). local, _, domain = norm_email.partition("@") sibling_domain = _PINGCAP_EMAIL_SIBLING.get(domain) candidates = [norm_email] if sibling_domain and local: candidates.append(f"{local}@{sibling_domain}") data = await self._request( "POST", "/contact/v3/users/batch_get_id?user_id_type=user_id", json_body={"emails": candidates}, ) users = [u for u in data.get("user_list", []) if u.get("user_id")] # Prefer an exact-email match; fall back to a local-part match within # the pingcap .com / .com.cn sibling pair. match = next( (u for u in users if (u.get("email") or "").lower() == norm_email), None, ) if match is None and sibling_domain: for u in users: ul, _, ud = (u.get("email") or "").lower().partition("@") if ul == local and ud in (domain, sibling_domain): match = u break if match is not None: self._user_id_cache[cache_key] = match["user_id"] return match["user_id"] raise AppException( code="FEISHU_USER_NOT_FOUND", message=f"Feishu user id not found for email {norm_email}", status_code=400, ) async def resolve_user_id_by_mobile(self, mobile: str) -> str: """Resolve a Feishu ``user_id`` by mobile (E.164, e.g. ``+8613800138000``). Mobile is globally UNIQUE on Feishu, so a single ``batch_get_id`` hit is authoritative — no domain-sibling tolerance is needed (unlike email). Used as the fallback when an email lookup misses because the account's registered email differs from work_email. Raises ``FEISHU_USER_NOT_FOUND`` on miss. """ norm = (mobile or "").strip() if not norm: raise AppException( code="FEISHU_USER_NOT_FOUND", message="Feishu user id not found for empty mobile", status_code=400, ) cache_key = (self.app_id, "user_id_mobile", norm) cached = self._user_id_cache.get(cache_key) if cached is not None: return cached data = await self._request( "POST", "/contact/v3/users/batch_get_id?user_id_type=user_id", json_body={"mobiles": [norm]}, ) users = [u for u in data.get("user_list", []) if u.get("user_id")] if users: uid = users[0]["user_id"] self._user_id_cache[cache_key] = uid return uid raise AppException( code="FEISHU_USER_NOT_FOUND", message=f"Feishu user id not found for mobile {norm}", status_code=400, ) async def resolve_open_id_by_email(self, email: str) -> str: """Messaging API wants ``open_id`` (not user_id); contact.v3 returns both with ``user_id_type=open_id``.""" norm_email = (email or "").strip().lower() cache_key = (self.app_id, "open_id", norm_email) cached = self._user_id_cache.get(cache_key) if cached is not None: return cached # Same .com/.com.cn corporate-sibling tolerance as # resolve_user_id_by_email — query both addresses, match by # local-part within the pingcap pair (never cross to another domain). local, _, domain = norm_email.partition("@") sibling_domain = _PINGCAP_EMAIL_SIBLING.get(domain) candidates = [norm_email] if sibling_domain and local: candidates.append(f"{local}@{sibling_domain}") data = await self._request( "POST", "/contact/v3/users/batch_get_id?user_id_type=open_id", json_body={"emails": candidates}, ) # with user_id_type=open_id, the field is still ``user_id``. users = [u for u in data.get("user_list", []) if u.get("user_id")] match = next( (u for u in users if (u.get("email") or "").lower() == norm_email), None, ) if match is None and sibling_domain: for u in users: ul, _, ud = (u.get("email") or "").lower().partition("@") if ul == local and ud in (domain, sibling_domain): match = u break if match is not None: self._user_id_cache[cache_key] = match["user_id"] return match["user_id"] raise AppException( code="FEISHU_USER_NOT_FOUND", message=f"Feishu open_id not found for email {email}", status_code=400, ) async def resolve_email_by_open_id(self, open_id: str) -> str | None: """``open_id`` → 企业邮箱,用于把一次 @提及 变成一个 Helix 的人。 群里「转给 @张三」里的张三,消息信封里只有一个 ``open_id``;而 Helix 认 人靠邮箱。先取 ``enterprise_email``,退回 ``email``。 ⚠️ **两个字段都可能是空的**(2026-09 实测:确有在职员工的飞书账号邮箱 字段为空),所以返回 ``None`` 是一个正常结果,不是错误 —— 调用方要能说出 「我认不出这个人」,而不是崩掉。 """ user = await self._fetch_user(id_type="open_id", id_value=open_id) if not user: return None for key in ("enterprise_email", "email"): value = str(user.get(key) or "").strip() if value: return value.lower() return None async def _fetch_user( self, *, id_type: str, id_value: str, ) -> dict[str, Any] | None: """Fetch a single Feishu contact user object by id. ``id_type`` must be ``open_id`` or ``user_id`` (the values Feishu's ``user_id_type`` query param accepts that we use here). Returns the raw ``user`` dict on success, ``None`` on any failure path — missing user (404), Feishu-side error, or a malformed response. Callers are expected to tolerate ``None`` (this is an informational, best-effort lookup, never a critical path). """ # URL-encode the path segment as defence-in-depth — Feishu's # open_id / user_id formats don't normally include slashes or # other reserved chars, but a malformed task_list entry (Feishu # tenant migration corner case, future schema change) shouldn't # be able to escape the path segment. safe_id = quote(id_value, safe="") try: data = await self._request( "GET", f"/contact/v3/users/{safe_id}?user_id_type={id_type}", ) except AppException as e: # Feishu codes that mean "this id doesn't resolve to a user # right now" — deleted account, never-existed id (Feishu's # ``user_id`` namespace is short and reusable, so a stale # task_list entry from a deactivated approver lands here), # tenant boundary mismatch. Logging once at debug avoids # spamming the log for an approval flow that legitimately # has a deleted approver in its history. logger.debug( "feishu: user lookup miss (%s=%s): %s", id_type, id_value, e, ) return None except Exception as e: # noqa: BLE001 logger.warning( "feishu: user lookup transient error (%s=%s): %s", id_type, id_value, e, ) return None user = (data or {}).get("user") return user if isinstance(user, dict) else None async def _fetch_user_name( self, *, id_type: str, id_value: str, ) -> str | None: """Resolve a single Feishu user id to its display name. Returns the name on success, ``None`` on any failure path. The caller is expected to tolerate ``None`` (fall back to the raw id) rather than treat it as a hard failure: the modal rendering this name is informational, not a critical path, and the alternative ("the admin modal won't open at all because Feishu briefly hiccuped") is a worse UX. """ user = await self._fetch_user(id_type=id_type, id_value=id_value) if user is None: return None name = user.get("name") or user.get("en_name") if isinstance(name, str) and name.strip(): return name.strip() return None async def list_pending_approver_emails(self, instance_code: str) -> list[str]: """Lowercased work-emails of the approvers a live Feishu instance is *currently waiting on* — the tasks whose status is ``PENDING`` / ``TRANSFERRED`` in the instance ``task_list``. This is "who needs to act right now", as opposed to everyone who ever touched (or could touch) the flow. Used by 催办 so an urge only nudges the current decision owner, not the whole approval chain. Best-effort: returns ``[]`` on any fetch/resolve failure (Feishu hiccup, deleted approver, empty task list) so the caller can fall back gracefully rather than treating it as a hard error. Emails are deduped and lowercased to match the case-insensitive UNIQUE index on ``identity_persons.work_email``. """ try: detail = await self.get_instance_detail(instance_code) except Exception as e: # noqa: BLE001 logger.warning( "feishu: pending-approver fetch failed for %s: %s", instance_code, e, ) return [] task_list = detail.get("task_list") if not isinstance(task_list, list) or not task_list: return [] pending_statuses = {"PENDING", "TRANSFERRED"} # (id_type, id_value) for each currently-pending task, open_id # preferred (stable across the lifecycle — see DNS 4-eyes check). wanted: list[tuple[str, str]] = [] seen_ids: set[str] = set() for task in task_list[:500]: if not isinstance(task, dict): continue if str(task.get("status") or "").upper() not in pending_statuses: continue open_id = task.get("open_id") user_id = task.get("user_id") if isinstance(open_id, str) and open_id.strip(): key = open_id.strip() id_type, id_value = "open_id", key elif isinstance(user_id, str) and user_id.strip(): key = user_id.strip() id_type, id_value = "user_id", key else: continue if key in seen_ids: continue seen_ids.add(key) wanted.append((id_type, id_value)) if not wanted: return [] sem = asyncio.Semaphore(_USER_LOOKUP_CONCURRENCY) async def _bounded(id_type: str, id_value: str) -> str | None: async with sem: user = await self._fetch_user(id_type=id_type, id_value=id_value) if user is None: return None email = user.get("enterprise_email") or user.get("email") if isinstance(email, str) and email.strip(): return email.strip().lower() return None gathered = await asyncio.gather( *(_bounded(t, v) for t, v in wanted), return_exceptions=True, ) out: list[str] = [] for item in gathered: if isinstance(item, str) and item not in out: out.append(item) return out async def batch_get_user_names( self, *, open_ids: list[str] | None = None, user_ids: list[str] | None = None, ) -> dict[str, str]: """Resolve a batch of open_ids / user_ids to display names. Returns a flat ``{id_value: display_name}`` map. open_ids and user_ids are merged into ONE map keyed by raw id string — the caller is expected to know which is which from context. When both an open_id and a user_id refer to the same person, both keys map to the same name string. Background — used by the admin approval detail modal. Feishu's ``/approval/v4/instances/{code}`` response carries ``task_list`` with both ``open_id`` and ``user_id`` per task, but ``user_name`` / ``name`` is null on every tenant we've observed (pending AND approved tasks alike — verified across multiple approval types in prod, 2026-05-15). Without this resolution, the flow tab renders raw short ids like ``d8d2646c`` for every approver, and the frontend dedup logic — which keys on ``user_name`` — duplicates each row. Performance / safety: * Class-level positive cache, so repeated modal opens for the same approver are a single in-process dict lookup. * Concurrency bounded by ``_USER_LOOKUP_CONCURRENCY`` so a 20-approver flow doesn't burst rate limits. * Lookups never raise — partial maps are valid output. An id absent from the returned dict means "couldn't resolve; caller should keep the raw id". * Empty / blank ids are silently filtered out before any HTTP call. """ wanted: list[tuple[str, str]] = [] # (id_type, id_value) pairs to fetch result: dict[str, str] = {} def _intake(id_type: str, ids: list[str] | None) -> None: if not ids: return seen_in_call: set[str] = set() for raw in ids: if not isinstance(raw, str): continue v = raw.strip() if not v or v in seen_in_call: continue seen_in_call.add(v) cache_key = (self.app_id, id_type, v) cached = type(self)._user_name_cache.get(cache_key) if cached is not None: result[v] = cached continue wanted.append((id_type, v)) _intake("open_id", open_ids) _intake("user_id", user_ids) if not wanted: return result sem = asyncio.Semaphore(_USER_LOOKUP_CONCURRENCY) async def _bounded(id_type: str, id_value: str) -> tuple[str, str | None]: async with sem: name = await self._fetch_user_name( id_type=id_type, id_value=id_value, ) return id_value, name # ``return_exceptions=True`` for defense-in-depth: ``_fetch_user_name`` # already swallows AppException + Exception, but if anything new # ever raises out of it, we still want partial results back. gathered = await asyncio.gather( *(_bounded(t, v) for t, v in wanted), return_exceptions=True, ) # Re-walk wanted in order so we can pair up ``(id_type, id_value)`` # with each gathered result for cache write-back. for (id_type, id_value), outcome in zip(wanted, gathered, strict=True): if isinstance(outcome, BaseException): continue _, name = outcome if name is None: continue result[id_value] = name type(self)._user_name_cache[ (self.app_id, id_type, id_value) ] = name return result async def send_text_to_user( self, *, open_id: str, text: str, idempotency_key: str | None = None, ) -> dict[str, Any]: """Send text, optionally deduped by Feishu for one hour.""" return await self._request( "POST", "/im/v1/messages?receive_id_type=open_id", json_body={ "receive_id": open_id, "msg_type": "text", "content": json.dumps({"text": text}, ensure_ascii=False), **_message_dedup_field(idempotency_key), }, ) async def send_post_to_user( self, *, open_id: str, markdown: str, idempotency_key: str | None = None, ) -> dict[str, Any]: """Send a rich-text (``post``) message whose body is markdown. A ``post`` message's ``md`` element renders a markdown subset — bold, links, lists, inline code — so a model's reply reaches the employee with its structure intact instead of being flattened to a wall of characters first. Verified against the live API. """ return await self._request( "POST", "/im/v1/messages?receive_id_type=open_id", json_body={ "receive_id": open_id, "msg_type": "post", "content": json.dumps( {"zh_cn": {"title": "", "content": [ [{"tag": "md", "text": markdown}], ]}}, ensure_ascii=False, ), **_message_dedup_field(idempotency_key), }, ) async def send_card_to_user( self, *, open_id: str, card: dict[str, Any], idempotency_key: str | None = None, ) -> dict[str, Any]: """Send a card, optionally deduped by Feishu for one hour.""" return await self._request( "POST", "/im/v1/messages?receive_id_type=open_id", json_body={ "receive_id": open_id, "msg_type": "interactive", "content": json.dumps(card, ensure_ascii=False), **_message_dedup_field(idempotency_key), }, ) async def upload_message_image(self, *, content: bytes) -> str: """Upload an image for use INSIDE a chat message, returning its key. ``POST /im/v1/images`` with ``image_type=message`` — the chat-message image API, deliberately distinct from :meth:`upload_image`, which hits the approval-form upload endpoint and returns a UUID-shaped approval file code that a card cannot render (see that method's note). The returned ``img_v3_*`` key is what a card 2.0 header's ``custom_icon`` takes. **It is scoped to this app** — a key minted by one Feishu app is not resolvable by another, which is why nothing caches it in source. See :mod:`app.notifications.card_icon`. """ data = await self._request( "POST", "/im/v1/images", data={"image_type": "message"}, files={"image": ("image.png", content, "image/png")}, ) key = data.get("image_key") if isinstance(data, dict) else None if not isinstance(key, str) or not key: raise AppException( code="FEISHU_API_ERROR", message="Feishu image upload returned no image_key", status_code=502, ) return key async def create_group_chat( self, *, name: str, member_open_ids: list[str], description: str = "", ) -> str: """Open a private group chat with these people in it. Returns ``chat_id``. This is what 「转人工」 needs: an employee whose question the assistant could not answer gets a room with the person who owns that area, and the exchange is forwarded in so nobody retypes it. Requires the ``im:chat`` scope, granted 2026-09-07. Before that the API answered ``99991672`` naming the scope — which is also the cheapest permission probe there is (see the module's error-code notes). ``chat_mode="group"`` + ``chat_type="private"``: a real group (not a p2p thread), invisible in search. ``edit_permission="all_members"`` so the owner can rename it to something meaningful. """ data = await self._request( "POST", "/im/v1/chats?user_id_type=open_id", json_body={ "name": name[:60], "description": description[:100], "user_id_list": member_open_ids, "chat_mode": "group", "chat_type": "private", "edit_permission": "all_members", }, ) chat_id = data.get("chat_id") if isinstance(data, dict) else None if not isinstance(chat_id, str) or not chat_id: raise AppException( code="FEISHU_API_ERROR", message="Feishu chat creation returned no chat_id", status_code=502, ) return chat_id async def add_chat_members( self, *, chat_id: str, member_open_ids: list[str], ) -> None: """Pull more people into an existing group.""" if not member_open_ids: return await self._request( "POST", f"/im/v1/chats/{chat_id}/members?member_id_type=open_id", json_body={"id_list": member_open_ids}, ) async def send_text_to_chat( self, *, chat_id: str, text: str, idempotency_key: str | None = None, ) -> dict[str, Any]: """Post a plain-text message into a group.""" return await self._request( "POST", "/im/v1/messages?receive_id_type=chat_id", json_body={ "receive_id": chat_id, "msg_type": "text", "content": json.dumps({"text": text}, ensure_ascii=False), **_message_dedup_field(idempotency_key), }, ) async def send_post_to_chat( self, *, chat_id: str, markdown: str, idempotency_key: str | None = None, ) -> dict[str, Any]: """Post a rich-text (``post``) message into a group. The group twin of :meth:`send_post_to_user`; same ``md`` element, same markdown subset. Kept as its own method rather than a ``receive_id_type`` parameter on one send: the two receiver kinds are different enough at the call sites (an employee's open_id versus a chat id) that a single polymorphic sender reads as "pass the right kind of id here" — exactly the sort of thing that gets passed wrong once and posts a private answer into a room. """ return await self._request( "POST", "/im/v1/messages?receive_id_type=chat_id", json_body={ "receive_id": chat_id, "msg_type": "post", "content": json.dumps( {"zh_cn": {"title": "", "content": [ [{"tag": "md", "text": markdown}], ]}}, ensure_ascii=False, ), **_message_dedup_field(idempotency_key), }, ) async def send_card_to_chat( self, *, chat_id: str, card: dict[str, Any], idempotency_key: str | None = None, ) -> dict[str, Any]: """Post an interactive card into a group. Everyone in the room sees the same card and any of them can tap it — which is why every group card's handler re-checks who tapped rather than trusting the card's mere presence in front of someone. """ return await self._request( "POST", "/im/v1/messages?receive_id_type=chat_id", json_body={ "receive_id": chat_id, "msg_type": "interactive", "content": json.dumps(card, ensure_ascii=False), **_message_dedup_field(idempotency_key), }, ) async def reply_message( self, *, message_id: str, msg_type: str, content: dict[str, Any], reply_in_thread: bool = False, idempotency_key: str | None = None, ) -> dict[str, Any]: """Reply UNDER an existing message rather than posting a new one. In a busy group a bare post is a message everyone has to correlate back to whoever asked; a reply carries the question with it. ``reply_in_thread`` additionally collapses the exchange into a Feishu thread, which is what keeps an assistant answering three people at once from turning the room into interleaved fragments. Deliberately generic over ``msg_type``: text, post and interactive all reply the same way, and three near-identical wrappers would drift. """ return await self._request( "POST", f"/im/v1/messages/{message_id}/reply", json_body={ "msg_type": msg_type, "content": json.dumps(content, ensure_ascii=False), "reply_in_thread": reply_in_thread, **_message_dedup_field(idempotency_key), }, ) async def get_bot_info(self) -> dict[str, Any]: """This app's own bot identity — ``open_id``, ``app_name``, avatar. ``GET /bot/v3/info`` takes the app's own token and needs no scope: it is the app asking who it is. The answer matters because a group message only belongs to the assistant when the assistant was **@-mentioned**, and a mention is an ``open_id`` in the message envelope — with no ``is_bot`` flag next to it. Without this call the only way to tell "they addressed me" from "they addressed a colleague and I happened to be delivered the message" is guessing from display names. """ # ``envelope_key="bot"``:这个端点**没有 ``data`` 信封**,结果直接挂在 # ``bot`` 下。用默认信封读它会安静地拿到 ``{}`` —— 而空结果在上层是 # 「身份解析不出来」,那条路又是刻意 fail-open 的(有人被 @ 就算在跟我 # 说话),所以整件事的表现不是报错,而是**机器人会去回答别人被 @ 的那条 # 消息**,同时「@ 的到底是不是我」这个判断永久失效。 bot = await self._request("GET", "/bot/v3/info", envelope_key="bot") return bot if isinstance(bot, dict) else {} async def disband_chat(self, *, chat_id: str) -> None: """Delete a group chat. Irreversible — the history goes with it. Only ever called on a chat Helix itself opened and still has recorded, never on a room somebody merely invited the bot into. """ await self._request("DELETE", f"/im/v1/chats/{chat_id}") async def list_chat_messages( self, *, chat_id: str, start_time: int | None = None, page_size: int = 50, page_token: str | None = None, ) -> dict[str, Any]: """Read a page of a group's message history, oldest first. On demand only. The assistant does not subscribe to every message in the rooms it sits in — it reads the room when somebody asks it to summarise, and not otherwise. That is a privacy stance, not an optimisation: a bot that silently records every conversation it can see is a different product from one that answers when addressed. """ params: dict[str, Any] = { "container_id_type": "chat", "container_id": chat_id, "sort_type": "ByCreateTimeAsc", "page_size": max(1, min(page_size, 50)), } if start_time is not None: params["start_time"] = str(start_time) if page_token: params["page_token"] = page_token return await self._request("GET", "/im/v1/messages", params=params) async def forward_message_to_chat( self, *, message_id: str, chat_id: str, ) -> None: """Forward an existing message into a group, as itself. Why forwarding rather than pasting a transcript: the answer card the employee is looking at carries its own formatting and its 参考文档 links. Re-rendering it as quoted text in a new message loses both, and the owner then reads a flattened copy of something the employee can see in full one screen away. """ await self._request( "POST", f"/im/v1/messages/{message_id}/forward?receive_id_type=chat_id", json_body={"receive_id": chat_id}, ) async def update_card_message( self, *, message_id: str, card: dict[str, Any], ) -> None: """Replace the card content of a message this app already sent. ``PATCH /im/v1/messages/{message_id}`` — the same message, re-rendered in place for everyone who can see it. This is what turns a 「去审批」 card into its own receipt once the underlying request is decided, instead of leaving a button that refuses the tap it invites. Verified against the live API on 2026-09-07 with the app credentials Helix already holds: no additional scope, no app-version review, and a second PATCH of the same message is accepted too (so a card can go pending → approved → superseded without special-casing the first update). Only cards sent BY this app can be patched, and only while Feishu still holds the message — a failure here is normal for old or foreign messages, so callers treat it as "this card stays as it was" rather than as an error worth failing a flow over. """ await self._request( "PATCH", f"/im/v1/messages/{message_id}", json_body={"content": json.dumps(card, ensure_ascii=False)}, ) async def create_instance( self, *, approval_code: str, user_id: str | None = None, open_id: str | None = None, form: str, node_approver_user_id_list: list[dict[str, Any]] | None = None, node_approver_open_id_list: list[dict[str, Any]] | None = None, node_cc_user_id_list: list[dict[str, Any]] | None = None, node_cc_open_id_list: list[dict[str, Any]] | None = None, instance_uuid: str | None = None, ) -> dict[str, Any]: if not (user_id or open_id): raise AppException( code="FEISHU_USER_ID_REQUIRED", message="user_id or open_id is required for create_instance", status_code=400, ) await self.ensure_subscribed(approval_code) payload = { "approval_code": approval_code, # Feishu documents ``uuid`` as the native idempotency key for # instance creation. Callers with a durable submission intent # pass the persisted value; legacy callers retain a fresh UUID. "uuid": instance_uuid or str(uuid4()), "form": form, } if user_id: payload["user_id"] = user_id if open_id: payload["open_id"] = open_id if node_approver_user_id_list: payload["node_approver_user_id_list"] = node_approver_user_id_list if node_approver_open_id_list: payload["node_approver_open_id_list"] = node_approver_open_id_list if node_cc_user_id_list: payload["node_cc_user_id_list"] = node_cc_user_id_list if node_cc_open_id_list: payload["node_cc_open_id_list"] = node_cc_open_id_list return await self._request("POST", "/approval/v4/instances", json_body=payload) async def find_instance_by_uuid( self, *, approval_code: str, instance_uuid: str, start_time: datetime, end_time: datetime, max_pages: int = 2, ) -> dict[str, Any] | None: """Resolve Feishu's create UUID to its provider ``instance_code``. The detail endpoint accepts only an instance code; the UUID supplied to create is an idempotency key, not an alternate detail key. Feishu also returns 60012 (rather than the original response) when that UUID is submitted again. Recovery therefore lists codes in the bounded create-time window and inspects their details until the UUID matches. ``None`` means the fully paginated window was authoritatively exhausted. A page cap is fail-closed instead of pretending absence. """ if max_pages < 1: raise ValueError("max_pages must be positive") start = ( start_time.replace(tzinfo=UTC) if start_time.tzinfo is None else start_time.astimezone(UTC) ) end = ( end_time.replace(tzinfo=UTC) if end_time.tzinfo is None else end_time.astimezone(UTC) ) if end < start or end - start > timedelta(hours=10): raise ValueError("Feishu instance lookup window must be 0-10 hours") expected_uuid = instance_uuid.strip().lower() page_token: str | None = None for page_number in range(1, max_pages + 1): params: dict[str, Any] = { "approval_code": approval_code, "start_time": str(int(start.timestamp() * 1000)), "end_time": str(int(end.timestamp() * 1000)), "page_size": 50, } if page_token: params["page_token"] = page_token page = await self._request( "GET", "/approval/v4/instances", params=params, ) for instance_code in page.get("instance_code_list") or []: if not instance_code: continue try: detail = await self.get_instance_detail(str(instance_code)) except AppException as exc: # An instance can be deleted between list and detail. if parse_feishu_error_code(exc) == 1390003: continue raise if str(detail.get("uuid") or "").strip().lower() == expected_uuid: return detail if not page.get("has_more"): return None page_token = str(page.get("page_token") or "").strip() or None if page_token is None: raise AppException( code="FEISHU_INSTANCE_UUID_LOOKUP_INCOMPLETE", message="Feishu instance list reported more pages without a page token", status_code=502, ) raise AppException( code="FEISHU_INSTANCE_UUID_LOOKUP_INCOMPLETE", message=( "Feishu instance UUID lookup exceeded its bounded page limit " f"({max_pages})" ), status_code=502, ) async def get_instance_detail(self, instance_code: str) -> dict[str, Any]: return await self._request("GET", f"/approval/v4/instances/{instance_code}") async def cancel_instance( self, *, instance_code: str, user_id: str | None = None, open_id: str | None = None, approval_code: str, ) -> dict[str, Any]: """Withdraw a Feishu approval instance. ``user_id`` must be the Feishu user_id variant (from :meth:`resolve_user_id_by_email`) or ``open_id`` must be the Feishu open_id variant (from :meth:`resolve_open_id_by_email`). Feishu's body field is still named ``user_id``; the query parameter tells Feishu which id namespace that value belongs to. """ actor_id = user_id or open_id if not actor_id: raise AppException( code="FEISHU_USER_ID_REQUIRED", message="user_id or open_id is required for cancel_instance", status_code=400, ) user_id_type = "user_id" if user_id else "open_id" payload = { "approval_code": approval_code, "instance_code": instance_code, "user_id": actor_id, } return await self._request( "POST", f"/approval/v4/instances/cancel?user_id_type={user_id_type}", json_body=payload, ) async def cc_instance( self, *, approval_code: str, instance_code: str, user_id: str | None = None, open_id: str | None = None, cc_user_ids: list[str], comment: str | None = None, ) -> dict[str, Any]: """CC an existing approval instance to additional users. Feishu's body uses ``user_id`` for the actor and ``cc_user_ids`` for recipients; the query parameter selects whether those values are employee ``user_id`` or app-scoped ``open_id``. """ actor_id = user_id or open_id if not actor_id: raise AppException( code="FEISHU_USER_ID_REQUIRED", message="user_id or open_id is required for cc_instance", status_code=400, ) clean_cc_user_ids = [ str(value).strip() for value in cc_user_ids if str(value).strip() ] if not clean_cc_user_ids: raise AppException( code="FEISHU_CC_USER_IDS_REQUIRED", message="cc_user_ids is required for cc_instance", status_code=400, ) user_id_type = "user_id" if user_id else "open_id" payload = { "approval_code": approval_code, "instance_code": instance_code, "user_id": actor_id, "cc_user_ids": clean_cc_user_ids, } if comment: payload["comment"] = comment return await self._request( "POST", f"/approval/v4/instances/cc?user_id_type={user_id_type}", json_body=payload, ) async def approve_task( self, *, approval_code: str, instance_code: str, user_id: str, task_id: str, form: str | None = None, comment: str | None = None, ) -> dict[str, Any]: """Approve a PENDING task as its assigned operator, optionally carrying ``form`` widget values. Endpoint: ``POST /approval/v4/tasks/approve``. ``user_id`` MUST be the task's operator (from the instance detail ``task_list``) in the ``user_id`` variant — same convention as :meth:`cancel_instance`. ``form`` is the same JSON-array-as-string shape as :meth:`create_instance`'s form. The official docs only promise it feeds condition-branch routing, but when the approval template grants the node "编辑" permission on a widget, the values land in the form itself — this is how the laptop_replacement 回购 template's 提交人本人 nodes (上传恢复 出厂设置视频 etc.) accept the file the requester supplements through Helix. Callers MUST verify the write by re-fetching the instance detail afterwards (the API silently ignores values for widgets the node cannot edit) and fall back to the comment channel for anything that didn't stick. """ payload: dict[str, Any] = { "approval_code": approval_code, "instance_code": instance_code, "user_id": user_id, "task_id": task_id, } if form is not None: payload["form"] = form if comment: payload["comment"] = comment return await self._request( "POST", "/approval/v4/tasks/approve?user_id_type=user_id", json_body=payload, ) async def add_instance_comment( self, *, instance_code: str, user_id: str, content_text: str, attachments: list[dict[str, Any]] | None = None, ) -> dict[str, Any]: """Post a comment with optional attachments on an in-flight approval instance. Returns the Feishu ``data`` block. Feishu forbids modifying the form fields of a submitted approval — the only sanctioned in-flight channel for the requester to attach more material (the "上传 buyback 视频/转账截图，但 创建时不强制" UX) is the instance comment stream. Approvers see the comment + its attachments in the approval-card timeline. ``attachments`` entries MUST follow the documented v4 comment ``files`` schema — **URL-based, not token-based**:: {"url": "", "fileSize": , "title": "", "type": "image" | "attachment"} The viewable ``url`` comes from the approval upload endpoint's response (``/approval/openapi/v2/file/upload`` returns ``{"code", "url"}``). Do NOT pass upload tokens/codes here: verified against prod instance A526A8A8 on 2026-06-11 — the v4 comment API **silently drops** an ``images`` key entirely (not part of its schema; the stored comment came back as ``{"text": ..., "files": null}``), and a ``files`` entry shaped ``{"file": , "name": ...}`` doesn't match the documented ``{url, fileSize, title, type}`` shape either — both failure modes lose the attachment without any API error, which is exactly how supplement uploads shipped invisible to approvers for weeks. ``?user_id_type=user_id`` matches our other instance endpoints — ``user_id`` must come from :meth:`resolve_user_id_by_email`. Endpoint: ``POST /approval/v4/instances/{instance_code}/comments``. Feishu's ``content`` field is a JSON string (not a dict) — same encoding shape as ``form`` on :meth:`create_instance`. Files ride alongside the text in the same ``content`` JSON so approvers see one timeline entry per upload batch. """ content_payload: dict[str, Any] = {"text": content_text} if attachments: content_payload["files"] = attachments payload = { "content": json.dumps(content_payload, ensure_ascii=False), "user_id": user_id, } # The commenter ``user_id`` MUST ride in the QUERY string — Feishu's # ``/comments`` endpoint validates it there, not in the JSON body. # Sending it only in the body returns 400 ``99992402 user_id is # required`` (verified against prod approval instances 2026-06-09), # which the asset supplement-upload flow has been silently hitting. # We keep it in the body too (harmless) but the query param is what # actually satisfies the API. return await self._request( "POST", f"/approval/v4/instances/{instance_code}/comments" f"?user_id_type=user_id&user_id={quote(user_id)}", json_body=payload, ) # ── Upload helpers used by the asset-request form ─ # Multipart uploads to Feishu push the whole file body across the # wire; on flaky / throttled office networks a 5–10 MB phone photo # routinely exceeds the default 30 s write timeout (laptop_replacement # 上传凭证 surfaced this — multiple WriteTimeout 500s). Give writes a # much longer budget while keeping connect / read snappy so we still # fail fast on a stuck DNS or stalled response. _UPLOAD_TIMEOUT = httpx.Timeout(connect=10.0, read=30.0, write=180.0, pool=10.0) async def upload_attachment(self, upload: UploadFile) -> dict[str, Any]: """Upload a file for an approval ``attachmentV2`` widget. Same endpoint as :meth:`upload_image`, but with ``type=attachment`` so Feishu lets attachmentV2 widgets reference the resulting code. Returns ``{"code": "...", "url": "..."}``. """ try: data = await read_upload_bounded( upload, _MAX_APPROVAL_FILE_BYTES, message="approval attachment exceeds 50 MB", ) return await self.upload_attachment_bytes( filename=upload.filename or "attachment.bin", content=data, content_type=upload.content_type or "application/octet-stream", ) finally: await upload.close() async def upload_image(self, upload: UploadFile) -> dict[str, Any]: """Upload a file for use in a Feishu approval form's ``image`` widget. Hits the **approval-specific** endpoint at ``https://www.feishu.cn/approval/openapi/v2/file/upload`` — NOT ``/im/v1/images`` on ``open.feishu.cn/open-apis``. Why this distinction matters (2026-05-26): - ``/im/v1/images`` returns an ``image_key`` (``img_v3_*`` format) usable for chat messages but Feishu approval instance creation accepts the form payload, then renders the image widget as EMPTY in the approval card — the IM token isn't a valid approval file reference. - ``/approval/openapi/v2/file/upload`` returns a ``code`` (UUID-shaped, e.g. ``D93653C3-2609-4EE0-8041-61DC1D84F0B5``) that is the canonical file identifier for the image / attachment widget value array, per the official 创建审批 实例 widget value spec. Note the host is ``www.feishu.cn`` (not ``open.feishu.cn``); this is one of the few Feishu endpoints that doesn't live under ``open.feishu.cn/open-apis``. The bearer token is the same tenant_access_token used everywhere else. Returns the Feishu ``data`` block: ``{"code": "...", "url": "..."}``. Caller pulls ``code`` and shoves it into the widget value array. """ try: data = await read_upload_bounded( upload, _MAX_APPROVAL_IMAGE_BYTES, message="approval image exceeds 10 MB", ) return await self.upload_image_bytes( filename=upload.filename or "image.png", content=data, content_type=upload.content_type or "image/png", ) finally: await upload.close() async def _upload_approval_file( self, *, upload_type: str, filename: str, content: bytes, content_type: str, ) -> dict[str, Any]: """Multipart upload to Feishu approval's special file endpoint ``https://www.feishu.cn/approval/openapi/v2/file/upload``. ``upload_type`` is ``"image"`` for the image widget, ``"attachment"`` for the attachment / attachmentV2 widget — Feishu uses this to gate which widget types can reference the resulting code. """ token = await self.get_tenant_access_token() url = "https://www.feishu.cn/approval/openapi/v2/file/upload" async with httpx.AsyncClient( timeout=self._UPLOAD_TIMEOUT, trust_env=False, follow_redirects=False, ) as client: response = await client.post( url, headers={"Authorization": f"Bearer {token}"}, data={"name": filename, "type": upload_type}, files={"content": (filename, content, content_type)}, ) payload = response.json() if payload.get("code") != 0: logger.warning( "feishu approval-file upload non-zero payload: %s", payload, ) raise AppException( code="FEISHU_API_ERROR", message=( f"Feishu API error {payload.get('code')}: " f"{payload.get('msg', 'approval file upload failed')}" ), status_code=502, ) return payload.get("data", {}) async def upload_attachment_bytes( self, *, filename: str, content: bytes, content_type: str, ) -> dict[str, Any]: """Upload already-bounded attachment bytes without another copy.""" if len(content) > _MAX_APPROVAL_FILE_BYTES: raise AppException( code="UPLOAD_TOO_LARGE", message="approval attachment exceeds 50 MB", status_code=413, ) return await self._upload_approval_file( upload_type="attachment", filename=filename, content=content, content_type=content_type, ) async def upload_image_bytes( self, *, filename: str, content: bytes, content_type: str, ) -> dict[str, Any]: """Upload already-bounded image bytes without another copy.""" if len(content) > _MAX_APPROVAL_IMAGE_BYTES: raise AppException( code="UPLOAD_TOO_LARGE", message="approval image exceeds 10 MB", status_code=413, ) return await self._upload_approval_file( upload_type="image", filename=filename, content=content, content_type=content_type, ) async def upload_video_bytes( self, *, filename: str, content: bytes, content_type: str, ) -> dict[str, Any]: """Upload already-bounded video bytes without another copy.""" if len(content) > _MAX_APPROVAL_FILE_BYTES: raise AppException( code="UPLOAD_TOO_LARGE", message="approval video exceeds 50 MB", status_code=413, ) return await self._upload_approval_file( upload_type="attachment", filename=filename, content=content, content_type=content_type, ) async def upload_video(self, upload: UploadFile) -> dict[str, Any]: """Upload a video for an approval ``image`` widget with the 视频 checkbox enabled. Goes to the same approval upload endpoint as :meth:`upload_image`, but with ``type=attachment`` — the Feishu approval upload endpoint's ``type=image`` only accepts image MIMEs; videos must upload as ``type=attachment``. The returned ``code`` is still acceptable as a value entry for the image widget when the admin enabled the 视频 checkbox (Feishu renders by widget config, not by upload type). Returns ``{"code": "...", "url": "..."}``. """ try: data = await read_upload_bounded( upload, _MAX_APPROVAL_FILE_BYTES, message="approval video exceeds 50 MB", ) return await self.upload_video_bytes( filename=upload.filename or "video.mp4", content=data, content_type=upload.content_type or "video/mp4", ) finally: await upload.close() # ── Contact / custom-attribute helpers used by HRBP resolver ── async def list_custom_attrs(self) -> list[dict[str, Any]]: """Return every enterprise-level custom attribute definition. Calls ``GET /open-apis/contact/v3/custom_attrs`` with full pagination. Each entry shape (Feishu OData): {"id": "C-1", "type": "USER" | "TEXT" | "PICTURE_ENUM" | …, "options": {…}, "i18n_name": [{"locale": "zh_cn", "value": "HRBP"}, {"locale": "en_us", "value": "HRBP"}, …]} The HRBP resolver iterates this list and matches by the ``i18n_name`` whose ``value`` equals "HRBP" (or contains it case-insensitively, since some tenants spell it "HR Business Partner"). Network failures propagate — the caller is responsible for deciding the policy on missing custom_attrs (the resolver falls back to the it_admin pool and writes an audit row). """ items: list[dict[str, Any]] = [] page_token: str | None = None while True: qs = "page_size=100" if page_token: qs = f"{qs}&page_token={page_token}" data = await self._request( "GET", f"/contact/v3/custom_attrs?{qs}", ) page_items = data.get("items") or [] for item in page_items: if isinstance(item, dict): items.append(item) if not data.get("has_more"): break page_token = data.get("page_token") or None if not page_token: break return items async def get_user_with_custom_attrs( self, user_id: str, *, user_id_type: str = "user_id", ) -> dict[str, Any]: """Return the full Feishu user record including ``custom_attrs``. ``GET /open-apis/contact/v3/users/{user_id}`` — Feishu returns the user object under ``data.user``. ``user_id_type`` ∈ {``user_id``, ``open_id``, ``union_id``}; pick to match what the caller has in hand. The returned dict carries: {"user_id": "...", "name": "...", "email": "...", "custom_attrs": [{"type": "USER", "id": "C-1", "value": {"user": {"user_id": "...", "name": "..."}}}, ...]} ``custom_attrs`` is absent when the user has none defined (vs. present-but-empty); the resolver treats both as "no HRBP" and falls back to the it_admin pool. """ if not user_id: raise AppException( code="FEISHU_USER_ID_REQUIRED", message="user_id is required for get_user_with_custom_attrs", status_code=400, ) data = await self._request( "GET", f"/contact/v3/users/{user_id}?user_id_type={user_id_type}", ) user = data.get("user") if not isinstance(user, dict): raise AppException( code="FEISHU_USER_NOT_FOUND", message=f"Feishu user record not found for {user_id_type}={user_id}", status_code=404, ) return user