personal-consumption-service/app/handler/logs_handler.py

# @description: 
# @author: licanglong
# @date: 2025/11/20 14:05
import logging
import os
import sys

from concurrent_log import ConcurrentTimedRotatingFileHandler

from app.core import ColorConsoleFormatter, CTX, EM
from app.handler import ApplicationStartupEvent

_log = logging.getLogger(__name__)
custom_stream_handler = logging.StreamHandler()
if not getattr(sys, 'frozen', False):  # 开发环境
    custom_stream_handler.setFormatter(ColorConsoleFormatter())
# 确保日志路径
logging.basicConfig(
    level=logging.INFO,
    format="{asctime} {levelname:>7} {threadName:^10} [{filename}#{funcName}:{lineno}]: {message}",
    style="{",
    encoding='utf-8',
    handlers=[custom_stream_handler]
)


@EM.subscribe(ApplicationStartupEvent, priority=sys.maxsize - 1)
def init_logger_onstartup(event: ApplicationStartupEvent):
    """加载并初始化配置"""
    log_handlers = []
    logpath = CTX.ENV.getprop('log.path', CTX.DEFAULT_LOG_FILE)
    custom_stream_handler = logging.StreamHandler()
    if getattr(sys, 'frozen', False):  # 打包后的环境
        log_path = os.path.join(os.path.dirname(sys.executable), logpath)
    else:  # 开发环境
        custom_stream_handler.setFormatter(ColorConsoleFormatter())
        log_path = os.path.join(os.getenv('APP_PATH'), logpath)
    if not os.path.exists(os.path.dirname(log_path)):
        os.makedirs(os.path.dirname(log_path), exist_ok=True)
    log_handlers.append(custom_stream_handler)

    file_handler = ConcurrentTimedRotatingFileHandler(
        filename=log_path,
        when='midnight',  # 每天午夜切割
        backupCount=30,  # 保留最近30天的日志
        encoding='utf-8',
    )
    _log.info(f"日志文件路径：{log_path}")
    log_handlers.append(file_handler)

    # loki_domain = _APP.ENV.get("log.loki.domain")
    # if loki_domain:
    #     import logging_loki
    #     loki_handler = logging_loki.LokiHandler(
    #         url=loki_domain + "/loki/api/v1/push",
    #         tags={"application": _APP.ENV.get("log.loki.app-name", "paddleocr-service")},
    #         # auth=("username", "password"),
    #         version="2",
    #     )
    #     logging.getLogger("urllib3").propagate = False
    #     logging.getLogger("requests").propagate = False
    #     log_handlers.append(loki_handler)

    # 确保日志路径
    logging.basicConfig(
        level=CTX.ENV.getprop('log.level', logging.DEBUG),
        format="{asctime} {levelname:>7} {threadName:^10} [{filename}#{funcName}:{lineno}]: {message}",
        style="{",
        encoding='utf-8',
        handlers=log_handlers,
        force=True
    )
    _log.info(f"日志级别：{logging.getLevelName(logging.getLogger().level)}")