python监控接口告警模板

问题：接口出现问题或者超时响应，没法第一时间响应

解决思路：通过返回数据或者返回状态判断接口是否可用  +  钉钉机器人告警  

缺陷： 重复性代码太多了，但是每个接口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()