68 lines
2.7 KiB
Python
68 lines
2.7 KiB
Python
# SPDX-FileCopyrightText: 2026 Art Sea LLC
|
|
#
|
|
# SPDX-License-Identifier: AGPL-3.0-only
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
|
|
import time
|
|
|
|
from band_aider.ui.dto.log_message import LogMessageDTO
|
|
from band_aider.ui.dto.logger_namespaces import LoggerNamespacesDTO
|
|
from band_aider.ui.status_publisher import StatusPublisher
|
|
|
|
|
|
class QueueLoggingHandler(logging.Handler):
|
|
"""QueueLoggingHandler adapts LogRecord instances to LogMessageDTO snapshots."""
|
|
|
|
def __init__(self, publisher: StatusPublisher) -> None:
|
|
super().__init__()
|
|
self.publisher = publisher
|
|
self._namespaces: set[str] = {"ALL"}
|
|
self._last_namespace_publish = 0.0
|
|
|
|
def emit(self, record: logging.LogRecord) -> None:
|
|
try:
|
|
# Track namespaces from seen records
|
|
name = record.name or ""
|
|
if name:
|
|
# include full name and its dot-prefixes as namespaces
|
|
parts = name.split(".")
|
|
for i in range(1, len(parts) + 1):
|
|
self._namespaces.add(".".join(parts[:i]))
|
|
|
|
for line in record.getMessage().splitlines() or [""]:
|
|
if not line.strip():
|
|
continue
|
|
self.publisher.publish(
|
|
LogMessageDTO(
|
|
created=record.created,
|
|
level=record.levelname,
|
|
logger_name=name,
|
|
message=line,
|
|
)
|
|
)
|
|
|
|
# Periodically publish the currently known namespaces by also
|
|
# interrogating the logging subsystem for registered loggers.
|
|
now = time.monotonic()
|
|
if now - self._last_namespace_publish >= 1.0:
|
|
self._last_namespace_publish = now
|
|
try:
|
|
logger_dict = getattr(logging.root.manager, "loggerDict", {})
|
|
for logger_name in list(logger_dict.keys()):
|
|
if not isinstance(logger_name, str):
|
|
continue
|
|
if logger_name:
|
|
parts = logger_name.split(".")
|
|
for i in range(1, len(parts) + 1):
|
|
self._namespaces.add(".".join(parts[:i]))
|
|
except Exception:
|
|
# do not break logging on unexpected structures
|
|
pass
|
|
# Remove the synthetic ALL from published list; UI adds it
|
|
namespaces = tuple(sorted(n for n in self._namespaces if n != "ALL"))
|
|
self.publisher.publish(LoggerNamespacesDTO(namespaces=namespaces))
|
|
except Exception:
|
|
self.handleError(record)
|