import json import time from typing import Dict, Any, Optional, Tuple, List from openai import AsyncOpenAI, APIError, APITimeoutError import structlog from ..exceptions.base import ExternalServiceException, RateLimitException from ..services.ai_analytics import AiAnalyticsService logger = structlog.get_logger("openai_client") ANALYTICS_ENABLED_SERVICES = { "translation", "wine_pairing", "beer_pairing", "menu_extraction", "create_restaurant_offer", "generate_ai_image", } class UnifiedOpenAIClient: """Unified OpenAI client for all services""" def __init__(self): self._clients: Dict[str, AsyncOpenAI] = {} self._analytics_service: Optional[AiAnalyticsService] = None def set_analytics_service(self, analytics_service: Optional[AiAnalyticsService]) -> None: """Inject analytics service for best-effort usage logging.""" self._analytics_service = analytics_service def get_client(self, api_key: str) -> AsyncOpenAI: """Get or create OpenAI client for specific API key""" if api_key not in self._clients: self._clients[api_key] = AsyncOpenAI(api_key=api_key) return self._clients[api_key] async def json_completion( self, *, api_key: str, model: str, prompt: Optional[str] = None, messages: Optional[List[Dict[str, Any]]] = None, operation_id: str, service: str, max_tokens: Optional[int] = None, temperature: float = 0, timeout: Optional[float] = None, json_schema: Optional[Dict[str, Any]] = None, analytics_context: Optional[Dict[str, Any]] = None, skip_analytics: bool = False, ) -> Tuple[Dict[str, Any], Dict[str, Any]]: """ Make JSON completion request Returns: Tuple of (result_data, metrics) """ if not prompt and not messages: raise ValueError("Either 'prompt' or 'messages' must be provided.") client = self.get_client(api_key) logger.info( "openai_request", service=service, operation_id=operation_id, model=model ) started_at = time.perf_counter() try: kwargs: Dict[str, Any] = { "model": model, "temperature": temperature, "response_format": {"type": "json_object"} } if messages: kwargs["messages"] = messages else: kwargs["messages"] = [{"role": "user", "content": prompt}] if max_tokens: kwargs["max_tokens"] = max_tokens if json_schema: kwargs["response_format"] = { "type": "json_schema", "json_schema": json_schema } if timeout: kwargs["timeout"] = timeout response = await client.chat.completions.create(**kwargs) execution_time_ms = int((time.perf_counter() - started_at) * 1000) # Get raw response content raw_content = response.choices[0].message.content # Log raw response for debugging (truncated if too long) logger.debug( "openai_raw_response", service=service, operation_id=operation_id, content_length=len(raw_content) if raw_content else 0, content_preview=raw_content[:500] if raw_content else None, finish_reason=response.choices[0].finish_reason ) # Parse JSON response try: result = json.loads(raw_content) except json.JSONDecodeError as parse_error: # Log the problematic JSON for debugging logger.error( "openai_json_parse_failed", service=service, operation_id=operation_id, error=str(parse_error), content_length=len(raw_content) if raw_content else 0, finish_reason=response.choices[0].finish_reason, # Show context around the error position error_context=raw_content[max(0, parse_error.pos-100):parse_error.pos+100] if raw_content and hasattr(parse_error, 'pos') else None ) raise metrics = { "input_tokens": response.usage.prompt_tokens, "output_tokens": response.usage.completion_tokens, "total_tokens": response.usage.total_tokens, "model": response.model, "cost": self._calculate_cost( response.model, response.usage.prompt_tokens, response.usage.completion_tokens ) } logger.info( "openai_response", service=service, operation_id=operation_id, **metrics ) await self._log_analytics( response=response, parsed_result=result, raw_content=raw_content, prompt=prompt, messages=messages, operation_id=operation_id, service=service, max_tokens=max_tokens, temperature=temperature, json_schema=json_schema, execution_time=execution_time_ms, analytics_context=analytics_context, skip_analytics=skip_analytics, ) return result, metrics except APITimeoutError as e: logger.error( "openai_timeout_error", service=service, operation_id=operation_id, error=str(e) ) raise ExternalServiceException( "OpenAI", f"Request timed out: {str(e)}", {"error_type": "timeout"} ) except APIError as e: status_code = getattr(e, 'status_code', None) if status_code == 429: logger.warning( "openai_rate_limit", service=service, operation_id=operation_id ) raise RateLimitException() logger.error( "openai_api_error", service=service, operation_id=operation_id, status_code=status_code, error=str(e) ) raise ExternalServiceException( "OpenAI", str(e), {"status_code": status_code} if status_code else {} ) except json.JSONDecodeError as e: logger.error( "openai_json_decode_error", service=service, operation_id=operation_id, error=str(e) ) raise ExternalServiceException("OpenAI", f"Invalid JSON response: {e}") except Exception as e: logger.error( "openai_unexpected_error", service=service, operation_id=operation_id, error=str(e) ) raise ExternalServiceException("OpenAI", str(e)) async def close(self): """Close all OpenAI clients""" for client in self._clients.values(): await client.close() if self._analytics_service is not None: await self._analytics_service.close() async def _log_analytics( self, *, response: Any, parsed_result: Dict[str, Any], raw_content: Optional[str], prompt: Optional[str], messages: Optional[List[Dict[str, Any]]], operation_id: str, service: str, max_tokens: Optional[int], temperature: float, json_schema: Optional[Dict[str, Any]], execution_time: int, analytics_context: Optional[Dict[str, Any]], skip_analytics: bool = False, ) -> None: if skip_analytics: return if service not in ANALYTICS_ENABLED_SERVICES: return if self._analytics_service is None or not self._analytics_service.enabled: return usage = getattr(response, "usage", None) prompt_tokens_details = getattr(usage, "prompt_tokens_details", None) completion_tokens_details = getattr(usage, "completion_tokens_details", None) choices = getattr(response, "choices", None) or [] finish_reason = getattr(choices[0], "finish_reason", None) if choices else None context = analytics_context or {} try: await self._analytics_service.submit_log_request( city=self._string_or_default(context.get("city")), zavedenia_id=self._zavedenia_id_or_default(context.get("zavedenia_id")), service=self._string_or_default(service), model=self._string_or_default(response.model), tokens_input={ "prompt_tokens": self._int_or_default(getattr(usage, "prompt_tokens", None)), "cached_tokens": self._int_or_default(getattr(prompt_tokens_details, "cached_tokens", None)), "total_tokens": self._int_or_default(getattr(usage, "prompt_tokens", None)), }, tokens_output={ "completion_tokens": self._int_or_default(getattr(usage, "completion_tokens", None)), "reasoning_tokens": self._int_or_default(getattr(completion_tokens_details, "reasoning_tokens", None)), "total_tokens": self._int_or_default(getattr(usage, "completion_tokens", None)), }, data_input=self._build_analytics_input( context=context, max_tokens=max_tokens, prompt=prompt, messages=messages, ), data_output=self._build_analytics_output( parsed_result=parsed_result, raw_content=raw_content, finish_reason=finish_reason, ), execution_time=self._int_or_default(execution_time), ) except Exception as exc: logger.warning( "ai_analytics_failed", service=service, operation_id=operation_id, error=str(exc), ) async def log_manual_analytics( self, *, city: Optional[str], zavedenia_id: Optional[int], service: str, model: str, tokens_input: Dict[str, Any], tokens_output: Dict[str, Any], data_input: Dict[str, Any], data_output: Dict[str, Any], execution_time: int, ) -> None: if self._analytics_service is None or not self._analytics_service.enabled: return await self._analytics_service.submit_log_request( city=city, zavedenia_id=zavedenia_id, service=service, model=model, tokens_input=tokens_input, tokens_output=tokens_output, data_input=data_input, data_output=data_output, execution_time=execution_time, ) def _build_analytics_input( self, *, context: Dict[str, Any], max_tokens: Optional[int], prompt: Optional[str], messages: Optional[List[Dict[str, Any]]], ) -> Dict[str, Any]: extra_data = context.get("data_input", {}) request_body = context.get("request_body") if request_body is None: if extra_data: request_body = extra_data elif messages: request_body = {"messages": messages} elif prompt is not None: request_body = {"prompt": prompt} return { "type": "chat_completion", "user_message": self._json_string_or_default(request_body), "language": self._string_or_default(extra_data.get("language")), "max_tokens": self._int_or_default(max_tokens), } def _build_analytics_output( self, *, parsed_result: Dict[str, Any], raw_content: Optional[str], finish_reason: Optional[str], ) -> Dict[str, Any]: return { "type": "chat_completion_response", "content": self._json_string_or_default(parsed_result or raw_content), "finish_reason": self._string_or_default(finish_reason), } @staticmethod def _truncate_text(value: Optional[str], limit: int = 4000) -> Optional[str]: if value is None or len(value) <= limit: return value return f"{value[:limit]}...[truncated]" @staticmethod def _string_or_default(value: Any) -> str: if value is None: return "missing content" string_value = str(value).strip() return string_value or "missing content" @classmethod def _json_string_or_default(cls, value: Any) -> str: if value is None: return "missing content" if isinstance(value, str): stripped = value.strip() return stripped or "missing content" try: return json.dumps(value, ensure_ascii=False, separators=(",", ":")) except (TypeError, ValueError): return cls._string_or_default(value) @staticmethod def _int_or_default(value: Any) -> int: if value is None: return 0 try: parsed = int(value) return parsed if parsed >= 0 else 0 except (TypeError, ValueError): return 0 @staticmethod def _zavedenia_id_or_default(value: Any) -> int: if value is None: return 2 try: parsed = int(value) return parsed if parsed > 1 else 2 except (TypeError, ValueError): return 2 @staticmethod def _calculate_cost(model: str, input_tokens: int, output_tokens: int) -> float: """Calculate cost based on model pricing (per 1M tokens)""" pricing = { "gpt-4o": {"input": 2.50, "output": 10.00}, "gpt-4o-mini": {"input": 0.150, "output": 0.600}, "gpt-4-turbo": {"input": 10.00, "output": 30.00}, "gpt-3.5-turbo": {"input": 0.50, "output": 1.50}, } model_key = next((k for k in pricing.keys() if k in model), "gpt-4o-mini") rates = pricing[model_key] cost = ( (input_tokens / 1_000_000) * rates["input"] + (output_tokens / 1_000_000) * rates["output"] ) return round(cost, 6) # Global instance openai_client = UnifiedOpenAIClient()