Feat 公众号被动回复部分实现

nonebot/adapters/wxmp/__init__.py

@@ -1,4 +1,5 @@
 from .bot import Bot
 from .event import *
+from .file import File
 from .adapter import Adapter
 from .message import Message, MessageSegment

nonebot/adapters/wxmp/adapter.py

@@ -1,6 +1,6 @@
-from typing import Any, Union, Callable, Optional, cast, Type
-from typing_extensions import override
+from typing import Any, Union, Callable, Optional, cast, Type, ClassVar
 from pydantic import BaseModel, Field, ValidationError
+from typing_extensions import override
 from yarl import URL
 import xmltodict
 import asyncio
@@ -31,6 +31,7 @@
 from .bot import Bot
 from .event import *
 from .utils import log
+from .store import OfficialReplyResult
 from .config import Config, BotInfo
 from .exception import (
     ActionFailed,
@@ -52,6 +53,8 @@
 
 
 class Adapter(BaseAdapter):
+    _result: ClassVar[OfficialReplyResult] = OfficialReplyResult()
+
     @override
     def __init__(self, driver: Driver, **kwargs: Any):
         super().__init__(driver, **kwargs)
@@ -114,7 +117,7 @@ def setup(self) -> None:
                     )
                 )
             if not (bot := self.bots.get(bot_info.appid, None)):
-                bot = Bot(self, bot_info.appid, bot_info)
+                bot = Bot(self, self_id=bot_info.appid, bot_info=bot_info, official_timeout=self.wxmp_config.wxmp_official_timeout)
                 self.bot_connect(bot)
                 log("INFO", f"<y>Bot {escape_tag(bot_info.appid)}</y> connected")
 
@@ -162,17 +165,22 @@ async def _handle_event(self, request: Request) -> Response:
             if not secrets.compare_digest(sha1_signature, signature):
                 return Response(403, content="Invalid signature")
             else:
-                if bot.bot_info.callback:
+                if bot.bot_info.callback:  # 转发事件推送到指定 URL
                     await self._callback(bot.bot_info.callback, request)
 
                 payload: dict = self.parse_body(request.content)
-                self.dispatch_event(bot, payload)
-                return Response(200, content="success")
+                return await self.dispatch_event(bot, payload, self.wxmp_config.wxmp_official_timeout)
         else:
             return Response(400, content="Invalid request body")
 
-    def dispatch_event(self, bot: Bot, payload: dict):
-        """ 分发事件 """
+    async def dispatch_event(self, bot: Bot, payload: dict, timeout: float) -> Response:
+        """ 分发事件 
+
+        参数：
+        - `bot`: Bot 对象
+        - `payload`: 事件数据
+        - `timeout`: 公众号响应超时时间
+        """
         try:
             event = self.payload_to_event(bot, payload)
         except Exception as e:
@@ -184,6 +192,17 @@ def dispatch_event(self, bot: Bot, payload: dict):
             task.add_done_callback(self.tasks.discard)
             self.tasks.add(task)
 
+        if isinstance(event, OfficalEvent):
+            try:
+                resp = self._result.get_resp(event_id=event.get_event_id(), timeout=timeout)
+            except asyncio.TimeoutError as e:
+                self._result.clear(event.get_event_id())
+                return Response(200, content="success")
+            else:
+                return resp
+        else:
+            return Response(200, content="success")
+
     def payload_to_event(self, bot: Bot,  payload: dict) -> type[Event]:
         """ 将微信公众平台的事件数据转换为 Event 对象 """
         if bot.bot_info.type == "miniprogram":

nonebot/adapters/wxmp/bot.py

@@ -1,5 +1,6 @@
 from typing import Union, Any, Optional, Type, TYPE_CHECKING, cast, Literal
 from typing_extensions import override
+from xmltodict import unparse
 from pathlib import Path
 import json
 import time
@@ -13,11 +14,11 @@
     Response,
 )
 
+from .event import *
 from .file import File
 from .utils import log
-from .event import Event
 from .config import BotInfo
-from .exception import ActionFailed
+from .exception import ActionFailed, OfficialReplyError
 from .message import (
     Text,
     Link,
@@ -37,11 +38,12 @@ class Bot(BaseBot):
     adapter: "Adapter"
 
     @override
-    def __init__(self, adapter: "Adapter", self_id: str, bot_info: BotInfo):
+    def __init__(self, adapter: "Adapter", self_id: str, bot_info: BotInfo, official_timeout: float):
         super().__init__(adapter, self_id)
 
         # Bot 配置信息
         self.bot_info: BotInfo = bot_info
+        self.official_timeout = official_timeout
 
         # Bot 鉴权信息
         self._access_token: Optional[str] = None
@@ -55,7 +57,14 @@ async def send(
         **kwargs,
     ) -> Any:
         """ 发送消息 """
-        return await self.send_custom_message(event.user_id, message)
+        if isinstance(event, OfficalEvent) and not self.bot_info.approve:  # 未完成微信认证的公众号
+            try:
+                return await self.reply_message(event=event, message=message)
+            except OfficialReplyError as e:
+                return await self.send_custom_message(user_id=event.get_user_id(), message=message)
+
+        else:  # 小程序、已认证的公众号 直接发客服消息
+            return await self.send_custom_message(user_id=event.get_user_id(), message=message)
 
     async def handle_event(self, event: Type[Event]):
         """ 处理事件 """
@@ -138,8 +147,22 @@ async def download_file(self, url: str) -> bytes:
         resp: Response = await self.adapter.request(Request("GET", url))
         return resp.content
 
+    async def create_menu(self, data: dict) -> None:
+        """ 创建自定义菜单
+
+        用法：[官方文档](https://developers.weixin.qq.com/doc/offiaccount/Custom_Menus/Creating_Custom-Defined_Menu.html)
+        """
+        await self.call_json_api(
+            "/menu/create",
+            json=data,
+        )
+
     async def send_custom_message(self, user_id: str, message: Message | MessageSegment | str) -> dict:
-        """ 发送 客服消息 """
+        """ 发送 客服消息
+
+        注意：
+        公众号需要微信认证
+        """
         if isinstance(message, str):
             message = Message(MessageSegment.text(message))
         elif isinstance(message, MessageSegment):
@@ -246,12 +269,64 @@ async def send_custom_message(self, user_id: str, message: Message | MessageSegm
             else:
                 raise NotImplementedError()
 
-    async def create_menu(self, data: dict) -> None:
-        """ 创建自定义菜单 
+    async def reply_message(self, event: Type[Event], message: Message | MessageSegment | str) -> None:
+        """ 公众号被动回复 [微信文档](https://developers.weixin.qq.com/doc/offiaccount/Message_Management/Passive_user_reply_message.html)
 
-        用法：[官方文档](https://developers.weixin.qq.com/doc/offiaccount/Custom_Menus/Creating_Custom-Defined_Menu.html)
+        注意：
+        - 需要在5秒内回复\n
+        - 只能回复一次\n
         """
-        await self.call_json_api(
-            "/menu/create",
-            json=data,
-        )
+        if isinstance(message, str):
+            message = Message(MessageSegment.text(message))
+        elif isinstance(message, MessageSegment):
+            message = Message(message)
+        elif not isinstance(message, Message):
+            raise ValueError("Unsupported message type")
+
+        resp = {
+            "ToUserName": event.user_id,
+            "FromUserName": event.to_user_id,
+            "CreateTime": int(time.time()),
+        }
+
+        MSG = "Passive replies have a shorter time limit, please upload in advance and use media_id"
+
+        segment = message[0]
+        if isinstance(segment, Text):
+            resp |= {
+                "MsgType": "text",
+                "Content": segment.data["text"]
+            }
+
+        elif isinstance(segment, Image):
+            if segment.data["media_id"]:
+                media_id = segment.data["media_id"]
+            else:
+                raise ValueError(MSG)
+
+            resp |= {
+                "MsgType": "image",
+                "Image": {
+                    "MediaId": media_id,
+                }
+            }
+
+        elif isinstance(segment, Voice):
+            if segment.data["media_id"]:
+                media_id = segment.data["media_id"]
+            else:
+                raise ValueError(MSG)
+
+            resp |= {
+                "MsgType": "voice",
+                "Voice": {
+                    "MediaId": media_id,
+                }
+            }
+
+        else:
+            raise NotImplementedError()
+
+        return Response(200, content=unparse({
+            "xml": resp,
+        }))

nonebot/adapters/wxmp/config.py

@@ -14,3 +14,4 @@ class BotInfo(BaseModel):
 class Config(BaseModel):
     wxmp_bots: list[BotInfo] = Field(default_factory=list)
     wxmp_verify: bool = Field(default=True)  # 是否开启消息签名验证
+    wxmp_official_timeout: float = Field(default=4.5)  # 公众号响应超时时间

nonebot/adapters/wxmp/event.py

@@ -2,6 +2,7 @@
 from pydantic import Field, ConfigDict, ValidationError, BaseModel
 from typing_extensions import override
 import datetime
+import random
 
 from nonebot.adapters import Event as BaseEvent
 from nonebot.compat import model_dump
@@ -55,6 +56,14 @@ def get_user_id(self) -> str:
     def get_session_id(self) -> str:
         return f"{self.user_id}_{self.to_user_id}"
 
+    def get_event_id(self) -> str:
+        """ 随机生成 event_id """
+        if event_id := getattr(self, "_event_id", None):
+            return event_id
+        else:
+            self._event_id = f"{self.get_session_id()}_{random.randint(10e5, 10e20)}"
+            return self._event_id
+
 
 class NoticeEvent(Event):
     """ 通知事件 """

nonebot/adapters/wxmp/exception.py

@@ -85,3 +85,9 @@ def __repr__(self):
 
     def __str__(self):
         return self.__repr__()
+
+
+class OfficialReplyError(AdapterException):
+    """ 公众号被动回复错误 \n
+    超时 / 已经回复过
+    """

nonebot/adapters/wxmp/store.py

@@ -0,0 +1,49 @@
+from typing import Callable, Any, Union, Awaitable, Type, Optional
+from asyncio import to_thread, iscoroutinefunction
+import asyncio
+
+from nonebot.drivers import (
+    Request,
+    Response,
+    ASGIMixin,
+    WebSocket,
+    HTTPServerSetup,
+    HTTPClientMixin,
+    WebSocketServerSetup
+)
+
+from .event import OfficalEvent
+from .utils import log
+
+
+class OfficialReplyResult:
+    """ 公众号被动回复内容储存 """
+
+    def __init__(self) -> None:
+        self._futures: dict[int, asyncio.Future] = {}
+
+    def set_resp(self, event_id: str, resp: Response) -> None:
+        """ 设置响应 """
+        if future := self._futures.get(event_id):
+            future.set_result(resp)
+
+    async def get_resp(self, event_id: str, timeout: float) -> Response:
+        """ 获取响应 """
+        future = asyncio.get_event_loop().create_future()
+        self._futures[event_id] = future
+        try:
+            return await asyncio.wait_for(future, timeout)
+        finally:
+            try:
+                del self._futures[event_id]
+            except:
+                pass
+
+    def clear(self, event_id: str) -> None:
+        """ 清除响应 """
+        if future := self._futures.get(event_id):
+            try:
+                future.cancel()
+                del self._futures[event_id]
+            except:
+                pass