276 lines
10 KiB
Python
276 lines
10 KiB
Python
"""
|
|
Advanced, queue-based, thread-safe logging system for applications.
|
|
Provides handlers for console, file, and a Tkinter Text widget.
|
|
"""
|
|
import logging
|
|
import logging.handlers
|
|
import tkinter as tk
|
|
from tkinter.scrolledtext import ScrolledText
|
|
import queue
|
|
from typing import Optional, Dict, Any
|
|
|
|
# --- Module-level globals for the centralized logging system ---
|
|
_global_log_queue: Optional[queue.Queue[logging.LogRecord]] = None
|
|
_actual_console_handler: Optional[logging.StreamHandler] = None
|
|
_actual_file_handler: Optional[logging.handlers.RotatingFileHandler] = None
|
|
_actual_tkinter_handler: Optional["TkinterTextHandler"] = None
|
|
|
|
_log_processor_after_id: Optional[str] = None
|
|
_logging_system_active: bool = False
|
|
_tk_root_instance_for_processing: Optional[tk.Tk] = None
|
|
_base_formatter: Optional[logging.Formatter] = None
|
|
|
|
GLOBAL_LOG_QUEUE_POLL_INTERVAL_MS = 100
|
|
|
|
|
|
class TkinterTextHandler(logging.Handler):
|
|
"""A logging handler that directs log messages to a Tkinter Text widget."""
|
|
|
|
def __init__(
|
|
self,
|
|
text_widget: tk.Text,
|
|
level_colors: Dict[int, str],
|
|
root_tk_instance_for_widget_update: tk.Tk,
|
|
internal_poll_interval_ms: int = 100,
|
|
):
|
|
super().__init__()
|
|
self.text_widget = text_widget
|
|
self.level_colors = level_colors
|
|
self._is_active = True
|
|
self._widget_update_queue = queue.Queue()
|
|
self._root_for_widget_update = root_tk_instance_for_widget_update
|
|
self._internal_poll_interval_ms = internal_poll_interval_ms
|
|
self._internal_after_id: Optional[str] = None
|
|
|
|
if not (self.text_widget and self.text_widget.winfo_exists()):
|
|
logging.getLogger(__name__).error(
|
|
"TkinterTextHandler initialized with an invalid text_widget."
|
|
)
|
|
self._is_active = False
|
|
return
|
|
|
|
self._configure_tags()
|
|
self._process_widget_update_queue()
|
|
|
|
def _configure_tags(self):
|
|
"""Configure color tags for different log levels in the text widget."""
|
|
for level, color_value in self.level_colors.items():
|
|
level_name = logging.getLevelName(level)
|
|
if color_value:
|
|
try:
|
|
self.text_widget.tag_config(level_name, foreground=color_value)
|
|
except tk.TclError:
|
|
pass # Widget might not be ready, or color is invalid
|
|
|
|
def emit(self, record: logging.LogRecord):
|
|
"""Put the formatted log message into the internal queue."""
|
|
if not self._is_active:
|
|
return
|
|
try:
|
|
msg = self.format(record)
|
|
self._widget_update_queue.put_nowait((record.levelname, msg))
|
|
except Exception as e:
|
|
# Avoid logging loops, print directly
|
|
print(f"Error in TkinterTextHandler.emit: {e}")
|
|
|
|
def _process_widget_update_queue(self):
|
|
"""Periodically update the text widget from the internal queue."""
|
|
if not self._is_active or not self._root_for_widget_update.winfo_exists():
|
|
self.close()
|
|
return
|
|
|
|
try:
|
|
while not self._widget_update_queue.empty():
|
|
level_name, msg = self._widget_update_queue.get_nowait()
|
|
if self.text_widget.winfo_exists():
|
|
self.text_widget.configure(state=tk.NORMAL)
|
|
self.text_widget.insert(tk.END, msg + "\n", (level_name,))
|
|
self.text_widget.configure(state=tk.DISABLED)
|
|
self.text_widget.see(tk.END)
|
|
else:
|
|
self._is_active = False
|
|
break
|
|
self._widget_update_queue.task_done()
|
|
except tk.TclError:
|
|
self._is_active = False
|
|
except Exception:
|
|
self._is_active = False
|
|
|
|
if self._is_active:
|
|
self._internal_after_id = self._root_for_widget_update.after(
|
|
self._internal_poll_interval_ms, self._process_widget_update_queue
|
|
)
|
|
|
|
def close(self):
|
|
"""Cleanly stop the handler and its update loop."""
|
|
self._is_active = False
|
|
if self._internal_after_id and self._root_for_widget_update.winfo_exists():
|
|
try:
|
|
self._root_for_widget_update.after_cancel(self._internal_after_id)
|
|
except tk.TclError:
|
|
pass # Can happen if root is being destroyed
|
|
self._internal_after_id = None
|
|
super().close()
|
|
|
|
|
|
class QueuePuttingHandler(logging.Handler):
|
|
"""A simple handler that puts any received LogRecord into a global queue."""
|
|
|
|
def __init__(self, handler_queue: queue.Queue[logging.LogRecord]):
|
|
super().__init__()
|
|
self.handler_queue = handler_queue
|
|
|
|
def emit(self, record: logging.LogRecord):
|
|
try:
|
|
self.handler_queue.put_nowait(record)
|
|
except queue.Full:
|
|
print("CRITICAL: Global log queue is full! Log record might be lost.")
|
|
|
|
|
|
def _process_global_log_queue():
|
|
"""GUI Thread: Periodically processes logs from the global queue."""
|
|
global _logging_system_active, _log_processor_after_id
|
|
|
|
if not _logging_system_active:
|
|
return
|
|
|
|
if not (_tk_root_instance_for_processing and _tk_root_instance_for_processing.winfo_exists()):
|
|
_logging_system_active = False
|
|
return
|
|
|
|
try:
|
|
while not _global_log_queue.empty():
|
|
record = _global_log_queue.get_nowait()
|
|
if _actual_console_handler:
|
|
_actual_console_handler.handle(record)
|
|
if _actual_file_handler:
|
|
_actual_file_handler.handle(record)
|
|
if _actual_tkinter_handler:
|
|
_actual_tkinter_handler.handle(record)
|
|
_global_log_queue.task_done()
|
|
except queue.Empty:
|
|
pass # This is expected
|
|
except Exception as e:
|
|
print(f"Critical error in log processor: {e}")
|
|
|
|
if _logging_system_active:
|
|
_log_processor_after_id = _tk_root_instance_for_processing.after(
|
|
GLOBAL_LOG_QUEUE_POLL_INTERVAL_MS, _process_global_log_queue
|
|
)
|
|
|
|
|
|
def setup_basic_logging(
|
|
root_tk_instance: tk.Tk,
|
|
logging_config: Dict[str, Any],
|
|
):
|
|
"""Initializes the queue, console, and file handlers."""
|
|
global _global_log_queue, _actual_console_handler, _actual_file_handler
|
|
global _logging_system_active, _tk_root_instance_for_processing, _base_formatter
|
|
|
|
if _logging_system_active:
|
|
return
|
|
|
|
_base_formatter = logging.Formatter(
|
|
logging_config.get("format", "%(asctime)s [%(levelname)-8s] %(name)-25s : %(message)s"),
|
|
datefmt=logging_config.get("date_format", "%Y-%m-%d %H:%M:%S"),
|
|
)
|
|
|
|
_global_log_queue = queue.Queue()
|
|
_tk_root_instance_for_processing = root_tk_instance
|
|
|
|
root_logger = logging.getLogger()
|
|
root_logger.handlers = [] # Clear any pre-existing handlers
|
|
root_logger.setLevel(logging_config.get("default_root_level", logging.DEBUG))
|
|
|
|
if logging_config.get("enable_console", True):
|
|
_actual_console_handler = logging.StreamHandler()
|
|
_actual_console_handler.setFormatter(_base_formatter)
|
|
_actual_console_handler.setLevel(logging.DEBUG)
|
|
|
|
if logging_config.get("enable_file", False):
|
|
try:
|
|
file_path = logging_config.get("file_path", "app.log")
|
|
_actual_file_handler = logging.handlers.RotatingFileHandler(
|
|
file_path,
|
|
maxBytes=logging_config.get("file_max_bytes", 5 * 1024 * 1024),
|
|
backupCount=logging_config.get("file_backup_count", 3),
|
|
encoding="utf-8",
|
|
)
|
|
_actual_file_handler.setFormatter(_base_formatter)
|
|
except Exception as e:
|
|
print(f"Failed to create file handler: {e}")
|
|
_actual_file_handler = None
|
|
|
|
queue_putter = QueuePuttingHandler(handler_queue=_global_log_queue)
|
|
root_logger.addHandler(queue_putter)
|
|
|
|
_logging_system_active = True
|
|
_process_global_log_queue()
|
|
|
|
get_logger(__name__).info("Basic logging system initialized.")
|
|
|
|
|
|
def add_tkinter_handler(
|
|
gui_log_widget: tk.Text,
|
|
root_tk_instance: tk.Tk,
|
|
logging_config: Dict[str, Any],
|
|
):
|
|
"""Adds the Tkinter handler to the active logging system."""
|
|
global _actual_tkinter_handler
|
|
|
|
if not _logging_system_active or not _base_formatter:
|
|
print("ERROR: Cannot add Tkinter handler, basic logging is not set up.")
|
|
return
|
|
|
|
if _actual_tkinter_handler:
|
|
_actual_tkinter_handler.close()
|
|
|
|
try:
|
|
level_colors = logging_config.get("colors", {})
|
|
_actual_tkinter_handler = TkinterTextHandler(
|
|
text_widget=gui_log_widget,
|
|
level_colors=level_colors,
|
|
root_tk_instance_for_widget_update=root_tk_instance,
|
|
)
|
|
_actual_tkinter_handler.setFormatter(_base_formatter)
|
|
get_logger(__name__).info("Tkinter logging handler added.")
|
|
except Exception as e:
|
|
get_logger(__name__).error(f"Failed to create Tkinter handler: {e}")
|
|
_actual_tkinter_handler = None
|
|
|
|
|
|
def get_logger(name: str) -> logging.Logger:
|
|
"""Retrieves a logger instance by name."""
|
|
return logging.getLogger(name)
|
|
|
|
|
|
def shutdown_logging_system():
|
|
"""Processes remaining logs and cleanly shuts down all handlers."""
|
|
global _logging_system_active, _log_processor_after_id, _tk_root_instance_for_processing
|
|
|
|
if not _logging_system_active:
|
|
return
|
|
|
|
get_logger(__name__).info("Shutting down logging system...")
|
|
_logging_system_active = False
|
|
|
|
if _log_processor_after_id and _tk_root_instance_for_processing.winfo_exists():
|
|
try:
|
|
_tk_root_instance_for_processing.after_cancel(_log_processor_after_id)
|
|
except tk.TclError:
|
|
pass # Root window might be already destroyed
|
|
_log_processor_after_id = None
|
|
|
|
# Final processing of the queue
|
|
if _global_log_queue:
|
|
try:
|
|
while not _global_log_queue.empty():
|
|
record = _global_log_queue.get_nowait()
|
|
if _actual_console_handler: _actual_console_handler.handle(record)
|
|
if _actual_file_handler: _actual_file_handler.handle(record)
|
|
# Don't try to write to Tkinter widget during shutdown
|
|
except queue.Empty:
|
|
pass
|
|
|
|
logging.shutdown()
|
|
_tk_root_instance_for_processing = None |