import logging import json import uuid from datetime import datetime, timezone from flask import request, g, has_request_context from typing import Dict, Any class StructuredFormatter(logging.Formatter): """Formateur de logs structurés en JSON.""" def format(self, record): log_data = { 'timestamp': datetime.now(timezone.utc).isoformat(), 'level': record.levelname, 'logger': record.name, 'message': record.getMessage(), 'module': record.module, 'function': record.funcName, 'line': record.lineno } # Ajouter le contexte de la requête si disponible if has_request_context() and request: log_data['request'] = { 'method': request.method, 'url': request.url, 'remote_addr': request.remote_addr, 'user_agent': request.headers.get('User-Agent', '') } # Ajouter l'ID de corrélation si disponible if hasattr(g, 'correlation_id'): log_data['correlation_id'] = g.correlation_id # Ajouter les données d'exception si présentes if record.exc_info: log_data['exception'] = { 'type': record.exc_info[0].__name__, 'message': str(record.exc_info[1]), 'traceback': self.formatException(record.exc_info) } # Ajouter les données personnalisées if hasattr(record, 'extra_data'): log_data['extra'] = record.extra_data return json.dumps(log_data, ensure_ascii=False) def setup_logging(app): """Configure le logging structuré.""" # Configuration du formateur formatter = StructuredFormatter() # Créer le dossier logs si nécessaire import os if not os.path.exists('logs'): os.makedirs('logs') # Handler pour fichier file_handler = logging.FileHandler('logs/notytex.log') file_handler.setFormatter(formatter) file_handler.setLevel(logging.INFO) # Handler pour console (développement) if app.debug: console_handler = logging.StreamHandler() console_formatter = logging.Formatter( '%(asctime)s - %(name)s - %(levelname)s - %(message)s' ) console_handler.setFormatter(console_formatter) console_handler.setLevel(logging.DEBUG) app.logger.addHandler(console_handler) app.logger.addHandler(file_handler) app.logger.setLevel(logging.INFO) # Middleware pour générer un ID de corrélation pour chaque requête @app.before_request def before_request(): g.correlation_id = str(uuid.uuid4()) app.logger.info(f"Début de requête {request.method} {request.url}") @app.after_request def after_request(response): app.logger.info(f"Fin de requête - Status: {response.status_code}") return response def log_business_event(event_type: str, details: Dict[str, Any]): """Log un événement métier.""" logger = logging.getLogger('notytex.business') extra_data = { 'event_type': event_type, 'details': details } logger.info(f"Événement métier : {event_type}", extra={'extra_data': extra_data})