国产xxxx99真实实拍_久久不雅视频_高清韩国a级特黄毛片_嗯老师别我我受不了了小说

資訊專欄INFORMATION COLUMN

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

shevy / 2484人閱讀

摘要:發送到企業微信,微信實現即時告警的系統十分好用,使用,之類的可以快捷地實現保存日志到文件,發送錯誤日志給管理員的功能。發送的到企業微信配置的系統接下來我們需要配置一下的系統,來發送級別的到企業微信。微信接口實現運維報警

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

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


注冊企業微信1

首先需要注冊一個企業微信賬號,地址是:
https://work.weixin.qq.com/we...

**注意**:
注冊過程如果是為企業使用要選擇企業,然后上傳企業的資質證明,如果是個人注冊,選擇團隊,然后輸入自己的身份證號即可完成注冊.

然后進入企業應用頁面,添加一個應用,添加完成后,進入應用頁面:

這里的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": "",
       "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的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.
    """

    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+微信接口實現運維報警 ?

文章版權歸作者所有,未經允許請勿轉載,若此文章存在違規行為,您可以聯系管理員刪除。

轉載請注明本文地址:http://specialneedsforspecialkids.com/yun/8010.html

相關文章

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

    摘要:發送到企業微信,微信實現即時告警的系統十分好用,使用,之類的可以快捷地實現保存日志到文件,發送錯誤日志給管理員的功能。發送的到企業微信配置的系統接下來我們需要配置一下的系統,來發送級別的到企業微信。微信接口實現運維報警 發送Django error log 到企業微信,python+微信實現即時告警 Django的logging系統十分好用,使用file,mail_admins之類的h...

    Lavender 評論0 收藏0
  • ElastAlert日志告警(郵件、企業微信

    摘要:工作原理周期性的查詢并且將數據傳遞給規則類型,規則類型定義了需要查詢哪些數據。要做根據頻率變化的告警。 ElastAlert 工作原理 It works by combining Elasticsearch with two types of components, rule types and alerts. Elasticsearch is periodically queried...

    Yuanf 評論0 收藏0
  • Django WebSocket Redis 在線聊天室

    摘要:它采用了請求響應模型。通信請求只能由客戶端發起,服務端對請求做出應答處理弊端協議無法實現服務器主動向客戶端發起消息。如何使用客戶端創建對象屬性表示連接狀態可選值表示連接尚未建立。表示連接正在進行關閉。 一言不合就上效果圖演示showImg(https://segmentfault.com/img/bVbkUDl?w=1920&h=638); 項目:http://112.74.164.1...

    ranwu 評論0 收藏0
  • 【容器云 UK8S】日志監控方案:監控中心操作指南之監控中心概述,開啟監控中心,添加監控目標和添加接

    摘要:添加接收人監控中心支持添加郵箱及微信兩種告警,需要注意的是,添加郵箱告警的話,需要預先配置發件服務器。由于監控中心配置了一條告警規則,只要企業微信的信息填寫正確,一般分鐘以內均可從企業微信中獲取到告警信息。監控中心概述監控中心是UK8S提供的產品化監控方案,提供基于Prometheus的產品解決方案,涵蓋Prometheus集群的全生命周期管理,以及告警規則配置、報警設置等功能,省去了自行搭...

    Tecode 評論0 收藏0

發表評論

0條評論

shevy

|高級講師

TA的文章

閱讀更多
最新活動
閱讀需要支付1元查看
<