mirror of
https://github.com/KenanZhu/AutoLibrary.git
synced 2026-06-18 07:23:03 +08:00
02463f087e
- 为 _showTrace 方法添加 no_log 参数,支持控制日志写入 - 在主窗口各关键操作点添加日志输出 - 优化错误信息输出策略,分离 trace 和 log 输出 - 改进配置目录初始化过程的日志记录
103 lines
2.6 KiB
Python
103 lines
2.6 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""
|
|
Copyright (c) 2025 - 2026 KenanZhu.
|
|
All rights reserved.
|
|
|
|
This software is provided "as is", without any warranty of any kind.
|
|
You may use, modify, and distribute this file under the terms of the MIT License.
|
|
See the LICENSE file for details.
|
|
"""
|
|
import logging
|
|
import queue
|
|
import datetime
|
|
|
|
from utils.LogManager import getLogger
|
|
|
|
|
|
class MsgBase:
|
|
"""
|
|
Base class for message and trace abilities (thread-safe).
|
|
|
|
This class provides the foundation for message handling and tracing
|
|
abilities based on the provided input and output queues. It enables
|
|
thread-safe communication between components using queue-based messaging.
|
|
|
|
Args:
|
|
input_queue (queue.Queue): The input queue for receiving messages.
|
|
output_queue (queue.Queue): The output queue for sending messages.
|
|
|
|
Usage:
|
|
This class must be initialized with input and output queues. The queue
|
|
provider (the caller of this class or its subclasses) must explicitly
|
|
implement queue polling to retrieve and process messages.
|
|
"""
|
|
|
|
class TraceLevel:
|
|
"""
|
|
Enum class for trace levels.
|
|
|
|
This class provides the trace levels for the logger.
|
|
"""
|
|
DEBUG = logging.DEBUG
|
|
INFO = logging.INFO
|
|
WARNING = logging.WARNING
|
|
ERROR = logging.ERROR
|
|
CRITICAL = logging.CRITICAL
|
|
|
|
def __init__(
|
|
self,
|
|
input_queue: queue.Queue,
|
|
output_queue: queue.Queue
|
|
):
|
|
|
|
self._class_name = self.__class__.__name__
|
|
self._input_queue = input_queue
|
|
self._output_queue = output_queue
|
|
try:
|
|
self._logger = getLogger(self._class_name)
|
|
except RuntimeError:
|
|
self._logger = None
|
|
|
|
|
|
def _showMsg(
|
|
self,
|
|
msg: str
|
|
):
|
|
|
|
self._output_queue.put(f"[{self._class_name:<15}] >>> : {msg}")
|
|
|
|
|
|
def _showTrace(
|
|
self,
|
|
msg: str,
|
|
level: int = logging.INFO,
|
|
no_log: bool = False
|
|
):
|
|
|
|
timestamp = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S.%f")[:-3]
|
|
self._output_queue.put(f"{timestamp}-[{self._class_name:<15}] : {msg}")
|
|
if self._logger and not no_log:
|
|
self._logger.log(level, msg)
|
|
|
|
|
|
def _showLog(
|
|
self,
|
|
msg: str,
|
|
level: int = logging.INFO
|
|
):
|
|
|
|
if self._logger:
|
|
self._logger.log(level, msg)
|
|
|
|
|
|
def _waitMsg(
|
|
self,
|
|
timeout: float = 1.0
|
|
) -> str:
|
|
|
|
try:
|
|
msg = self._input_queue.get(timeout=timeout)
|
|
return msg
|
|
except queue.Empty:
|
|
return None
|