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

摘要：发送到企业微信，微信实现即时告警的系统十分好用，使用，之类的可以快捷地实现保存日志到文件，发送错误日志给管理员的功能。发送的到企业微信配置的系统接下来我们需要配置一下的系统，来发送级别的到企业微信。微信接口实现运维报警

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

首先需要注册一个企业微信账号，地址是：
https://work.weixin.qq.com/we...

**注意**：

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

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

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

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

测试使用企业微信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": "",
       "corpsecret": "",
      }
   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": "",
       "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的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.
    """

    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)
        message = self.format(record)
        self.send_msg(subject, message)
        
    def send_msg(self, subject, message=None, *args, **kwargs):
        WechatAlarm().send_msg("{}

{}".format(subject, message))


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


class WechatAlarm:
    def __init__(self, corpid="",
                 corpsecret="",
                 agentid="", 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+微信接口实现运维报警 ↩