准备工作：

根据以下脚本安装相应依赖模块，以下脚本只是简单的做了个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" +

#"> ![screenshot](https://gw.alicdn.com/tfs/TB1ut3xxbsrBKNjSZFpXXcXhFXa-846-786.png)\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")