ai_qback/tools/loglog.py

# -*- coding=utf-8 -*-
import time, os
import traceback
from loguru import logger
from threading import Lock
from tools.ali_log import put_logs


class SimpleLogger:
    """
    简易日志：存放几乎没用的大量gpt日志
    """

    def __init__(self, base_file_name: str = "ai_log"):
        self.base_file_name = "log/" + base_file_name
        self.file_ext = ".txt"
        self.max_size = 10 * 1024 * 1024
        self.current_file = self._get_current_file()
        self.lock = Lock()
        if not os.path.exists("log"):
            os.mkdir("log")

    def _get_current_file(self):
        """获取当前应该写入的文件"""
        i = 1
        while True:
            file_name = f"{self.base_file_name}_{i}{self.file_ext}"
            if not os.path.exists(file_name) or os.path.getsize(file_name) < self.max_size:
                return file_name
            i += 1

    def _check_file_size(self):
        """检查文件大小，如果需要则切换到新文件"""
        try:
            if os.path.getsize(self.current_file) >= self.max_size:
                self.current_file = self._get_current_file()
        except FileNotFoundError:
            with open(f"{self.base_file_name}_1{self.file_ext}", "a", encoding="utf-8") as log_file:
                log_file.write(f"Hello World\n")

    def log(self, message: str, level="INFO"):
        """记录日志到文件"""
        self._check_file_size()
        date_time = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
        with self.lock:
            with open(self.current_file, "a", encoding="utf-8") as log_file:
                log_file.write(f"{date_time} 【{level}】 {str(message)}\n\n")

    def info(self, message: str):
        """记录INFO级别的日志"""
        self.log(message, "INFO")

    def warning(self, message: str):
        """记录WARNING级别的日志"""
        self.log(message, "WARNING")

    def error(self, message: str):
        """记录ERROR级别的日志"""
        message = "\n" + "-" * 20 + "\n" + message + "\n" + "-" * 20
        self.log(message, "ERROR")

    def debug(self, message: str):
        """记录DEBUG级别的日志"""
        self.log(message, "DEBUG")


logger.remove(handler_id=None)
logger.add('log/log.log', level="INFO", rotation="5 MB", encoding="utf-8", retention="7 days")
logger.add('log/error.log', level="ERROR", rotation="5 MB", encoding="utf-8", retention="7 days")
simple_logger = SimpleLogger()


def exception_handler(func):
    def wrapper(*args, **kwargs):
        try:
            return func(*args, **kwargs)
        except Exception as e:
            logger.error(f"{type(e).__name__}: {e}")
            traceback_str = traceback.format_exc()
            logger.error(f"错误追溯:{traceback_str}")

    return wrapper


def log_err_e(e: Exception, msg=None):
    if msg:
        logger.error(f"{msg}{type(e).__name__}:{e}")
    traceback_str = traceback.format_exc()
    logger.error(traceback_str)


class AliyunLogHandler:
    @staticmethod
    def write(message):
        put_logs(message)


if os.getenv("env") != "development":
    print("这是正式环境，加载阿里云日志")
    aliyun_log_handler = AliyunLogHandler()
    logger.add(aliyun_log_handler, enqueue=True)

if __name__ == '__main__':
    #

    import os

    os.chdir("..")
    logger.error("test信息0123456789.*/-")