Fastapi微信公众号开发简要

calendar 2024-01-19 · 4 分钟 · python  ·
分享到: linkedin copy

去年底申请了个微信公众号，网站更新的时候，顺带更新微信公众号，体验有好有坏。

坏的方面：

  • 那个编辑文章的后台，不支持原生 markdown 文件；
  • 文章发布后，不能像网站那样灵活更改了。

好的方面：

  • 微信用户多，开流量主能搞钱，能不能火起来就看你有多大本领了，我这种站点基本不可能发达；
  • 能在公众号号平台开发额外功能，方便自己也能为用户提供一些轻量服务。

本文主要讲如何使用 Python 进行简单微信公众号开发，内容点为：

  • 文本自动回复

  • 关注公众号自动回复

  • 文本换行实现

wechat-return1

框架我选的 Fastapi，因为我的一个站点的后端实现的为 Fastapi，包括网站的自动更新逻辑也是。

且公众号主要是 api 交互，Fastapi 写 api 简单快速。

怎么注册公众号，绑定服务器就不说了，网上一大堆，没必要再写一份，直接上代码节约时间。

友情提示：代码不一定能跑起来，因为很多网站业务相关敏感信息删掉了，也懒得费力去测试，只展示了重要的部分，代码不见得漂亮但大部分是在线上跑的。

1. 微信签名验证

微信有自己的一套签名算法，证明连的服务器是对的，不然的话，连到一个黑服务器，乱搞一通，岂不天下大乱。只有通过签名认证，表明你是正确的服务提供方，才能做后续的内容服务。

 1
 2from fastapi import BackgroundTasks, FastAPI, Response, Request
 3import hashlib
 4import logging
 5
 6import reply
 7import receive
 8
 9WX_TOKEN = "your_token" #你的token，微信后台里面填的。
10
11logger = logging.getLogger("__name__")
12
13app = FastAPI()
14
15@app.get("/api/wx")
16async def handle_wx(signature, timestamp, nonce, echostr):
17    try:
18        wx_params= [WX_TOKEN, timestamp, nonce]
19        wx_params.sort()
20        hashcode = hashlib.sha1("".join(wx_params).encode('utf-8')).hexdigest()
21        logger.info(f"hashcode:{hashcode},signature:{signature}")
22        if hashcode == signature:
23            logger.info("验证成功")
24            return int(echostr)
25        else:
26            logger.error("加密字符串 不等于 微信返回字符串，验证失败！！！")
27            return "验证失败！"
28    except Exception as error:
29        return f"微信服务器配置验证出现异常:{error}"

这里不得不说，微信官方 webpy 文档还是 python2 的写法，早就过时了。反正你按照自己理解，不停测试验证通过了就行。

2. 文本回复

 1
 2@app.post("/api/wx")
 3async def handle_wx(request: Request):
 4    try:
 5        webData = await request.body()  # Fastapi下获取请求body, 和webpy不一样哦，毕竟框架不一样，写法就不一样，牛吃草，人吃饭。
 6        print("Handle Post webdata is ", webData)
 7        recMsg = receive.parse_xml(webData)  # 交给receive.py去解析
 8        toUser = recMsg.FromUserName
 9        fromUser = recMsg.ToUserName
10        if isinstance(recMsg, receive.Msg):
11            Content = recMsg.Content
12            Content = Content.decode("utf-8")
13            if recMsg.MsgType == 'text':
14                if Content == "芝麻开门": # 这里是给wine微信的用户提供关键文件下载的(雷锋附体,真正的为人民服务), 真代码不贴了。
15                    pass
16                elif Content.isnumeric() and len(Content)==6: # 这里是股票名单相关，返回缅A信息，读者不用关心。
17                    print("content:", Content)
18                    logger.info("查询列表中...")
19                    content = await check_stock(Content)
20                    logger.info("开始发送列表信息...")
21                    replyMsg = reply.TextMsg(toUser, fromUser, content)
22                    # return replyMsg.send()
23                    # print("reply send:", type(replyMsg.send()))  #这里返回的是字符串类型
24                    return Response(content=replyMsg.send(), media_type="application/xml") # 一定要指定‘application/xml’，不然换行有问题，很关键！！！
25                else:
26                    logger.info("该文本信息没匹配中，暂不处理")
27                    return "success"
28            # 目前没想好需要提供什么图片服务，人穷，小站带宽有限，不处理...
29            elif recMsg.MsgType == 'image':
30                logger.info("该图片信息没匹配中，暂不处理")
31                return "success"
32            else:
33                return reply.Msg().send()
34        elif isinstance(recMsg, receive.EventMsg):   # 订阅和退订属于微信定义的 ‘事件’ 类型
35            # 关注自动回复内容，一般向用户返公众号都有些什么服务，好比饭馆的菜单。当然你有自己的独特思路，当我没说。
36            if recMsg.Event == 'subscribe':
37                logger.info("又有人订阅啦")
38                content = "【🤗终于等到你啦，感谢关注！】\n① 输入6位数缅A代码，可查询是否被标记\n② 输入'芝麻开门'可获得wine微信相关依赖包临时下载地址\n③ 其它功能待开发..."
39                replyMsg = reply.TextMsg(toUser, fromUser, content)
40                return Response(content=replyMsg.send(), media_type="application/xml")
41            # 退订消息实际不发，但是日志还是可以打的
42            if recMsg.Event == 'unsubscribe':
43                logger.info("我们失去了一个伙伴")
44                content = "【🤝后会有期，江湖再见！】\n"
45                replyMsg = reply.TextMsg(toUser, fromUser, content)
46                return Response(content=replyMsg.send(), media_type="application/xml")
47            else:
48                logger.info("该事件没匹配中，暂不处理")
49                return "success"
50        else:
51            logger.info("不是图片、文本、事件，暂不处理")
52            return reply.Msg().send()
53    except Exception as Argment:
54        return Argment

网上搜索微信公众号开发如何换行，会发现很多人处理不了，到处求助，而且回答者大部分不提供明确代码示例，我这个可以说是良心之作了，简单一看就懂。

代码不漂亮，但是它真能换行，有图有真相，其它编程语言思路差不多的。

wechat-return2

一定要指定为信息类型为application/xml，否则当作普通文本，按照微信的规则，是无法换行的。

receive.py 、reply.py 和微信官方的差不多，只稍微改动，多写代码多掉头发，能跑的代码不用是傻子 😄

receive.py 文件内容：

 1
 2# -*- coding: utf-8 -*-#
 3# filename: receive.py
 4import xml.etree.ElementTree as ET
 5
 6
 7def parse_xml(web_data):
 8    if len(web_data) == 0:
 9        return None
10    xmlData = ET.fromstring(web_data)
11    msg_type = xmlData.find("MsgType").text
12    if msg_type == "text":
13        return TextMsg(xmlData)
14    elif msg_type == "image":
15        return ImageMsg(xmlData)
16    elif msg_type == 'event':
17        event_type = xmlData.find('Event').text
18        if event_type == 'CLICK':
19            return Click(xmlData)
20        elif event_type in ('subscribe', 'unsubscribe'):
21            return Subscribe(xmlData)
22        #elif event_type == 'VIEW':
23            #return View(xmlData)
24        #elif event_type == 'LOCATION':
25            #return LocationEvent(xmlData)
26        #elif event_type == 'SCAN':
27            #return Scan(xmlData)
28
29
30
31class Msg(object):
32    def __init__(self, xmlData):
33        self.ToUserName = xmlData.find("ToUserName").text
34        self.FromUserName = xmlData.find("FromUserName").text
35        self.CreateTime = xmlData.find("CreateTime").text
36        self.MsgType = xmlData.find("MsgType").text
37        self.MsgId = xmlData.find("MsgId").text
38
39
40class TextMsg(Msg):
41    def __init__(self, xmlData):
42        Msg.__init__(self, xmlData)
43        self.Content = xmlData.find("Content").text.encode("utf-8")
44
45
46class ImageMsg(Msg):
47    def __init__(self, xmlData):
48        Msg.__init__(self, xmlData)
49        self.PicUrl = xmlData.find("PicUrl").text
50        self.MediaId = xmlData.find("MediaId").text
51
52
53class EventMsg(object):
54    def __init__(self, xmlData):
55        self.ToUserName = xmlData.find("ToUserName").text
56        self.FromUserName = xmlData.find("FromUserName").text
57        self.CreateTime = xmlData.find("CreateTime").text
58        self.MsgType = xmlData.find("MsgType").text
59        self.Event = xmlData.find("Event").text
60
61
62class Click(EventMsg):
63    def __init__(self, xmlData):
64        EventMsg.__init__(self, xmlData)
65        self.Eventkey = xmlData.find("EventKey").text
66
67
68class Subscribe(EventMsg):
69    def __init__(self, xmlData):
70        EventMsg.__init__(self, xmlData)
71        self.Eventkey = xmlData.find("EventKey").text

replay.py 文件内容：

 1# -*- coding: utf-8 -*-#
 2# filename: reply.py
 3import time
 4
 5class Msg(object):
 6    def __init__(self):
 7        pass
 8
 9    def send(self):
10        return "success"
11
12class TextMsg(Msg):
13    def __init__(self, toUserName, fromUserName, content):
14        self.__dict = dict()
15        self.__dict['ToUserName'] = toUserName
16        self.__dict['FromUserName'] = fromUserName
17        self.__dict['CreateTime'] = int(time.time())
18        self.__dict['Content'] = content
19
20    def send(self):
21        XmlForm = """
22            <xml>
23                <ToUserName><![CDATA[{ToUserName}]]></ToUserName>
24                <FromUserName><![CDATA[{FromUserName}]]></FromUserName>
25                <CreateTime>{CreateTime}</CreateTime>
26                <MsgType><![CDATA[text]]></MsgType>
27                <Content><![CDATA[{Content}]]></Content>
28            </xml>
29            """
30        return XmlForm.format(**self.__dict)
31
32class ImageMsg(Msg):
33    def __init__(self, toUserName, fromUserName, mediaId):
34        self.__dict = dict()
35        self.__dict['ToUserName'] = toUserName
36        self.__dict['FromUserName'] = fromUserName
37        self.__dict['CreateTime'] = int(time.time())
38        self.__dict['MediaId'] = mediaId
39
40    def send(self):
41        XmlForm = """
42            <xml>
43                <ToUserName><![CDATA[{ToUserName}]]></ToUserName>
44                <FromUserName><![CDATA[{FromUserName}]]></FromUserName>
45                <CreateTime>{CreateTime}</CreateTime>
46                <MsgType><![CDATA[image]]></MsgType>
47                <Image>
48                <MediaId><![CDATA[{MediaId}]]></MediaId>
49                </Image>
50            </xml>
51            """
52        return XmlForm.format(**self.__dict)

上面代码都很简单实用，如果你实在理解不了，可以先搜索下各种文档，再微信公众号联系也可以。

往后如果实现值得分享的功能，再写后续篇章。

最后修改于： Friday, January 19, 2024
欢迎关注微信公众号，留言交流。

版权申明:

  • 未标注来源的内容全部为原创，未经授权请勿转载（因转载后排版往往错乱、内容不可控、无法持续更新等）;
  • 非营利为目的，演绎本博客任何内容，请以'原文出处'或者'参考链接'等方式给出本站相关网页地址(方便读者)。

相关文章:

翻译: