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

版權申明:

  • 未標註來源的內容皆為原創，未經授權請勿轉載（因轉載後排版往往錯亂、內容不可控、無法持續更新等）;
  • 非營利為目的，演繹本博客任何內容，請以'原文出處'或者'參考鏈接'等方式給出本站相關網頁地址(方便讀者)。

相關文章:

翻譯: