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"""
{create_time}
"""
@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"\n"
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"""
"""
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"""
"""
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"""
"""
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"""
"""
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"{count}\n\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"""-
\n"""
xml += "\n"
return xml
@staticmethod
def single_article(msg: Dict[str, Any], create_time=None, title: str, description: str, image: str, url: str) -> 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"""
{create_time}
"""
return xml