201 lines
7.1 KiB
Python
201 lines
7.1 KiB
Python
import time
|
||
from typing import Optional, List, Dict, Any
|
||
import time
|
||
from typing import List, Dict, Any, Optional
|
||
|
||
class ReplyBuilder:
|
||
"""
|
||
原生微信消息回复构建器 (Native WeChat Reply Builder)
|
||
|
||
功能：
|
||
1. 生成所有标准类型的被动回复 XML (明文)。
|
||
2. 自动处理 ToUserName/FromUserName 的反转。
|
||
3. 支持文本、图片、语音、视频、音乐、图文、客服转发。
|
||
|
||
注意：
|
||
- 返回的是明文字符串。如果公众号开启了加密模式，需在主程序中对返回结果进行 AES 加密。
|
||
- 所有 media_id 必须是通过微信接口上传后获得的有效 ID。
|
||
"""
|
||
|
||
@staticmethod
|
||
def _get_base_xml(to_user: str, from_user: str, create_time: int, msg_type: str) -> str:
|
||
"""生成公共的 XML 头部"""
|
||
return f"""<xml>
|
||
<ToUserName><![CDATA[{to_user}]]></ToUserName>
|
||
<FromUserName><![CDATA[{from_user}]]></FromUserName>
|
||
<CreateTime>{create_time}</CreateTime>
|
||
<MsgType><![CDATA[{msg_type}]]></MsgType>
|
||
"""
|
||
|
||
@staticmethod
|
||
def text(msg: Dict[str, Any], content: str, create_time=None) -> str:
|
||
"""
|
||
构造文本回复
|
||
:param msg: 原始接收消息字典
|
||
:param content: 回复的文本内容
|
||
"""
|
||
to_user = msg.get('FromUserName')
|
||
from_user = msg.get('ToUserName')
|
||
create_time = create_time or int(time.time())
|
||
|
||
xml = ReplyBuilder._get_base_xml(to_user, from_user, create_time, 'text')
|
||
xml += f"<Content><![CDATA[{content}]]></Content>\n</xml>"
|
||
return xml
|
||
|
||
@staticmethod
|
||
def image(msg: Dict[str, Any], media_id: str, create_time=None) -> str:
|
||
"""
|
||
构造图片回复
|
||
:param msg: 原始接收消息字典
|
||
:param media_id: 微信服务器返回的图片媒体 ID
|
||
"""
|
||
to_user = msg.get('FromUserName')
|
||
from_user = msg.get('ToUserName')
|
||
create_time = create_time or int(time.time())
|
||
|
||
xml = ReplyBuilder._get_base_xml(to_user, from_user, create_time, 'image')
|
||
xml += f"""<Image>
|
||
<MediaId><![CDATA[{media_id}]]></MediaId>
|
||
</Image>
|
||
</xml>"""
|
||
return xml
|
||
|
||
@staticmethod
|
||
def voice(msg: Dict[str, Any], media_id: str, create_time=None) -> str:
|
||
"""
|
||
构造语音回复
|
||
:param msg: 原始接收消息字典
|
||
:param media_id: 微信服务器返回的语音媒体 ID
|
||
"""
|
||
to_user = msg.get('FromUserName')
|
||
from_user = msg.get('ToUserName')
|
||
create_time = create_time or int(time.time())
|
||
|
||
xml = ReplyBuilder._get_base_xml(to_user, from_user, create_time, 'voice')
|
||
xml += f"""<Voice>
|
||
<MediaId><![CDATA[{media_id}]]></MediaId>
|
||
</Voice>
|
||
</xml>"""
|
||
return xml
|
||
|
||
@staticmethod
|
||
def video(msg: Dict[str, Any], media_id: str, title: str = "", description: str = "", create_time=None) -> str:
|
||
"""
|
||
构造视频回复
|
||
:param msg: 原始接收消息字典
|
||
:param media_id: 微信服务器返回的视频媒体 ID
|
||
:param title: 视频标题 (可选，建议填写)
|
||
:param description: 视频描述 (可选)
|
||
"""
|
||
to_user = msg.get('FromUserName')
|
||
from_user = msg.get('ToUserName')
|
||
create_time = create_time or int(time.time())
|
||
|
||
xml = ReplyBuilder._get_base_xml(to_user, from_user, create_time, 'video')
|
||
xml += f"""<Video>
|
||
<MediaId><![CDATA[{media_id}]]></MediaId>
|
||
<Title><![CDATA[{title}]]></Title>
|
||
<Description><![CDATA[{description}]]></Description>
|
||
</Video>
|
||
</xml>"""
|
||
return xml
|
||
|
||
@staticmethod
|
||
def music(msg: Dict[str, Any], title: str, description: str, music_url: str, hq_music_url: str, thumb_media_id: str, create_time=None) -> str:
|
||
"""
|
||
构造音乐回复
|
||
:param msg: 原始接收消息字典
|
||
:param title: 音乐标题
|
||
:param description: 音乐描述
|
||
:param music_url: 音乐播放链接 (普通质量，http/https)
|
||
:param hq_music_url: 音乐播放链接 (高质量，http/https)
|
||
:param thumb_media_id: 封面图的媒体 ID (必须已上传到微信)
|
||
"""
|
||
to_user = msg.get('FromUserName')
|
||
from_user = msg.get('ToUserName')
|
||
create_time = create_time or int(time.time())
|
||
|
||
xml = ReplyBuilder._get_base_xml(to_user, from_user, create_time, 'music')
|
||
xml += f"""<Music>
|
||
<Title><![CDATA[{title}]]></Title>
|
||
<Description><![CDATA[{description}]]></Description>
|
||
<MusicUrl><![CDATA[{music_url}]]></MusicUrl>
|
||
<HQMusicUrl><![CDATA[{hq_music_url}]]></HQMusicUrl>
|
||
<ThumbMediaId><![CDATA[{thumb_media_id}]]></ThumbMediaId>
|
||
</Music>
|
||
</xml>"""
|
||
return xml
|
||
|
||
@staticmethod
|
||
def news(msg: Dict[str, Any], articles: List[Dict[str, str]], create_time=None) -> str:
|
||
"""
|
||
构造图文消息回复 (支持 1-8 篇)
|
||
:param msg: 原始接收消息字典
|
||
:param articles: 文章列表，每项包含 title, description, image, url
|
||
"""
|
||
to_user = msg.get('FromUserName')
|
||
from_user = msg.get('ToUserName')
|
||
create_time = create_time or int(time.time())
|
||
|
||
if not articles:
|
||
# 如果没有文章，返回空字符串或默认文本，避免生成非法 XML
|
||
return ""
|
||
|
||
count = len(articles)
|
||
if count > 8:
|
||
count = 8
|
||
articles = articles[:8]
|
||
|
||
xml = ReplyBuilder._get_base_xml(to_user, from_user, create_time, 'news')
|
||
xml += f"<ArticleCount>{count}</ArticleCount>\n<Articles>\n"
|
||
|
||
for item in articles:
|
||
title = item.get('title', '无标题')
|
||
desc = item.get('description', '')
|
||
img_url = item.get('image', '')
|
||
link_url = item.get('url', '#')
|
||
|
||
xml += f"""<item>
|
||
<Title><![CDATA[{title}]]></Title>
|
||
<Description><![CDATA[{desc}]]></Description>
|
||
<PicUrl><![CDATA[{img_url}]]></PicUrl>
|
||
<Url><![CDATA[{link_url}]]></Url>
|
||
</item>\n"""
|
||
|
||
xml += "</Articles>\n</xml>"
|
||
return xml
|
||
|
||
@staticmethod
|
||
def single_article(msg: Dict[str, Any], title: str, description: str, image: str, url: str, create_time=None) -> str:
|
||
"""
|
||
快捷方法：构造单篇图文消息
|
||
"""
|
||
article = {
|
||
"title": title,
|
||
"description": description,
|
||
"image": image,
|
||
"url": url
|
||
}
|
||
return ReplyBuilder.news(msg, [article], create_time=create_time)
|
||
|
||
@staticmethod
|
||
def transfer_customer_service(msg: Dict[str, Any], create_time=None) -> str:
|
||
"""
|
||
构造转发客服消息指令
|
||
用途：当机器人无法回答时，将此消息返回给微信，用户消息会进入客服队列，
|
||
由人工客服或多客服系统接管。
|
||
:param msg: 原始接收消息字典
|
||
"""
|
||
to_user = msg.get('FromUserName')
|
||
from_user = msg.get('ToUserName')
|
||
create_time = create_time or int(time.time())
|
||
|
||
# 这种类型的回复没有 Content 或其他节点，只有 MsgType
|
||
xml = f"""<xml>
|
||
<ToUserName><![CDATA[{to_user}]]></ToUserName>
|
||
<FromUserName><![CDATA[{from_user}]]></FromUserName>
|
||
<CreateTime>{create_time}</CreateTime>
|
||
<MsgType><![CDATA[transfer_customer_service]]></MsgType>
|
||
</xml>"""
|
||
return xml
|