feat(logger, telegram_bot): Add real-time Telegram logging capability

- Add TelegramLogHandler class to stream logs to Telegram in real-time
- Implement enable_telegram_logging() method to activate Telegram log streaming
- Implement disable_telegram_logging() method to deactivate Telegram log streaming
- Implement is_telegram_logging_enabled() method to check logging status
- Add /logs_live command to enable real-time log push notifications
- Add /logs_stop command to disable real-time log push notifications
- Update help text to document new real-time logging commands
- Remove obsolete team.json.example file
- Enables administrators to monitor system logs in real-time through Telegram bot
This commit is contained in:
2026-01-18 01:16:20 +08:00
parent b902922d22
commit ad120687b3
3 changed files with 92 additions and 40 deletions

View File

@@ -72,6 +72,38 @@ class FileFormatter(logging.Formatter):
return f"[{timestamp}] [{level}] {icon}{record.getMessage()}"
class TelegramLogHandler(logging.Handler):
"""Telegram 实时日志处理器"""
def __init__(self, callback, level=logging.INFO):
"""初始化 Telegram 日志处理器
Args:
callback: 回调函数,接收 (message: str, level: str) 参数
level: 日志级别
"""
super().__init__(level)
self.callback = callback
self.setFormatter(FileFormatter())
def emit(self, record):
"""发送日志记录到 Telegram"""
try:
msg = self.format(record)
# 映射日志级别到 BotNotifier 的级别
level_map = {
logging.DEBUG: "debug",
logging.INFO: "info",
logging.WARNING: "warning",
logging.ERROR: "error",
logging.CRITICAL: "error"
}
level = level_map.get(record.levelno, "info")
self.callback(msg, level)
except Exception:
self.handleError(record)
class Logger:
"""统一日志输出 (基于 Python logging 模块)"""
@@ -113,6 +145,7 @@ class Logger:
self.name = name
self.use_color = use_color
self.enable_file_log = enable_file_log
self._telegram_handler = None # Telegram 日志处理器
# 从环境变量读取日志级别,默认 INFO
if level is None:
@@ -162,6 +195,27 @@ class Logger:
return self.ICONS.get(icon, icon)
return ""
def enable_telegram_logging(self, callback, level: int = logging.INFO):
"""启用 Telegram 实时日志
Args:
callback: 回调函数,接收 (message: str, level: str) 参数
level: 日志级别,默认 INFO
"""
if self._telegram_handler is None:
self._telegram_handler = TelegramLogHandler(callback, level)
self._logger.addHandler(self._telegram_handler)
def disable_telegram_logging(self):
"""禁用 Telegram 实时日志"""
if self._telegram_handler is not None:
self._logger.removeHandler(self._telegram_handler)
self._telegram_handler = None
def is_telegram_logging_enabled(self) -> bool:
"""检查 Telegram 实时日志是否启用"""
return self._telegram_handler is not None
def info(self, msg: str, icon: str = None, indent: int = 0):
"""信息日志"""
prefix = " " * indent