"""Feishu wiki → KB mirror (``source='feishu_doc'``). The Feishu wiki stays the ONLY authoring surface. This module mirrors the docx text under the configured wiki subtree(s) into ``support_kb_articles`` as machine-owned published articles, so the existing retrieval stack (semantic + keyword search, copilot grounding, deflection, MCP ``search_support_kb``) grounds on them with zero consumer changes. The mirror is disposable: no human edits these rows (guarded in ``SupportKbService``) and they are rebuildable from the wiki at any time. Wire shapes verified against the live tenant (2026-08-16, legacy-doc support verified 2026-08-20): - wiki node ``obj_edit_time`` is epoch **seconds** as a string (the helpdesk ticket API uses milliseconds — do not mix them up); - node listing paginates with ``has_more`` + ``page_token`` (the helpdesk API's 1-based ``page`` differs); - child node items carry a ready-made ``url``; the root node from ``get_node`` does not (built from the node token instead); - both upgraded ``docx`` and legacy ``doc`` nodes are searchable documents. Their raw-content endpoints return plain text that starts with the doc title; container/placeholder docs come back essentially empty and are skipped (they would only add retrieval noise). """ from __future__ import annotations import logging from collections import Counter from datetime import UTC, datetime from typing import Any import httpx from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from app.approval.feishu_client import FeishuApprovalClient from app.core.exceptions import AppException from app.domains.support.models import ( KB_ARCHIVED, KB_PUBLISHED, KB_SOURCE_FEISHU_DOC, SupportKbArticle, ) logger = logging.getLogger(__name__) SETTING_ROOT_TOKENS = "support.feishu_kb.root_tokens" SETTING_LAST_SYNC = "support.feishu_kb.last_sync" SETTING_INCLUDE_POLICY = "support.feishu_kb.include_policy_notices" # Sentinel external_root_token grouping every mirrored 制度中心 doc — the # archive sweep treats the policy library as one all-or-nothing scope. POLICY_SCOPE = "policy_notices" # 制度中心 token provider — REGISTERED FROM ABOVE (the coordinator layer # plugs itself in via ``register_policy_token_provider`` at process start). # Domains never import coordinators, so the dependency is inverted: this # module only holds the hook. When None (e.g. a stripped-down test app), # policy mirroring degrades to "no policy docs" rather than erroring. PolicyTokenProvider = Any # Callable[[AsyncSession], Awaitable[tuple[list[str], int]]] policy_token_provider: PolicyTokenProvider | None = None _BASE_URL = "https://open.feishu.cn/open-apis" _PAGE_SIZE = 50 # Runaway guards: a KB subtree is dozens of docs, not thousands. Hitting a # cap is reported in the summary (never a silent truncation). _MAX_NODES_PER_ROOT = 2000 _MAX_ROOTS = 20 _TITLE_MAX = 255 _BODY_MAX = 16000 # keep in sync with kb_service._BODY_MAX # raw_content of a container/placeholder doc is just its title + whitespace. # Anything with fewer than this many non-title characters carries no # answerable content and is excluded from the published mirror. _MIN_BODY_CHARS = 20 # Vectors produced in one sync run. One embedding is one HTTP round-trip, so # this bounds the run's tail (a full first mirror of ~300 docs finishes inside # it); anything left over is reported as ``embed_pending`` and picked up by the # next run. _MAX_EMBEDS_PER_SYNC = 1000 class FeishuDocsClient: """Read-only wiki/docx client on the existing chat-approval app (token cache reused via ``FeishuApprovalClient`` — no new credential storage). Needs tenant scopes ``wiki:wiki:readonly`` + ``docx:document:readonly`` and space membership for the KB subtree.""" def __init__(self, *, app_id: str, app_secret: str) -> None: self._token_client = FeishuApprovalClient( app_id=app_id, app_secret=app_secret, ) async def _get(self, path: str, *, params: dict[str, Any] | None = None) -> dict[str, Any]: token = await self._token_client.get_tenant_access_token() async with httpx.AsyncClient( timeout=30, trust_env=False, follow_redirects=False, ) as client: resp = await client.get( f"{_BASE_URL}{path}", params=params or {}, headers={"Authorization": f"Bearer {token}"}, ) payload = resp.json() if payload.get("code") != 0: logger.warning("feishu docs: GET %s returned non-zero: %s", path, payload) raise AppException( code="FEISHU_DOCS_API_ERROR", message=( f"Feishu API error {payload.get('code')}: " f"{payload.get('msg', 'wiki/docx request failed')}" ), status_code=502, ) return payload.get("data") or {} async def get_node(self, node_token: str) -> dict[str, Any]: data = await self._get( "/wiki/v2/spaces/get_node", params={"token": node_token}, ) return data.get("node") or {} async def list_child_nodes( self, space_id: str, parent_node_token: str, ) -> list[dict[str, Any]]: items: list[dict[str, Any]] = [] page_token: str | None = None while True: params: dict[str, Any] = { "page_size": _PAGE_SIZE, "parent_node_token": parent_node_token, } if page_token: params["page_token"] = page_token data = await self._get( f"/wiki/v2/spaces/{space_id}/nodes", params=params, ) items.extend(data.get("items") or []) if not data.get("has_more"): return items page_token = data.get("page_token") async def raw_content(self, doc_token: str, *, obj_type: str = "docx") -> str: """Fetch plain text from either Feishu document generation. A wiki tree can mix old ``doc`` with upgraded ``docx`` nodes. The prior mirror only included docx, silently making every legacy document look like it did not exist. Keep the type decision at this boundary; the rest of the sync intentionally treats both as the same KB source. """ path = ( f"/doc/v2/{doc_token}/raw_content" if obj_type == "doc" else f"/docx/v1/documents/{doc_token}/raw_content" ) data = await self._get(path) return str(data.get("content") or "") async def _build_client(db: AsyncSession) -> FeishuDocsClient: from app.integrations.compat import get_feishu_config config = await get_feishu_config(session=db) app_id = str(config.get("app_id") or "") app_secret = str(config.get("app_secret") or "") if not app_id or not app_secret: raise AppException( code="FEISHU_NOT_CONFIGURED", message="飞书应用凭证未配置(chat_approval 连接器)", status_code=503, ) return FeishuDocsClient(app_id=app_id, app_secret=app_secret) def _node_url(node: dict[str, Any]) -> str: url = str(node.get("url") or "").strip() if url: return url[:512] return f"https://pingcap.feishu.cn/wiki/{node.get('node_token')}"[:512] def _edit_time(node: dict[str, Any]) -> int: try: return int(node.get("obj_edit_time") or 0) except (TypeError, ValueError): return 0 def _strip_title(content: str, title: str) -> str: """The doc body without its leading title line — the emptiness signal.""" text = (content or "").strip() t = (title or "").strip() if t and text.startswith(t): text = text[len(t) :] return text.strip() async def _walk_root( client: FeishuDocsClient, root_token: str, ) -> tuple[list[dict[str, Any]], Counter, bool]: """BFS the subtree under ``root_token``. Returns document nodes (``doc`` and ``docx``; root itself included when applicable), skipped-by-type counter, truncated?. API-only: no DB access, so callers keep transactions closed around it.""" root = await client.get_node(root_token) if not root: raise AppException( code="FEISHU_DOCS_NODE_NOT_FOUND", message=f"知识库根节点不存在或应用无权限: {root_token}", status_code=502, ) space_id = str(root.get("space_id") or "") docs: list[dict[str, Any]] = [] skipped: Counter = Counter() if root.get("obj_type") in {"doc", "docx"}: docs.append(root) else: skipped[str(root.get("obj_type"))] += 1 queue: list[str] = [str(root.get("node_token"))] if root.get("has_child") else [] seen_nodes = 0 truncated = False while queue: parent = queue.pop(0) for item in await client.list_child_nodes(space_id, parent): seen_nodes += 1 if seen_nodes > _MAX_NODES_PER_ROOT: truncated = True queue.clear() break if item.get("has_child"): queue.append(str(item.get("node_token"))) if item.get("obj_type") in {"doc", "docx"}: docs.append(item) else: skipped[str(item.get("obj_type"))] += 1 return docs, skipped, truncated async def sync_feishu_kb( db: AsyncSession, *, client: FeishuDocsClient | None = None, ) -> dict[str, Any]: """Mirror every configured wiki root subtree into the KB. Incremental by ``obj_edit_time``; per-doc commits (no transaction spans a Feishu call); the disappeared→archive sweep runs ONLY over roots whose walk fully succeeded (a half-failed walk must never archive live docs). Returns a summary dict suitable for ``last_result_summary``.""" from app.integrations.models import PlatformSetting from app.integrations.service import get_platform_setting roots_raw = await get_platform_setting(SETTING_ROOT_TOKENS, [], session=db) or [] roots = [] for raw in roots_raw: if isinstance(raw, dict): token = str(raw.get("token") or raw.get("root_token") or "").strip() else: token = str(raw).strip() if token and token not in roots: roots.append(token) roots = roots[:_MAX_ROOTS] # 制度中心 integration (PR-5): published policy entries' wiki docs join # the mirror under one sentinel scope, so policy questions ground on the # actual policy text. Toggle-off is an active state: the sentinel scope # is then swept with zero seen docs, archiving previously mirrored # policy rows (out of search, resurrect on re-enable). include_policy = bool( await get_platform_setting(SETTING_INCLUDE_POLICY, True, session=db), ) policy_tokens: list[str] = [] policy_skipped_non_wiki = 0 if include_policy and policy_token_provider is not None: policy_tokens, policy_skipped_non_wiki = await policy_token_provider(db) # Snapshot existing mirror state into plain values, then CLOSE the # transaction — the walk below is a long chain of external calls and this # repo has been burned by transactions idling across them (MySQL 2013). existing_rows = ( await db.execute( select( SupportKbArticle.id, SupportKbArticle.external_obj_token, SupportKbArticle.external_root_token, SupportKbArticle.external_edited_at, SupportKbArticle.status, SupportKbArticle.title, SupportKbArticle.body, ).where(SupportKbArticle.source == KB_SOURCE_FEISHU_DOC) ) ).all() existing = { row.external_obj_token: { "id": row.id, "root": row.external_root_token, "edited_at": row.external_edited_at, "status": row.status, "empty_stub": len(_strip_title(row.body or "", row.title or "")) < _MIN_BODY_CHARS, } for row in existing_rows if row.external_obj_token } if not roots and not policy_tokens and not existing: # Nothing to mirror and nothing previously mirrored to sweep. await db.commit() return {"skipped": "not_configured", "roots": 0} if client is None: # Build the client (its config read is a SELECT on this session) # BEFORE closing the transaction — anything after this commit must # not touch the DB until Phase 2, or the walk rides an idle # transaction across minutes of external calls (MySQL 2013). client = await _build_client(db) await db.commit() # Phase 1 — API only: walk each target, fetch content for new/changed # docs. Targets are (token, scope): configured roots scope to themselves; # every policy doc scopes to the POLICY_SCOPE sentinel so the archive # sweep treats the policy library as one all-or-nothing group. per_root: list[dict[str, Any]] = [] ok_roots: list[str] = [] # complete walks only — the archive-sweep scope walked_roots = 0 # walks that returned at all (truncated ones included) seen_tokens: set[str] = set() pending: list[dict[str, Any]] = [] # docs to upsert (content fetched) unchanged = 0 skipped_empty = 0 policy_docs = 0 policy_ok = True # any policy walk failing/truncating vetoes the group targets: list[tuple[str, str]] = [(r, r) for r in roots] targets += [(tok, POLICY_SCOPE) for tok in policy_tokens] async def _ingest(node: dict[str, Any], scope: str) -> int: """Queue one docx node for upsert; returns 1 if content was fetched.""" nonlocal unchanged, skipped_empty, policy_docs obj_token = str(node.get("obj_token") or "") if not obj_token or obj_token in seen_tokens: return 0 seen_tokens.add(obj_token) if scope == POLICY_SCOPE: policy_docs += 1 edited_at = _edit_time(node) prior = existing.get(obj_token) # An archived row is not evidence that this node is absent: it may # have been falsely swept by an older sync, then still carry the same # Feishu edit watermark. Treating it as unchanged forever made a live # document remain invisible even while the configured subtree listed # it on every run. Only a *published* row at the same watermark can # skip the fetch; an archived row must be re-read and resurrected. if ( prior is not None and ( prior["status"] == KB_PUBLISHED or (prior["status"] == KB_ARCHIVED and prior["empty_stub"]) ) and prior["edited_at"] == edited_at ): unchanged += 1 return 0 title = str(node.get("title") or "").strip() or obj_token content = await client.raw_content(obj_token, obj_type=str(node.get("obj_type") or "docx")) empty = len(_strip_title(content, title)) < _MIN_BODY_CHARS if empty: # Container/placeholder doc — excluded from the published # mirror, but still upserted as an ARCHIVED stub so its # watermark persists (otherwise every run re-fetches it). skipped_empty += 1 pending.append( { "obj_token": obj_token, "empty": empty, "title": title[:_TITLE_MAX], "body": content.strip()[:_BODY_MAX] or title[:_BODY_MAX], "url": _node_url(node), "edited_at": edited_at, "root": scope, } ) return 1 walked_config_roots = 0 # configured-root walks that returned at all for root_token, scope in targets: try: docs, skipped, truncated = await _walk_root(client, root_token) fetched = 0 for node in docs: fetched += await _ingest(node, scope) # A truncated walk is an INCOMPLETE walk: its upserts are kept, # but it must never qualify the scope for the disappeared→archive # sweep — docs beyond the cap were not visited, not deleted. walked_roots += 1 if scope == POLICY_SCOPE: policy_ok = policy_ok and not truncated else: walked_config_roots += 1 if not truncated: ok_roots.append(scope) per_root.append( { "root": root_token, **({"scope": scope} if scope != root_token else {}), "docs": len(docs), "fetched": fetched, "skipped_types": dict(skipped), **({"truncated": True} if truncated else {}), } ) except Exception as e: # noqa: BLE001 — root isolation; summary carries the error logger.exception("feishu kb sync: walk failed for root %s", root_token) if scope == POLICY_SCOPE: policy_ok = False per_root.append( { "root": root_token, **({"scope": scope} if scope != root_token else {}), "error": f"{type(e).__name__}: {e}", } ) # The policy group qualifies for the archive sweep as one unit: every # policy walk completed (or the toggle is OFF, which actively sweeps # previously mirrored policy rows out of search). A process with NO # provider registered must never sweep — an empty token list there # means "wiring absent", not "no published policies". if include_policy: if policy_token_provider is not None and policy_ok: ok_roots.append(POLICY_SCOPE) else: ok_roots.append(POLICY_SCOPE) # Phase 2 — DB writes, one commit per doc (short transactions; a re-run # after a mid-batch failure continues idempotently by watermark). created = updated = archived = vectors_cleared = 0 for item in pending: prior = existing.get(item["obj_token"]) if item["empty"]: if prior is None: # Archived stub: carries the watermark, never enters search # (search is published-only), resurrects if content arrives. db.add( SupportKbArticle( title=item["title"], body=item["body"], category="other", status=KB_ARCHIVED, source=KB_SOURCE_FEISHU_DOC, external_obj_token=item["obj_token"], external_url=item["url"], external_root_token=item["root"], external_edited_at=item["edited_at"], ) ) else: art = await db.get(SupportKbArticle, prior["id"]) if art is not None: if art.status != KB_ARCHIVED: art.status = KB_ARCHIVED archived += 1 if art.embedding is not None: art.embedding = None art.embedding_model = None art.embedded_at = None vectors_cleared += 1 art.external_edited_at = item["edited_at"] art.external_root_token = item["root"] await db.commit() continue if prior is None: db.add( SupportKbArticle( title=item["title"], body=item["body"], category="other", status=KB_PUBLISHED, source=KB_SOURCE_FEISHU_DOC, external_obj_token=item["obj_token"], external_url=item["url"], external_root_token=item["root"], external_edited_at=item["edited_at"], published_at=datetime.now(UTC), ) ) created += 1 else: art = await db.get(SupportKbArticle, prior["id"]) if art is None: await db.commit() continue body_changed = art.body != item["body"] or art.title != item["title"] art.title = item["title"] art.body = item["body"] art.external_url = item["url"] art.external_root_token = item["root"] art.external_edited_at = item["edited_at"] # Status is machine-owned for mirrored rows: a doc that exists # with real content is published (this also resurrects rows the # empty/disappeared sweeps archived once the doc comes back). if art.status != KB_PUBLISHED: art.status = KB_PUBLISHED art.published_at = art.published_at or datetime.now(UTC) if body_changed: art.embedding = None art.embedding_model = None art.embedded_at = None updated += 1 await db.commit() # Phase 3 — archive mirrored articles whose doc disappeared (scoped to # roots that walked successfully end-to-end) PLUS rows whose root was # REMOVED from the config: the mirror follows the config, so deleting a # root token is an active decommission — without this, its articles # lingered published-and-stale forever with no way to take them down # (manual archive is refused for mirrored rows by design). known_scopes = set(roots) | {POLICY_SCOPE} if ok_roots or existing: gone = [ info["id"] for token, info in existing.items() if ( (info["root"] in ok_roots and token not in seen_tokens) or info["root"] not in known_scopes ) ] for art_id in gone: art = await db.get(SupportKbArticle, art_id) if art is not None: if art.status != KB_ARCHIVED: art.status = KB_ARCHIVED archived += 1 if art.embedding is not None: art.embedding = None art.embedding_model = None art.embedded_at = None vectors_cleared += 1 # Reset the watermark: a swept doc that later REAPPEARS # unchanged (same obj_edit_time — e.g. a policy entry # unpublished then republished, or a wiki node moved back) # must not be skipped as "unchanged" forever; a null # watermark forces a refetch → resurrection on next sight. art.external_edited_at = None if gone: await db.commit() # Phase 4 — embed anything published that lacks a vector (covers the rows # this run created/invalidated; reuses the bounded, per-row-commit backfill, # looped until the backlog is empty so a first sync of a 300-doc library # doesn't stop at the 100-per-batch cap; graceful no-op when the embeddings # backend is off). ``embed_pending`` in the summary reports whatever the # per-run ceiling left behind — never a silent truncation. from app.domains.support.kb_service import SupportKbService embed_result = await SupportKbService(db).drain_embeddings( batch=100, max_total=_MAX_EMBEDS_PER_SYNC, ) vector_stats = await SupportKbService(db).embedding_stats() summary = { "roots": len(roots), "roots_ok": len([r for r in ok_roots if r != POLICY_SCOPE]), "include_policy_notices": include_policy, "policy_docs": policy_docs, "policy_skipped_non_wiki": policy_skipped_non_wiki, "docs_seen": len(seen_tokens), "created": created, "updated": updated, "unchanged": unchanged, "archived": archived, "skipped_empty": skipped_empty, "embedded": embed_result.get("embedded", 0), "embed_pending": embed_result.get("pending", 0), "vectorized_total": vector_stats["vectorized_total"], "vectorized_capacity": vector_stats["vectorized_capacity"], "vectors_cleared": vectors_cleared, "details": per_root, "ran_at": datetime.now(UTC).isoformat(), } await db.merge(PlatformSetting(key=SETTING_LAST_SYNC, value=summary)) await db.commit() failed_walks = [d["root"] for d in per_root if d.get("error")] if failed_walks: # Partial failure must be LOUD: a root whose walk keeps erroring # (revoked permission, deleted node) would otherwise stay status=ok # forever while its subtree silently stops updating — the exact # "skip recorded as ok" failure mode this repo has been burned by. # The summary above is already persisted, and the sync is idempotent, # so raising only affects the job status + failure notifier. raise RuntimeError( f"feishu kb sync: {len(failed_walks)} walk(s) failed: " + ", ".join(failed_walks[:5]), ) if (roots and not walked_config_roots) or (targets and not walked_roots): # Total failure must raise so the scheduled-job run lands # status=failed and the failure notifier fires (handlers must raise, # not return a failed-looking dict). Two total-failure shapes: every # walk raised, OR every CONFIGURED root walk raised (a lucky policy # doc succeeding must not mask the root mirror being down). A # truncated-but-returned walk is a partial success, not a failure. raise RuntimeError( f"feishu kb sync: root mirror failed " f"({walked_config_roots}/{len(roots)} roots walked, " f"{walked_roots}/{len(targets)} walks returned)", ) return summary async def run_feishu_kb_sync_once() -> None: """Background entrypoint for the admin 「立即同步」 button — fresh session, serialized against the cron run by the same distributed lock.""" from app.core.database import async_session_factory from app.core.distributed_lock import LockBusy, distributed_lock try: # auto_renew: a first full sync (walk + per-doc embedding) can exceed # the TTL; a lapsed lock lets cron and the admin button race and # collide on ux_support_kb_external_obj (1062). async with distributed_lock("support_feishu_kb_sync", ttl=600, auto_renew=True): async with async_session_factory() as db: await sync_feishu_kb(db) except LockBusy: logger.info("feishu kb sync: another run holds the lock — skipping") except Exception: # noqa: BLE001 — spawn() logs, but keep the reason local too logger.exception("feishu kb sync (manual) failed") raise __all__ = [ "POLICY_SCOPE", "SETTING_INCLUDE_POLICY", "FeishuDocsClient", "SETTING_LAST_SYNC", "SETTING_ROOT_TOKENS", "run_feishu_kb_sync_once", "sync_feishu_kb", ]