from flask import Flask
from loguru import logger
# import requests
# import sys
import os
import logging
import json
import urllib.request
# from datetime import datetime
from flask import request
from flask import make_response
import time
# from tornado.wsgi import WSGIContainer
# from tornado.httpserver import HTTPServer
# from tornado.ioloop import IOLoop
app = Flask(__name__)
def send_msg(content):
"""发送钉钉消息"""
# 1、构建url
# url为机器人的webhook
url = 'https://oapi.dingtalk.com/robot/send?access_token='\
'token_val'
# 2、构建一下请求头部
header = {"Content-Type": "application/json", "Charset": "UTF-8"}
# 3、构建请求数据
data = {
"msgtype": "text",
"text": {
"content": str(content)
},
"at": {
"isAtAll": False # @全体成员(在此可设置@特定某人)
}
}
# 4、对请求的数据进行json封装
sendData = json.dumps(data) # 将字典类型数据转化为json格式
sendData = sendData.encode("utf-8") # python3的Request要求data为byte类型
# 5、发送请求
request = urllib.request.Request(url=url, data=sendData, headers=header)
# 6、将请求发回的数据构建成为文件格式
opener = urllib.request.urlopen(request)
# 7、打印返回的结果
r = opener.read() # b'{"errmsg":"ok","errcode":0}'
return r
def runtime(func):
"""接口执行时间"""
def wrapper(*args, **kwargs):
begintime = time.time()
temp_fun = func(*args, **kwargs)
resp = make_response(temp_fun)
span_time = round((time.time() - begintime), 3)
resp.headers['RunTime'] = str(span_time) + 's'
# if span_time > 2:
# send_msg('[WARN]接口:{}, 执行时间{}s'.format(request.url, span_time))
return resp
return wrapper # 返回
@app.route('/')
@logger.catch
@runtime
def hello_world():
# time.sleep(1)
return 'Hello World!'
@app.route('/invoke')
@logger.catch
def invoke():
r = request.args.get('action')
return r
@app.route('/testlog')
@logger.catch
def testlog():
raise Exception('testlog error')
class DingDingHandler(logging.Handler):
def emit(self, record):
errstr = record.getMessage()
# 发送钉钉错误消息
send_msg(errstr)
@app.before_request
def access_log():
"""每次请求前记录请求路径"""
logger.info(request.url)
@logger.catch
def main():
parentPath = os.path.abspath(
os.path.join(os.path.dirname(os.path.realpath(__file__)), ".."))
logger.add(parentPath + "/logs/file_debug.log",
rotation="10 days",
level="DEBUG")
logger.add(parentPath + "/logs/file_error.log",
rotation="10 days",
level="ERROR")
logger.add(DingDingHandler(), format="{message}", level="ERROR")
logger.debug("main begin...")
app.run("0.0.0.0")
# 使用Tornado 调用flask的时候,只有一个并发
# http_server = HTTPServer(WSGIContainer(app))
# http_server.listen(5000)
# IOLoop.instance().start()
if __name__ == "__main__":
main()