準備工作：
根據(jù)以下腳本安裝相應依賴模塊，以下腳本只是簡單的做了個DEMO供初學者參考沈贝！舉一反三~
上腳本：
腳本中有多線程和非多線程的寫法杠人，已實釘釘發(fā)送報告和運行日志記錄（腳本中釘釘?shù)腤ebhook換成你自己的即可）可根據(jù)需要自行調整優(yōu)化！
# 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):
#釘釘發(fā)送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)
#支持多線程-發(fā)送釘釘
? ? 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'])
# 判斷請求結果發(fā)送報告至釘釘
? ? ? ? ? ? ? ? if res.status_code !=200:
dingHook().markdown('系統(tǒng)健康檢查', '請核實此鏈接訪問是否正常:' + Row_value['url'], '當前返回結果：' +str(res.text))
else:
pass
? ? ? ? ? ? else:
# 調用請求方法
? ? ? ? ? ? ? ? res = requests.get(Row_value['url'], timeout=Row_value['timeout'])
# 判斷請求結果發(fā)送報告至釘釘
? ? ? ? ? ? ? ? if res.status_code !=200:
dingHook().markdown('系統(tǒng)健康檢查', '請核實此鏈接訪問是否正常:' + 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('系統(tǒng)健康檢查', '請核實此鏈接訪問是否正常:' + 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行數(shù)
? ? ? ? print(len(data))
url=[]
result=[]
#根據(jù)行數(shù)獲取參婁
? ? ? ? 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'])
#判斷請求結果發(fā)送報告至釘釘
? ? ? ? ? ? if res.status_code !=200:
dingHook().markdown('系統(tǒng)健康檢查','請核實此URL訪問是否正常'+Row_value['url'],'當前返回狀態(tài)：'+str(res.status_code))
else:
pass
? ? ? ? ? ? #將相關數(shù)據(jù)及結果添加進列表
? ? ? ? ? ? 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('腳本執(zhí)行開始')
? ? ? ? dingHook().markdown('健康檢查', f'@定時任務&開始啟動...', f'>>>>>>({execl_name}.xlsx文件)-腳本執(zhí)行中>>>>>>')
Logger().write(time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()) +f"? 檢查腳本定時執(zhí)行開始\n")
#執(zhí)行多線程操作
? ? ? ? for i in range(len(data)):
system_check().run(data)
#判斷腳本是否執(zhí)行完畢
? ? ? ? if i == max(range(len(data))):
time.sleep(5)
dingHook().markdown('健康檢查', f'@定時任務#({execl_name}.xlsx文件)-腳本執(zhí)行完成(共{len(data)}條用例)', '請留意釘釘信送消息！')
Logger().write(time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()) +f"? 定時任務執(zhí)行完成(共{len(data)}條用例)学歧。\n")
