mirror of
https://github.com/MeoProject/lx-music-api-server.git
synced 2025-05-23 19:17:41 +08:00
314 lines
12 KiB
Python
314 lines
12 KiB
Python
# ----------------------------------------
|
|
# - mode: python -
|
|
# - author: helloplhm-qwq -
|
|
# - name: log.py -
|
|
# - project: lx-music-api-server -
|
|
# - license: MIT -
|
|
# ----------------------------------------
|
|
# This file is part of the "lx-music-api-server" project.
|
|
|
|
import logging
|
|
import colorlog
|
|
import os
|
|
import sys
|
|
import re
|
|
import io
|
|
import traceback
|
|
import time
|
|
from pygments import highlight
|
|
from pygments.lexers import PythonLexer
|
|
from pygments.formatters import TerminalFormatter
|
|
from .utils import filterFileName, setGlobal, require
|
|
from .variable import debug_mode, log_length_limit, log_file, log_files
|
|
from colorama import Fore, Style
|
|
from colorama import init as clinit
|
|
|
|
clinit() # 初始化 colorama
|
|
|
|
if ((not os.path.exists("logs")) and log_file):
|
|
try:
|
|
os.mkdir("logs")
|
|
except:
|
|
pass
|
|
|
|
class Color:
|
|
"""
|
|
彩色文字处理器
|
|
"""
|
|
|
|
def __getattr__(self, k):
|
|
return lambda x: f"{getattr(Fore, k.upper())}{x}{Style.RESET_ALL}"
|
|
|
|
color = Color()
|
|
# purple_parentheses = color.purple('(') + '{}' + color.purple(')')
|
|
|
|
def is_rubbish(input_string):
|
|
return bool(re.match(r'^(\~|\^)*$', input_string))
|
|
|
|
def stack_error(exception):
|
|
stack_trace = traceback.format_exception(type(exception), exception, exception.__traceback__)
|
|
return ''.join(stack_trace)
|
|
|
|
def python_highlight(code):
|
|
return highlight(code, PythonLexer(), TerminalFormatter())
|
|
|
|
def read_code(file_path, target_line_number):
|
|
try:
|
|
with open(file_path, 'r', encoding='utf-8') as file:
|
|
lines = file.readlines()
|
|
start = max(0, target_line_number - 4)
|
|
end = min(target_line_number + 4, len(lines))
|
|
lineMap = {
|
|
'current': lines[target_line_number - 1],
|
|
'result': lines[start:end]
|
|
}
|
|
# print(lineMap)
|
|
return lineMap
|
|
except FileNotFoundError:
|
|
sys.stderr.write("日志模块出错，本次日志可能无法记录，请报告给开发者: 处理错误语法高亮时找不到源文件")
|
|
except Exception as e:
|
|
sys.stderr.write(f"日志模块出错，本次日志可能无法记录，请报告给开发者: \n" + traceback.format_exc())
|
|
|
|
def stack_info(stack_trace_line):
|
|
try:
|
|
parts = stack_trace_line.split(', ')
|
|
file_path = ' '.join(parts[0].split(' ')[1:]) # 提取文件路径
|
|
line_number = int(parts[1].split(' ')[1]) # 提取行号
|
|
function_name = parts[2].split(' ')[1] # 提取函数名
|
|
return file_path, line_number, function_name
|
|
except Exception as e:
|
|
sys.stderr.write(f"日志模块出错，本次日志可能无法记录，请报告给开发者: \n" + traceback.format_exc())
|
|
|
|
def highlight_error(e):
|
|
try:
|
|
if (isinstance(e, Exception)):
|
|
error = stack_error(e)
|
|
else:
|
|
error = e
|
|
lines = [i.strip() for i in error.split("\n") if i.strip()]
|
|
final = []
|
|
ign = False
|
|
for i in lines:
|
|
if (ign):
|
|
ign = False
|
|
continue
|
|
|
|
if (i.startswith("Traceback (most recent call last):")):
|
|
final.append(color.cyan(i))
|
|
elif (i.startswith("During handling of the above exception, another exception occurred:")):
|
|
final.append(color.cyan(i))
|
|
elif (i.startswith("The above exception was the direct cause of the following exception:")):
|
|
final.append(color.cyan(i))
|
|
elif (i.startswith("File")):
|
|
ign = True
|
|
p, l, f = stack_info(i)
|
|
p = p[1:-1]
|
|
|
|
if (p.startswith('<') or not os.path.isfile(p)):
|
|
final.append(i)
|
|
final.append(" " if (lines[lines.index(l) + 1].startswith("File")) else (" " + lines[lines.index(l) + 1]))
|
|
continue
|
|
|
|
code = read_code(p, l)
|
|
cc = []
|
|
viewed = False
|
|
firstLineNumber = max(l - 3, 1)
|
|
line_number = firstLineNumber
|
|
for i in range(len(code['result'])):
|
|
c = code["result"][i]
|
|
if (c.startswith(code['current']) and (not viewed) and (i <= 3)):
|
|
viewed = True
|
|
line_number = line_number + 1
|
|
# print(line_number)
|
|
cc.append((' ' * (10 - len(str(l))) + f'{color.red(str(l))} >|' + python_highlight(c)))
|
|
else:
|
|
cc.append((' ' * (10 - len(str(line_number))) + f'{color.blue(str(line_number))} |' + python_highlight(c)))
|
|
line_number = line_number + 1
|
|
# print(line_number)
|
|
code = "\n".join(cc)
|
|
p = '"' + p + '"'
|
|
final.append(f" File {color.yellow(f'{p}')} in {color.cyan(f) + '()' if (not f.startswith('<')) else ''}\n\n\n{code}\n")
|
|
else:
|
|
try:
|
|
if (is_rubbish(i)):
|
|
continue
|
|
if (issubclass(require(("builtins." if ("." not in i.split(":")[0]) else "") + i.split(":")[0]), Exception)):
|
|
exc = i.split(":")[0]
|
|
desc = "" if (len(i.split(":")) == 1) else ':'.join(i.split(":")[1:]).strip()
|
|
final.append(color.red(exc) + (": " + color.yellow(desc)) if (desc) else "")
|
|
else:
|
|
final.append(color.cyan(i))
|
|
except:
|
|
# traceback.print_exc()
|
|
final.append(i)
|
|
return "\n".join(final).replace('\n\n', '\n')
|
|
'''
|
|
lines = [i for i in error.split("\n") if (i.strip() and not is_rubbish(i))]
|
|
|
|
final = []
|
|
ispass = False # pass handle next line
|
|
for index in range(len(lines)):
|
|
i = lines[index]
|
|
i2 = i.strip()
|
|
if (ispass):
|
|
ispass = False
|
|
continue
|
|
if (i2.startswith("Traceback (most recent call last):")):
|
|
final.append(color.cyan(i))
|
|
elif (i2.startswith("During handling of the above exception, another exception occurred:")):
|
|
final.append(color.cyan(i))
|
|
elif (i2.startswith("The above exception was the direct cause of the following exception:")):
|
|
final.append(color.cyan(i))
|
|
elif (i2.startswith("File")):
|
|
p, l, f = stack_info(i2)
|
|
p = p[1:-1]
|
|
if (p.startswith('<') or not os.path.isfile(p)):
|
|
final.append(f' File {color.yellow("\"" + p + "\"")} in {color.cyan(f)}() [line {color.blue(l)}]\n')
|
|
_probably_err_name = lines[index + 1].split(':')[0]
|
|
if not (lines.get(index + 1, '').strip().startswith("File") or require(
|
|
('builtins.' + _probably_err_name) if ("." not in _probably_err_name) else _probably_err_name
|
|
) is Exception):
|
|
final.append(f" > {color.red(l)} |" + python_highlight(lines[index + 1]))
|
|
ispass = True
|
|
continue
|
|
codeMap = read_code(p, l)
|
|
line_length = len(codeMap['result'])
|
|
|
|
'''
|
|
|
|
|
|
except:
|
|
sys.stderr.write('格式化错误失败，使用默认格式\n' + traceback.format_exc())
|
|
if (isinstance(e, Exception)):
|
|
return stack_error(e)
|
|
else:
|
|
return e
|
|
|
|
class LogHelper(logging.Handler):
|
|
# 日志转接器
|
|
def __init__(self, custom_logger):
|
|
super().__init__()
|
|
self.custom_logger = custom_logger
|
|
|
|
def emit(self, record):
|
|
# print(record)
|
|
log_message = self.format(record)
|
|
self.custom_logger.info(log_message)
|
|
|
|
class fileWriter(logging.Handler):
|
|
def __init__(self, f: io.TextIOWrapper, f2: logging.Formatter):
|
|
self.file = f
|
|
self.formatter = f2
|
|
|
|
def emit(self, record: logging.LogRecord):
|
|
self.file.write(self.format(record) + '\n')
|
|
self.file.flush()
|
|
|
|
class log:
|
|
# 主类
|
|
def __init__(self, module_name='Not named logger', output_level='INFO', filename=''):
|
|
self.name = module_name
|
|
self._logger = logging.getLogger(module_name)
|
|
if not output_level.upper() in dir(logging):
|
|
raise NameError('Unknown loglevel: '+output_level)
|
|
if not debug_mode:
|
|
self._logger.setLevel(getattr(logging, output_level.upper()))
|
|
else:
|
|
self._logger.setLevel(logging.DEBUG)
|
|
formatter = colorlog.ColoredFormatter(
|
|
'%(log_color)s%(asctime)s|[%(name)s/%(levelname)s]|%(message)s',
|
|
datefmt='%Y-%m-%d %H:%M:%S',
|
|
log_colors={
|
|
'DEBUG': 'cyan',
|
|
'INFO': 'white',
|
|
'WARNING': 'yellow',
|
|
'ERROR': 'red',
|
|
'CRITICAL': 'red,bg_white',
|
|
})
|
|
if log_file:
|
|
if filename:
|
|
filename = filterFileName(filename)
|
|
else:
|
|
filename = './logs/' + module_name + '.log'
|
|
self.file = open(filename, 'a+', encoding='utf-8')
|
|
log_files.append(self.file)
|
|
console_handler = logging.StreamHandler()
|
|
console_handler.setFormatter(formatter)
|
|
self.module_name = module_name
|
|
self._logger.addHandler(console_handler)
|
|
debug_handler = logging.StreamHandler()
|
|
debug_handler.setFormatter(formatter)
|
|
|
|
def debug(self, message, allow_hidden=True):
|
|
if (log_file):
|
|
self.file.write('{time}|[{name}/DEBUG]{msg}'.format(time = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()), name = self.module_name, msg = message) + '\n')
|
|
|
|
if len(str(message)) > log_length_limit and allow_hidden:
|
|
message = str(message)[:log_length_limit] + " ..."
|
|
self._logger.debug(message)
|
|
|
|
def log(self, message, allow_hidden=True):
|
|
if (log_file):
|
|
self.file.write('{time}|[{name}/INFO]{msg}'.format(time = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()), name = self.module_name, msg = message) + '\n')
|
|
|
|
if len(str(message)) > log_length_limit and allow_hidden:
|
|
message = str(message)[:log_length_limit] + " ..."
|
|
self._logger.info(message)
|
|
|
|
def info(self, message, allow_hidden=True):
|
|
if (log_file):
|
|
self.file.write('{time}|[{name}/INFO]{msg}'.format(time = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()), name = self.module_name, msg = message) + '\n')
|
|
|
|
if len(str(message)) > log_length_limit and allow_hidden:
|
|
message = str(message)[:log_length_limit] + "..."
|
|
self._logger.info(message)
|
|
|
|
def warning(self, message):
|
|
if (log_file):
|
|
self.file.write('{time}|[{name}/WARNING]{msg}'.format(time = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()), name = self.module_name, msg = message) + '\n')
|
|
|
|
if (message.strip().startswith('Traceback')):
|
|
self._logger.warning('\n' + highlight_error(message))
|
|
else:
|
|
self._logger.warning(message)
|
|
|
|
def error(self, message):
|
|
if (log_file):
|
|
self.file.write('{time}|[{name}/ERROR]{msg}'.format(time = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()), name = self.module_name, msg = message) + '\n')
|
|
|
|
if (message.startswith('Traceback')):
|
|
self._logger.error('\n' + highlight_error(message))
|
|
else:
|
|
self._logger.error(message)
|
|
|
|
def critical(self, message):
|
|
if (log_file):
|
|
self.file.write('{time}|[{name}/CRITICAL]{msg}'.format(time = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()), name = self.module_name, msg = message) + '\n')
|
|
|
|
if (message.startswith('Traceback')):
|
|
self._logger.critical('\n' + highlight_error(message))
|
|
else:
|
|
self._logger.critical(message)
|
|
|
|
def set_level(self, loglevel):
|
|
loglevel_upper = loglevel.upper()
|
|
if not loglevel_upper in dir(logging):
|
|
raise NameError('Unknown loglevel: ' + loglevel)
|
|
self._logger.setLevel(getattr(logging, loglevel_upper))
|
|
|
|
def getLogger(self):
|
|
return self._logger
|
|
|
|
def addHandler(self, handler):
|
|
self._logger.addHandler(handler)
|
|
|
|
|
|
printlogger = log('print')
|
|
|
|
|
|
def logprint(*args, sep=' ', end='', file=None, flush=None):
|
|
printlogger.info(sep.join(str(arg) for arg in args), allow_hidden=False)
|
|
|
|
|
|
setGlobal(logprint, 'print')
|