update

.env

@@ -1 +1,10 @@
 HUOBANYUN_API_KEY=emdYCszTIUrczBf2wOPGQ553J3OO9NCKKnLGJEK9
+FEISHU_APPROVAL_CODES=ECD8CE34-AA80-4A4F-B4C8-8510A7126490,BB944139-432F-4AC2-AD27-81C2F738E7C3,D7252659-47B6-4312-AC16-ECDE87FDB553,93F09E2D-B418-458D-A92D-10B56B53F45E
+FEISHU_APPROVAL_CODE_PUBLIC=ECD8CE34-AA80-4A4F-B4C8-8510A7126490,BB944139-432F-4AC2-AD27-81C2F738E7C3
+FEISHU_APPROVAL_CODE_PRIVATE=D7252659-47B6-4312-AC16-ECDE87FDB553,93F09E2D-B418-458D-A92D-10B56B53F45E
+FEISHU_APPROVAL_EVENT_KEY=approval_instance
+FEISHU_PROJECT_NO_FIELD_CODE=proj_id
+HUOBANYUN_ORDER_STATUS_DONE_ID=3
+HUOBANYUN_ORDER_STATUS_DONE_NAME=已完结
+FEISHU_APP_ID=cli_a90b035fd4799cb5
+FEISHU_APP_SECRET=O729hnbQARM2DHncWUd51eFHF6TDZAc3

app/clients/feishu_client.py

@@ -44,3 +44,23 @@ class FeishuClient:
             resp = await client.get(url, headers=headers)
             resp.raise_for_status()
             return resp.json()
+
+    async def subscribe_approval(self, approval_code: str) -> Dict[str, Any]:
+        token = await self._get_tenant_token()
+        url = (
+            "https://open.feishu.cn/open-apis/approval/v4/approvals/"
+            f"{approval_code}/subscribe"
+        )
+        headers = {"Authorization": f"Bearer {token}"}
+        async with httpx.AsyncClient(timeout=self.settings.request_timeout) as client:
+            resp = await client.post(url, headers=headers)
+            data = {}
+            try:
+                data = resp.json()
+            except Exception:
+                data = {}
+            code = data.get("code")
+            if resp.status_code != 200 and code != 1390007:
+                logger.error("订阅审批定义失败: status=%s body=%s", resp.status_code, resp.text)
+                resp.raise_for_status()
+            return data

app/clients/feishu_ws_client.py

@@ -3,51 +3,75 @@ from __future__ import annotations
 import asyncio
 import json
 import logging
+import threading
 from typing import Awaitable, Callable, Optional
 
+import lark_oapi as lark
+
 from app.config.settings import get_settings
 
 logger = logging.getLogger(__name__)
 
 
 class FeishuWsClient:
-    def __init__(self, handler: Callable[[dict], Awaitable[None]]) -> None:
+    def __init__(
+        self,
+        handler: Callable[[dict], Awaitable[None]],
+        main_loop: Optional[asyncio.AbstractEventLoop] = None,
+    ) -> None:
         self.settings = get_settings()
         self._handler = handler
-        self._task: Optional[asyncio.Task] = None
-        self._stopped = asyncio.Event()
+        self._main_loop = main_loop
+        self._thread: Optional[threading.Thread] = None
 
-    async def _run(self) -> None:
+    def _run(self) -> None:
+        loop = asyncio.new_event_loop()
+        asyncio.set_event_loop(loop)
         try:
-            import websockets  # type: ignore
+            from lark_oapi.ws import client as ws_client  # type: ignore
+            ws_client.loop = loop
         except Exception as exc:
-            logger.error("WebSocket 依赖缺失: %s", exc)
+            logger.error("设置飞书 WS 事件循环失败: %s", exc)
+        if not self.settings.feishu_app_id or not self.settings.feishu_app_secret:
+            logger.error("缺少飞书 APP_ID 或 APP_SECRET，无法启动长连接")
+            return
+        if not self.settings.feishu_approval_event_key:
+            logger.error("缺少 FEISHU_APPROVAL_EVENT_KEY，无法订阅审批事件")
             return
 
-        url = self.settings.feishu_ws_url
-        if not url:
-            logger.info("未配置飞书 WebSocket 地址，跳过连接")
-            return
+        def do_customized_event(data: lark.CustomizedEvent) -> None:
+            try:
+                payload = lark.JSON.marshal(data)
+                if isinstance(payload, str):
+                    payload = json.loads(payload)
+                logger.info("收到飞书事件: %s", payload)
+                if self._main_loop and self._main_loop.is_running():
+                    asyncio.run_coroutine_threadsafe(self._handler(payload), self._main_loop)
+                    logger.info("飞书事件已投递主循环")
+                else:
+                    logger.error("主事件循环不可用，无法处理飞书回调")
+            except Exception as exc:
+                logger.error("处理飞书事件失败: %s", exc)
 
-        while not self._stopped.is_set():
-            try:
-                async with websockets.connect(url, ping_interval=20, ping_timeout=20) as ws:
-                    logger.info("飞书 WebSocket 已连接")
-                    async for message in ws:
-                        try:
-                            payload = json.loads(message)
-                            await self._handler(payload)
-                        except Exception as exc:
-                            logger.error("处理 WebSocket 消息失败: %s", exc)
-            except Exception as exc:
-                logger.error("WebSocket 连接异常: %s", exc)
-                await asyncio.sleep(3)
+        event_handler = (
+            lark.EventDispatcherHandler.builder("", "")
+            .register_p1_customized_event(
+                self.settings.feishu_approval_event_key, do_customized_event
+            )
+            .build()
+        )
+        client = lark.ws.Client(
+            self.settings.feishu_app_id,
+            self.settings.feishu_app_secret,
+            event_handler=event_handler,
+            log_level=lark.LogLevel.INFO,
+        )
+        client.start()
 
     def start(self) -> None:
-        if self._task is None:
-            self._task = asyncio.create_task(self._run())
+        if self._thread is None:
+            self._thread = threading.Thread(target=self._run, daemon=True)
+            self._thread.start()
 
     async def stop(self) -> None:
-        self._stopped.set()
-        if self._task:
-            await self._task
+        return

app/clients/huobanyun_client.py

@@ -60,3 +60,7 @@ class HuobanyunClient:
     async def list_items(self, payload: Dict[str, Any]) -> Dict[str, Any]:
         url = f"{self._base_url()}/openapi/v1/item/list"
         return await self._request("POST", url, payload)
+
+    async def update_item(self, item_id: str, payload: Dict[str, Any]) -> Dict[str, Any]:
+        url = f"{self._base_url()}/openapi/v1/item/{item_id}"
+        return await self._request("PUT", url, payload)

app/config/settings.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import os
+from typing import List
 
 from dotenv import load_dotenv
 from dataclasses import dataclass
@@ -14,12 +15,20 @@ class Settings:
     feishu_verify_token: str
     feishu_encrypt_key: str
     feishu_ws_url: str
+    feishu_approval_code: str
+    feishu_approval_codes: List[str]
+    feishu_approval_code_public: List[str]
+    feishu_approval_code_private: List[str]
+    feishu_approval_event_key: str
+    feishu_project_no_field_code: str
 
     huobanyun_app_id: str
     huobanyun_app_secret: str
     huobanyun_token: str
     huobanyun_api_key: str
     huobanyun_base_url: str
+    huobanyun_order_status_done_id: str
+    huobanyun_order_status_done_name: str
 
     log_level: str
     log_rotation: str
@@ -35,6 +44,11 @@ def _env(name: str, default: str = "") -> str:
     return os.getenv(name, default).strip()
 
 
+def _env_list(name: str) -> List[str]:
+    raw = os.getenv(name, "")
+    return [item.strip() for item in raw.split(",") if item.strip()]
+
+
 @lru_cache(maxsize=1)
 def get_settings() -> Settings:
     load_dotenv()
@@ -44,11 +58,19 @@ def get_settings() -> Settings:
         feishu_verify_token=_env("FEISHU_VERIFY_TOKEN"),
         feishu_encrypt_key=_env("FEISHU_ENCRYPT_KEY"),
         feishu_ws_url=_env("FEISHU_WS_URL"),
+        feishu_approval_code=_env("FEISHU_APPROVAL_CODE"),
+        feishu_approval_codes=_env_list("FEISHU_APPROVAL_CODES"),
+        feishu_approval_code_public=_env_list("FEISHU_APPROVAL_CODE_PUBLIC"),
+        feishu_approval_code_private=_env_list("FEISHU_APPROVAL_CODE_PRIVATE"),
+        feishu_approval_event_key=_env("FEISHU_APPROVAL_EVENT_KEY"),
+        feishu_project_no_field_code=_env("FEISHU_PROJECT_NO_FIELD_CODE"),
         huobanyun_app_id=_env("HUOBANYUN_APP_ID"),
         huobanyun_app_secret=_env("HUOBANYUN_APP_SECRET"),
         huobanyun_token=_env("HUOBANYUN_TOKEN"),
         huobanyun_api_key=_env("HUOBANYUN_API_KEY"),
         huobanyun_base_url=_env("HUOBANYUN_BASE_URL"),
+        huobanyun_order_status_done_id=_env("HUOBANYUN_ORDER_STATUS_DONE_ID"),
+        huobanyun_order_status_done_name=_env("HUOBANYUN_ORDER_STATUS_DONE_NAME", "已完成"),
         log_level=_env("LOG_LEVEL", "INFO"),
         log_rotation=_env("LOG_ROTATION", "size"),
         log_max_bytes=int(_env("LOG_MAX_BYTES", "10485760")),

app/main.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import asyncio
 import logging
 import os
 import time
@@ -30,12 +31,18 @@ async def on_startup() -> None:
     setup_logging(LOG_DIR)
 
     approval_service = ApprovalSyncService()
+    await approval_service.ensure_approval_subscription()
 
     async def handler(payload: dict) -> None:
+        logger.info("开始处理飞书事件")
+        try:
             event_payload = payload.get("event", payload)
             await approval_service.handle_event(event_payload)
+            logger.info("完成处理飞书事件")
+        except Exception as exc:
+            logger.exception("处理飞书事件异常: %s", exc)
 
-    ws_client = FeishuWsClient(handler)
+    ws_client = FeishuWsClient(handler, asyncio.get_running_loop())
     ws_client.start()
     app.state.ws_client = ws_client
 

app/schemas/feishu_external.py

@@ -18,6 +18,7 @@ class FeishuExternalQueryRequest(BaseModel):
     user_id: Optional[str] = None
     employee_id: Optional[str] = None
     token: Optional[str] = None
+    approval_code: Optional[str] = None
     linkage_params: Dict[str, Any] = {}
     page_token: Optional[str] = None
     query: Optional[str] = None

app/services/approval_sync_service.py

@@ -5,15 +5,17 @@ import time
 from typing import Any, Dict, Optional, Tuple
 
 from app.clients.feishu_client import FeishuClient
-from app.clients.huobanyun_client import HuobanyunClient
+from app.config.settings import get_settings
+from app.services.huobanyun_service import HuobanyunService
 
 logger = logging.getLogger(__name__)
 
 
 class ApprovalSyncService:
     def __init__(self) -> None:
+        self.settings = get_settings()
         self.feishu_client = FeishuClient()
-        self.huobanyun_client = HuobanyunClient()
+        self.huobanyun_service = HuobanyunService()
         self._seen: Dict[Tuple[str, str], float] = {}
 
     def _cleanup_seen(self) -> None:
@@ -27,32 +29,127 @@ class ApprovalSyncService:
             return False
         return status.upper() in {"APPROVED", "REJECTED", "CANCELED", "CANCELLED", "DONE"}
 
-    async def handle_event(self, event: Dict[str, Any]) -> None:
-        instance_id = event.get("instance_id") or event.get("instanceId") or ""
-        status = event.get("status") or event.get("instance_status") or ""
-        if not instance_id or not self._is_done(status):
+    def _extract_project_no_from_detail(self, detail: Dict[str, Any]) -> str:
+        field_code = self.settings.feishu_project_no_field_code
+        if not field_code:
+            return ""
+        data = detail.get("data", {})
+        form = data.get("form")
+        if isinstance(form, str):
+            try:
+                import json
+                form = json.loads(form)
+            except Exception:
+                form = None
+        if isinstance(form, list):
+            for item in form:
+                if not isinstance(item, dict):
+                    continue
+                code = (
+                    item.get("custom_id")
+                    or item.get("id")
+                    or item.get("field_code")
+                    or item.get("code")
+                )
+                if code == field_code:
+                    value = item.get("value") or item.get("text") or item.get("name")
+                    if isinstance(value, list) and value:
+                        return str(value[0])
+                    return "" if value is None else str(value)
+        if isinstance(form, dict):
+            value = form.get(field_code)
+            if isinstance(value, list) and value:
+                return str(value[0])
+            return "" if value is None else str(value)
+        return ""
+
+    async def ensure_approval_subscription(self) -> None:
+        codes = self.settings.feishu_approval_codes or (
+            [self.settings.feishu_approval_code] if self.settings.feishu_approval_code else []
+        )
+        if not codes:
+            logger.error("未配置 FEISHU_APPROVAL_CODE，无法订阅审批定义")
+            return
+        for approval_code in codes:
+            try:
+                resp = await self.feishu_client.subscribe_approval(approval_code)
+                code = resp.get("code")
+                msg = resp.get("msg")
+                if code in (0, 1390007):
+                    logger.info("审批定义订阅成功或已存在: %s", msg)
+                else:
+                    logger.error("审批定义订阅失败: code=%s msg=%s", code, msg)
+            except Exception as exc:
+                logger.error("订阅审批定义异常: %s", exc)
+
+    async def handle_approval_event(self, event: Dict[str, Any]) -> None:
+        payload = event.get("event", event)
+        logger.info("审批事件内容: %s", payload)
+        approval_code = payload.get("approval_code") or payload.get("approvalCode")
+        codes = self.settings.feishu_approval_codes or (
+            [self.settings.feishu_approval_code] if self.settings.feishu_approval_code else []
+        )
+        if codes and approval_code not in codes:
+            logger.info("审批定义不匹配，跳过: %s", approval_code)
+            return
+
+        instance_id = (
+            payload.get("instance_id")
+            or payload.get("instanceId")
+            or payload.get("instance_code")
+            or payload.get("instanceCode")
+            or ""
+        )
+        status = payload.get("status") or payload.get("instance_status") or payload.get("instanceStatus")
+        if not instance_id or not status:
             logger.info("审批事件未结束或缺少实例ID，跳过处理")
             return
 
-        dedupe_key = (instance_id, status)
+        logger.info("审批事件实例: id=%s status=%s", instance_id, status)
+
+        dedupe_key = (instance_id, str(status))
         self._cleanup_seen()
         if dedupe_key in self._seen:
             logger.info("审批事件重复，跳过: %s", dedupe_key)
             return
         self._seen[dedupe_key] = time.time()
 
-        detail = {}
         try:
             detail = await self.feishu_client.get_approval_instance(instance_id)
+            logger.info("审批实例详情: %s", detail)
         except Exception as exc:
             logger.error("拉取审批详情失败: %s", exc)
+            return
 
-        record_id = event.get("record_id") or instance_id
-        fields = event.get("fields") or detail.get("data", {}).get("form", {})
-        payload = {"record_id": record_id, "fields": fields}
+        detail_status = (
+            detail.get("data", {}).get("status")
+            or detail.get("data", {}).get("instance_status")
+            or status
+            or ""
+        )
+        if str(detail_status).upper() != "APPROVED":
+            logger.info("审批实例未通过，跳过回写: %s", detail_status)
+            return
+
+        project_no = self._extract_project_no_from_detail(detail)
+        if not project_no:
+            logger.error("未能从审批详情提取项目单号")
+            return
+        logger.info("提取项目单号: %s", project_no)
+
+        item_id = await self.huobanyun_service.find_item_by_project_no(project_no)
+        if not item_id:
+            logger.error("未找到对应伙伴云项目: %s", project_no)
+            return
+        logger.info("找到伙伴云项目: %s", item_id)
 
         try:
-            await self.huobanyun_client.writeback(payload)
-            logger.info("审批写回伙伴云成功: %s", record_id)
+            await self.huobanyun_service.update_order_status(item_id)
+            if approval_code:
+                await self.huobanyun_service.update_linked_flags(item_id, approval_code)
+            logger.info("审批完成回写成功: %s", item_id)
         except Exception as exc:
-            logger.error("审批写回伙伴云失败: %s", exc)
+            logger.error("审批完成回写失败: %s", exc)
+
+    async def handle_event(self, event: Dict[str, Any]) -> None:
+        await self.handle_approval_event(event)

app/services/huobanyun_service.py

@@ -3,6 +3,7 @@ from __future__ import annotations
 import logging
 from typing import Any, Dict, List, Optional
 
+from app.config.settings import get_settings
 from app.clients.huobanyun_client import HuobanyunClient
 from app.schemas.feishu_external import (
     FeishuExternalItem,
@@ -16,6 +17,7 @@ logger = logging.getLogger(__name__)
 
 class HuobanyunService:
     def __init__(self) -> None:
+        self.settings = get_settings()
         self.client = HuobanyunClient()
 
     def _extract_value(self, value: Any) -> Any:
@@ -50,6 +52,50 @@ class HuobanyunService:
         }
         return mapping.get(key, key)
 
+    async def find_item_by_project_no(self, project_no: str) -> str:
+        table_id = "2100000015544940"
+        project_no_field_id = self._resolve_field_key("项目单号")
+        payload = {
+            "table_id": table_id,
+            "limit": 1,
+            "offset": 0,
+            "filter": {
+                "and": [
+                    {"field": project_no_field_id, "query": {"eq": str(project_no)}}
+                ]
+            },
+        }
+        data = await self.client.list_items(payload)
+        data_block = data.get("data", {}) if isinstance(data, dict) else {}
+        items = data_block.get("items", []) if isinstance(data_block, dict) else []
+        if not items:
+            return ""
+        return str(items[0].get("item_id", ""))
+
+    async def update_order_status(self, item_id: str) -> None:
+        status_field_id = self._resolve_field_key("订单状态")
+        status_id = self.settings.huobanyun_order_status_done_id
+        status_name = self.settings.huobanyun_order_status_done_name or "已完成"
+        status_value = status_id or status_name
+        if not status_value:
+            logger.error("未配置订单状态已完成选项值，跳过更新")
+            return
+        payload = {"fields": {status_field_id: [str(status_value)]}}
+        await self.client.update_item(item_id, payload)
+
+    async def update_linked_flags(self, item_id: str, approval_code: str) -> None:
+        public_codes = set(self.settings.feishu_approval_code_public)
+        private_codes = set(self.settings.feishu_approval_code_private)
+        fields: Dict[str, Any] = {}
+        if approval_code in public_codes:
+            fields[self._resolve_field_key("是否已关联对公付款审批")] = True
+        if approval_code in private_codes:
+            fields[self._resolve_field_key("是否已关联对私付款审批")] = True
+        if not fields:
+            return
+        payload = {"fields": fields}
+        await self.client.update_item(item_id, payload)
+
     def _resolve_field_label(self, key: str) -> str:
         reverse = {
             "2200000149785345": "项目单号",
@@ -189,7 +235,24 @@ class HuobanyunService:
         else:
             offset = 0
 
-        key = req.key or raw.get("key") or raw.get("field") or req.token or ""
+        approval_code = (
+            req.approval_code
+            or raw.get("approval_code")
+            or raw.get("approvalCode")
+            or (req.linkage_params or {}).get("approval_code")
+            or (req.linkage_params or {}).get("approvalCode")
+            or ""
+        )
+        token_key = req.token or ""
+        key = req.key or raw.get("key") or raw.get("field") or token_key or ""
+        approval_hint = ""
+        if isinstance(token_key, str):
+            if token_key.startswith("对公"):
+                approval_hint = "public"
+                key = token_key.replace("对公", "", 1)
+            elif token_key.startswith("对私"):
+                approval_hint = "private"
+                key = token_key.replace("对私", "", 1)
         key = self._resolve_field_key(str(key).strip()) if key else ""
         query_value = req.query or req.keyword or ""
         linkage_params = req.linkage_params or raw.get("linkage_params") or {}
@@ -203,6 +266,21 @@ class HuobanyunService:
             {"field": project_no_field_id, "query": {"em": False}},
             {"field": order_status_field, "query": {"ne": ["已完成"]}},
         ]
+        if approval_code or approval_hint:
+            if approval_hint == "public" or approval_code in set(self.settings.feishu_approval_code_public):
+                base_filters.append(
+                    {
+                        "field": self._resolve_field_key("是否已关联对公付款审批"),
+                        "query": {"eq": False},
+                    }
+                )
+            if approval_hint == "private" or approval_code in set(self.settings.feishu_approval_code_private):
+                base_filters.append(
+                    {
+                        "field": self._resolve_field_key("是否已关联对私付款审批"),
+                        "query": {"eq": False},
+                    }
+                )
         if linkage_project_no:
             base_filters.append(
                 {
@@ -264,6 +342,11 @@ class HuobanyunService:
                 project_key = key or self._resolve_field_key("项目单号")
                 value = self._extract_value(fields.get(project_key, ""))
                 value_str = "" if value is None else str(value)
+                if project_key == self._resolve_field_key("项目单号"):
+                    if not value_str:
+                        continue
+                    if value_str.startswith("2300") and len(value_str) >= 10:
+                        continue
                 i18n_key = f"@i18n@{item_id}" if item_id else f"@i18n@{value_str}"
                 options.append({"id": str(item_id), "value": i18n_key, "isDefault": False})
                 texts[i18n_key] = value_str

requirements.txt

@@ -3,3 +3,5 @@ uvicorn
 httpx
 websockets
 python-dotenv
+lark-oapi
+python-socks