refactor: Enhance Wecom adapter with API delegation pattern

- Introduce WechatApiDelegate interface for handling different WeChat API types
- Implement CorpWechatApiDelegate and PublicWechatApiDelegate for enterprise and public APIs respectively
- Refactor WecomAdapter to utilize the new delegation structure for improved maintainability
- Update message handling methods to leverage the new API delegates
- Ensure proper setup and signature verification for both API types

Author: lss233

kirara_ai/plugins/im_wecom_adapter/__init__.py

@@ -1,13 +1,14 @@
 import os
 
-from im_wecom_adapter.adapter import WecomAdapter, WecomConfig
-
 from kirara_ai.logger import get_logger
 from kirara_ai.plugin_manager.plugin import Plugin
 from kirara_ai.web.app import WebServer
 
+from .adapter import WecomAdapter, WecomConfig
+
 logger = get_logger("Wecom-Adapter")
 
+__all__ = ["WecomAdapter", "WecomConfig"]
 
 class WecomAdapterPlugin(Plugin):
     web_server: WebServer

kirara_ai/plugins/im_wecom_adapter/adapter.py

@@ -8,6 +8,7 @@
 import aiohttp
 from fastapi import FastAPI, HTTPException, Request, Response
 from pydantic import BaseModel, ConfigDict, Field
+from wechatpy.client import BaseWeChatClient
 from wechatpy.exceptions import InvalidSignatureException
 from wechatpy.replies import create_reply
 
@@ -18,6 +19,8 @@
 from kirara_ai.web.app import WebServer
 from kirara_ai.workflow.core.dispatch.dispatcher import WorkflowDispatcher
 
+from .delegates import CorpWechatApiDelegate, PublicWechatApiDelegate
+
 WECOM_TEMP_DIR = os.path.join(os.getcwd(), 'data', 'temp', 'wecom')
 
 WEBHOOK_URL_PREFIX = "/im/webhook/wechat"
@@ -68,9 +71,13 @@ def __init__(self, **kwargs: Any):
 class WeComUtils:
     """企业微信相关的工具类"""
 
-    def __init__(self, access_token: str):
-        self.access_token = access_token
+    def __init__(self, client: BaseWeChatClient):
+        self.client = client
         self.logger = get_logger("WeComUtils")
+        
+    @property
+    def access_token(self) -> str:
+        return self.client.access_token
 
     async def download_and_save_media(self, media_id: str, file_name: str) -> Optional[str]:
         """下载并保存媒体文件到本地"""
@@ -114,7 +121,6 @@ def __init__(self, config: WecomConfig):
         else:
             self.app = self.web_server.app
 
-        self.setup_wechat_api()
         self.logger = get_logger("Wecom-Adapter")
         self.is_running = False
         if not self.config.host:
@@ -127,25 +133,20 @@ def __init__(self, config: WecomConfig):
 
         self.reply_tasks = {}
 
+        # 根据配置选择合适的API代理
+        self.setup_wechat_api()
+        
     def setup_wechat_api(self):
+        """根据配置设置微信API代理"""
         if self.config.corp_id:
-            from wechatpy.enterprise import parse_message
-            from wechatpy.enterprise.client import WeChatClient
-            from wechatpy.enterprise.crypto import WeChatCrypto
-            self.crypto = WeChatCrypto(
-                self.config.token, self.config.encoding_aes_key, self.config.corp_id
-            )
-            self.client = WeChatClient(self.config.corp_id, self.config.secret)
-            self.parse_message = parse_message
+            self.api_delegate = CorpWechatApiDelegate()
         else:
-            from wechatpy import WeChatClient
-            from wechatpy.crypto import WeChatCrypto
-            from wechatpy.parser import parse_message
-            self.crypto = WeChatCrypto(
-                self.config.token, self.config.encoding_aes_key, self.config.app_id
-            )
-            self.client = WeChatClient(self.config.app_id, self.config.secret)
-            self.parse_message = parse_message
+            self.api_delegate = PublicWechatApiDelegate()
+        
+        self.api_delegate.setup_api(self.config)
+        
+        # 设置工具类
+        self.wecom_utils = WeComUtils(self.api_delegate.client)
 
     def setup_routes(self):
         if self.config.host:
@@ -165,19 +166,16 @@ async def handle_check_request(request: Request):
                 raise HTTPException(status_code=404)
 
             signature = request.query_params.get("msg_signature", "")
+            if not signature:
+                signature = request.query_params.get("signature", "")
             timestamp = request.query_params.get("timestamp", "")
             nonce = request.query_params.get("nonce", "")
             echo_str = request.query_params.get("echostr", "")
 
-
             try:            
-                if self.config.corp_id:
-                    echo_str = self.crypto.check_signature(
-                        signature, timestamp, nonce, echo_str
-                    )
-                else:
-                    from wechatpy.utils import check_signature
-                    check_signature(self.config.token, signature, timestamp, nonce)
+                echo_str = self.api_delegate.check_signature(
+                    signature, timestamp, nonce, echo_str
+                )
                 return Response(content=echo_str, media_type="text/plain")
             except InvalidSignatureException:
                 self.logger.error("failed to check signature, please check your settings.")
@@ -190,16 +188,18 @@ async def handle_message(request: Request):
                 self.logger.warning("Wecom-Adapter is not running, skipping message request.")
                 raise HTTPException(status_code=404)
             signature = request.query_params.get("msg_signature", "")
+            if not signature:
+                signature = request.query_params.get("signature", "")
             timestamp = request.query_params.get("timestamp", "")
             nonce = request.query_params.get("nonce", "")
             try:
-                msg = self.crypto.decrypt_message(
+                msg = self.api_delegate.decrypt_message(
                     await request.body(), signature, timestamp, nonce
                 )
             except InvalidSignatureException:
                 self.logger.error("failed to check signature, please check your settings.")
                 raise HTTPException(status_code=403)
-            msg = self.parse_message(msg)
+            msg = self.api_delegate.parse_message(msg)
 
             if msg.id in self.reply_tasks:
                 self.logger.debug(f"skip processing due to duplicate msgid: {msg.id}")
@@ -262,7 +262,7 @@ def convert_to_message(self, raw_message: Any, media_path: Optional[str] = None)
     async def _send_text(self, user_id: str, text: str):
         """发送文本消息"""
         try:
-            return self.client.message.send_text(self.config.app_id, user_id, text)
+            return await self.api_delegate.send_text(self.config.app_id, user_id, text)
         except Exception as e:
             self.logger.error(f"Failed to send text message: {e}")
             raise e
@@ -271,10 +271,7 @@ async def _send_media(self, user_id: str, media_data: str, media_type: str):
         """发送媒体消息的通用方法"""
         try:
             media_bytes = BytesIO(base64.b64decode(media_data))
-            media_id = self.client.media.upload(
-                media_type, media_bytes)["media_id"]
-            send_method = getattr(self.client.message, f"send_{media_type}")
-            return send_method(self.config.app, user_id, media_id)
+            return await self.api_delegate.send_media(self.config.app_id, user_id, media_type, media_bytes)
         except Exception as e:
             self.logger.error(f"Failed to send {media_type} message: {e}")
             raise e

kirara_ai/plugins/im_wecom_adapter/delegates.py

@@ -0,0 +1,117 @@
+from abc import ABC, abstractmethod
+from io import BytesIO
+from typing import TYPE_CHECKING, Any
+
+from kirara_ai.logger import get_logger
+
+if TYPE_CHECKING:
+    from .adapter import WecomConfig
+
+class WechatApiDelegate(ABC):
+    """微信API代理接口，用于处理不同类型的微信API调用"""
+    
+    @abstractmethod
+    def setup_api(self, config: "WecomConfig"):
+        """设置API相关组件"""
+    
+    @abstractmethod
+    def check_signature(self, signature: str, timestamp: str, nonce: str, echo_str: str) -> str:
+        """验证签名"""
+    
+    @abstractmethod
+    def decrypt_message(self, message: bytes, signature: str, timestamp: str, nonce: str) -> str:
+        """解密消息"""
+    
+    @abstractmethod
+    def parse_message(self, message: str) -> Any:
+        """解析消息"""
+    
+    @abstractmethod
+    async def send_text(self, app_id: str, user_id: str, text: str) -> Any:
+        """发送文本消息"""
+    
+    @abstractmethod
+    async def send_media(self, app_id: str, user_id: str, media_type: str, media_bytes: BytesIO) -> Any:
+        """发送媒体消息"""
+
+
+class CorpWechatApiDelegate(WechatApiDelegate):
+    """企业微信API代理实现"""
+    
+    def setup_api(self, config: "WecomConfig"):
+        """设置企业微信API相关组件"""
+        from wechatpy.enterprise import parse_message
+        from wechatpy.enterprise.client import WeChatClient
+        from wechatpy.enterprise.crypto import WeChatCrypto
+        
+        self.crypto = WeChatCrypto(
+            config.token, config.encoding_aes_key, config.corp_id
+        )
+        self.client = WeChatClient(config.corp_id, config.secret)
+        self.parse_message_func = parse_message
+        self.logger = get_logger("CorpWechatApiDelegate")
+        
+    def check_signature(self, signature: str, timestamp: str, nonce: str, echo_str: str) -> str:
+        """验证企业微信签名"""
+        return self.crypto.check_signature(signature, timestamp, nonce, echo_str)
+    
+    def decrypt_message(self, message: bytes, signature: str, timestamp: str, nonce: str) -> str:
+        """解密企业微信消息"""
+        return self.crypto.decrypt_message(message, signature, timestamp, nonce)
+    
+    def parse_message(self, message: str) -> Any:
+        """解析企业微信消息"""
+        return self.parse_message_func(message)
+    
+    async def send_text(self, app_id: str, user_id: str, text: str) -> Any:
+        """发送企业微信文本消息"""
+        return self.client.message.send_text(app_id, user_id, text)
+    
+    async def send_media(self, app_id: str, user_id: str, media_type: str, media_bytes: BytesIO) -> Any:
+        """发送企业微信媒体消息"""
+        media_id = self.client.media.upload(media_type, media_bytes)["media_id"]
+        send_method = getattr(self.client.message, f"send_{media_type}")
+        return send_method(app_id, user_id, media_id)
+
+
+class PublicWechatApiDelegate(WechatApiDelegate):
+    """公众号微信API代理实现"""
+    
+    def setup_api(self, config: "WecomConfig"):
+        """设置公众号API相关组件"""
+        from wechatpy import WeChatClient
+        from wechatpy.crypto import WeChatCrypto
+        from wechatpy.parser import parse_message
+        
+        self.crypto = WeChatCrypto(
+            config.token, config.encoding_aes_key, config.app_id
+        )
+        self.client = WeChatClient(config.app_id, config.secret)
+        self.parse_message_func = parse_message
+        self.logger = get_logger("PublicWechatApiDelegate")
+        
+    def check_signature(self, signature: str, timestamp: str, nonce: str, echo_str: str) -> str:
+        """验证公众号签名"""
+        from wechatpy.utils import check_signature as wechat_check_signature
+        wechat_check_signature(self.crypto.token, signature, timestamp, nonce)
+        return echo_str
+    
+    def decrypt_message(self, message: bytes, signature: str, timestamp: str, nonce: str) -> str:
+        """解密公众号消息"""
+        return self.crypto.decrypt_message(message, signature, timestamp, nonce)
+    
+    def parse_message(self, message: str) -> Any:
+        """解析公众号消息"""
+        return self.parse_message_func(message)
+    
+    async def send_text(self, app_id: str, user_id: str, text: str) -> Any:
+        """发送公众号文本消息"""
+        # 公众号API不需要app_id参数
+        return self.client.message.send_text(user_id, text)
+    
+    async def send_media(self, app_id: str, user_id: str, media_type: str, media_bytes: BytesIO) -> Any:
+        """发送公众号媒体消息"""
+        media_id = self.client.media.upload(media_type, media_bytes)["media_id"]
+        send_method = getattr(self.client.message, f"send_{media_type}")
+        # 公众号API不需要app_id参数
+        return send_method(user_id, media_id)