基于FastApi的微信公众号开发

疯云变换 發表於 2022-10-22 20:15:00

个人申请的订阅号，未认证，可用功能可在 微信公众号平台 - 接口权限 处查看 使用代码开发，首先需要在 基础配置-服务器配置 中进行设置
<img src="https://img2022.cnblogs.com/blog/1160412/202210/1160412-20221022200916991-836231763.png">
 填写服务器地址（URL）、Token和EncodingAESKey，其中 URL 是用来接收微信消息和事件的接口URL。Token可以任意填写，用作生成签名（该 Token 会和接口 URL 中包含的 Token 进行比对，从而验证安全性）。EncodingAESKey手动填写或随机生成，将用作消息体加解密密钥在点击【提交】按钮后，微信会向填写的服务器地址发送一个GET请求，携带signature、timestamp、nonce、echostr四个参数
<img src="https://img2022.cnblogs.com/blog/1160412/202210/1160412-20221022200948344-135455123.png">
 微信开发文档上给的加密规则：
　　1）将token、timestamp、nonce三个参数进行字典序排序     
　　2）将三个参数字符串拼接成一个字符串进行sha1加密     
　　3）开发者获得加密后的字符串可与 signature 对比，标识该请求来源于微信
<div class="cnblogs_code">
<pre>def sign_sha1(signature, timestamp, nonce):
"""
服务器配置验证
:param signature:
:param timestamp:
:param nonce:
:return:
"""
temp =
temp.sort()
hashcode = hashlib.sha1("".join(temp).encode('utf-8')).hexdigest()
logger.info(f"加密：{hashcode}，微信返回：{signature}")
if hashcode == signature:
 return True
# 返回的 echostr 需要是整型。int(echostr)</pre>
</div>
fastapi接收接口
<div class="cnblogs_code">
<pre># main.py
from fastapi import FastAPI
from api import wechat

app = FastAPI(title="EasyTest接口项目", description="这是一个接口文档", version="1.0.0", docs_url=None, redoc_url=None)

app.include_router(
wechat.router,
prefix="/wx",
tags=["wx"],
responses=response_error
)
# wechat.py
from fastapi import APIRouter
from public.wx_public import sign_sha1

router = APIRouter()

@router.get("/", summary="微信服务器配置验证")
async def handle_wx(signature, timestamp, nonce, echostr):
try:
 if sign_sha1(signature, timestamp, nonce):
 return int(echostr)
 else:
 logger.error("加密字符串不等于微信返回字符串，验证失败！！！")
 return "验证失败！"
except Exception as error:
 return f"微信服务器配置验证出现异常:{error}"</pre>
</div>
关注/取消事件和被动回复用户消息
用户在关注与取消关注公众号时，微信会发送一个POST请求，携带xml数据包，把这个事件推送给服务器地址
<img src="https://img2022.cnblogs.com/blog/1160412/202210/1160412-20221022201126084-1622014913.png">
 fastapi接收接口
<div class="cnblogs_code">
<pre>import aiohttp

from starlette.responses import HTMLResponse, Response
# parse_xml, Message 参考地址：https://github.com/vastsa/Wechat-Fastapi
from public.wx_message import parse_xml, Message
from public.wx_public import sign_sha1, send_wx_msg

@router.post("/", summary="回复微信消息")
async def wx_msg(request: Request, signature, timestamp, nonce, openid):
if sign_sha1(signature, timestamp, nonce):
 try:
 async with aiohttp.ClientSession() as session:
 async with session.get("http://121.41.54.234/wx/login") as resp:
 res = await resp.json()
 token = res["result"]["access_token"]
 except Exception as error:
 logger.error(f"获取微信登录token出现异常：{error}")
 token = ""
 try:
 rec_msg = parse_xml(await request.body())
 to_user = rec_msg.FromUserName
 from_user = rec_msg.ToUserName
 content, media_id = send_wx_msg(rec_msg, token)
 if rec_msg.MsgType == 'text' and not media_id:
 return Response(
 Message(to_user, from_user, content=content).send(),
 media_type="application/xml")
 elif rec_msg.MsgType == 'event' and content:
 return Response(
 Message(to_user, from_user, content=content).send(), media_type="application/xml")
 elif rec_msg.MsgType == "image" or media_id:
 return Response(
 Message(to_user, from_user, media_id=media_id, msg_type="image").send(),
 media_type="application/xml")
 except Exception as error:
 logger.error(f"微信回复信息报错：{error}")
 return HTMLResponse('success')

# wx_public.py
def send_wx_msg(rec_msg, token):
content, media_id = "", ""
if rec_msg.MsgType == 'text':
 logger.info(f"文本信息：{rec_msg.Content}")
 patt = r"[\d+]{4}.[\d+]{1,2}.[\d+]{1,2}"
 content = re.findall(patt, rec_msg.Content)
 if content:
 # 输入出生年月日，回复信息
 content = age_content(content)
 else:
 content = rec_msg.Content
 if content in ["图片", "小七"] and token:
 # 输入 "图片", "小七" 回复图片
 media_id = wx_media(token)
 elif content in ["今天", "today"]:
 # 输入 "今天", "today" 回复信息
 content = fishing(make=True)
 elif content == "放假":
 # 输入 "放假" 回复信息
 content = fishing()
 else:
 # 输入股票名称、股票代码回复信息
 content = shares(stock_code=rec_msg.Content)
 if not content:
 # 根据输入值返回
 content = rec_msg.Content
elif rec_msg.MsgType == 'event':
 if rec_msg.Event == "subscribe":
 # 关注公众号回复信息
 content = FOLLOW
 elif rec_msg.Event == "unsubscribe":
 # 取消关注
 logger.info(f"用户 {rec_msg.FromUserName} 取消关注了！！！")
elif rec_msg.MsgType == "image":
 # 收到图片返回
 if token:
 media_id = wx_media(token)
 else:
 media_id = rec_msg.MediaId
return content, media_id</pre>
</div>
log日志打印信息
<img src="https://img2022.cnblogs.com/blog/1160412/202210/1160412-20221022201202327-63204145.png">
 公众号效果
<img src="https://img2022.cnblogs.com/blog/1160412/202210/1160412-20221022201216437-1392844215.png">
 >>>关注公众号试试吧~
<img src="https://img2022.cnblogs.com/blog/1160412/202210/1160412-20221022204124209-14816703.png">
 

</div>
<div id="MySignature" role="contentinfo">
愿你走出半生，归来仍是少年！ 
来源：https://www.cnblogs.com/changqing8023/p/16817207.html

頁: [1]

琼殿技术论坛's Archiver

基于FastApi的微信公众号开发