发送Django error log 到企业微信，python+微信实现即时告警

发送Django error log 到企业微信，python+微信实现即时告警

Django的logging系统十分好用，使用file，mail_admins之类的handlers可以快捷地实现保存日志到文件，发送错误日志给管理员的功能。但是，如果能直接将应用的错误日志发送到手机上，实现即时告警，岂不是更好？

发送Django error log 到企业微信python微信实现即时告警
注册企业微信1

使用企业微信API发送消息

发送django的log到企业微信

注册企业微信1

首先需要注册一个企业微信账号，地址是：

https://work.weixin.qq.com/wework_admin/register_wx?from=myhome

注意：

注册过程如果是为企业使用要选择企业，然后上传企业的资质证明，如果是个人注册，选择团队，然后输入自己的身份证号即可完成注册.

然后进入企业应用页面，添加一个应用，添加完成后，进入应用页面：



这里的agentid和secret需要留意， 后续的发送信息api需要它们。

另外我们还需要corpid，在我的企业-企业信息中可以找到。

使用企业微信API发送消息

测试使用企业微信API发送消息：

企业微信API：发送消息

例子：

#! /usr/bin/env python

# -*- coding: utf-8 -*-

import requests
import json

def get_token():

url = 'https://qyapi.weixin.qq.com/cgi-bin/gettoken'
values = {
'corpid': '<YOUR　CORPID>',
'corpsecret': '<YOUR SECRET>',
}
req = requests.post(url, params=values)
data = json.loads(req.text)
return data["access_token"]

def send_msg():
url = ("https://qyapi.weixin.qq.com/cgi-bin/message/send"
"?access_token={}").format(get_token())
values = {
# "touser": "@all",
"toparty": "2",
"msgtype": "text",
"agentid": "<YOUR AGENTID>",
"text": {
"content": u"报警测试，toparty: 2"
},
}

req = requests.post(url, json.dumps(values))
print(values, req)

if __name__ == '__main__':
send_msg()

其中toparty：2，意为向id为2的部门的所有成员发送消息。

使用touser：@all, 可以向所有人发送信息。

发送django的log到企业微信

配置Django的logging系统

接下来我们需要配置一下django的logging系统，来发送ERROR级别的log到企业微信。

我的思路是可以参照django自带的AdminEmailHandler写一个WechatAlarmHandler。

代码如下：

import logging
import requests
import json
from copy import copy

from django.core.cache import cache
from django.views.debug import ExceptionReporter

class WechatAlarmHandler(logging.Handler):
"""An exception log handler that sends log entries to wechat alarm bot.

If the request is passed as the first argument to the log record,
request data will be provided in the email report.
"""

def __init__(self):
logging.Handler.__init__(self)

def emit(self, record):
try:
request = record.request
subject = '%s (%s IP): %s' % (
record.levelname,
('internal' if request.META.get('REMOTE_ADDR') in settings.INTERNAL_IPS # NOQA
else 'EXTERNAL'),
record.getMessage()
)
except Exception:
subject = '%s: %s' % (
record.levelname,
record.getMessage()
)
request = None
subject = self.format_subject(subject)

# Since we add a nicely formatted traceback on our own, create a copy
# of the log record without the exception data.
no_exc_record = copy(record)
no_exc_record.exc_info = None
no_exc_record.exc_text = None

if record.exc_info:
exc_info = record.exc_info
else:
exc_info = (None, record.getMessage(), None)

reporter = ExceptionReporter(request, is_email=True, *exc
4000
_info)
message = "%s\n\n%s" % (
self.format(no_exc_record), reporter.get_traceback_text())
self.send_msg(subject, message)

def send_msg(self, subject, message, *args, **kwargs):
WechatAlarm().send_msg('{}'.format(subject))

def format_subject(self, subject):
"""
Escape CR and LF characters.
"""
return subject.replace('\n', '\\n').replace('\r', '\\r')

class WechatAlarm:
def __init__(self, corpid='<YOUR CORPID>',
corpsecret='<YOUR SECRET>',
agentid='<YOUR AGENTID>', partyid='<YOUR PARTYID>'):
self.corpid = corpid
self.partyid = partyid
self.key = 'wechat_send_alarm_key'
self.corpsecret = corpsecret
self.agentid = agentid

def get_token(self):
token = cache.get(self.key)
if token:
return token
else:
url = 'https://qyapi.weixin.qq.com/cgi-bin/gettoken'
values = {
'corpid': self.corpid,
'corpsecret': self.corpsecret,
}
req = requests.post(url, params=values)
data = json.loads(req.text)
cache.set(self.key, data["access_token"], 7200)
return data["access_token"]

def send_msg(self, content=None):
url = ("https://qyapi.weixin.qq.com/cgi-bin/message/send"
"?access_token={}").format(self.get_token())
values = {
# "touser": "@all",
"toparty": self.partyid,
"msgtype": "text",
"agentid": self.agentid,
"text": {
"content": content,
},
}

return requests.post(url, json.dumps(values))

再配置一下django的settings里的LOGGING：

GGING = {
'version': 1,
'disable_existing_loggers': False,
'formatters': {
'verbose': {
'format': '[%(asctime)s](%(levelname)s)<%(name)s.%(funcName)s>{%(process)d/%(thread)d} : %(message)s'
},
'simple': {
'format': '%(levelname)s %(message)s'
},
},
'filters': {
'require_debug_false': {
'()': 'django.utils.log.RequireDebugFalse'
}
},
'handlers': {
'null': {
'level': 'DEBUG',
'class': 'django.utils.log.NullHandler',
},
'console': {
'level': 'DEBUG',
'class': 'logging.StreamHandler',
'formatter': 'simple'
},
'mail_admins': {
'level': 'ERROR',
'class': 'django.utils.log.AdminEmailHandler',
'filters': ['require_debug_false'],
},
'send_wechat': {
'level': 'ERROR',
'class': 'utils.log.WechatAlarmHandler',   # your handler path
},
'file': {
'level': 'INFO',
'class': 'logging.handlers.TimedRotatingFileHandler',
'formatter': 'verbose',
'filename': webservice_logfile,
'when': 'D'
},
},
'loggers': {
'': {
'handlers': ['file', 'mail_admins', 'send_wechat'],
'propagate': True,
'level': 'ERROR',
},
'django': {
'handlers': ['file', 'mail_admins'],
'propagate': True,
'level': 'ERROR',
},
'django.request': {
'handlers': ['file', 'mail_admins', ],
'level': 'ERROR',
'propagate': True,
},
}
}

即，在handler里增加了一个send_wechat, 在loggers里的handers里增加了send_wechat。

这样django的error log就会通过企业微信发送到手机了。

Python+微信接口实现运维报警 ↩