Spaces:
Paused
Paused
File size: 3,202 Bytes
3b9a6b5 2004c79 918bdb4 3b9a6b5 918bdb4 3b9a6b5 918bdb4 3b9a6b5 918bdb4 3b9a6b5 918bdb4 3b9a6b5 918bdb4 3b9a6b5 918bdb4 3b9a6b5 918bdb4 3b9a6b5 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 |
from datetime import datetime
from typing import List
import logging, threading
from utils.logging_config import setup_logging, get_logger, is_debug_enabled
# Initialize logging
setup_logging()
logger = get_logger(__name__)
class LogCapture:
"""Helper class to capture logs for streaming to UI"""
def __init__(self):
self.logs: List[str] = []
self.lock = threading.Lock()
def add_log(self, message: str) -> None:
"""Add a log message with timestamp"""
with self.lock:
timestamp = datetime.now().strftime("%H:%M:%S")
self.logs.append(f"[{timestamp}] {message}")
def get_logs(self) -> str:
"""Get all accumulated logs as a single string"""
with self.lock:
return "\n".join(self.logs)
def clear(self) -> None:
"""Clear all accumulated logs"""
with self.lock:
self.logs.clear()
class StreamingLogHandler(logging.Handler):
"""Custom log handler that captures logs for UI streaming"""
def __init__(self, log_capture: LogCapture):
super().__init__()
self.log_capture = log_capture
def emit(self, record: logging.LogRecord) -> None:
try:
msg = self.format(record)
self.log_capture.add_log(msg)
except Exception:
self.handleError(record)
class LoggingService:
"""Service for managing log streaming and capture for the UI"""
def __init__(self):
self.log_capture = LogCapture()
self._handler_added = False
def setup_log_streaming(self) -> None:
"""Set up log streaming to capture logs for UI"""
# Use the root logger which is configured by our centralized system
root_logger = logging.getLogger()
# Remove existing streaming handlers to avoid duplicates
for handler in root_logger.handlers[:]:
if isinstance(handler, StreamingLogHandler):
root_logger.removeHandler(handler)
# Add our streaming handler
stream_handler = StreamingLogHandler(self.log_capture)
# Respect the debug flag when setting the handler level
if is_debug_enabled():
stream_handler.setLevel(logging.DEBUG)
logger.debug("UI log streaming configured for DEBUG level")
else:
stream_handler.setLevel(logging.INFO)
logger.debug("UI log streaming configured for INFO level")
# Use a more detailed formatter for UI streaming
formatter = logging.Formatter("%(levelname)s - %(name)s - %(message)s")
stream_handler.setFormatter(formatter)
root_logger.addHandler(stream_handler)
self._handler_added = True
logger.debug("UI log streaming handler added to root logger")
def get_streaming_logs(self) -> str:
"""Get accumulated logs for streaming to UI"""
return self.log_capture.get_logs()
def clear_streaming_logs(self) -> None:
"""Clear accumulated logs"""
logger.debug("Clearing UI streaming logs")
self.log_capture.clear()
def is_setup(self) -> bool:
"""Check if log streaming is set up"""
return self._handler_added
|