# Author: sea 2019
import requests
import json
import time
class WebchatUtil:
corpid = ‘必须填写你自己申请的‘
secret = ‘固定填写你自己申请的‘
access_token = ‘‘
@staticmethod
def init_access_token():
# 获取token,必须最长两个小时换一次7200秒
url = ‘https://qyapi.weixin.qq.com/cgi-bin/gettoken?corpid={}&corpsecret={}‘
getr = requests.get(url=url.format(WebchatUtil.corpid, WebchatUtil.secret))
WebchatUtil.access_token = getr.json().get(‘access_token‘)
def sendMarkdownMessage(chatid,*,app_name,ip,err_count,errs_str):
n_time = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
data = {
"chatid": chatid,
"msgtype": "markdown",
"markdown": {
"content": " # 日志风险预警
n ## 错误主体:<font color="info">{app_name}({ip}) </font>
n ### 统计时间: <font color="comment"> {n_time}</font>
n ### 错误总数: <font color="warning">{err_count}</font>
n ### 详情摘要:
n #### {errs_str}
n
n ### 每个错误都是质量风险,请相应负责人对每个错误及时跟进处理好,如需了解更多日志详情,请登录服务器查看具体日志文件".format(app_name=app_name, n_time=n_time,ip=ip, err_count=err_count, errs_str=errs_str)
},
"safe": 0
}
m_url = ‘https://qyapi.weixin.qq.com/cgi-bin/appchat/send?access_token={}‘
result = requests.post(url=m_url.format(WebchatUtil.access_token), data=json.dumps(data))
return result.json()
@staticmethod
def sendTextMessage(chatid,content):
data = {
"chatid": chatid,
"msgtype": "text",
"text": {
"content": content
},
"safe": 0
}
m_url = ‘https://qyapi.weixin.qq.com/cgi-bin/appchat/send?access_token={}‘
result = requests.post(url=m_url.format(WebchatUtil.access_token), data=json.dumps(data))
return result.json()
if __name__ == ‘__main__‘:
WebchatUtil.init_access_token()
WebchatUtil.sendMarkdownMessage( ‘xxx‘ , app_name=xxx, ip=xxx,err_count=xxx, errs_str=xxx )