最近在搞一个自动化运行的小项目，基于某些原因，运行过程中可能会时不时的报错什么的，开始是每天或隔几天登录机器查看运行情况，修复问题，重新运行程序，不过这样搞感觉就不能称之为自动化了，就想着能不能搞个什么消息通知，报错或者运行状态日志实时发送，有问题可以及时收到，然后进行修复，最初的想法是通过邮件发送，不过后来发现telegram机器人更简单好用一些，就搞了一个telegram机器人来接收python logging日志消息，跑了几天，效果貌似还不错。

准备一个telegram账号

可以使用google voice来注册，注册的时候要用telegram手机app，telegram desktop版本无法发送SMS短信验证码，当然也可以搞一个虚拟机，使用 android-x86，建议虚拟机内存搞的大一点，不然真是卡卡卡，安装的时候，不能安装最新版的tg app，不然还是无法发送SMS短信验证码，也不知道是不是google voice虚拟号的问题，不过最终用telegram andriod app 8.9.0版本成功注册，是在 apkapure 下载的，然后就可以安装 telegram desktop版本，用app接收验证码登录了。

创建一个telegram bot机器人

  1. 首先搜索联系 @BotFather，注意，官方 Telegram 机器人的名称旁边有一个蓝色复选标记，单击Start激活 BotFather 机器人。
  2. 选择或输入/newbot命令并发送。
  3. 给机器人起一个闪亮霸气的名字，并且设置机器人的username，username需要以bot结尾。
  4. 机器人创建完成后，会收到机器人链接t.me/<bot_username>和token，这个token要保存好。
  5. 点击收到的机器人链接，打开和机器人的聊天窗口，随便发送点什么，比如 hello bot。
  6. 浏览器访问 https://api.telegram.org/bot<token>/getUpdates，把<token>替换成第4步收到的token，注意一定先给机器人发个消息，才能通过这个链接获取chat_id。
  7. 把机器人token和chat_id保存好。

python安装telegram-logging包

通过Telegram bot机器人发消息，可以用requests包，直接使用telegram的http api发送，https://core.telegram.org/bots/api#sendmessage

不过为了省事，还是用现成的轮子，telegram-logging，可以直接通过pip安装

pip install telegram-logging

可以参考telegram-logging包给的示例代码，把之前保存的bot token和chat_id替换到对应的位置。

但是有一个点要注意，日志消息中不能有 <>& 这几个html的特殊符号，不然tg消息会发送失败，会报错400 BAD_REQUEST。

telegram-logging包中没有对<>&进行处理，所以直接使用这个包的话，日志消息中就不要有这几个字符。

不过因为这个包比较简单，可以直接把包中的telegram.py文件复制出来，进行改造一下，把<>&分别替换成对应的html实体字符。

在TelegramFormatter.format中增加代码

        if record.msg:
            # telegram 消息机器人，telegram的消息中不能包含<>，不然会报错
            # https://core.telegram.org/bots/api#html-style
            record.msg = record.msg.replace('&', '&amp;').replace('<', '&lt;').replace('>', '&gt;')

可以新建一个日志消息的包目录 log，新建一个文件 __init__py，把改造过的 telegram.py也复制到这个目录，在程序运行开始时，先导入一下 log，然后使用logging记录日志消息时，就会把设置的对应级别消息通过bot机器人直接发送到telegram，这样就可以在telegram里一边愉快的看小姐姐，一边处理程序问题，两不耽误，哈哈。

具体代码如下：

# __init__.py


import logging

from .telegram import TelegramFormatter, TelegramHandler


logger = logging.getLogger()
logger.handlers.clear()  # 每次被调用后，清空已经存在handler，防止多次调用该函数时重复生成日志
# 设置为DEBUG级别
logger.setLevel(logging.DEBUG)
formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s: %(message)s')


# 标准流处理器，设置的级别为WARAING
stream_handler = logging.StreamHandler()
stream_handler.setLevel(logging.WARAING)
stream_handler.setFormatter(formatter)
logger.addHandler(stream_handler)


# 文件处理器，设置的级别为DEBUG
file_handler = logging.FileHandler(filename=filename, encoding='utf-8')
file_handler.setLevel(logging.DEBUG)
file_handler.setFormatter(formatter)
logger.addHandler(file_handler)

# telegram bot处理器，设置的级别为INFO
telegram_formatter = TelegramFormatter(
    fmt="[%(asctime)s %(name)s] %(levelname)8s\n\n%(message)s",
    datefmt="%Y-%m-%d %H:%M:%S",
    use_emoji=True,
    # (Optional) If you want to use custom emojis:
    emoji_map={
        logging.DEBUG: "\U0001F41B",
        logging.INFO: "\U0001F4A1",
        logging.WARNING: "\U0001F6A8",
        logging.ERROR: "\U0000274C",
        logging.CRITICAL: "\U0001F4A5",
    })

telegram_handler = TelegramHandler(bot_token="<Your Telegram Bot Token>", chat_id="<Your Telegram Chat ID>")
telegram_handler.setLevel(logging.INFO)
telegram_handler.setFormatter(telegram_formatter)
telegram_handler.addFilter(tg_filter)
logger.addHandler(telegram_handler)


# telegram.py


"""A simple Telegram logging module with Handler and Formatter.
因为有一些hmlt字符“<>&”需要预处理，所以对原包进行修改
https://core.telegram.org/bots/api#html-style
"""

import logging
from urllib import error, parse, request


class TelegramFormatter(logging.Formatter):
    """TelegramFormatter.
    """

    EMOJI_MAP = {
        logging.DEBUG: "\u26aa",
        logging.INFO: "\U0001f535",
        logging.WARNING: "\U0001F7E0",
        logging.ERROR: "\U0001F534",
        logging.CRITICAL: "\U0001f525",
    }

    def __init__(self,
                 fmt: str = '%(asctime)s - %(levelname)s - %(message)s',
                 datefmt: str = None,
                 use_emoji: bool = True,
                 emoji_map: dict = None):
        """:fmt: str, default: '%(asctime)s - %(levelname)s - %(message)s'\n
        :datefmt: str, default: None\n
        :use_emoji: bool, default: True\n
        :emoji_map: dict, default: None\n
        """
        super().__init__(fmt, datefmt)
        self.use_emoji = use_emoji
        self.emojis = self.EMOJI_MAP
        if emoji_map:
            self.emojis.update(emoji_map)

    def format(self, record):
        if self.use_emoji and record.levelno in self.emojis:
            record.levelname = self.emojis[record.levelno]

        if record.msg:
            # telegram 消息机器人，telegram的消息中不能包含<>，不然会报错
            # https://core.telegram.org/bots/api#html-style
            record.msg = record.msg.replace('&', '&amp;').replace('<', '&lt;').replace('>', '&gt;')
        return super().format(record)


class TelegramHandler(logging.Handler):
    """Send log messages to Telegram.
    https://core.telegram.org/bots/api#sendmessage
    """
    def __init__(self,
                 bot_token: str,
                 chat_id: str,
                 timeout: int = 5,
                 **params):
        """:bot_token: Telegram bot_token\n
        :chat_id: Telegram chat_id\n
        :params: https://core.telegram.org/bots/api#sendmessage
        """
        logging.Handler.__init__(self)
        self.bot_token = bot_token
        self.chat_id = chat_id
        self.timeout = timeout
        self.kwargs = params
        self.kwargs["parse_mode"] = "HTML"

    def emit(self, record):
        try:
            url = f"https://api.telegram.org/bot{self.bot_token}/sendMessage"
            params = {
                "chat_id": self.chat_id,
                "text": self.format(record),
            }
            params.update(self.kwargs)
            data = parse.urlencode(params).encode()
            req = request.Request(url, data=data)
            with request.urlopen(req, timeout=self.timeout):
                pass

        except error.URLError:
            self.handleError(record)

标签: python, logging, telegram, google voice

添加新评论