准备工作：
根据以下脚本安装相应依赖模块，以下脚本只是简单的做了个DEMO供初学者参考！举一反三~
上脚本：
脚本中有多线程和非多线程的写法，已实钉钉发送报告和运行日志记录（脚本中钉钉的Webhook换成你自己的即可）可根据需要自行调整优化！
# coding:utf-8
import json,time,sys,os,re
import requests
import pandasas pd
from tomorrow3import threads
class Logger():
def __init__(self, fileN="python_sendreport.log"):
self.terminal = sys.stdout
self.log =open(fileN, "a")
def write(self, message):
self.terminal.write(message)
self.log.write(message)
def flush(self):
pass
class dingHook(object):
#钉钉发送markdown方法
def markdown(self,project,text,text1):
self.url ="https://oapi.dingtalk.com/robot/send?access_token=5af67a5b8f879cfe4a9f5fd4f7e1aef8d83c23e36aaa8e45860d55fef17edb87"
data={
"msgtype":"markdown",
"markdown": {
"title":"吕布",
"text":f'### **{project}**\n'+
f"> #### {text}\n\n" +
#"> \n"+
f"> ###### {text1} \n"
#"> ###### 具体可查看邮件[链接](C:/Users/lenovo/Desktop/12.html) \n"
},
"at": {
"atMobiles": [
"1860176XXXX",
""
],
"isAtAll":False
}
}
data = json.dumps(data).encode(encoding='UTF8')
head = {"Content-Type":"application/json"}
response = requests.post(self.url,data=data,headers=head)
#调用钉钉返回结果转json格式
res_code = json.loads(response.text)
#print(res_code['errcode'])
if res_code['errcode'] ==0:
#print('钉钉消息推送成功！')
Logger().write(time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())+' ' +"钉钉消息推送成功\n")
else:
print('钉钉消息推送失败!')
Logger().write(time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()) +"钉钉消息推送失败\n")
#查询当前路径下所有xlsx
class read_execlfile(object):
def search(self,dirname, filename):
result = []
for itemin os.listdir(dirname):
item_path = os.path.join(dirname, item)
if os.path.isdir(item_path):
read_execlfile().search(item_path, filename)
elif os.path.isfile(item_path):
if filenamein item:
result.append(item_path)
return result
class system_check(object):
@threads(5)
#支持多线程-发送钉钉
def run(self,data):
try:
# 取每一行的值
Row_value = data.loc[i]
# 定义headers变量
headers = Row_value['headers']
# 判断header是否为NAN，当相同时为真
if headers == headers:
# 调用请求方法
res = requests.get(Row_value['url'], headers=eval(headers), timeout=Row_value['timeout'])
# 判断请求结果发送报告至钉钉
if res.status_code !=200:
dingHook().markdown('系统健康检查', '请核实此链接访问是否正常:' + Row_value['url'], '当前返回结果：' +str(res.text))
else:
pass
else:
# 调用请求方法
res = requests.get(Row_value['url'], timeout=Row_value['timeout'])
# 判断请求结果发送报告至钉钉
if res.status_code !=200:
dingHook().markdown('系统健康检查', '请核实此链接访问是否正常:' + Row_value['url'], '当前返回结果：' +str(res.status_code))
else:
pass
except Exception as e:
Logger().write(time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()) +f" 异常{e}\n")
dingHook().markdown('系统健康检查', '请核实此链接访问是否正常:' + Row_value['url'], f'当前返回结果:{e}')
#非多线程写法-生成报告写EXECL
def run_1(self):
#报告文件地址
EX_PATH =r".\\url.xlsx"
otherStyleTime = time.strftime("%Y%m%d%H%M%S", time.localtime(time.time()))
#随机生成报告
EX_PATH1 =f'.\\report\\report_{str(otherStyleTime)}.xlsx'
# 读取excel里面的内容
data = pd.DataFrame(pd.read_excel(EX_PATH))
#获取execl行数
print(len(data))
url=[]
result=[]
#根据行数获取参娄
for iin range(len(data)):
#取每一行的值
Row_value=data.loc[i]
#定义headers变量
headers=Row_value['headers']
#判断header是否为真
if headers ==headers:
#调用请求方法
res=requests.get(Row_value['url'],headers=eval(headers),timeout=Row_value['timeout'])
else:
#调用请求方法
res=requests.get(Row_value['url'],timeout=Row_value['timeout'])
#判断请求结果发送报告至钉钉
if res.status_code !=200:
dingHook().markdown('系统健康检查','请核实此URL访问是否正常'+Row_value['url'],'当前返回状态：'+str(res.status_code))
else:
pass
#将相关数据及结果添加进列表
url.append(Row_value['url'])
result.append(res.status_code)
# 写文件件操作
dic1 = {'url': url,
'result': result
}
df = pd.DataFrame(dic1)
df.to_excel(EX_PATH1, index=False)
if __name__ =='__main__':
#获取当前路径
path = os.path.split(os.path.realpath(__file__))[0]
#print(path)
# 报告文件地址
EX_PATH = read_execlfile().search(path, ".xlsx")
#遍历所有execl文件
for iin EX_PATH:
#获取execl文件名
execl_name_list = re.findall(r'health_check\\(.*?).xlsx',i,re.S)
execl_name=''.join(execl_name_list)
#print(execl_name)
# 读取excel里面的内容
data = pd.DataFrame(pd.read_excel(i))
#print('脚本执行开始')
dingHook().markdown('健康检查', f'@定时任务&开始启动...', f'>>>>>>({execl_name}.xlsx文件)-脚本执行中>>>>>>')
Logger().write(time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()) +f" 检查脚本定时执行开始\n")
#执行多线程操作
for i in range(len(data)):
system_check().run(data)
#判断脚本是否执行完毕
if i == max(range(len(data))):
time.sleep(5)
dingHook().markdown('健康检查', f'@定时任务#({execl_name}.xlsx文件)-脚本执行完成(共{len(data)}条用例)', '请留意钉钉信送消息！')
Logger().write(time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()) +f" 定时任务执行完成(共{len(data)}条用例)。\n")
