去年底申请了个微信公众号，网站更新的时候，顺带更新微信公众号，体验有好有坏。
坏的方面：
好的方面：
本文主要讲如何使用 Python 进行简单微信公众号开发，内容点为：
文本自动回复
关注公众号自动回复
文本换行实现
框架我选的 Fastapi，因为我的一个站点的后端实现的为 Fastapi，包括网站的自动更新逻辑也是。
且公众号主要是 api 交互，Fastapi 写 api 简单快速。
怎么注册公众号，绑定服务器就不说了，网上一大堆，没必要再写一份，直接上代码节约时间。
友情提示：代码不一定能跑起来，因为很多网站业务相关敏感信息删掉了，也懒得费力去测试，只展示了重要的部分，代码不见得漂亮但大部分是在线上跑的。
微信有自己的一套签名算法，证明连的服务器是对的，不然的话，连到一个黑服务器，乱搞一通，岂不天下大乱。只有通过签名认证，表明你是正确的服务提供方，才能做后续的内容服务。
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 的写法，早就过时了。反正你按照自己理解，不停测试验证通过了就行。
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
网上搜索微信公众号开发如何换行，会发现很多人处理不了，到处求助，而且回答者大部分不提供明确代码示例，我这个可以说是良心之作了，简单一看就懂。
代码不漂亮，但是它真能换行，有图有真相，其它编程语言思路差不多的。
一定要指定为信息类型为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)
上面代码都很简单实用，如果你实在理解不了，可以先搜索下各种文档，再微信公众号联系也可以。
往后如果实现值得分享的功能，再写后续篇章。
此内容由惯性聚合(RSS阅读器)自动聚合整理，仅供阅读参考。 原文来自 — 版权归原作者所有。