#!/usr/bin/env python # -*- coding: utf-8 -*- import logging import os import sys from datetime import datetime from functools import wraps from flask import request, g, current_app from flask_login import current_user import traceback import json import time # ANSI Color Codes für farbige Terminal-Ausgabe class Colors: # Standard Colors RESET = '\033[0m' BOLD = '\033[1m' DIM = '\033[2m' # Foreground Colors BLACK = '\033[30m' RED = '\033[31m' GREEN = '\033[32m' YELLOW = '\033[33m' BLUE = '\033[34m' MAGENTA = '\033[35m' CYAN = '\033[36m' WHITE = '\033[37m' # Bright Colors BRIGHT_RED = '\033[91m' BRIGHT_GREEN = '\033[92m' BRIGHT_YELLOW = '\033[93m' BRIGHT_BLUE = '\033[94m' BRIGHT_MAGENTA = '\033[95m' BRIGHT_CYAN = '\033[96m' BRIGHT_WHITE = '\033[97m' # Background Colors BG_RED = '\033[41m' BG_GREEN = '\033[42m' BG_YELLOW = '\033[43m' BG_BLUE = '\033[44m' BG_MAGENTA = '\033[45m' BG_CYAN = '\033[46m' class LoggerConfig: """Konfiguration für das Logging-System""" LOG_DIR = 'logs' MAX_LOG_SIZE = 10 * 1024 * 1024 # 10MB BACKUP_COUNT = 5 LOG_FORMAT = '%(asctime)s | %(levelname)s | %(name)s | %(message)s' DATE_FORMAT = '%Y-%m-%d %H:%M:%S' class ColoredFormatter(logging.Formatter): """Custom Formatter für farbige Log-Ausgaben mit schönen Emojis""" LEVEL_COLORS = { 'DEBUG': Colors.BRIGHT_CYAN, 'INFO': Colors.BRIGHT_GREEN, 'WARNING': Colors.BRIGHT_YELLOW, 'ERROR': Colors.BRIGHT_RED, 'CRITICAL': Colors.BG_RED + Colors.BRIGHT_WHITE } COMPONENT_COLORS = { 'AUTH': Colors.BLUE, 'API': Colors.GREEN, 'DB': Colors.MAGENTA, 'SOCIAL': Colors.CYAN, 'SYSTEM': Colors.YELLOW, 'ERROR': Colors.RED, 'SECURITY': Colors.BRIGHT_MAGENTA, 'PERFORMANCE': Colors.BRIGHT_BLUE, 'ACTIVITY': Colors.BRIGHT_CYAN } # Erweiterte Emoji-Mappings für verschiedene Komponenten und Aktionen COMPONENT_EMOJIS = { 'AUTH': '🔐', 'API': '🌐', 'DB': '🗄️', 'SOCIAL': '👥', 'SYSTEM': '⚙️', 'ERROR': '💥', 'SECURITY': '🛡️', 'PERFORMANCE': '⚡', 'ACTIVITY': '🎯' } # Spezielle Emojis für verschiedene Log-Level LEVEL_EMOJIS = { 'DEBUG': '🔍', 'INFO': '✅', 'WARNING': '⚠️', 'ERROR': '❌', 'CRITICAL': '🚨' } # Action-spezifische Emojis ACTION_EMOJIS = { 'login': '🚪', 'logout': '🚪', 'register': '📝', 'like': '❤️', 'unlike': '💔', 'comment': '💬', 'share': '🔄', 'follow': '➕', 'unfollow': '➖', 'bookmark': '🔖', 'unbookmark': '📑', 'post_created': '📝', 'post_deleted': '🗑️', 'upload': '📤', 'download': '📥', 'search': '🔍', 'notification': '🔔', 'message': '💌', 'profile_update': '👤', 'settings': '⚙️', 'admin': '👑', 'backup': '💾', 'restore': '🔄', 'migration': '🚚', 'cache': '⚡', 'email': '📧', 'password_reset': '🔑', 'verification': '✅', 'ban': '🚫', 'unban': '✅', 'report': '🚩', 'moderate': '🛡️' } def _get_action_emoji(self, message: str) -> str: """Ermittelt das passende Emoji basierend auf der Nachricht""" message_lower = message.lower() for action, emoji in self.ACTION_EMOJIS.items(): if action in message_lower: return emoji return '📝' def format(self, record): # Zeitstempel mit schöner Formatierung timestamp = datetime.fromtimestamp(record.created).strftime('%H:%M:%S.%f')[:-3] colored_timestamp = f"{Colors.DIM}⏰ {timestamp}{Colors.RESET}" # Level mit Farbe und Emoji level_color = self.LEVEL_COLORS.get(record.levelname, Colors.WHITE) level_emoji = self.LEVEL_EMOJIS.get(record.levelname, '📝') colored_level = f"{level_color}{level_emoji} {record.levelname:<8}{Colors.RESET}" # Component mit Farbe und Emoji component = getattr(record, 'component', 'SYSTEM') component_color = self.COMPONENT_COLORS.get(component, Colors.WHITE) component_emoji = self.COMPONENT_EMOJIS.get(component, '📝') colored_component = f"{component_color}{component_emoji} [{component:<11}]{Colors.RESET}" # Message mit Action-spezifischem Emoji message = record.getMessage() action_emoji = self._get_action_emoji(message) # User-Info hinzufügen falls verfügbar user_info = "" if hasattr(record, 'user') and record.user: user_info = f" {Colors.BRIGHT_BLUE}👤 {record.user}{Colors.RESET}" # IP-Info hinzufügen falls verfügbar ip_info = "" if hasattr(record, 'ip') and record.ip: ip_info = f" {Colors.DIM}🌍 {record.ip}{Colors.RESET}" # Duration-Info hinzufügen falls verfügbar duration_info = "" if hasattr(record, 'duration') and record.duration: if record.duration > 1000: duration_color = Colors.BRIGHT_RED duration_emoji = "🐌" elif record.duration > 500: duration_color = Colors.BRIGHT_YELLOW duration_emoji = "⏱️" else: duration_color = Colors.BRIGHT_GREEN duration_emoji = "⚡" duration_info = f" {duration_color}{duration_emoji} {record.duration:.2f}ms{Colors.RESET}" # Separator für bessere Lesbarkeit separator = f"{Colors.DIM}│{Colors.RESET}" # Finale formatierte Nachricht mit schöner Struktur formatted_message = ( f"{colored_timestamp} {separator} " f"{colored_level} {separator} " f"{colored_component} {separator} " f"{action_emoji} {message}" f"{user_info}{ip_info}{duration_info}" ) return formatted_message class JSONFormatter(logging.Formatter): """JSON-Formatter für strukturierte Logs""" def format(self, record): log_entry = { 'timestamp': datetime.now().isoformat(), 'level': record.levelname, 'module': record.module, 'function': record.funcName, 'line': record.lineno, 'message': record.getMessage(), 'logger': record.name } # Benutzerinformationen hinzufügen - nur im Application Context try: from flask import has_app_context, g, current_user if has_app_context(): if hasattr(g, 'user_id'): log_entry['user_id'] = g.user_id elif current_user and hasattr(current_user, 'id') and current_user.is_authenticated: log_entry['user_id'] = current_user.id except (ImportError, RuntimeError): # Flask ist nicht verfügbar oder kein App-Context pass # Request-Informationen hinzufügen - nur im Request Context try: from flask import has_request_context, request if has_request_context() and request: log_entry['request'] = { 'method': getattr(request, 'method', None), 'path': getattr(request, 'path', None), 'remote_addr': getattr(request, 'remote_addr', None), 'user_agent': str(getattr(request, 'user_agent', '')) } except (ImportError, RuntimeError): # Flask ist nicht verfügbar oder kein Request-Context pass # Performance-Informationen hinzufügen - nur im Application Context try: from flask import has_app_context, g if has_app_context() and hasattr(g, 'start_time'): duration = (datetime.now() - g.start_time).total_seconds() * 1000 log_entry['duration_ms'] = round(duration, 2) except (ImportError, RuntimeError): # Flask ist nicht verfügbar oder kein App-Context pass # Exception-Informationen hinzufügen if record.exc_info: log_entry['exception'] = { 'type': record.exc_info[0].__name__, 'message': str(record.exc_info[1]), 'traceback': self.formatException(record.exc_info) } return json.dumps(log_entry) class SocialNetworkLogger: """Hauptklasse für das Social Network Logging""" def __init__(self, name: str = 'SysTades'): self.name = name self.logger = logging.getLogger(name) self.logger.setLevel(logging.DEBUG) # Log-Verzeichnis erstellen os.makedirs(LoggerConfig.LOG_DIR, exist_ok=True) # Handler nur einmal hinzufügen if not self.logger.handlers: self._setup_handlers() def _setup_handlers(self): """Setup für verschiedene Log-Handler""" # Console Handler mit Farben console_handler = logging.StreamHandler() console_handler.setLevel(logging.DEBUG) console_handler.setFormatter(ColoredFormatter()) self.logger.addHandler(console_handler) # File Handler für alle Logs from logging.handlers import RotatingFileHandler file_handler = RotatingFileHandler( os.path.join(LoggerConfig.LOG_DIR, 'app.log'), maxBytes=LoggerConfig.MAX_LOG_SIZE, backupCount=LoggerConfig.BACKUP_COUNT, encoding='utf-8' ) file_handler.setLevel(logging.DEBUG) file_formatter = logging.Formatter( '%(asctime)s | %(levelname)s | %(name)s | %(component)s | %(message)s', datefmt=LoggerConfig.DATE_FORMAT ) file_handler.setFormatter(file_formatter) self.logger.addHandler(file_handler) # Error Handler für nur Fehler error_handler = RotatingFileHandler( os.path.join(LoggerConfig.LOG_DIR, 'errors.log'), maxBytes=LoggerConfig.MAX_LOG_SIZE, backupCount=LoggerConfig.BACKUP_COUNT, encoding='utf-8' ) error_handler.setLevel(logging.ERROR) error_handler.setFormatter(file_formatter) self.logger.addHandler(error_handler) # API Handler für API-spezifische Logs api_handler = RotatingFileHandler( os.path.join(LoggerConfig.LOG_DIR, 'api.log'), maxBytes=LoggerConfig.MAX_LOG_SIZE, backupCount=LoggerConfig.BACKUP_COUNT, encoding='utf-8' ) api_handler.setLevel(logging.INFO) api_handler.addFilter(lambda record: hasattr(record, 'component') and record.component == 'API') api_handler.setFormatter(file_formatter) self.logger.addHandler(api_handler) def _log_with_context(self, level: str, message: str, component: str = 'SYSTEM', **kwargs): """Log mit erweiterten Kontext-Informationen""" extra = {'component': component} # User-Info hinzufügen if 'user' in kwargs: extra['user'] = kwargs['user'] # IP-Info hinzufügen if 'ip' in kwargs: extra['ip'] = kwargs['ip'] # Duration-Info hinzufügen if 'duration' in kwargs: extra['duration'] = kwargs['duration'] # Weitere Context-Daten extra.update({k: v for k, v in kwargs.items() if k not in ['user', 'ip', 'duration']}) getattr(self.logger, level.lower())(message, extra=extra) def debug(self, message: str, component: str = 'SYSTEM', **kwargs): """Debug-Level Logging mit erweiterten Infos""" self._log_with_context('DEBUG', message, component, **kwargs) def info(self, message: str, component: str = 'SYSTEM', **kwargs): """Info-Level Logging mit erweiterten Infos""" self._log_with_context('INFO', message, component, **kwargs) def warning(self, message: str, component: str = 'SYSTEM', **kwargs): """Warning-Level Logging mit erweiterten Infos""" self._log_with_context('WARNING', message, component, **kwargs) def error(self, message: str, component: str = 'ERROR', **kwargs): """Error-Level Logging mit erweiterten Infos""" self._log_with_context('ERROR', message, component, **kwargs) def critical(self, message: str, component: str = 'ERROR', **kwargs): """Critical-Level Logging mit erweiterten Infos""" self._log_with_context('CRITICAL', message, component, **kwargs) # Erweiterte spezielle Logging-Methoden für Social Network def auth_success(self, username: str, ip: str = None, method: str = 'password'): """Erfolgreiche Authentifizierung mit Details""" message = f"Benutzer '{username}' erfolgreich angemeldet" if method != 'password': message += f" (Methode: {method})" self.info(message, 'AUTH', user=username, ip=ip) def auth_failure(self, username: str, ip: str = None, reason: str = None, method: str = 'password'): """Fehlgeschlagene Authentifizierung mit Details""" message = f"Anmeldung fehlgeschlagen für '{username}'" if reason: message += f" - Grund: {reason}" if method != 'password': message += f" (Methode: {method})" self.warning(message, 'AUTH', user=username, ip=ip) def user_action(self, username: str, action: str, details: str = None, target: str = None): """Erweiterte Benutzer-Aktion mit mehr Details""" message = f"{username}: {action}" if target: message += f" → {target}" if details: message += f" ({details})" self.info(message, 'ACTIVITY', user=username) def api_request(self, method: str, endpoint: str, user: str = None, status: int = None, duration: float = None, size: int = None): """Erweiterte API Request Logging""" message = f"{method} {endpoint}" # Status-spezifische Emojis und Farben if status: if status >= 500: message = f"Server Error: {message}" component = 'ERROR' elif status >= 400: message = f"Client Error: {message}" component = 'API' elif status >= 300: message = f"Redirect: {message}" component = 'API' else: message = f"Success: {message}" component = 'API' else: component = 'API' # Zusätzliche Infos extras = {} if user: extras['user'] = user if duration: extras['duration'] = duration * 1000 # Convert to ms if size: message += f" ({self._format_bytes(size)})" if status and status >= 400: self.warning(message, component, **extras) else: self.info(message, component, **extras) def database_operation(self, operation: str, table: str, success: bool = True, details: str = None, affected_rows: int = None): """Erweiterte Datenbank-Operation Logging""" message = f"DB {operation.upper()} auf '{table}'" if affected_rows is not None: message += f" ({affected_rows} Zeilen)" if details: message += f" - {details}" if success: self.info(message, 'DB') else: self.error(message, 'DB') def security_event(self, event: str, user: str = None, ip: str = None, severity: str = 'warning', details: str = None): """Erweiterte Sicherheitsereignis Logging""" message = f"Security Event: {event}" if details: message += f" - {details}" extras = {} if user: extras['user'] = user if ip: extras['ip'] = ip if severity == 'critical': self.critical(message, 'SECURITY', **extras) elif severity == 'error': self.error(message, 'SECURITY', **extras) else: self.warning(message, 'SECURITY', **extras) def performance_metric(self, metric_name: str, value: float, unit: str = 'ms', threshold: dict = None): """Erweiterte Performance-Metrik Logging""" message = f"Performance: {metric_name} = {value}{unit}" # Threshold-basierte Bewertung if threshold and unit == 'ms': if value > threshold.get('critical', 2000): self.critical(message, 'PERFORMANCE', duration=value) elif value > threshold.get('warning', 1000): self.warning(message, 'PERFORMANCE', duration=value) else: self.info(message, 'PERFORMANCE', duration=value) else: self.info(message, 'PERFORMANCE') def social_interaction(self, user: str, action: str, target: str, target_type: str = 'post', target_user: str = None): """Erweiterte Social Media Interaktion Logging""" message = f"{user} {action} {target_type}" if target_user and target_user != user: message += f" von {target_user}" message += f" (ID: {target})" self.info(message, 'SOCIAL', user=user) def system_startup(self, version: str = None, environment: str = None, port: int = None): """Erweiterte System-Start Logging""" message = "🚀 SysTades Social Network gestartet" if version: message += f" (v{version})" if environment: message += f" in {environment} Umgebung" if port: message += f" auf Port {port}" self.info(message, 'SYSTEM') def system_shutdown(self, reason: str = None, uptime: float = None): """Erweiterte System-Shutdown Logging""" message = "🛑 SysTades Social Network beendet" if uptime: message += f" (Laufzeit: {self._format_duration(uptime)})" if reason: message += f" - Grund: {reason}" self.info(message, 'SYSTEM') def file_operation(self, operation: str, filename: str, success: bool = True, size: int = None, user: str = None): """Datei-Operation Logging""" message = f"File {operation.upper()}: {filename}" if size: message += f" ({self._format_bytes(size)})" extras = {} if user: extras['user'] = user if success: self.info(message, 'SYSTEM', **extras) else: self.error(message, 'SYSTEM', **extras) def cache_operation(self, operation: str, key: str, hit: bool = None, size: int = None): """Cache-Operation Logging""" message = f"Cache {operation.upper()}: {key}" if hit is not None: message += f" ({'HIT' if hit else 'MISS'})" if size: message += f" ({self._format_bytes(size)})" self.debug(message, 'SYSTEM') def email_sent(self, recipient: str, subject: str, success: bool = True, error: str = None): """E-Mail Versand Logging""" message = f"E-Mail an {recipient}: '{subject}'" if not success and error: message += f" - Fehler: {error}" if success: self.info(message, 'SYSTEM') else: self.error(message, 'SYSTEM') def _format_bytes(self, bytes_count: int) -> str: """Formatiert Byte-Anzahl in lesbare Form""" for unit in ['B', 'KB', 'MB', 'GB']: if bytes_count < 1024.0: return f"{bytes_count:.1f}{unit}" bytes_count /= 1024.0 return f"{bytes_count:.1f}TB" def _format_duration(self, seconds: float) -> str: """Formatiert Dauer in lesbare Form""" if seconds < 60: return f"{seconds:.1f}s" elif seconds < 3600: return f"{seconds/60:.1f}min" else: return f"{seconds/3600:.1f}h" def exception(self, exc: Exception, context: str = None, user: str = None): """Erweiterte Exception Logging mit mehr Details""" message = f"Exception: {type(exc).__name__}: {str(exc)}" if context: message = f"{context} - {message}" # Stack-Trace hinzufügen stack_trace = traceback.format_exc() message += f"\n{stack_trace}" extras = {} if user: extras['user'] = user self.error(message, 'ERROR', **extras) def log_execution_time(component: str = 'SYSTEM'): """Decorator für Ausführungszeit-Logging""" def decorator(func): @wraps(func) def wrapper(*args, **kwargs): logger = SocialNetworkLogger() start_time = time.time() try: result = func(*args, **kwargs) execution_time = (time.time() - start_time) * 1000 logger.performance_metric(f"{func.__name__} Ausführungszeit", execution_time, 'ms') return result except Exception as e: execution_time = (time.time() - start_time) * 1000 logger.exception(e, f"Fehler in {func.__name__} nach {execution_time:.2f}ms") raise return wrapper return decorator def log_api_call(func): """Decorator für API-Call Logging""" @wraps(func) def wrapper(*args, **kwargs): from flask import request, current_user logger = SocialNetworkLogger() start_time = time.time() # Request-Informationen sammeln method = request.method endpoint = request.endpoint or request.path user = current_user.username if hasattr(current_user, 'username') and current_user.is_authenticated else 'Anonymous' try: result = func(*args, **kwargs) duration = time.time() - start_time # Status-Code ermitteln status = getattr(result, 'status_code', 200) if hasattr(result, 'status_code') else 200 logger.api_request(method, endpoint, user, status, duration) return result except Exception as e: duration = time.time() - start_time logger.api_request(method, endpoint, user, 500, duration) logger.exception(e, f"API-Fehler in {endpoint}") raise return wrapper def performance_monitor(operation_name: str = None): """Erweiterte Decorator für Performance-Monitoring mit schönen Logs""" def decorator(func): @wraps(func) def wrapper(*args, **kwargs): logger = SocialNetworkLogger() start_time = time.time() op_name = operation_name or func.__name__ # User-Info ermitteln falls verfügbar user = None try: from flask import current_user if hasattr(current_user, 'username') and current_user.is_authenticated: user = current_user.username except: pass try: result = func(*args, **kwargs) duration = (time.time() - start_time) * 1000 # Performance-Kategorisierung mit Emojis if duration > 2000: logger.critical(f"Kritisch langsame Operation: {op_name}", 'PERFORMANCE', user=user, duration=duration) elif duration > 1000: logger.warning(f"Langsame Operation: {op_name}", 'PERFORMANCE', user=user, duration=duration) elif duration > 500: logger.info(f"Mäßige Operation: {op_name}", 'PERFORMANCE', user=user, duration=duration) else: logger.debug(f"Schnelle Operation: {op_name}", 'PERFORMANCE', user=user, duration=duration) return result except Exception as e: duration = (time.time() - start_time) * 1000 logger.error(f"Fehler in Operation: {op_name} nach {duration:.2f}ms", 'PERFORMANCE', user=user, duration=duration) logger.exception(e, f"Performance Monitor - {op_name}", user=user) raise return wrapper return decorator def log_user_activity(activity_name: str): """Erweiterte Decorator für User-Activity Logging mit Details""" def decorator(func): @wraps(func) def wrapper(*args, **kwargs): from flask import current_user, request logger = SocialNetworkLogger() start_time = time.time() # User und Request-Info sammeln username = 'Anonymous' ip = None user_agent = None try: if hasattr(current_user, 'username') and current_user.is_authenticated: username = current_user.username if request: ip = request.remote_addr user_agent = str(request.user_agent)[:100] # Begrenzen except: pass try: result = func(*args, **kwargs) duration = (time.time() - start_time) * 1000 # Erfolgreiche Aktivität loggen details = f"Erfolgreich in {duration:.2f}ms" if user_agent: details += f" (Browser: {user_agent.split('/')[0] if '/' in user_agent else user_agent})" logger.user_action(username, activity_name, details=details) return result except Exception as e: duration = (time.time() - start_time) * 1000 logger.error(f"Fehler in User-Activity '{activity_name}' für {username} nach {duration:.2f}ms: {str(e)}", 'ACTIVITY', user=username, ip=ip, duration=duration) logger.exception(e, f"User Activity - {activity_name}", user=username) raise return wrapper return decorator # Globale Logger-Instanz social_logger = SocialNetworkLogger() def get_logger(name: str = None) -> SocialNetworkLogger: """Factory-Funktion für Logger-Instanzen""" if name: return SocialNetworkLogger(name) return social_logger # Convenience-Funktionen für häufige Log-Operationen def log_user_login(username: str, ip: str = None, success: bool = True): """Shortcut für Login-Logging""" if success: social_logger.auth_success(username, ip) else: social_logger.auth_failure(username, ip) def log_user_action(username: str, action: str, details: str = None): """Shortcut für Benutzer-Aktionen""" social_logger.user_action(username, action, details) def log_social_action(user: str, action: str, target: str, target_type: str = 'post'): """Shortcut für Social Media Aktionen""" social_logger.social_interaction(user, action, target, target_type) def log_error(message: str, exception: Exception = None): """Shortcut für Error-Logging""" if exception: social_logger.exception(exception, message) else: social_logger.error(message) def log_performance(metric_name: str, value: float, unit: str = 'ms'): """Shortcut für Performance-Logging""" social_logger.performance_metric(metric_name, value, unit) # Setup-Funktion für initiale Konfiguration def setup_logging(app=None, log_level: str = 'INFO'): """Setup-Funktion für die Flask-App""" if app: # Flask App Logging konfigurieren app.logger.handlers.clear() app.logger.addHandler(social_logger.logger.handlers[0]) # Console Handler app.logger.setLevel(getattr(logging, log_level.upper())) # System-Start loggen social_logger.system_startup() return social_logger if __name__ == "__main__": # Test des Logging-Systems logger = SocialNetworkLogger() logger.info("🧪 Teste das Logging-System") logger.auth_success("testuser", "192.168.1.1") logger.user_action("testuser", "Post erstellt", "Neuer Gedanke geteilt") logger.social_interaction("user1", "like", "post_123") logger.api_request("GET", "/api/social/posts", "testuser", 200, 0.045) logger.database_operation("INSERT", "social_posts", True, "Neuer Post gespeichert") logger.performance_metric("Seitenladezeit", 1234.5) logger.warning("⚠️ Test-Warnung") logger.error("❌ Test-Fehler") logger.debug("🔍 Debug-Information") print(f"\n{Colors.BRIGHT_GREEN}✅ Logging-System erfolgreich getestet!{Colors.RESET}") print(f"{Colors.CYAN}📁 Logs gespeichert in: {LoggerConfig.LOG_DIR}/{Colors.RESET}")