""" Professional Logging Module Provides consistent, structured logging across the application with: - Timing measurements - Request tracking - Processing stage markers - Performance metrics """ import time import os from datetime import datetime from functools import wraps from typing import Optional, Callable, Any DEBUG_MODE = os.getenv('DEBUG_MODE', '').lower() == 'true' # ANSI color codes for console output class Colors: RESET = '\033[0m' BOLD = '\033[1m' DIM = '\033[2m' # Foreground CYAN = '\033[96m' GREEN = '\033[92m' YELLOW = '\033[93m' RED = '\033[91m' BLUE = '\033[94m' WHITE = '\033[97m' # Background BG_RED = '\033[41m' def _get_timestamp() -> str: """Get formatted timestamp for logs.""" return datetime.now().strftime('%Y-%m-%d %H:%M:%S.%f')[:-3] def _format_duration(seconds: float) -> str: """Format duration in human-readable form.""" if seconds < 0.001: return f"{seconds*1000000:.1f}µs" elif seconds < 1: return f"{seconds*1000:.1f}ms" elif seconds < 60: return f"{seconds:.1f}s" else: return f"{seconds/60:.1f}m" def _format_size(bytes_: int) -> str: """Format file size in human-readable form.""" for unit in ['B', 'KB', 'MB', 'GB']: if bytes_ < 1024: return f"{bytes_:.1f}{unit}" bytes_ /= 1024 return f"{bytes_:.1f}TB" # ==================== REQUEST CONTEXT ==================== _request_context = {} def set_request_id(request_id: str): """Set the current request ID for logging context.""" _request_context['request_id'] = request_id _request_context['start_time'] = time.time() def get_request_id() -> str: """Get the current request ID.""" return _request_context.get('request_id', 'N/A') def get_request_elapsed() -> float: """Get elapsed time since request started.""" start = _request_context.get('start_time') if not start: return 0 return time.time() - start def clear_request_context(): """Clear the request context.""" _request_context.clear() # ==================== LOGGING FUNCTIONS ==================== def log_section(title: str, length: int = 70): """Log a section header.""" timestamp = _get_timestamp() filler = "=" * (length - len(title) - len(timestamp) - 3) print(f"{Colors.CYAN}{Colors.BOLD}═ {title} {filler} {timestamp}{Colors.RESET}") def log_start(title: str, request_id: Optional[str] = None): """Log the start of a processing stage.""" timestamp = _get_timestamp() rid = request_id or get_request_id() print(f"{Colors.BLUE}[{timestamp}] [{rid}] ▶ {title}{Colors.RESET}") def log_info(message: str, detail: str = None, indent: int = 2): """Log informational message.""" timestamp = _get_timestamp() rid = get_request_id() indent_str = " " * indent if detail: print(f"{Colors.WHITE}[{timestamp}] [{rid}] {indent_str}ℹ {message}{Colors.RESET}") print(f"{Colors.DIM}[{timestamp}] [{rid}] {indent_str} └─ {detail}{Colors.RESET}") else: print(f"{Colors.WHITE}[{timestamp}] [{rid}] {indent_str}ℹ {message}{Colors.RESET}") def log_success(message: str, detail: str = None, timing: float = None): """Log successful completion.""" timestamp = _get_timestamp() rid = get_request_id() if timing: duration = _format_duration(timing) detail = f"{detail} ({duration})" if detail else f"took {duration}" if detail: print(f"{Colors.GREEN}[{timestamp}] [{rid}] ✓ {message}{Colors.RESET}") print(f"{Colors.DIM}[{timestamp}] [{rid}] └─ {detail}{Colors.RESET}") else: print(f"{Colors.GREEN}[{timestamp}] [{rid}] ✓ {message}{Colors.RESET}") def log_warning(message: str, detail: str = None): """Log warning message.""" timestamp = _get_timestamp() rid = get_request_id() if detail: print(f"{Colors.YELLOW}[{timestamp}] [{rid}] ⚠ {message}{Colors.RESET}") print(f"{Colors.DIM}[{timestamp}] [{rid}] └─ {detail}{Colors.RESET}") else: print(f"{Colors.YELLOW}[{timestamp}] [{rid}] ⚠ {message}{Colors.RESET}") def log_error(message: str, detail: str = None, exc: Exception = None): """Log error message.""" timestamp = _get_timestamp() rid = get_request_id() error_detail = detail or (str(exc) if exc else None) if error_detail: print(f"{Colors.RED}[{timestamp}] [{rid}] ✗ {message}{Colors.RESET}") print(f"{Colors.DIM}[{timestamp}] [{rid}] └─ {error_detail}{Colors.RESET}") else: print(f"{Colors.RED}[{timestamp}] [{rid}] ✗ {message}{Colors.RESET}") def log_metric(label: str, value: Any, unit: str = ""): """Log a performance metric.""" timestamp = _get_timestamp() rid = get_request_id() # Format value if isinstance(value, (int, float)): if unit == "bytes": formatted_value = _format_size(int(value)) elif unit == "seconds": formatted_value = _format_duration(value) else: formatted_value = f"{value}{unit}" if unit else str(value) else: formatted_value = str(value) print(f"{Colors.YELLOW}[{timestamp}] [{rid}] 📊 {label}: {formatted_value}{Colors.RESET}") def log_summary(title: str, metrics: dict): """Log a summary of processing with metrics.""" timestamp = _get_timestamp() rid = get_request_id() elapsed = get_request_elapsed() print(f"{Colors.CYAN}{Colors.BOLD}[{timestamp}] [{rid}] ═ {title} completed in {_format_duration(elapsed)}{Colors.RESET}") for key, value in metrics.items(): if isinstance(value, float) and value < 10: # Likely a duration formatted = _format_duration(value) elif isinstance(value, int) and value > 1000: # Likely a size formatted = _format_size(value) else: formatted = str(value) print(f"{Colors.BLUE}[{timestamp}] [{rid}] • {key}: {formatted}{Colors.RESET}") # ==================== DECORATORS ==================== def measure_time(name: str) -> Callable: """Decorator to measure function execution time.""" def decorator(func: Callable) -> Callable: @wraps(func) def wrapper(*args, **kwargs) -> Any: start_time = time.time() log_start(f"{name}...") try: result = func(*args, **kwargs) elapsed = time.time() - start_time log_success(f"{name}", timing=elapsed) return result except Exception as e: elapsed = time.time() - start_time log_error(f"{name} failed", detail=str(e), exc=e) raise return wrapper return decorator # ==================== DEBUG LOGGING ==================== def debug_log(message: str): """Log only if DEBUG_MODE is enabled.""" if DEBUG_MODE: timestamp = _get_timestamp() rid = get_request_id() print(f"{Colors.DIM}[{timestamp}] [{rid}] 🔧 {message}{Colors.RESET}") def debug_info(label: str, value: Any): """Log debug information (key-value pair).""" if DEBUG_MODE: timestamp = _get_timestamp() rid = get_request_id() print(f"{Colors.DIM}[{timestamp}] [{rid}] {label}: {value}{Colors.RESET}")