Python logging日志通过Telegram bot机器人发送消息通知
最近在搞一个自动化运行的小项目，基于某些原因，运行过程中可能会时不时的报错什么的，开始是每天或隔几天登录机器查看运行情况，修复问题，重新运行程序，不过这样搞感觉就不能称之为自动化了，就想着能不能搞个什么消息通知，报错或者运行状态日志实时发送，有问题可以及时收到，然后进行修复，最初的想法是通过邮件发送，不过后来发现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机器人
- 首先搜索联系 @BotFather，注意，官方 Telegram 机器人的名称旁边有一个蓝色复选标记，单击
Start激活 BotFather 机器人。 - 选择或输入
/newbot命令并发送。 - 给机器人起一个闪亮霸气的名字，并且设置机器人的username，username需要以
bot结尾。 - 机器人创建完成后，会收到机器人链接
t.me/<bot_username>和token，这个token要保存好。 - 点击收到的机器人链接，打开和机器人的聊天窗口，随便发送点什么，比如
hello bot。 - 浏览器访问
https://api.telegram.org/bot<token>/getUpdates，把<token>替换成第4步收到的token，注意一定先给机器人发个消息，才能通过这个链接获取chat_id。 - 把机器人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('&', '&').replace('<', '<').replace('>', '>')可以新建一个日志消息的包目录 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('&', '&').replace('<', '<').replace('>', '>')
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)