Fastapi WeChat public account development brief

 1
 2from fastapi import BackgroundTasks, FastAPI, Response, Request
 3import hashlib
 4import logging
 5
 6import reply
 7import receive
 8
 9WX_TOKEN = "your_token" #Your token, filled in the WeChat backend.
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("Verification successful")
24             return int(echostr)
25         else:
26             logger.error("The encrypted string is not equal to the string returned by WeChat, verification failed!!!")
27             return "Verification failed!"
28     except Exception as error:
29         return f"An exception occurred in the WeChat server configuration verification: {error}"

 1
 2@app.post("/api/wx")
 3async def handle_wx(request: Request):
 4     try:
 5         webData = await request.body() # Get the request body under Fastapi, which is different from webpy. After all, the framework is different and the writing method is different. Cows eat grass and people eat.
 6         print("Handle Post webdata is ", webData)
 7         recMsg = receive.parse_xml(webData) #Leave it to receive.py to parse
 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 == "Open Sesame": # This is for users of Wine WeChat to download key files (possessed by Lei Feng, truly serving the people), the real code is not posted.
15                     pass
16                 elif Content.isnumeric() and len(Content)==6: # This is related to the stock list and returns Myanmar A information. Readers do not need to care.
17                     print("content:", Content)
18                     logger.info("Query list...")
19                     content = await check_stock(Content)
20                     logger.info("Start sending list information...")
21                     replyMsg = reply.TextMsg(toUser, fromUser, content)
22                     # return replyMsg.send()
23                     # print("reply send:", type(replyMsg.send())) #What is returned here is the string type
24                     return Response(content=replyMsg.send(), media_type="application/xml") # Be sure to specify 'application/xml', otherwise there will be problems with line breaks, this is very important! ! !
25                 else:
26                     logger.info("The text information is not matched and will not be processed yet")
27                     return "success"
28             # I haven’t thought about what picture service I need to provide at the moment. People are poor and the small station has limited bandwidth and cannot handle...
29             elif recMsg.MsgType == 'image':
30                 logger.info("The image information is not matched and will not be processed yet")
31                 return "success"
32             else:
33                 return reply.Msg().send()
34         elif isinstance(recMsg, receive.EventMsg): # Subscription and unsubscription belong to the ‘event’ type defined by WeChat
35             # Pay attention to the automatic reply content. Generally, what services are provided by the public account to return to users, such as the menu of a restaurant. Of course you have your own unique ideas, just pretend I didn't tell you.
36             if recMsg.Event == 'subscribe':
37                 logger.info("Someone else subscribed")
38                 content = "[🤗Finally waiting for you, thank you for your attention!]\n① Enter the 6-digit Myanmar A code to check whether it is marked\n② Enter 'Open Sesame' to get the temporary download address of wine WeChat related dependency packages\n③ Other functions To be developed..."
39                 replyMsg = reply.TextMsg(toUser, fromUser, content)
40                 return Response(content=replyMsg.send(), media_type="application/xml")
41             # The unsubscription message is not actually sent, but the log can still be printed.
42             if recMsg.Event == 'unsubscribe':
43                 logger.info("We lost a partner")
44                 content = "【🤝See you later, see you in the world!】\n"
45                 replyMsg = reply.TextMsg(toUser, fromUser, content)
46                 return Response(content=replyMsg.send(), media_type="application/xml")
47             else:
48                 logger.info("This event is not matched and will not be processed yet")
49                 return "success"
50         else:
51             logger.info("It is not a picture, text or event, so it will not be processed yet")
52             return reply.Msg().send()
53     except Exception as Argment:
54         return Argment

 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

 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)