最近想拿微信公众号接一个消息处理玩玩，这样就能自定义回复公众号的私聊消息了。

公众号开通注册的部分就略过了，直接从接入消息处理开始。

首先进入公众号管理后台mp.weixin.qq.com

然后点击左边栏的设置与开发-基本配置

设置好开发者密码和ip白名单，ip白名单可以设置现在你在用的ip便于开发，还有服务器的ip。

添加并验证服务器

然后是下面的服务器配置。

添加一个服务器配置。

服务器配置的官方文档：开始开发 / 接入指南 (qq.com)

url就是服务器地址，微信公众号每次收到消息或者事件（比如关注公众号、取关公众号）都会往这个地址发送一个POST请求。

只能用http和https，分别对应80和443端口，也就是说不能随意指定端口。

而且好像域名还需要是通过备案的才能用。

token可以自己随便设置。

EncodingAESKey可以让他随机生成，记得保存。

由于新添加配置微信需要发送一个请求来验证，所以要先开发好验证的部分。

根据开发文档的示例，他验证的时候会给你发送一个GET请求，包含signature、timestamp、nonce、echostr这四个参数。

在代码中根据开发文档将token、timestamp、nonce三个参数进行字典序排序，然后将三个参数字符串拼接成一个字符串进行sha1加密，最后将加密后的字符串可与signature对比，如果一致的话验证成功，验证成功后直接返回echostr，这样就能通过验证了。（当然你懒的话也可以不验证直接返回echostr，不过要确保你的服务器地址保密或者你能确认这个请求确实是微信发的，防止别人滥用给你乱发请求）

感觉框架比较简单，我试着使用flask框架来实现。

以下是这一部分的代码：

"""
@FILE_NAME : flask_wechat
-*- coding : utf-8 -*-
@Author : Zhaokugua
@Time : 2023/12/5 18:20
"""
from flask import Flask, request
import requests
import hashlib

app = Flask(__name__)
@app.route('/', methods=['GET'])
def index():
    signature = req_info.args.get('signature')
    timestamp = req_info.args.get('timestamp')
    nonce = req_info.args.get('nonce')
    echostr = req_info.args.get('echostr')
    token = '你刚才写的token'
    aes_key = '你刚才随机生成的EncodingAESKey'
    
    list_verify = [token, timestamp, nonce]
    list_verify.sort()
    list_sha1 = hashlib.sha1()
    map(list_sha1.update, list_verify)
    hashcode = list_sha1.hexdigest()
    print("handle/GET func: hashcode, signature: ", hashcode, signature)
    if hashcode == signature:
        return echostr
    else:
        return echostr
app.run(host='127.0.0.1', port=5203)

	

在本地运行之后就可以在5023端口访问到了，不过由于微信的要求需要直接在域名访问，所以我还做了内网穿透和反向代理。

先用frp把本地的5023端口映射到服务器上的一个端口，然后再到服务器里用nginx反向代理了刚刚映射的端口，这样就能满足微信的要求直接用域名访问到我的接口了。

通过验证之后就添加服务器成功了，可以进行下一步的开发。

接收回复消息

官方文档：基础消息能力 / 接收普通消息 (qq.com)

根据官方文档的说明，当有用户向公众号发送消息时，微信就会往你设置的服务器地址发送一个POST请求，body里面是xml格式的消息数据。

接受到消息后需要在5秒内返回一个xml格式的响应，超过5秒微信会认为超时并且重试，如果三次都超过5秒超时则微信返回失败消息。

由于我平时都是对json格式的数据处理比较熟悉，所以我写了两个函数用于xml和json数据的互转。

def parse_xml_to_dict(xml_string):
    root = ET.fromstring(xml_string)
    result = {}
    for child in root:
        result[child.tag] = child.text
    return result


def dict_to_xml(d, tag='xml'):
    elem = ET.Element(tag)
    for key, val in d.items():
        child = ET.Element(key)
        child.text = str(val)
        elem.append(child)
    xml_text = ET.tostring(elem, encoding='utf8')
    # xml_text = str(xml_text).replace("&lt;MediaId&gt;", "<MediaId>").replace("&lt;/MediaId&gt;", "</MediaId>")
    xml_text = xml_text.decode('utf8').replace('&lt;', '<').replace('&gt;', '>')
    return xml_text

（代码还有点问题，嵌套格式的还转不了，先暂时凑活用）

然后就可以进行消息解析回复部分了。

关于回复消息的官方文档：基础消息能力 / 被动回复用户消息 (qq.com)

@app.route('/', methods=['GET', 'POST'])
def index():
 【这一部分是上面的验证部分，验证成功返回echostr便于后续修改服务器配置验证，由于重复了就不写了】
    echostr = request.args.get('echostr')
    if echostr:
        return echostr
    if request.method == 'GET':
        return '这里是赵苦瓜测试处理消息。'
    # 获取原始的xml消息数据
    raw_xml_message = request.data.decode('utf8')
    # 转换为json便于处理
    message_info = parse_xml_to_dict(raw_xml_message)
    print(message_info)
    # 此时消息是用户发给公众号的，所以self_id是公众号id
    self_uid = message_info['ToUserName']
    #user_id自然就是发送消息的用户的id
    user_uid = message_info['FromUserName']
    # 有text voice image event link等类型
    message_type = message_info['MsgType']
    # 消息发送的时间戳
    create_time = message_info['CreateTime']
    
    # 提前准备好回复消息的模板
    return_message_dict = {
        'ToUserName': user_uid,
        'FromUserName': self_uid,
        'CreateTime': create_time,
        'MsgType': 'text',
    }

    default_message = '你好！这里是赵苦瓜。'
    
    if message_type == 'text':
        message = message_info['Content']
        # 随机一言功能
        if message == '一言':
            res_info = requests.get('https://hitokoto.jijidown.com/v2/api/hitokoto').json()['res'][0]
            return_message_dict['Content'] = f'{res_info["hitokoto"]}\n                        ——{res_info["source"]}'
        else:
            return_message_dict['Content'] = default_message
    elif message_type == 'voice':
        return_message_dict['Content'] = '暂不支持语音消息！'
    elif message_type == 'image':
        return_message_dict['Content'] = '暂不支持图片消息！'
    elif message_type == ' ':
        return_message_dict['Content'] = default_message
    elif message_type == 'link':
        message = message_info['Url']
        return_message_dict['Content'] = f'链接消息，链接{message}'
    elif message_type == 'video':
        return_message_dict['Content'] = '暂不支持视频消息！'
    else:
        return_message_dict['Content'] = f'暂不支持的消息类型！{message_type}'
    print(dict_to_xml(return_message_dict))
    # 将消息转换为xml并返回
    return dict_to_xml(return_message_dict)

这样就做好了一个小功能：一言功能。

对公众号发送【一言】这两个字，公众号就会从https://hitokoto.jijidown.com/v2/api/hitokoto请求一个名言然后将结果发送回去。

功能测试成功后就可以测试其他消息了，比如发送图片，需要先把图片上传到微信的素材库（官方开发文档：素材管理 / 新增临时素材 (qq.com)），然后使用素材的id发送消息。发送图片的时候就把MsgType改成image，发送音乐消息就改成music并且附带一个嵌套的Music信息等等。

比如发送图片时的上传图片的封装：

"""
@FILE_NAME : wechat_plugin
-*- coding : utf-8 -*-
@Author : Zhaokugua
@Time : 2023/12/17 22:02
"""
import requests
from requests_toolbelt import MultipartEncoder


def get_token():
    req_url = 'https://api.weixin.qq.com/cgi-bin/token?grant_type=client_credential&appid=【你的appid】&secret=【你的appsecret】'
    res = requests.get(req_url).json()
    if access_token := res.get('access_token'):
        return access_token
    return None


def upload_file(file, file_type, access_token):
    """
    图片（image）: 10M，支持PNG\JPEG\JPG\GIF格式
    语音（voice）：2M，播放长度不超过60s，支持AMR\MP3格式
    视频（video）：10MB，支持MP4格式
    缩略图（thumb）：64KB，支持JPG格式
    :param file:
    :param file_type:
    :param access_token:
    :return:
    """
    req_url = f'https://api.weixin.qq.com/cgi-bin/media/upload?access_token={access_token}&type={file_type}'

    encoder = MultipartEncoder(
        {
            'image': ('test.jpg', file, "image/jpeg"),
        }
    )

    req_headers = {
         'Content-Type': encoder.content_type,
    }

    res = requests.post(req_url, data=encoder, headers=req_headers).json()
    if media_id := res.get('media_id'):
        return media_id
    else:
        return res


if __name__ == '__main__':
    # print(get_token())
    access_token = get_token()
    image_file = requests.get('https://moe.jitsu.top/r18').content
    upload_file(image_file, 'image', access_token)

比如涩图功能：

        # 随机涩图
        elif message == '我要涩图':
            # python3.8新特性海象运算符，判断的同时赋值
            # 可以等效为两行，第一行access_token = wechat_plugin.get_token() 第二行if access_token:
            if access_token := wechat_plugin.get_token():
                # 获取随机图片的二进制数据
                image_file = requests.get('https://moe.jitsu.top/api').content
                # 将图片数据上传到微信临时素材库
                image_info = wechat_plugin.upload_file(image_file, 'image', access_token)
                if type(image_info) == 'dict':
                    return_message_dict['Content'] = f'图片上传失败！\n{image_info}'
                else:
                    # 将上传到素材库返回的MeidaId返回给结果以发送图片
                    return_message_dict['MsgType'] = 'image'
                    return_message_dict['Image'] = '<MediaId>' + image_info + '</MediaId>'
            else:
                return_message_dict['Content'] = 'token获取失败！请联系开发者修复。'

这样基础的消息回复功能就做好了！可以在此基础上进行拓展，开发出功能更丰富的公众号！

搜索