210 Zeilen
8.0 KiB
Python
210 Zeilen
8.0 KiB
Python
import logging
|
|
import datetime
|
|
import os
|
|
from pathlib import Path
|
|
from typing import Optional, Any
|
|
import json
|
|
from threading import Lock
|
|
import traceback
|
|
import inspect
|
|
|
|
class ApplicationLogger:
|
|
_instance = None
|
|
_lock = 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 not hasattr(self, 'initialized'):
|
|
self.initialized = True
|
|
self.log_entries = []
|
|
self.max_entries = 50000 # Increased for more detailed logging
|
|
self.interaction_count = 0
|
|
|
|
# Create logs directory if it doesn't exist
|
|
self.log_dir = Path("logs")
|
|
self.log_dir.mkdir(exist_ok=True)
|
|
|
|
# Setup file logging
|
|
self.log_file = self.log_dir / f"cpm_{datetime.datetime.now().strftime('%Y%m%d_%H%M%S')}.log"
|
|
|
|
# Configure logger
|
|
self.logger = logging.getLogger('CPM')
|
|
self.logger.setLevel(logging.DEBUG)
|
|
|
|
# File handler
|
|
file_handler = logging.FileHandler(self.log_file, encoding='utf-8')
|
|
file_handler.setLevel(logging.DEBUG)
|
|
|
|
# Console handler
|
|
console_handler = logging.StreamHandler()
|
|
console_handler.setLevel(logging.INFO)
|
|
|
|
# Detailed Formatter
|
|
formatter = logging.Formatter(
|
|
'%(asctime)s - %(name)s - %(levelname)s - [%(filename)s:%(lineno)d] - %(message)s',
|
|
datefmt='%Y-%m-%d %H:%M:%S.%f'[:-3] # Include milliseconds
|
|
)
|
|
file_handler.setFormatter(formatter)
|
|
console_handler.setFormatter(formatter)
|
|
|
|
# Add handlers
|
|
self.logger.addHandler(file_handler)
|
|
self.logger.addHandler(console_handler)
|
|
|
|
# Custom handler to store in memory
|
|
self.memory_handler = MemoryHandler(self)
|
|
self.memory_handler.setFormatter(formatter)
|
|
self.logger.addHandler(self.memory_handler)
|
|
|
|
self.info("Claude Project Manager - Logger initialized")
|
|
|
|
def debug(self, message: str, extra: Optional[dict] = None):
|
|
self._log('DEBUG', message, extra)
|
|
|
|
def info(self, message: str, extra: Optional[dict] = None):
|
|
self._log('INFO', message, extra)
|
|
|
|
def warning(self, message: str, extra: Optional[dict] = None):
|
|
self._log('WARNING', message, extra)
|
|
|
|
def error(self, message: str, extra: Optional[dict] = None, exc_info=None):
|
|
self._log('ERROR', message, extra, exc_info)
|
|
|
|
def critical(self, message: str, extra: Optional[dict] = None, exc_info=None):
|
|
self._log('CRITICAL', message, extra, exc_info)
|
|
|
|
def _log(self, level: str, message: str, extra: Optional[dict] = None, exc_info=None):
|
|
# Get caller information
|
|
frame = inspect.currentframe()
|
|
if frame and frame.f_back and frame.f_back.f_back:
|
|
caller = frame.f_back.f_back
|
|
caller_info = f"{caller.f_code.co_filename}:{caller.f_lineno}"
|
|
else:
|
|
caller_info = "unknown"
|
|
|
|
log_method = getattr(self.logger, level.lower())
|
|
if extra:
|
|
message = f"{message} | Extra: {json.dumps(extra, default=str)}"
|
|
|
|
if exc_info:
|
|
log_method(message, exc_info=exc_info)
|
|
else:
|
|
log_method(message)
|
|
|
|
def add_entry(self, timestamp: str, level: str, message: str):
|
|
"""Add log entry to memory buffer"""
|
|
with self._lock:
|
|
self.log_entries.append({
|
|
'timestamp': timestamp,
|
|
'level': level,
|
|
'message': message
|
|
})
|
|
|
|
# Keep only the latest entries
|
|
if len(self.log_entries) > self.max_entries:
|
|
self.log_entries = self.log_entries[-self.max_entries:]
|
|
|
|
def get_log_content(self) -> str:
|
|
"""Get all log entries as formatted string"""
|
|
with self._lock:
|
|
lines = []
|
|
for entry in self.log_entries:
|
|
lines.append(f"{entry['timestamp']} - {entry['level']} - {entry['message']}")
|
|
return '\n'.join(lines)
|
|
|
|
def export_logs(self, filepath: str, include_system_info: bool = True):
|
|
"""Export logs to a file with optional system information"""
|
|
content = []
|
|
|
|
if include_system_info:
|
|
content.append("=" * 80)
|
|
content.append("CLAUDE PROJECT MANAGER - LOG EXPORT")
|
|
content.append("=" * 80)
|
|
content.append(f"Export Date: {datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
|
|
content.append(f"Log File: {self.log_file}")
|
|
content.append(f"Total Interactions: {self.interaction_count}")
|
|
content.append(f"Total Log Entries: {len(self.log_entries)}")
|
|
content.append("=" * 80)
|
|
content.append("")
|
|
|
|
content.append(self.get_log_content())
|
|
|
|
with open(filepath, 'w', encoding='utf-8') as f:
|
|
f.write('\n'.join(content) if isinstance(content, list) else content)
|
|
|
|
def clear_logs(self):
|
|
"""Clear in-memory log entries"""
|
|
with self._lock:
|
|
self.log_entries.clear()
|
|
self.info("Log entries cleared")
|
|
|
|
class MemoryHandler(logging.Handler):
|
|
"""Custom handler to store logs in memory"""
|
|
def __init__(self, logger_instance):
|
|
super().__init__()
|
|
self.logger_instance = logger_instance
|
|
|
|
def emit(self, record):
|
|
try:
|
|
msg = self.format(record)
|
|
# Extract timestamp, level and message
|
|
parts = msg.split(' - ', 3)
|
|
if len(parts) >= 4:
|
|
timestamp = parts[0]
|
|
level = parts[2]
|
|
message = parts[3]
|
|
self.logger_instance.add_entry(timestamp, level, message)
|
|
except Exception:
|
|
self.handleError(record)
|
|
|
|
def log_interaction(self, component: str, action: str, details: Optional[dict] = None):
|
|
"""Log UI interactions"""
|
|
self.interaction_count += 1
|
|
interaction_msg = f"UI_INTERACTION #{self.interaction_count} - Component: {component}, Action: {action}"
|
|
if details:
|
|
interaction_msg += f", Details: {json.dumps(details, default=str)}"
|
|
self.info(interaction_msg)
|
|
|
|
def log_git_operation(self, operation: str, status: str, details: Optional[dict] = None):
|
|
"""Log Git operations with details"""
|
|
git_msg = f"GIT_OPERATION - Operation: {operation}, Status: {status}"
|
|
if details:
|
|
git_msg += f", Details: {json.dumps(details, default=str)}"
|
|
self.info(git_msg)
|
|
|
|
def log_exception(self, exception: Exception, context: str = ""):
|
|
"""Log exceptions with full traceback"""
|
|
tb = traceback.format_exc()
|
|
error_msg = f"EXCEPTION in {context}: {type(exception).__name__}: {str(exception)}\nTraceback:\n{tb}"
|
|
self.error(error_msg)
|
|
|
|
def log_method_call(self, method_name: str, args: tuple = None, kwargs: dict = None):
|
|
"""Log method calls with arguments"""
|
|
call_msg = f"METHOD_CALL - {method_name}"
|
|
if args:
|
|
call_msg += f", Args: {args}"
|
|
if kwargs:
|
|
call_msg += f", Kwargs: {kwargs}"
|
|
self.debug(call_msg)
|
|
|
|
def log_state_change(self, component: str, old_state: Any, new_state: Any):
|
|
"""Log state changes in the application"""
|
|
state_msg = f"STATE_CHANGE - Component: {component}, Old: {old_state}, New: {new_state}"
|
|
self.info(state_msg)
|
|
|
|
def get_full_log_path(self) -> str:
|
|
"""Get the full path to the current log file"""
|
|
return str(self.log_file.absolute())
|
|
|
|
def get_all_log_files(self) -> list:
|
|
"""Get all log files in the log directory"""
|
|
return sorted([f for f in self.log_dir.glob('*.log')], reverse=True)
|
|
|
|
# Global logger instance
|
|
logger = ApplicationLogger() |