问题：接口出现问题或者超时响应，没法第一时间响应
解决思路：通过返回数据或者返回状态判断接口是否可用 + 钉钉机器人告警
缺陷： 重复性代码太多了，但是每个接口url跟告警模板又不一样，觉得代码不简洁
求助：有大佬有好的建议，可以留言，谢谢
脚本如下：
#_*_ coding:utf-8 _*_
import requests
import json,sys,os,datetime,time
import threading
now_time=time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
ding_webhook = 'https://oapi.dingtalk.com/robot/send?access_token=3a791cb4e33dae665b14aef9e2bf04c1810fd33eda3774cba7e4a9234831a817'
user = [18636021897]
headers = {'Content-Type': 'application/json'}
##通过cookie模拟登陆，获取coockie
def get_coockie():
url = 'https://xxxxxxxx/erp-set/index/login?account=xxxx&companyId=3&pwd=xxxx'
des = requests.post(url)
cookie_head = des.headers['Set-Cookie']
cookie = cookie_head.split(";")[0]
return cookie
def erp_base():
base_url = 'https://xxxxxxx/erp-base/cardrank/getAllList'
header = {"cookie": get_coockie(),'Content-Type':'application/json;charset=UTF-8'}
base_data={'type':'1'}
res = requests.post(url=base_url, headers=header,data=json.dumps(base_data))
result_code = (res.json()['status'])
dd_date = {"msgtype": "text", "text": {"content": '%s error: 资料模块接口响应出现问题，返回码为：%s'%(now_time,result_code)},"at": {"atMobiles": user,"isAtAll": False}}
try:
if res.status_code == 200 and result_code == 200:
with open("/mydata/monitor/erp_logs/base.log","at") as f:
print('INFO: %s base result_code:%s '%(now_time,result_code),file=f)
else:
requests.post(url=ding_webhook, data=json.dumps(dd_date), headers=headers)
except Exception as e:
with open("/mydata/monitor/erp_logs/base.log", "at") as f:
print('ERROR: %s error:%s' % (now_time, e), file=f)
except_date = {"msgtype": "text", "text": {"content": '%s 资料模块 login接口抛出异常: %s' % (now_time, e)},
"at": {"atMobiles": user, "isAtAll": False}}
requests.post(url=ding_webhook, data=json.dumps(except_date), headers=headers)
#erpset模块，根据返回的公司ID去判断服务是否存活
def erp_set_url():
set_url = 'https://xxxxxx/erp-set/personalSet/findName'
header = {"cookie": get_coockie()}
res = requests.post(url=set_url, headers=header)
company_id = (res.json()['data']['company_id'])
dd_date = {"msgtype": "text", "text": {"content": '%s error: 设置模块接口响应出现问题，返回码为：%s'%(now_time,res.status_code)},"at": {"atMobiles": user,"isAtAll": False}}
try:
if res.status_code == 200 and company_id == 3:
with open("/mydata/monitor/erp_logs/erp-set.log","at") as f:
print('INFO: %s set status_code:%s company_id:%s'%(now_time,res.status_code,company_id),file=f)
else:
requests.post(url=ding_webhook, data=json.dumps(dd_date), headers=headers)
except Exception as e:
with open("/mydata/monitor/erp_logs/erp-set.log", "at") as f:
print('ERROR: %s error:%s' % (now_time, e), file=f)
except_date = {"msgtype": "text", "text": {"content": '%s 设置模块 login接口抛出异常: %s' % (now_time, e)},
"at": {"atMobiles": user, "isAtAll": False}}
requests.post(url=ding_webhook, data=json.dumps(except_date), headers=headers)
def erp_customer():
set_url = 'https://xxxxxx/erp-customer/customerHomePage/getCustomerList'
customer_data={'mobile': '18636021897'}
header = {"cookie": get_coockie()}
cas = requests.post(url=set_url, headers=header,data=customer_data,timeout=2)
dd_date = {"msgtype": "text", "text": {"content": '%s customer接口返回码为：%s'%(now_time, cas.status_code)},"at": {"atMobiles": [user],"isAtAll": False}}
test_mobile=cas.json()['data']['entitys'][0]['mobile']
try:
if cas.status_code == 200 and test_mobile=='18636021897':
with open("/mydata/monitor/erp_logs/customer.log","at") as f:
print('INFO: %s status_code:%s byf_mobile:%s'%(now_time,cas.status_code,test_mobile),file=f)
else:
requests.post(url=ding_webhook, data=json.dumps(dd_date), headers=headers)
except Exception as e:
with open("/mydata/monitor/erp_logs/customer.log", "at") as f:
print('ERROR: %s error:%s' % (now_time, e), file=f)
except_date = {"msgtype": "text", "text": {"content": '%s 顾客模块 login接口抛出异常: %s' % (now_time, e)},
"at": {"atMobiles": user, "isAtAll": False}}
requests.post(url=ding_webhook, data=json.dumps(except_date), headers=headers)
def erp_logstatic():
logstatic_url = 'https://xxxxxxxxx/ERP-logistics/purchaseList/listNew.shtml'
header = {"cookie": get_coockie(), "Content-Type": "text/html;charset=UTF-8"}
cas = requests.post(url=logstatic_url, headers=header, timeout=1)
request_time = cas.elapsed.total_seconds()
url_result = cas.json()['msg']
try:
dd_date = {"msgtype": "text", "text": {"content": '%s erp_logstatic接口返回码为：%s' %(now_time,cas.status_code)},
"at": {"atMobiles": user, "isAtAll": False}}
if cas.status_code == 200 and url_result=='请求成功!':
with open("/mydata/monitor/erp_logs/logstatic.log","at") as f:
print('INFO: %s logstatic.url_result:%s 接口请求时间：%s '%(now_time,url_result,request_time),file=f)
else:
requests.post(url=ding_webhook, data=json.dumps(dd_date), headers=headers)
except Exception as e:
with open("/mydata/monitor/erp_logs/logstatic.log", "at") as f:
print('ERROR: %s error：%s ' % (now_time,e),file=f)
timeout_data = {"msgtype": "text", "text": {"content": '%s erp_logstatic接口抛出异常：%s'%(now_time,e)},
"at": {"atMobiles": user, "isAtAll": False}}
requests.post(url=ding_webhook, data=json.dumps(timeout_data), headers=headers)
#####监测接口返回码跟接口是否能通，接口调用超时抛出异常告警
def erp_cashier():
cashier_url = 'https://xxxxxxx/cashier/service/search'
header = {"cookie": get_coockie(), "Content-Type": "application/json;charset=UTF-8"}
head_data = {'departid': '239', 'pagenum': '1', 'pagesize': '1'}
cas = requests.post(url=cashier_url, headers=header, timeout=1, data=json.dumps(head_data))
request_time = cas.elapsed.total_seconds() ##接口请求所用时间
url_result = cas.json()['status']
try:
dd_date = {"msgtype": "text", "text": {"content": '%s cashier模块接口请求结果返回码：%s'%(now_time,url_result)},
"at": {"atMobiles": user, "isAtAll": False}}
if cas.status_code == 200 and url_result=='200':
with open("/mydata/monitor/erp_logs/cashier.log","at") as f:
print('INFO: %s cashier.status_code:%s 接口请求时间：%s '%(now_time,cas.status_code,request_time),file=f)
else:
requests.post(url=ding_webhook, data=json.dumps(dd_date), headers=headers)
except Exception as e:
with open("/mydata/monitor/erp_logs/cashier.log", "at") as f:
print('ERROR: %s error：%s ' % (now_time,e),file=f)
timeout_data = {"msgtype": "text", "text": {"content": '%s cashier接口抛出异常：%s'%(now_time,e)},
"at": {"atMobiles": user, "isAtAll": False}}
requests.post(url=ding_webhook, data=json.dumps(timeout_data), headers=headers)
###调用登陆接口，查看返回数据是否正常，不正常告警
def xibao_master():
xibao_url='https://xxxxxxx/xibao/login/login'
header={'contentType':'application/x-www-form-urlencoded'}
date={'account':'xxxxxx','password':'xxxxxx','companyId':'1','phoneType':'temp!' }
aceept_login=requests.post(url=xibao_url,data=date,headers=header,timeout=1)
byf_id=aceept_login.json()['data']['id']
#####告警联系人，告警接口，告警文本格式定义
now_time = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
user=[18636021897]
headers = {'Content-Type': 'application/json'}
dd_webhook = 'https://oapi.dingtalk.com/robot/send?access_token'
'=3a791cb4e33dae665b14aef9e2bf04c1810fd33eda3774cba7e4a9234831a817'
dd_date = {"msgtype": "text", "text": {"content": '%s xibao_app数据返回出现问题,为：%s'%(now_time,byf_id)},"at": {"atMobiles": user,"isAtAll": False}}
#####判断返回数据以及捕获异常
try:
if byf_id== 50002543:
with open("/mydata/monitor/erp_logs/xibao_master.log", "at") as f:
print('INFO: %s byf_id:%s' %( now_time,byf_id), file=f)
else:
requests.post(url=dd_webhook, data=json.dumps(dd_date),headers=headers)
except Exception as e:
dd_date = {"msgtype": "text", "text": {"content": '%s xibao_app login接口抛出异常: %s' % (now_time,e)},
"at": {"atMobiles": user, "isAtAll": False}}
with open("/mydata/monitor/erp_logs/xibao_master.log", "at") as f:
print('ERROR: %s error：%s ' % (now_time,e),file=f)
requests.post(url=dd_webhook, data=json.dumps(dd_date),headers=headers)
def weixin_xique():
xique_url='https://xxxxxxx/weixin-xique/index/getMallAdvertising'
header={'contentType':'application/x-www-form-urlencoded'}
date={'appid':'xxxxxxxx' }
xique_post=requests.post(url=xique_url,headers=header,data=date,timeout=1)
xique_image=xique_post.json()
dd_date = {"msgtype": "text", "text": {"content": '%s ERROR: xique_master status: %s msg:%s' % (now_time,xique_image['status'],xique_image['msg'])},"at": {"atMobiles": user, "isAtAll": False}}
try:
if xique_image['status'] == 200 and xique_image['msg']=='成功':
with open("/mydata/monitor/erp_logs/xique_master.log", "at") as f:
print('INFO: %s status:%s msg:%s' % (now_time, xique_image['status'],xique_image['msg']), file=f)
else:
requests.post(url=ding_webhook, data=json.dumps(dd_date), headers=headers)
except Exception as e:
except_date = {"msgtype": "text", "text": {"content": '%s xique_master login接口抛出异常: %s' % (now_time, e)},
"at": {"atMobiles": user, "isAtAll": False}}
with open("/mydata/monitor/erp_logs/xique_master.log", "at") as f:
print('ERROR: %s error：%s ' % (now_time, e), file=f)
requests.post(url=ding_webhook, data=json.dumps(except_date), headers=headers)
if __name__ == '__main__':
####创建多线程
base_thread=threading.Thread(target=erp_base,args='')
set_thread=threading.Thread(target=erp_set_url,args='')
customer_thread=threading.Thread(target=erp_customer,args='')
logstatic_thread=threading.Thread(target=erp_logstatic,args='')
cashier_thread=threading.Thread(target=erp_cashier,args='')
xibao_thread=threading.Thread(target=xibao_master,args='')
xique_thread=threading.Thread(target=weixin_xique,args='')
####启动多线程，并发执行
base_thread.start()
set_thread.start()
customer_thread.start()
logstatic_thread.start()
cashier_thread.start()
xibao_thread.start()
xique_thread.start()
base_thread.join()
set_thread.join()
customer_thread.join()
logstatic_thread.join()
cashier_thread.join()
xibao_thread.join()
xique_thread.join()