modbuslink.common.logging 源代码

"""
ModbusLink 日志系统

ModbusLink Logging System
"""

import sys
import logging
from typing import Optional
from .language import Language, set_language, get_message




class BilingualLogger:
    """
    双语日志适配器

    Bilingual log adapter
    """



    def __init__(self, name: str) -> None:
        self._logger = logging.getLogger(name)


    @staticmethod
    def _get_msg(cn: str, en: str) -> str:
        return get_message(cn, en)



    def debug(self, cn: str, en: str, *args, **kwargs) -> None:
        if self._logger.isEnabledFor(logging.DEBUG):
            msg = self._get_msg(cn, en)
            self._logger.debug(msg, *args, stacklevel=2, **kwargs)




    def info(self, cn: str, en: str, *args, **kwargs) -> None:
        if self._logger.isEnabledFor(logging.INFO):
            msg = self._get_msg(cn, en)
            self._logger.info(msg, *args, stacklevel=2, **kwargs)




    def warning(self, cn: str, en: str, *args, **kwargs) -> None:
        if self._logger.isEnabledFor(logging.WARNING):
            msg = self._get_msg(cn, en)
            self._logger.warning(msg, *args, stacklevel=2, **kwargs)




    def error(self, cn: str, en: str, *args, **kwargs) -> None:
        if self._logger.isEnabledFor(logging.ERROR):
            msg = self._get_msg(cn, en)
            self._logger.error(msg, *args, stacklevel=2, **kwargs)




    def critical(self, cn: str, en: str, *args, **kwargs) -> None:
        if self._logger.isEnabledFor(logging.CRITICAL):
            msg = self._get_msg(cn, en)
            self._logger.critical(msg, *args, stacklevel=2, **kwargs)




    def exception(self, cn: str, en: str, *args, **kwargs) -> None:
        msg = self._get_msg(cn, en)
        # exception方法本质上是 error(..., exc_info=True)
        self._logger.exception(msg, *args, stacklevel=2, **kwargs)




    def set_level(self, level) -> None:
        self._logger.setLevel(level)






class ModbusLogger:
    """
    ModbusLink日志管理器

    ModbusLink Log Manager
    """

    # 默认日志格式 | Default log format
    DEFAULT_FORMAT = "%(asctime)s - %(name)s - %(levelname)s - %(message)s"
    DEBUG_FORMAT = "%(asctime)s - %(name)s - %(levelname)s - [%(filename)s:%(lineno)d] - %(message)s"



    @staticmethod
    def setup_logging(
            level: int = logging.INFO,
            format_string: Optional[str] = None,
            enable_debug: bool = False,
            log_file: Optional[str] = None,
            language: str = Language.CN
    ) -> None:
        """
        设置ModbusLink的日志配置

        Setup ModbusLink logging configuration

        Args:
            level: 日志级别，默认INFO | Log level, default INFO
            format_string: 自定义日志格式 | Custom log format
            enable_debug: 是否启用调试模式（显示详细信息） | Enable debug mode (show detailed info)
            log_file: 日志文件路径，如果提供则同时输出到文件 | Log file path, if provided, also output to file
            language: 语言设置 | Language settings (Language.CN or Language.EN)
        """
        # 设置语言 | Set language
        set_language(language)

        # 选择日志格式 | Choose log format
        if format_string is None:
            format_string = ModbusLogger.DEBUG_FORMAT if enable_debug else ModbusLogger.DEFAULT_FORMAT

        # 创建格式化器 | Create formatter
        formatter = logging.Formatter(format_string)

        # 获取根日志器 | Get root logger
        root_logger = logging.getLogger("modbuslink")
        root_logger.setLevel(level)

        # 清除现有处理器 | Clear existing handlers
        root_logger.handlers.clear()

        # 添加控制台处理器 | Add console handler
        console_handler = logging.StreamHandler(sys.stdout)
        console_handler.setLevel(level)
        console_handler.setFormatter(formatter)
        root_logger.addHandler(console_handler)

        # 添加文件处理器（如果指定） | Add file handler (if specified)
        if log_file:
            file_handler = logging.FileHandler(log_file, encoding="utf-8")
            file_handler.setLevel(level)
            file_handler.setFormatter(formatter)
            root_logger.addHandler(file_handler)

        # 防止日志传播到根日志器 | Prevent log propagation to root logger
        root_logger.propagate = False

        # 使用标准logger打印初始化消息，因为此时还没有BilingualLogger实例
        init_msg = (
            f"ModbusLink日志系统已初始化 - Level: {logging.getLevelName(level)}"
            if language == Language.CN
            else f"ModbusLink logging system initialized - Level: {logging.getLevelName(level)}"
        )
        root_logger.info(init_msg)




    @staticmethod
    def get_logger(name: str) -> BilingualLogger:
        """
        获取指定名称的日志器

        Get logger with specified name

        Args:
            name: 日志器名称 | Logger name

        Returns:
            配置好的日志器实例

            Configured logger instance
        """
        # 内部使用 modbuslink. 前缀，但对用户隐藏 | Internally, it uses the "modbuslink." prefix, but hides it from the users.
        return BilingualLogger(f"modbuslink.{name}")




    @staticmethod
    def enable_protocol_debug() -> None:
        """
        启用协议级别的调试日志

        显示原始的十六进制数据包，用于调试通信问题。

        Enable protocol-level debug logging

        Shows raw hexadecimal packets for debugging communication issues.
        """
        # 设置传输层日志级别为DEBUG | Set transport layer log level to DEBUG
        logging.getLogger("modbuslink.transport").setLevel(logging.DEBUG)

        # 设置客户端日志级别为DEBUG | Set client layer log level to DEBUG
        logging.getLogger("modbuslink.client").setLevel(logging.DEBUG)

        # 设置服务端日志级别为DEBUG | Set server layer log level to DEBUG
        logging.getLogger("modbuslink.server").setLevel(logging.DEBUG)

        msg = get_message("协议调试模式已启用", "Protocol debug mode enabled")
        logging.getLogger("modbuslink").info(msg)




    @staticmethod
    def disable_protocol_debug() -> None:
        """
        禁用协议级别的调试日志

        Disable protocol-level debug logging
        """
        # 恢复传输层日志级别 | Restore transport layer log level
        logging.getLogger("modbuslink.transport").setLevel(logging.INFO)

        # 恢复客户端日志级别 | Restore client layer log level
        logging.getLogger("modbuslink.client").setLevel(logging.INFO)

        # 恢复服务端日志级别 | Restore server layer log level
        logging.getLogger("modbuslink.server").setLevel(logging.INFO)

        msg = get_message("协议调试模式已禁用", "Protocol debug mode disabled")
        logging.getLogger("modbuslink").info(msg)






def get_logger(name: str) -> BilingualLogger:
    """
    便捷函数：获取ModbusLink日志器

    Convenience function: get ModbusLink logger

    Args:
        name: 日志器名称 | Logger name

    Returns:
        配置好的日志器实例

        Configured logger instance
    """
    return ModbusLogger.get_logger(name)



logging.getLogger("modbuslink").addHandler(logging.NullHandler())