""" Request logging service for card generation requests """ import logging import json from datetime import datetime from typing import Optional, Dict, Any, List from pathlib import Path # Create logs directory if it doesn't exist logs_dir = Path("logs") logs_dir.mkdir(exist_ok=True) # Create a specific logger for request events request_logger = logging.getLogger("request_events") request_logger.setLevel(logging.INFO) # Create file handler for request logs request_log_file = logs_dir / "request_events.log" file_handler = logging.FileHandler(request_log_file) file_handler.setLevel(logging.INFO) # Create console handler for important events console_handler = logging.StreamHandler() console_handler.setLevel(logging.INFO) # Create formatter for structured logging formatter = logging.Formatter( '%(asctime)s - %(name)s - %(levelname)s - %(message)s', datefmt='%Y-%m-%d %H:%M:%S' ) file_handler.setFormatter(formatter) console_handler.setFormatter(formatter) # Add handlers to logger request_logger.addHandler(file_handler) request_logger.addHandler(console_handler) # Prevent propagation to root logger to avoid duplicate logs request_logger.propagate = False class RequestEventLogger: """Request event logging with structured file output""" @staticmethod def log_card_generation_request( user_id: Optional[str], username: str, client_ip: str, user_agent: str, terms: List[str], card_date: str, lang: str, card_design_id: Optional[int] = None, symbol_ids: Optional[List[int]] = None, request_id: Optional[str] = None ): """Log incoming card generation request""" event_data = { "event": "card_generation_request", "request_id": request_id, "user_id": user_id, "username": username, "client_ip": client_ip, "user_agent": user_agent, "terms": terms, "card_date": card_date, "lang": lang, "card_design_id": card_design_id, "symbol_ids": symbol_ids, "timestamp": datetime.utcnow().isoformat() } request_logger.info(f"CARD_REQUEST: {json.dumps(event_data)}") @staticmethod def log_generator_prompt( request_id: Optional[str], prompt: str, terms: List[str], card_date: str, constellation: str, lang: str ): """Log the prompt sent to the generator""" event_data = { "event": "generator_prompt", "request_id": request_id, "prompt": prompt, "terms": terms, "card_date": card_date, "constellation": constellation, "lang": lang, "timestamp": datetime.utcnow().isoformat() } request_logger.info(f"GENERATOR_PROMPT: {json.dumps(event_data)}") @staticmethod def log_llm_request( request_id: Optional[str], prompt: str, generation_params: Dict[str, Any], model_info: Dict[str, Any] ): """Log the request sent to the LLM via HF API""" event_data = { "event": "llm_request", "request_id": request_id, "prompt": prompt, "generation_params": generation_params, "model_info": model_info, "timestamp": datetime.utcnow().isoformat() } request_logger.info(f"LLM_REQUEST: {json.dumps(event_data)}") @staticmethod def log_llm_response( request_id: Optional[str], generated_text: str, attempt: int, constraints_met: bool, terms: List[str] ): """Log the response from the LLM""" event_data = { "event": "llm_response", "request_id": request_id, "generated_text": generated_text, "attempt": attempt, "constraints_met": constraints_met, "terms": terms, "timestamp": datetime.utcnow().isoformat() } request_logger.info(f"LLM_RESPONSE: {json.dumps(event_data)}") @staticmethod def log_card_renderer_params( request_id: Optional[str], card_design_id: int, symbol_ids: List[int], text: str, base_images_path: str, symbols_images_path: str, font_path: str, output_path: str ): """Log parameters sent to card renderer""" event_data = { "event": "card_renderer_params", "request_id": request_id, "card_design_id": card_design_id, "symbol_ids": symbol_ids, "text": text, "base_images_path": str(base_images_path), "symbols_images_path": str(symbols_images_path), "font_path": str(font_path), "output_path": str(output_path), "timestamp": datetime.utcnow().isoformat() } request_logger.info(f"CARD_RENDERER_PARAMS: {json.dumps(event_data)}") @staticmethod def log_card_generation_result( request_id: Optional[str], success: bool, card_id: Optional[str] = None, card_file_id: Optional[str] = None, qr_code_file_id: Optional[str] = None, qr_content_url: Optional[str] = None, error_message: Optional[str] = None ): """Log the final result of card generation""" event_data = { "event": "card_generation_result", "request_id": request_id, "success": success, "card_id": card_id, "card_file_id": card_file_id, "qr_code_file_id": qr_code_file_id, "qr_content_url": qr_content_url, "error_message": error_message, "timestamp": datetime.utcnow().isoformat() } if success: request_logger.info(f"CARD_GENERATION_SUCCESS: {json.dumps(event_data)}") else: request_logger.error(f"CARD_GENERATION_FAILED: {json.dumps(event_data)}")