405 lines
13 KiB
Python
405 lines
13 KiB
Python
|
"""
|
|||
|
|
日志辅助模块 - 使用loguru和rich提供美观的日志输出
|
|||
|
|
|
|||
|
|
此模块提供了基于loguru的彩色日志输出功能，支持在关键日志处添加emoji表情
|
|||
|
|
并增加了rich的一些高级功能
|
|||
|
|
"""
|
|||
|
|
import sys
|
|||
|
|
import os
|
|||
|
|
from typing import Optional, Dict, Any, Union
|
|||
|
|
from datetime import datetime
|
|||
|
|
import time
|
|||
|
|
|
|||
|
|
try:
|
|||
|
|
from loguru import logger
|
|||
|
|
from rich.console import Console
|
|||
|
|
from rich.theme import Theme
|
|||
|
|
from rich.table import Table
|
|||
|
|
from rich.progress import Progress, TextColumn, BarColumn, TimeElapsedColumn, SpinnerColumn
|
|||
|
|
from rich.panel import Panel
|
|||
|
|
from rich.syntax import Syntax
|
|||
|
|
from rich.json import JSON
|
|||
|
|
from rich.traceback import install as install_rich_traceback
|
|||
|
|
|
|||
|
|
# 安装rich错误追踪
|
|||
|
|
install_rich_traceback(show_locals=True)
|
|||
|
|
|
|||
|
|
# 创建rich控制台
|
|||
|
|
console = Console(theme=Theme({
|
|||
|
|
"info": "cyan",
|
|||
|
|
"warning": "yellow",
|
|||
|
|
"error": "bold red",
|
|||
|
|
"success": "bold green",
|
|||
|
|
"debug": "dim white"
|
|||
|
|
}))
|
|||
|
|
|
|||
|
|
RICH_AVAILABLE = True
|
|||
|
|
except ImportError:
|
|||
|
|
RICH_AVAILABLE = False
|
|||
|
|
|
|||
|
|
try:
|
|||
|
|
from loguru import logger
|
|||
|
|
LOGURU_AVAILABLE = True
|
|||
|
|
except ImportError:
|
|||
|
|
LOGURU_AVAILABLE = False
|
|||
|
|
import logging
|
|||
|
|
logger = logging.getLogger("机器人")
|
|||
|
|
|
|||
|
|
# Emoji常量定义
|
|||
|
|
EMOJI = {
|
|||
|
|
"成功": "✅",
|
|||
|
|
"失败": "❌",
|
|||
|
|
"警告": "⚠️",
|
|||
|
|
"错误": "🚫",
|
|||
|
|
"信息": "ℹ️",
|
|||
|
|
"启动": "🚀",
|
|||
|
|
"停止": "🛑",
|
|||
|
|
"旋转": "🔄",
|
|||
|
|
"扫描": "🔍",
|
|||
|
|
"计算": "🧮",
|
|||
|
|
"时间": "⏱️",
|
|||
|
|
"距离": "📏",
|
|||
|
|
"角度": "📐",
|
|||
|
|
"调试": "🐞",
|
|||
|
|
"关键": "🔑",
|
|||
|
|
"位置": "📍",
|
|||
|
|
"校准": "🎯",
|
|||
|
|
"完成": "🏁",
|
|||
|
|
"移动": "🚶",
|
|||
|
|
"检测": "👁️"
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
class LogHelper:
|
|||
|
|
"""日志辅助类，提供美观的日志输出"""
|
|||
|
|
|
|||
|
|
def __init__(self, name: str = "机器人控制系统"):
|
|||
|
|
"""
|
|||
|
|
初始化日志助手
|
|||
|
|
|
|||
|
|
参数:
|
|||
|
|
name: 日志名称
|
|||
|
|
"""
|
|||
|
|
self.name = name
|
|||
|
|
|
|||
|
|
if LOGURU_AVAILABLE:
|
|||
|
|
# 移除默认的处理器
|
|||
|
|
logger.remove()
|
|||
|
|
|
|||
|
|
# 添加彩色控制台处理器
|
|||
|
|
logger.add(
|
|||
|
|
sys.stdout,
|
|||
|
|
colorize=True,
|
|||
|
|
format="<green>{time:HH:mm:ss}</green> | <level>{level: <8}</level> | <cyan>{name}</cyan> - <level>{message}</level>",
|
|||
|
|
level="INFO"
|
|||
|
|
)
|
|||
|
|
|
|||
|
|
# 添加文件处理器 - 按日期分割日志文件
|
|||
|
|
log_dir = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "logs")
|
|||
|
|
os.makedirs(log_dir, exist_ok=True)
|
|||
|
|
|
|||
|
|
logger.add(
|
|||
|
|
os.path.join(log_dir, "robot_{time:YYYY-MM-DD}.log"),
|
|||
|
|
rotation="00:00", # 每天零点创建新文件
|
|||
|
|
retention="30 days", # 保留30天的日志
|
|||
|
|
compression="zip", # 压缩旧日志
|
|||
|
|
format="{time:YYYY-MM-DD HH:mm:ss} | {level: <8} | {name} - {message}",
|
|||
|
|
level="DEBUG"
|
|||
|
|
)
|
|||
|
|
|
|||
|
|
self.logger = logger.bind(name=name)
|
|||
|
|
else:
|
|||
|
|
# 使用标准logging作为备选
|
|||
|
|
self.logger = logging.getLogger(name)
|
|||
|
|
|
|||
|
|
# 配置基本的控制台处理器
|
|||
|
|
if not self.logger.handlers:
|
|||
|
|
handler = logging.StreamHandler(sys.stdout)
|
|||
|
|
formatter = logging.Formatter(
|
|||
|
|
'%(asctime)s | %(levelname)-8s | %(name)s - %(message)s',
|
|||
|
|
'%H:%M:%S'
|
|||
|
|
)
|
|||
|
|
handler.setFormatter(formatter)
|
|||
|
|
self.logger.addHandler(handler)
|
|||
|
|
self.logger.setLevel(logging.INFO)
|
|||
|
|
|
|||
|
|
def debug(self, message: str, emoji: Optional[str] = None):
|
|||
|
|
"""
|
|||
|
|
输出调试级别日志
|
|||
|
|
|
|||
|
|
参数:
|
|||
|
|
message: 日志消息
|
|||
|
|
emoji: 可选的emoji名称
|
|||
|
|
"""
|
|||
|
|
msg = f"{EMOJI.get(emoji, EMOJI['调试']) if emoji else ''} {message}"
|
|||
|
|
if LOGURU_AVAILABLE:
|
|||
|
|
self.logger.debug(msg)
|
|||
|
|
else:
|
|||
|
|
self.logger.debug(msg)
|
|||
|
|
|
|||
|
|
def info(self, message: str, emoji: Optional[str] = None):
|
|||
|
|
"""
|
|||
|
|
输出信息级别日志
|
|||
|
|
|
|||
|
|
参数:
|
|||
|
|
message: 日志消息
|
|||
|
|
emoji: 可选的emoji名称
|
|||
|
|
"""
|
|||
|
|
msg = f"{EMOJI.get(emoji, EMOJI['信息']) if emoji else ''} {message}"
|
|||
|
|
if LOGURU_AVAILABLE:
|
|||
|
|
self.logger.info(msg)
|
|||
|
|
else:
|
|||
|
|
self.logger.info(msg)
|
|||
|
|
|
|||
|
|
def warning(self, message: str, emoji: Optional[str] = None):
|
|||
|
|
"""
|
|||
|
|
输出警告级别日志
|
|||
|
|
|
|||
|
|
参数:
|
|||
|
|
message: 日志消息
|
|||
|
|
emoji: 可选的emoji名称
|
|||
|
|
"""
|
|||
|
|
msg = f"{EMOJI.get(emoji, EMOJI['警告']) if emoji else ''} {message}"
|
|||
|
|
if LOGURU_AVAILABLE:
|
|||
|
|
self.logger.warning(msg)
|
|||
|
|
else:
|
|||
|
|
self.logger.warning(msg)
|
|||
|
|
|
|||
|
|
def error(self, message: str, emoji: Optional[str] = None):
|
|||
|
|
"""
|
|||
|
|
输出错误级别日志
|
|||
|
|
|
|||
|
|
参数:
|
|||
|
|
message: 日志消息
|
|||
|
|
emoji: 可选的emoji名称
|
|||
|
|
"""
|
|||
|
|
msg = f"{EMOJI.get(emoji, EMOJI['错误']) if emoji else ''} {message}"
|
|||
|
|
if LOGURU_AVAILABLE:
|
|||
|
|
self.logger.error(msg)
|
|||
|
|
else:
|
|||
|
|
self.logger.error(msg)
|
|||
|
|
|
|||
|
|
def success(self, message: str, emoji: Optional[str] = None):
|
|||
|
|
"""
|
|||
|
|
输出成功级别日志
|
|||
|
|
|
|||
|
|
参数:
|
|||
|
|
message: 日志消息
|
|||
|
|
emoji: 可选的emoji名称
|
|||
|
|
"""
|
|||
|
|
msg = f"{EMOJI.get(emoji, EMOJI['成功']) if emoji else ''} {message}"
|
|||
|
|
if LOGURU_AVAILABLE:
|
|||
|
|
self.logger.success(msg)
|
|||
|
|
else:
|
|||
|
|
# 标准logging没有success级别，用info代替
|
|||
|
|
self.logger.info(f"[成功] {msg}")
|
|||
|
|
|
|||
|
|
def critical(self, message: str, emoji: Optional[str] = None):
|
|||
|
|
"""
|
|||
|
|
输出严重错误级别日志
|
|||
|
|
|
|||
|
|
参数:
|
|||
|
|
message: 日志消息
|
|||
|
|
emoji: 可选的emoji名称
|
|||
|
|
"""
|
|||
|
|
msg = f"{EMOJI.get(emoji, EMOJI['错误']) if emoji else ''} {message}"
|
|||
|
|
if LOGURU_AVAILABLE:
|
|||
|
|
self.logger.critical(msg)
|
|||
|
|
else:
|
|||
|
|
self.logger.critical(msg)
|
|||
|
|
|
|||
|
|
def timing(self, operation: str, elapsed_time: float):
|
|||
|
|
"""
|
|||
|
|
记录操作耗时
|
|||
|
|
|
|||
|
|
参数:
|
|||
|
|
operation: 操作名称
|
|||
|
|
elapsed_time: 耗时(秒)
|
|||
|
|
"""
|
|||
|
|
if elapsed_time < 0.1:
|
|||
|
|
time_str = f"{elapsed_time*1000:.1f}毫秒"
|
|||
|
|
else:
|
|||
|
|
time_str = f"{elapsed_time:.3f}秒"
|
|||
|
|
self.info(f"{operation}耗时: {time_str}", "时间")
|
|||
|
|
|
|||
|
|
def section(self, title: str, emoji: Optional[str] = None):
|
|||
|
|
"""
|
|||
|
|
输出一个分节标题
|
|||
|
|
|
|||
|
|
参数:
|
|||
|
|
title: 标题内容
|
|||
|
|
emoji: 可选的emoji名称
|
|||
|
|
"""
|
|||
|
|
emoji_char = EMOJI.get(emoji, "") if emoji else ""
|
|||
|
|
separator = "=" * (30 - len(title) * 2)
|
|||
|
|
message = f"{separator} {emoji_char} {title} {emoji_char} {separator}"
|
|||
|
|
|
|||
|
|
if LOGURU_AVAILABLE:
|
|||
|
|
self.logger.opt(colors=True).info(f"<yellow>{message}</yellow>")
|
|||
|
|
else:
|
|||
|
|
self.logger.info(message)
|
|||
|
|
|
|||
|
|
def table(self, title: str, columns: list, rows: list, caption: Optional[str] = None):
|
|||
|
|
"""
|
|||
|
|
打印表格数据
|
|||
|
|
|
|||
|
|
参数:
|
|||
|
|
title: 表格标题
|
|||
|
|
columns: 列名列表
|
|||
|
|
rows: 行数据列表，每行是一个包含多个值的列表
|
|||
|
|
caption: 可选的表格说明
|
|||
|
|
"""
|
|||
|
|
if RICH_AVAILABLE:
|
|||
|
|
table = Table(title=title, caption=caption, show_header=True, header_style="bold magenta")
|
|||
|
|
|
|||
|
|
# 添加列
|
|||
|
|
for col in columns:
|
|||
|
|
table.add_column(col)
|
|||
|
|
|
|||
|
|
# 添加行
|
|||
|
|
for row in rows:
|
|||
|
|
table.add_row(*[str(item) for item in row])
|
|||
|
|
|
|||
|
|
# 打印表格
|
|||
|
|
console.print(table)
|
|||
|
|
|
|||
|
|
# 同时记录日志
|
|||
|
|
self.info(f"{title} - 表格数据，共{len(rows)}行", "信息")
|
|||
|
|
else:
|
|||
|
|
# 不支持rich时，使用简单格式输出
|
|||
|
|
self.info(f"{title} - 表格数据:", "信息")
|
|||
|
|
self.info(f"列: {', '.join(columns)}", "信息")
|
|||
|
|
|
|||
|
|
for i, row in enumerate(rows):
|
|||
|
|
self.info(f"行 {i+1}: {' | '.join([str(item) for item in row])}", "信息")
|
|||
|
|
|
|||
|
|
def progress(self, items: list, desc: str = "处理中", *, unit: str = "项"):
|
|||
|
|
"""
|
|||
|
|
显示进度条并处理列表中的每个项
|
|||
|
|
|
|||
|
|
参数:
|
|||
|
|
items: 要处理的项列表
|
|||
|
|
desc: 进度条描述
|
|||
|
|
unit: 单位名称
|
|||
|
|
|
|||
|
|
用法:
|
|||
|
|
for item in logger.progress(items, "处理文件"):
|
|||
|
|
# 对每个item进行处理
|
|||
|
|
process(item)
|
|||
|
|
"""
|
|||
|
|
if RICH_AVAILABLE:
|
|||
|
|
with Progress(
|
|||
|
|
SpinnerColumn(),
|
|||
|
|
TextColumn("[bold blue]{task.description}"),
|
|||
|
|
BarColumn(),
|
|||
|
|
TextColumn("[progress.percentage]{task.percentage:>3.0f}%"),
|
|||
|
|
TextColumn(f"[bold green]{{task.completed}}/{{task.total}} {unit}"),
|
|||
|
|
TimeElapsedColumn(),
|
|||
|
|
console=console
|
|||
|
|
) as progress:
|
|||
|
|
task = progress.add_task(desc, total=len(items))
|
|||
|
|
for item in items:
|
|||
|
|
yield item
|
|||
|
|
progress.advance(task)
|
|||
|
|
else:
|
|||
|
|
# 不支持rich时直接迭代，只打印开始和结束
|
|||
|
|
self.info(f"{desc} - 开始处理 {len(items)} {unit}", "信息")
|
|||
|
|
start_time = time.time()
|
|||
|
|
|
|||
|
|
for i, item in enumerate(items):
|
|||
|
|
yield item
|
|||
|
|
if (i+1) % max(1, len(items)//10) == 0: # 每处理10%打印一次
|
|||
|
|
self.info(f"{desc} - 已处理 {i+1}/{len(items)} {unit}", "信息")
|
|||
|
|
|
|||
|
|
elapsed = time.time() - start_time
|
|||
|
|
self.timing(f"{desc} - 完成处理 {len(items)} {unit}", elapsed)
|
|||
|
|
|
|||
|
|
def syntax(self, code: str, language: str = "python", title: Optional[str] = None):
|
|||
|
|
"""
|
|||
|
|
打印语法高亮的代码
|
|||
|
|
|
|||
|
|
参数:
|
|||
|
|
code: 代码内容
|
|||
|
|
language: 代码语言
|
|||
|
|
title: 可选的标题
|
|||
|
|
"""
|
|||
|
|
if RICH_AVAILABLE:
|
|||
|
|
syntax = Syntax(code, language, theme="monokai", line_numbers=True, word_wrap=True)
|
|||
|
|
if title:
|
|||
|
|
console.print(Panel(syntax, title=title))
|
|||
|
|
else:
|
|||
|
|
console.print(syntax)
|
|||
|
|
else:
|
|||
|
|
# 不支持rich时直接打印代码，添加标记
|
|||
|
|
if title:
|
|||
|
|
self.info(f"代码块: {title}", "调试")
|
|||
|
|
lines = code.split("\n")
|
|||
|
|
for i, line in enumerate(lines):
|
|||
|
|
self.debug(f"{i+1:4d} | {line}")
|
|||
|
|
|
|||
|
|
def json(self, data: Any, title: Optional[str] = None):
|
|||
|
|
"""
|
|||
|
|
美化打印JSON数据
|
|||
|
|
|
|||
|
|
参数:
|
|||
|
|
data: 要打印的JSON数据对象
|
|||
|
|
title: 可选的标题
|
|||
|
|
"""
|
|||
|
|
if RICH_AVAILABLE:
|
|||
|
|
if title:
|
|||
|
|
console.print(f"[bold]{title}[/bold]")
|
|||
|
|
console.print(JSON.from_data(data))
|
|||
|
|
else:
|
|||
|
|
# 不支持rich时使用json模块直接打印
|
|||
|
|
import json
|
|||
|
|
if title:
|
|||
|
|
self.info(f"JSON数据: {title}", "信息")
|
|||
|
|
self.info(json.dumps(data, indent=2, ensure_ascii=False), "信息")
|
|||
|
|
|
|||
|
|
def panel(self, content: str, title: Optional[str] = None, style: str = "bold green"):
|
|||
|
|
"""
|
|||
|
|
打印带边框的面板
|
|||
|
|
|
|||
|
|
参数:
|
|||
|
|
content: 面板内容
|
|||
|
|
title: 可选的标题
|
|||
|
|
style: 样式字符串
|
|||
|
|
"""
|
|||
|
|
if RICH_AVAILABLE:
|
|||
|
|
console.print(Panel(content, title=title, style=style))
|
|||
|
|
else:
|
|||
|
|
# 不支持rich时直接输出
|
|||
|
|
if title:
|
|||
|
|
self.info(f"{title}", "信息")
|
|||
|
|
self.info(content, "信息")
|
|||
|
|
|
|||
|
|
# 创建默认实例
|
|||
|
|
log = LogHelper()
|
|||
|
|
|
|||
|
|
def get_logger(name: str = None) -> LogHelper:
|
|||
|
|
"""
|
|||
|
|
获取指定名称的日志器
|
|||
|
|
|
|||
|
|
参数:
|
|||
|
|
name: 日志器名称
|
|||
|
|
|
|||
|
|
返回:
|
|||
|
|
LogHelper实例
|
|||
|
|
"""
|
|||
|
|
if name:
|
|||
|
|
return LogHelper(name)
|
|||
|
|
return log
|
|||
|
|
|
|||
|
|
# 导出主要函数，方便直接使用
|
|||
|
|
debug = log.debug
|
|||
|
|
info = log.info
|
|||
|
|
warning = log.warning
|
|||
|
|
error = log.error
|
|||
|
|
success = log.success
|
|||
|
|
critical = log.critical
|
|||
|
|
timing = log.timing
|
|||
|
|
section = log.section
|
|||
|
|
table = log.table
|
|||
|
|
progress = log.progress
|
|||
|
|
syntax = log.syntax
|
|||
|
|
json_log = log.json
|
|||
|
|
panel = log.panel
|