import base64 import time from pathlib import Path from typing import Any, Dict, List, Optional import structlog from openai import APIError, APITimeoutError from ...shared.clients.openai_client import openai_client from ...shared.exceptions.base import ExternalServiceException, RateLimitException from .config import GenerateAiImageServiceConfig from .models.requests import GenerateAiImageRequest logger = structlog.get_logger("generate_ai_image.service") _MOCK_IMAGE_PATH = Path("menues/restaurant123/offer_9a8e94e6e2524de2abd34e584b315f79.png") class GenerateAiImageService: def __init__(self, config: GenerateAiImageServiceConfig): self.config = config async def generate_ai_image(self, request: GenerateAiImageRequest) -> Dict[str, Any]: if request.mock_data: logger.info("generate_ai_image_mock_mode") image_base64 = base64.b64encode(_MOCK_IMAGE_PATH.read_bytes()).decode("ascii") return { "image_base64": image_base64, "model": "mock", "input_tokens": 0, "cached_tokens": 0, "output_tokens": 0, "reasoning_tokens": 0, "total_tokens": 0, "price_usd": 0.0, "response_id": None, } selected_model = request.selected_model or self.config.generate_ai_image_model image_size = request.image_size or self.config.image_size image_quality = request.image_quality or self.config.image_quality input_payload = self._build_input_payload(request) tools = [{"type": "image_generation", "size": image_size, "quality": image_quality}] logger.info( "generate_ai_image_started", model=selected_model, image_size=image_size, image_quality=image_quality, language=request.language, has_content=bool((request.content or "").strip()), logo_url_present=bool(request.logo_image_url), reference_image_count=len(request.reference_image_urls), ) started_at = time.perf_counter() client = openai_client.get_client(self.config.openai_api_key) try: response = await client.responses.create( model=selected_model, input=input_payload, tools=tools, tool_choice={"type": "image_generation"}, timeout=self.config.timeout, ) except APITimeoutError as exc: logger.error("generate_ai_image_timeout", error=str(exc), model=selected_model) raise ExternalServiceException("OpenAI", f"Responses API image generation timed out: {exc}") except APIError as exc: status_code = getattr(exc, "status_code", None) logger.error("generate_ai_image_api_error", error=str(exc), model=selected_model, status_code=status_code) if status_code == 429: raise RateLimitException() raise ExternalServiceException("OpenAI", str(exc), {"status_code": status_code} if status_code else {}) execution_time_ms = int((time.perf_counter() - started_at) * 1000) image_base64 = self._extract_image_base64(response) usage_metrics = self._extract_usage_metrics(response) price_usd = self._calculate_price_usd( selected_model, usage_metrics["input_tokens"], usage_metrics["output_tokens"], ) await self._log_analytics( request=request, response=response, selected_model=selected_model, image_size=image_size, image_quality=image_quality, input_payload=input_payload, tools=tools, usage_metrics=usage_metrics, execution_time_ms=execution_time_ms, ) logger.info( "generate_ai_image_completed", model=selected_model, input_tokens=usage_metrics["input_tokens"], output_tokens=usage_metrics["output_tokens"], total_tokens=usage_metrics["total_tokens"], price_usd=price_usd, response_id=getattr(response, "id", None), ) return { "image_base64": image_base64, "model": selected_model, "price_usd": price_usd, "response_id": getattr(response, "id", None), **usage_metrics, } @staticmethod def _build_input_payload(request: GenerateAiImageRequest) -> List[Dict[str, Any]]: system_prompt = GenerateAiImageService._build_system_prompt(request) user_prompt = GenerateAiImageService._build_user_prompt(request) user_content: List[Dict[str, Any]] = [ {"type": "input_text", "text": user_prompt}, {"type": "input_text", "text": "Input image 1 is the restaurant logo. Use it as the brand mark."}, {"type": "input_image", "image_url": request.logo_image_url}, ] for index, url in enumerate(request.reference_image_urls, start=2): user_content.extend( [ { "type": "input_text", "text": ( f"Input image {index} is a required restaurant reference image. " "Use this image visibly as primary guidance for the generated poster's subject, " "setting, style, colors, lighting, materials, and composition." ), }, {"type": "input_image", "image_url": url}, ] ) return [ {"role": "system", "content": [{"type": "input_text", "text": system_prompt}]}, {"role": "user", "content": user_content}, ] @staticmethod def _build_system_prompt(request: GenerateAiImageRequest) -> str: system_prompt = GenerateAiImageService._relax_reference_prompt_restrictions(request.system_prompt) ultrarealistic_rules = ( "\n\nREALISM REQUIREMENTS:\n" "- The generated poster must be ultra realistic and photo-realistic.\n" "- Render food, restaurant interiors, lighting, materials, and table details as believable real-world photography.\n" "- Use natural shadows, realistic reflections, authentic textures, believable depth of field, and cinematic lighting.\n" "- Avoid illustration, painting, cartoon, CGI-looking, or obviously synthetic visual styles unless the request explicitly asks for them." ) system_prompt = f"{system_prompt}{ultrarealistic_rules}" language = (request.language or "").strip() if not language: return GenerateAiImageService._append_reference_image_rules(system_prompt, request) language_rules = ( "\n\nLANGUAGE AND TEXT QUALITY REQUIREMENTS:\n" f"- All visible text in the generated image must be written in {language}.\n" f"- Rewrite the provided content into natural, grammatically correct {language} before placing it in the design.\n" "- Correct spelling, grammar, capitalization, punctuation, and diacritics where appropriate.\n" "- Keep prices, numbers, brand names, restaurant names, product names, and URLs exactly as provided unless the user explicitly asks to translate them.\n" "- Do not mix languages in visible text unless a brand/product name is intentionally in another language.\n" "- Ensure every visible word is sharp, readable, professionally aligned, and free of typos." ) return GenerateAiImageService._append_reference_image_rules(f"{system_prompt}{language_rules}", request) @staticmethod def _append_reference_image_rules(system_prompt: str, request: GenerateAiImageRequest) -> str: if not request.reference_image_urls: return system_prompt reference_rules = ( "\n\nREFERENCE IMAGE REQUIREMENTS:\n" "- Reference images are mandatory visual inputs whenever provided.\n" "- The generated output must visibly reflect the provided reference images, including the main subject, setting, style, colors, lighting, materials, and composition.\n" "- Use the supplied reference images directly as visual guidance rather than replacing them with a generic scene." "\n- Do not redesign, restyle, recolor, or simplify key product objects shown in the reference images." "\n- Preserve bottle shape, label or etiquette color, packaging color, glass shape, object proportions, and distinctive visual details as faithfully as possible." ) return f"{system_prompt}{reference_rules}" @staticmethod def _build_user_prompt(request: GenerateAiImageRequest) -> str: user_prompt = GenerateAiImageService._relax_reference_prompt_restrictions(request.user_prompt) language = (request.language or "").strip() content = (request.content or "").strip() additions: List[str] = [] additions.append( "Image input usage:\n" "- Use the first attached image as the restaurant logo/brand asset. Keep it recognizable and integrate it cleanly.\n" "- Use every additional attached image as required primary visual guidance. The poster must clearly reflect the main subject, setting, style, colors, lighting, materials, and composition from these reference images.\n" "- If reference images are attached, match the generated poster to those images instead of using a generic scene.\n" "- Do not modify the identity of important objects from the reference images. Keep bottle shapes, label or etiquette colors, packaging colors, glassware forms, and distinctive product details unchanged.\n" "- If a reference image shows a specific bottle, box, candle, plate, or glass, preserve that object's visible design details instead of inventing a new variant.\n" "- Keep the final image ultra realistic with believable materials, natural shadows, realistic reflections, premium food styling, and authentic photography depth.\n" "- The provided content remains the source of truth for visible copy." ) if language: additions.append( "Visible text language requirement:\n" f"Write every visible text element in {language} with proper grammar, spelling, punctuation, and natural phrasing." ) if content: additions.append( "Content to write in the image:\n" f"{content}\n\n" "Use this content as the source of truth for visible copy. Correct grammar and spelling in the requested language, while preserving prices, numbers, brand names, and restaurant names." ) if not additions: return user_prompt return f"{user_prompt}\n\n" + "\n\n".join(additions) @staticmethod def _relax_reference_prompt_restrictions(prompt: str) -> str: replacements = { "Design only food, restaurant interior details, table settings, typography, and brand elements.": ( "Use the supplied reference images as required visual inputs, including their subject, atmosphere, " "restaurant setting, decor, colors, lighting, materials, food styling, and composition." ), "Do not create or depict people, faces, bodies, children, alcohol, cigarettes, weapons, nudity, medical claims, political symbols, or shocking content.": ( "Use the supplied reference images as required visual inputs." ), "Do not recreate identifiable people, faces, bodies, private details, unrelated text, or unsafe content from reference images.": ( "Use the supplied reference images as required visual inputs." ), "Do not include placeholder text, watermarks, QR codes, fake logos, unrelated brand names, people, faces, bodies, alcohol, smoking, or unsafe imagery.": ( "Use the supplied reference images as required visual inputs." ), } for old, new in replacements.items(): prompt = prompt.replace(old, new) return prompt @staticmethod def _extract_image_base64(response: Any) -> str: for output in getattr(response, "output", []) or []: output_type = getattr(output, "type", None) result = getattr(output, "result", None) if output_type == "image_generation_call" and result: return result if isinstance(output, dict) and output.get("type") == "image_generation_call" and output.get("result"): return output["result"] raise ExternalServiceException("OpenAI", "Responses API returned no image_generation_call result") @staticmethod def _extract_usage_metrics(response: Any) -> Dict[str, int]: usage = getattr(response, "usage", None) input_details = getattr(usage, "input_tokens_details", None) output_details = getattr(usage, "output_tokens_details", None) input_tokens = GenerateAiImageService._int_or_default(getattr(usage, "input_tokens", None)) output_tokens = GenerateAiImageService._int_or_default(getattr(usage, "output_tokens", None)) total_tokens = GenerateAiImageService._int_or_default(getattr(usage, "total_tokens", None)) if total_tokens == 0: total_tokens = input_tokens + output_tokens return { "input_tokens": input_tokens, "cached_tokens": GenerateAiImageService._int_or_default(getattr(input_details, "cached_tokens", None)), "output_tokens": output_tokens, "reasoning_tokens": GenerateAiImageService._int_or_default(getattr(output_details, "reasoning_tokens", None)), "total_tokens": total_tokens, } async def _log_analytics( self, *, request: GenerateAiImageRequest, response: Any, selected_model: str, image_size: str, image_quality: str, input_payload: List[Dict[str, Any]], tools: List[Dict[str, Any]], usage_metrics: Dict[str, int], execution_time_ms: int, ) -> None: try: await openai_client.log_manual_analytics( city=request.city, zavedenia_id=request.zavedenia_id, service="generate_ai_image", model=selected_model, tokens_input={ "prompt_tokens": usage_metrics["input_tokens"], "cached_tokens": usage_metrics["cached_tokens"], "total_tokens": usage_metrics["input_tokens"], }, tokens_output={ "completion_tokens": usage_metrics["output_tokens"], "reasoning_tokens": usage_metrics["reasoning_tokens"], "total_tokens": usage_metrics["output_tokens"], }, data_input={ "type": "responses_image_generation", "app_service": "generate_ai_image", "model": selected_model, "language": request.language, "content": request.content, "input": input_payload, "tools": tools, "image_size": image_size, "image_quality": image_quality, }, data_output={ "type": "responses_image_generation_response", "response_id": getattr(response, "id", None), "status": getattr(response, "status", None), "image_base64_present": True, }, execution_time=execution_time_ms, ) except Exception as exc: logger.warning("ai_analytics_failed", service="generate_ai_image", error=str(exc)) @staticmethod def _calculate_price_usd(model: str, input_tokens: int, output_tokens: int) -> Optional[float]: pricing = { "gpt-image-1": {"input": 5.0, "output": 40.0}, "gpt-image-1.5": {"input": 5.0, "output": 40.0}, } model_key = next((key for key in pricing if key in model), None) if model_key is None: return None rates = pricing[model_key] return round((input_tokens / 1_000_000) * rates["input"] + (output_tokens / 1_000_000) * rates["output"], 6) @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