319 lines
12 KiB
Python
319 lines
12 KiB
Python
"""
|
|
Logging configuration for ComfyUI_PromptManager.
|
|
|
|
This module provides centralized logging configuration with support for:
|
|
- Multiple log levels (DEBUG, INFO, WARNING, ERROR, CRITICAL)
|
|
- File rotation to prevent logs from growing too large
|
|
- Console and file output
|
|
- Structured logging with timestamps and context
|
|
- Log viewer API integration
|
|
"""
|
|
|
|
import logging
|
|
import logging.handlers
|
|
import os
|
|
import json
|
|
import threading
|
|
from datetime import datetime
|
|
from typing import Dict, Any, Optional, List
|
|
from pathlib import Path
|
|
|
|
|
|
class PromptManagerLogger:
|
|
"""
|
|
Centralized logging system for PromptManager.
|
|
|
|
Features:
|
|
- Configurable log levels
|
|
- File rotation with size limits
|
|
- Structured JSON logging for API consumption
|
|
- Thread-safe operations
|
|
- Memory buffer for recent logs
|
|
"""
|
|
|
|
_instance = None
|
|
_lock = threading.Lock()
|
|
|
|
def __new__(cls):
|
|
if cls._instance is None:
|
|
with cls._lock:
|
|
if cls._instance is None:
|
|
cls._instance = super().__new__(cls)
|
|
return cls._instance
|
|
|
|
def __init__(self):
|
|
if hasattr(self, '_initialized'):
|
|
return
|
|
|
|
self._initialized = True
|
|
self.log_dir = Path(__file__).parent.parent / "logs"
|
|
self.log_dir.mkdir(exist_ok=True)
|
|
|
|
# Configuration
|
|
self.config = {
|
|
'level': 'INFO',
|
|
'max_file_size': 10 * 1024 * 1024, # 10MB
|
|
'backup_count': 5,
|
|
'console_logging': True,
|
|
'file_logging': True,
|
|
'buffer_size': 1000 # Keep last 1000 log entries in memory
|
|
}
|
|
|
|
# Memory buffer for recent logs (for web viewer)
|
|
self._log_buffer = []
|
|
self._buffer_lock = threading.Lock()
|
|
|
|
# Initialize loggers
|
|
self._setup_loggers()
|
|
|
|
def _setup_loggers(self):
|
|
"""Set up all loggers with appropriate handlers."""
|
|
# Main logger
|
|
self.logger = logging.getLogger('prompt_manager')
|
|
self.logger.setLevel(getattr(logging, self.config['level']))
|
|
|
|
# Clear existing handlers
|
|
for handler in self.logger.handlers[:]:
|
|
self.logger.removeHandler(handler)
|
|
|
|
# Custom formatter
|
|
formatter = logging.Formatter(
|
|
'%(asctime)s - %(name)s - %(levelname)s - %(filename)s:%(lineno)d - %(message)s',
|
|
datefmt='%Y-%m-%d %H:%M:%S'
|
|
)
|
|
|
|
# Console handler
|
|
if self.config['console_logging']:
|
|
console_handler = logging.StreamHandler()
|
|
console_handler.setFormatter(formatter)
|
|
self.logger.addHandler(console_handler)
|
|
|
|
# File handler with rotation and safe encoding for Windows
|
|
if self.config['file_logging']:
|
|
log_file = self.log_dir / "prompt_manager.log"
|
|
file_handler = logging.handlers.RotatingFileHandler(
|
|
log_file,
|
|
maxBytes=self.config['max_file_size'],
|
|
backupCount=self.config['backup_count'],
|
|
encoding='utf-8',
|
|
errors='replace' # Replace problematic characters instead of crashing
|
|
)
|
|
file_handler.setFormatter(formatter)
|
|
self.logger.addHandler(file_handler)
|
|
|
|
# Memory handler for web viewer
|
|
memory_handler = MemoryBufferHandler(self)
|
|
memory_handler.setFormatter(formatter)
|
|
self.logger.addHandler(memory_handler)
|
|
|
|
# Component-specific loggers
|
|
self._setup_component_loggers()
|
|
|
|
def _setup_component_loggers(self):
|
|
"""Set up loggers for specific components."""
|
|
components = [
|
|
'prompt_manager.database',
|
|
'prompt_manager.api',
|
|
'prompt_manager.image_monitor',
|
|
'prompt_manager.prompt_tracker',
|
|
'prompt_manager.web_ui'
|
|
]
|
|
|
|
for component in components:
|
|
comp_logger = logging.getLogger(component)
|
|
comp_logger.setLevel(getattr(logging, self.config['level']))
|
|
# Child loggers inherit handlers from parent
|
|
|
|
def get_logger(self, name: str = 'prompt_manager') -> logging.Logger:
|
|
"""Get a logger instance for a specific component."""
|
|
return logging.getLogger(name)
|
|
|
|
def add_to_buffer(self, record: logging.LogRecord, formatted_message: str):
|
|
"""Add a log entry to the memory buffer for web viewer."""
|
|
with self._buffer_lock:
|
|
log_entry = {
|
|
'timestamp': datetime.fromtimestamp(record.created).isoformat(),
|
|
'level': record.levelname,
|
|
'logger': record.name,
|
|
'message': record.getMessage(),
|
|
'formatted': formatted_message,
|
|
'module': record.module,
|
|
'filename': record.filename,
|
|
'lineno': record.lineno,
|
|
'thread': record.thread,
|
|
'thread_name': record.threadName if hasattr(record, 'threadName') else '',
|
|
'process': record.process
|
|
}
|
|
|
|
self._log_buffer.append(log_entry)
|
|
|
|
# Maintain buffer size limit
|
|
if len(self._log_buffer) > self.config['buffer_size']:
|
|
self._log_buffer = self._log_buffer[-self.config['buffer_size']:]
|
|
|
|
def get_recent_logs(self, limit: int = 100, level: Optional[str] = None) -> List[Dict[str, Any]]:
|
|
"""Get recent log entries from memory buffer."""
|
|
with self._buffer_lock:
|
|
logs = self._log_buffer[:]
|
|
|
|
# Filter by level if specified
|
|
if level:
|
|
level_num = getattr(logging, level.upper(), None)
|
|
if level_num:
|
|
logs = [log for log in logs if getattr(logging, log['level']) >= level_num]
|
|
|
|
# Return most recent first
|
|
return list(reversed(logs[-limit:]))
|
|
|
|
def get_log_files(self) -> List[Dict[str, Any]]:
|
|
"""Get information about available log files."""
|
|
log_files = []
|
|
|
|
for log_file in self.log_dir.glob("*.log*"):
|
|
try:
|
|
stat = log_file.stat()
|
|
log_files.append({
|
|
'filename': log_file.name,
|
|
'path': str(log_file),
|
|
'size': stat.st_size,
|
|
'modified': datetime.fromtimestamp(stat.st_mtime).isoformat(),
|
|
'is_main': log_file.name == 'prompt_manager.log'
|
|
})
|
|
except Exception as e:
|
|
self.logger.error(f"Error getting info for log file {log_file}: {e}")
|
|
|
|
# Sort by modification time, newest first
|
|
log_files.sort(key=lambda x: x['modified'], reverse=True)
|
|
return log_files
|
|
|
|
def read_log_file(self, filename: str, lines: int = 100) -> List[str]:
|
|
"""Read lines from a specific log file."""
|
|
log_file = self.log_dir / filename
|
|
|
|
if not log_file.exists() or not log_file.is_file():
|
|
raise FileNotFoundError(f"Log file not found: {filename}")
|
|
|
|
# Security check - ensure file is in log directory
|
|
if not str(log_file.resolve()).startswith(str(self.log_dir.resolve())):
|
|
raise ValueError("Invalid log file path")
|
|
|
|
try:
|
|
# Use UTF-8 encoding with error handling for Windows compatibility
|
|
with open(log_file, 'r', encoding='utf-8', errors='replace') as f:
|
|
all_lines = f.readlines()
|
|
return all_lines[-lines:] if lines > 0 else all_lines
|
|
except Exception as e:
|
|
self.logger.error(f"Error reading log file {filename}: {e}")
|
|
raise
|
|
|
|
def truncate_logs(self) -> Dict[str, Any]:
|
|
"""Truncate all log files."""
|
|
results = {
|
|
'truncated': [],
|
|
'errors': []
|
|
}
|
|
|
|
for log_file in self.log_dir.glob("prompt_manager.log*"):
|
|
try:
|
|
if log_file.name == 'prompt_manager.log':
|
|
# For main log file, just clear it with safe encoding
|
|
with open(log_file, 'w', encoding='utf-8', errors='replace') as f:
|
|
f.write("")
|
|
else:
|
|
# For rotated files, delete them
|
|
log_file.unlink()
|
|
|
|
results['truncated'].append(log_file.name)
|
|
self.logger.info(f"Truncated log file: {log_file.name}")
|
|
|
|
except Exception as e:
|
|
error_msg = f"Failed to truncate {log_file.name}: {str(e)}"
|
|
results['errors'].append(error_msg)
|
|
self.logger.error(error_msg)
|
|
|
|
# Clear memory buffer
|
|
with self._buffer_lock:
|
|
self._log_buffer.clear()
|
|
|
|
return results
|
|
|
|
def update_config(self, new_config: Dict[str, Any]):
|
|
"""Update logging configuration."""
|
|
self.config.update(new_config)
|
|
|
|
# Reconfigure loggers if level changed
|
|
if 'level' in new_config:
|
|
level = getattr(logging, new_config['level'].upper(), logging.INFO)
|
|
self.logger.setLevel(level)
|
|
|
|
# Update all component loggers
|
|
for logger_name in logging.Logger.manager.loggerDict:
|
|
if logger_name.startswith('prompt_manager'):
|
|
logger = logging.getLogger(logger_name)
|
|
logger.setLevel(level)
|
|
|
|
# Re-setup loggers if handlers changed
|
|
if any(key in new_config for key in ['console_logging', 'file_logging', 'max_file_size', 'backup_count']):
|
|
self._setup_loggers()
|
|
|
|
self.logger.info(f"Updated logging configuration: {new_config}")
|
|
|
|
def get_config(self) -> Dict[str, Any]:
|
|
"""Get current logging configuration."""
|
|
return self.config.copy()
|
|
|
|
def get_log_stats(self) -> Dict[str, Any]:
|
|
"""Get logging statistics."""
|
|
with self._buffer_lock:
|
|
buffer_count = len(self._log_buffer)
|
|
level_counts = {}
|
|
for entry in self._log_buffer:
|
|
level = entry['level']
|
|
level_counts[level] = level_counts.get(level, 0) + 1
|
|
|
|
log_files = self.get_log_files()
|
|
total_log_size = sum(f['size'] for f in log_files)
|
|
|
|
return {
|
|
'buffer_count': buffer_count,
|
|
'level_counts': level_counts,
|
|
'log_files_count': len(log_files),
|
|
'total_log_size': total_log_size,
|
|
'log_directory': str(self.log_dir),
|
|
'current_level': self.config['level']
|
|
}
|
|
|
|
|
|
class MemoryBufferHandler(logging.Handler):
|
|
"""Custom logging handler that stores logs in memory for web viewer."""
|
|
|
|
def __init__(self, logger_manager: PromptManagerLogger):
|
|
super().__init__()
|
|
self.logger_manager = logger_manager
|
|
|
|
def emit(self, record: logging.LogRecord):
|
|
"""Handle a log record by adding it to the memory buffer."""
|
|
try:
|
|
formatted_message = self.format(record)
|
|
self.logger_manager.add_to_buffer(record, formatted_message)
|
|
except Exception:
|
|
# Don't let logging errors break the application
|
|
pass
|
|
|
|
|
|
# Global logger instance
|
|
_logger_manager = None
|
|
|
|
def get_logger_manager() -> PromptManagerLogger:
|
|
"""Get the global logger manager instance."""
|
|
global _logger_manager
|
|
if _logger_manager is None:
|
|
_logger_manager = PromptManagerLogger()
|
|
return _logger_manager
|
|
|
|
def get_logger(name: str = 'prompt_manager') -> logging.Logger:
|
|
"""Convenience function to get a logger."""
|
|
return get_logger_manager().get_logger(name)
|
|
|
|
# Initialize logging on import
|
|
get_logger_manager() |