log.py

# ----------------------------------------
# - 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')