539bebd58b
- 修正 send_voice 到真实 SendVoice 路由 - 为 864 补上名片发送与视频发送的初版适配入口 - 更新路线图，记录消息接口在未建立连接对象时返回该链接不存在的联调结论
336 lines
13 KiB
Python
336 lines
13 KiB
Python
import asyncio
|
||
import base64
|
||
import os
|
||
import time
|
||
from asyncio import Future, Queue, sleep
|
||
from typing import Union
|
||
|
||
import aiofiles
|
||
|
||
from utils.trace_context import format_trace_prefix
|
||
from wechat_ipad.providers.server_864.base import Server864APIClientBase
|
||
|
||
|
||
class MessageMixin(Server864APIClientBase):
|
||
"""864 消息发送与消息同步接口。"""
|
||
|
||
def __init__(self):
|
||
# 这里不能再走 `super().__init__()`：
|
||
# 1. provider 显式构造时已经先初始化过 `Server864APIClientBase`；
|
||
# 2. 若继续沿 MRO 调 `super()`，会再次命中基类构造并要求重复传 ip/port；
|
||
# 3. 因此消息模块只初始化自己独有的发送队列状态即可。
|
||
self._message_queue = Queue()
|
||
self._is_processing = False
|
||
|
||
async def _process_message_queue(self):
|
||
"""串行处理消息发送，保持现有项目的节流语义。"""
|
||
if self._is_processing:
|
||
return
|
||
|
||
self._is_processing = True
|
||
while True:
|
||
if self._message_queue.empty():
|
||
self._is_processing = False
|
||
break
|
||
|
||
func, args, kwargs, future = await self._message_queue.get()
|
||
try:
|
||
result = await func(*args, **kwargs)
|
||
future.set_result(result)
|
||
except Exception as e:
|
||
future.set_exception(e)
|
||
finally:
|
||
self._message_queue.task_done()
|
||
await sleep(1)
|
||
|
||
async def _queue_message(self, func, *args, **kwargs):
|
||
"""将发送动作排入本地发送队列。"""
|
||
future = Future()
|
||
await self._message_queue.put((func, args, kwargs, future))
|
||
if not self._is_processing:
|
||
asyncio.create_task(self._process_message_queue())
|
||
return await future
|
||
|
||
@staticmethod
|
||
def _normalize_base64_payload(data: Union[str, bytes, os.PathLike]) -> str:
|
||
"""把图片/语音等输入统一转成 base64 字符串。"""
|
||
if isinstance(data, str):
|
||
return data.split(",", 1)[1] if "," in data else data
|
||
if isinstance(data, bytes):
|
||
return base64.b64encode(data).decode()
|
||
raise ValueError("需要先读取文件路径后再处理")
|
||
|
||
async def _read_base64_payload(self, data: Union[str, bytes, os.PathLike]) -> str:
|
||
"""兼容路径/字节/字符串三种输入。"""
|
||
if isinstance(data, os.PathLike):
|
||
async with aiofiles.open(data, "rb") as f:
|
||
return base64.b64encode(await f.read()).decode()
|
||
return self._normalize_base64_payload(data)
|
||
|
||
def _extract_send_message_result(self, data, *, fallback_target: str = "") -> tuple[int, int, int]:
|
||
"""把 864 的发送回执尽量归一化成旧项目习惯的三元组。"""
|
||
if isinstance(data, list) and data:
|
||
item = dict(data[0] or {})
|
||
elif isinstance(data, dict):
|
||
item = dict(data)
|
||
else:
|
||
item = {}
|
||
|
||
resp = item.get("resp") or {}
|
||
base_response = resp.get("BaseResponse") or resp.get("baseResponse") or {}
|
||
|
||
client_msg_id = (
|
||
item.get("ClientMsgId")
|
||
or item.get("clientMsgId")
|
||
or resp.get("ClientMsgId")
|
||
or resp.get("clientMsgId")
|
||
or resp.get("MsgId")
|
||
or resp.get("msgId")
|
||
)
|
||
create_time = item.get("CreateTime") or item.get("createTime") or int(time.time())
|
||
new_msg_id = (
|
||
item.get("NewMsgId")
|
||
or item.get("newMsgId")
|
||
or resp.get("NewMsgId")
|
||
or resp.get("newMsgId")
|
||
or resp.get("MsgSvrId")
|
||
or resp.get("msgSvrId")
|
||
)
|
||
|
||
if client_msg_id is None and base_response.get("Ret") == 0:
|
||
return self._next_fallback_message_ids()
|
||
|
||
if client_msg_id is None:
|
||
client_msg_id, fallback_create_time, fallback_new_msg_id = self._next_fallback_message_ids()
|
||
return client_msg_id, fallback_create_time, fallback_new_msg_id
|
||
|
||
return int(client_msg_id), int(create_time or 0), int(new_msg_id or 0)
|
||
|
||
async def send_text_message(self, wxid: str, content: str, at: Union[list, str] = "") -> tuple[int, int, int]:
|
||
return await self._queue_message(self._send_text_message, wxid, content, at)
|
||
|
||
async def _send_text_message(self, wxid: str, content: str, at: Union[list, str] = "") -> tuple[int, int, int]:
|
||
at_list = [item for item in (at if isinstance(at, list) else str(at or "").split(",")) if item]
|
||
data = await self._request_data(
|
||
"post",
|
||
"/message/SendTextMessage",
|
||
json_body={"MsgItem": [{
|
||
"ToUserName": wxid,
|
||
"TextContent": content,
|
||
"ImageContent": "",
|
||
"MsgType": 1,
|
||
"AtWxIDList": at_list,
|
||
}]},
|
||
timeout=20,
|
||
)
|
||
return self._extract_send_message_result(data, fallback_target=wxid)
|
||
|
||
async def send_image_message(self, wxid: str, image: Union[str, bytes, os.PathLike]) -> tuple[int, int, int]:
|
||
return await self._queue_message(self._send_image_message, wxid, image)
|
||
|
||
async def _send_image_message(self, wxid: str, image: Union[str, bytes, os.PathLike]) -> tuple[int, int, int]:
|
||
image_base64 = await self._read_base64_payload(image)
|
||
data = await self._request_data(
|
||
"post",
|
||
"/message/SendImageMessage",
|
||
json_body={"MsgItem": [{
|
||
"ToUserName": wxid,
|
||
"TextContent": "",
|
||
"ImageContent": image_base64,
|
||
"MsgType": 2,
|
||
"AtWxIDList": [],
|
||
}]},
|
||
timeout=60,
|
||
)
|
||
return self._extract_send_message_result(data, fallback_target=wxid)
|
||
|
||
async def send_voice_message(
|
||
self,
|
||
wxid: str,
|
||
voice: Union[str, bytes, os.PathLike],
|
||
format: str = "silk",
|
||
) -> tuple[int, int, int]:
|
||
return await self._queue_message(self._send_voice_message, wxid, voice, format)
|
||
|
||
async def _send_voice_message(
|
||
self,
|
||
wxid: str,
|
||
voice: Union[str, bytes, os.PathLike],
|
||
format: str = "silk",
|
||
) -> tuple[int, int, int]:
|
||
voice_base64 = await self._read_base64_payload(voice)
|
||
data = await self._request_data(
|
||
"post",
|
||
# 864 的语音发送真实路由是 `/message/SendVoice`：
|
||
# 1. controller 方法名虽然叫 `UploadVoiceRequestApi`；
|
||
# 2. 但 router.go 暴露给外部的路径是 `SendVoice`；
|
||
# 3. 这里按真实注册路由适配，避免后续一接登录态就踩 404。
|
||
"/message/SendVoice",
|
||
json_body={
|
||
"ToUserName": wxid,
|
||
"VoiceData": voice_base64,
|
||
"VoiceSecond": 1,
|
||
"VoiceFormat": 4 if str(format).lower() in {"silk", "wav", "mp3"} else 0,
|
||
},
|
||
timeout=60,
|
||
)
|
||
return self._extract_send_message_result(data, fallback_target=wxid)
|
||
|
||
async def send_link_xml_message(self, xml: str, towxid: str) -> tuple[int, int, int]:
|
||
return await self._queue_message(self._send_link_xml_message, xml, towxid)
|
||
|
||
async def _send_link_xml_message(self, xml: str, towxid: str) -> tuple[int, int, int]:
|
||
data = await self._request_data(
|
||
"post",
|
||
"/message/SendAppMessage",
|
||
json_body={"AppList": [{
|
||
"ToUserName": towxid,
|
||
"ContentXML": xml,
|
||
"ContentType": 5,
|
||
}]},
|
||
timeout=20,
|
||
)
|
||
return self._extract_send_message_result(data, fallback_target=towxid)
|
||
|
||
async def send_link_message(
|
||
self,
|
||
wxid: str,
|
||
url: str,
|
||
title: str = "",
|
||
description: str = "",
|
||
thumb_url: str = "",
|
||
) -> tuple[int, int, int]:
|
||
xml = (
|
||
f"<msg><appmsg><title>{title}</title><des>{description}</des>"
|
||
f"<url>{url}</url><thumburl>{thumb_url}</thumburl></appmsg></msg>"
|
||
)
|
||
return await self.send_link_xml_message(xml, wxid)
|
||
|
||
async def send_emoji_message(self, wxid: str, md5: str, total_length: int):
|
||
return await self._queue_message(self._send_emoji_message, wxid, md5, total_length)
|
||
|
||
async def _send_emoji_message(self, wxid: str, md5: str, total_length: int):
|
||
return await self._request_data(
|
||
"post",
|
||
"/message/SendEmojiMessage",
|
||
json_body={"EmojiList": [{
|
||
"ToUserName": wxid,
|
||
"EmojiMd5": md5,
|
||
"EmojiSize": int(total_length),
|
||
}]},
|
||
timeout=20,
|
||
)
|
||
|
||
async def revoke_message(self, wxid: str, client_msg_id: int, create_time: int, new_msg_id: int) -> bool:
|
||
del create_time
|
||
await self._request_data(
|
||
"post",
|
||
"/message/RevokeMsgNew",
|
||
json_body={
|
||
"NewMsgId": str(new_msg_id),
|
||
"ClientMsgId": int(client_msg_id),
|
||
"CreateTime": int(time.time()),
|
||
"ToUserName": wxid,
|
||
},
|
||
timeout=20,
|
||
)
|
||
return True
|
||
|
||
async def send_video_message(self, wxid: str, video: Union[str, bytes, os.PathLike], image=None):
|
||
"""发送视频消息。
|
||
|
||
当前实现说明：
|
||
1. 864 走的是 `CdnUploadVideo` 路由，而不是 855 的 `SendVideo`；
|
||
2. body 里的 `VideoData` 在 JSON 中要传整数数组，因此这里会把视频字节展开成 `list[int]`；
|
||
3. 返回结构和旧项目三元组并不完全一致，所以仍复用本地兜底消息回执。
|
||
"""
|
||
return await self._queue_message(self._send_video_message, wxid, video, image)
|
||
|
||
async def _send_video_message(self, wxid: str, video: Union[str, bytes, os.PathLike], image=None):
|
||
if isinstance(video, str):
|
||
video_bytes = base64.b64decode(video.split(",", 1)[1] if "," in video else video)
|
||
elif isinstance(video, bytes):
|
||
video_bytes = video
|
||
elif isinstance(video, os.PathLike):
|
||
async with aiofiles.open(video, "rb") as f:
|
||
video_bytes = await f.read()
|
||
else:
|
||
raise ValueError("video should be str, bytes, or path")
|
||
|
||
thumb_base64 = ""
|
||
if image is not None:
|
||
thumb_base64 = await self._read_base64_payload(image)
|
||
|
||
data = await self._request_data(
|
||
"post",
|
||
"/message/CdnUploadVideo",
|
||
json_body={
|
||
"ToUserName": wxid,
|
||
"VideoData": list(video_bytes),
|
||
"ThumbData": thumb_base64,
|
||
},
|
||
timeout=120,
|
||
)
|
||
return self._extract_send_message_result(data, fallback_target=wxid)
|
||
|
||
async def send_card_message(self, wxid: str, card_wxid: str, card_nickname: str, card_alias: str = ""):
|
||
"""发送名片消息。"""
|
||
return await self._queue_message(self._send_card_message, wxid, card_wxid, card_nickname, card_alias)
|
||
|
||
async def _send_card_message(self, wxid: str, card_wxid: str, card_nickname: str, card_alias: str = ""):
|
||
data = await self._request_data(
|
||
"post",
|
||
"/message/ShareCardMessage",
|
||
json_body={
|
||
"ToUserName": wxid,
|
||
"CardWxId": card_wxid,
|
||
"CardNickName": card_nickname,
|
||
"CardAlias": card_alias,
|
||
"CardFlag": 0,
|
||
},
|
||
timeout=20,
|
||
)
|
||
return self._extract_send_message_result(data, fallback_target=wxid)
|
||
|
||
async def send_app_message(self, wxid: str, xml: str, type: int):
|
||
data = await self._request_data(
|
||
"post",
|
||
"/message/SendAppMessage",
|
||
json_body={"AppList": [{
|
||
"ToUserName": wxid,
|
||
"ContentXML": xml,
|
||
"ContentType": int(type),
|
||
}]},
|
||
timeout=20,
|
||
)
|
||
return self._extract_send_message_result(data, fallback_target=wxid)
|
||
|
||
async def sync_message(self, count: int = 0) -> dict:
|
||
"""轮询 864 的 HTTP 同步消息接口，并归一化成 `AddMsgs` 结构。"""
|
||
data = await self._request_data(
|
||
"post",
|
||
"/message/HttpSyncMsg",
|
||
json_body={"Count": int(count)},
|
||
timeout=25,
|
||
)
|
||
|
||
if isinstance(data, dict) and "AddMsgs" in data:
|
||
return data
|
||
|
||
if isinstance(data, list):
|
||
return {"AddMsgs": data, "raw": data}
|
||
|
||
if isinstance(data, dict):
|
||
add_msgs = (
|
||
data.get("AddMsgs")
|
||
or data.get("Msgs")
|
||
or data.get("MsgList")
|
||
or data.get("msgList")
|
||
or data.get("Messages")
|
||
or data.get("messages")
|
||
or []
|
||
)
|
||
return {"AddMsgs": list(add_msgs or []), "raw": data}
|
||
|
||
return {"AddMsgs": [], "raw": data}
|