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 |