792 lines
29 KiB
Python
792 lines
29 KiB
Python
#!/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}") |