简易前后端事件处理
基于 Flask 框架搭建的简易api处理框架
0. 构思：
1.利用前后端分离的设想
2.通过Flask框架搭建简易的api接口
3.端口监听事件
4.处理事件
5.推送相关日志
1. 环境：
windows 10 / 10+
Python3.8
Flask == 1.1.2
Tailscale (用于跨网访问)
2. 前端：
建立2个接口，处理启动和关闭事件
采用线程启动，避免卡顿
内网启动 / 外网启动（Tailscale）
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
|
@app.route("/xx", methods=["GET"])
def star_xx():
return_dict = {'return_code': '200', 'return_info': '处理成功', 'Push_data': None}
print("访问成功！")
return_dict['Push_data'] = '开始学习！'
try:
threading.Thread(target=add_cmd, args=()).start()
except:
return "发生了错误哦，有时间记得排查！"
return json.dumps(return_dict, ensure_ascii=False)
|
3. 后端：
后端消息推送，利用企业微信的api接口，推送后台处理事件的日志
发送重要日志消息，支持图片信息
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
|
def __init__(self, is_log=True, log_path="qywx.log", log_level=0):
"""初始化，需要传入企业ID和密钥，在企业微信的页面上有显示"""
self.corpid = 'your id'
self.corpsecret = 'your keys'
self.is_log = is_log
self.log_path = log_path
self.log_level = log_level
if is_log == True:
LOG_FORMAT = "%(asctime)s - %(levelname)s: %(message)s"
DATE_FORMAT = "%m/%d/%Y %H:%M:%S %p"
if log_level == 0:
logging.basicConfig(level=logging.INFO, filename=log_path, format=LOG_FORMAT, datefmt=DATE_FORMAT)
else:
logging.basicConfig(level=logging.ERROR, filename=log_path, format=LOG_FORMAT, datefmt=DATE_FORMAT)
def send_message(self, msg, msgtype, agid):
upload_token = self.get_upload_token(self.corpid, self.corpsecret)
if msgtype == "text":
data = self.msg_messages(msg, agid, msgtype='text', msgid="content")
elif msgtype == "image":
media_id = self.get_media_ID(msg, upload_token, msgtype="image")
data = self.msg_messages(media_id, agid, msgtype='image', msgid="media_id")
else:
raise Exception("msgtype参数错误，参数只能是image或text或voice或video或file")
url = "https://qyapi.weixin.qq.com"
token = self.get_token(url, self.corpid, self.corpsecret)
send_url = '%s/cgi-bin/message/send?access_token=%s' % (url, token)
respone = urllib.request.urlopen(urllib.request.Request(url=send_url, data=data)).read()
x = json.loads(respone.decode())['errcode']
if x == 0:
print("{} 发送成功".format(msg))
if self.is_log == True:
logging.info("{} 发送成功".format(msg))
else:
print("{} 发送失败".format(msg))
if self.is_log == True:
logging.info("{} 发送失败".format(msg))
|
说明
由于项目特殊性，仅说明思路，详细自行实践。
写在后面
最近一月超忙，里外忙不完，断更了。。。。
忙里偷闲，后续争取周更。尽力做一些有趣的小东西。在？😂
谨心
