發送Django error log 到企業微信，python+微信實現即時告警

發送Django error log 到企業微信，python+微信實現即時告警

Django的logging系統十分好用，使用file，mail_admins之類的handlers能夠快捷地實現保存日誌到文件，發送錯誤日誌給管理員的功能。可是，若是能直接將應用的錯誤日誌發送到手機上，實現即時告警，豈不是更好？

---

註冊企業微信¹

1. 首先須要註冊一個企業微信帳號，地址是：
   https://work.weixin.qq.com/we...

**注意**：

註冊過程若是是爲企業使用要選擇企業，而後上傳企業的資質證實，若是是我的註冊，選擇團隊，而後輸入本身的身份證號便可完成註冊.

1. 而後進入企業應用頁面，添加一個應用，添加完成後，進入應用頁面：
   

• 這裏的agentid和secret須要留意， 後續的發送信息api須要它們。

• 另外咱們還須要corpid，在個人企業-企業信息中能夠找到。

使用企業微信API發送消息

1. 測試使用企業微信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到企業微信

1. 配置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('{}\n\n{}'.format(subject, message))


    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就會經過企業微信發送到手機了。

---

1. Python+微信接口實現運維報警 ↩