import base64 import io import re from dataclasses import dataclass from pathlib import Path from typing import Dict, Any, Tuple, List, Optional import httpx import structlog import numpy as np from PIL import Image from openai import APIError, APITimeoutError from ...shared.clients.openai_client import openai_client from ...shared.exceptions.base import ExternalServiceException, ValidationException from .config import CreateRestaurantOfferServiceConfig from .models.requests import CreateRestaurantOfferRequest logger = structlog.get_logger("create_restaurant_offer.service") @dataclass class OfferTheme: palette: str background_scene: str decorative_elements: str font_style: str mood: str # Occasion keyword → theme mapping (keywords lowercase, checked via 'in') _OCCASION_THEMES: List[Tuple[List[str], OfferTheme]] = [ ( ["valentine", "love day"], OfferTheme( palette="deep crimson, rose gold, soft blush pink, ivory", background_scene="romantic candlelit restaurant table with red rose petals scattered, soft warm bokeh", decorative_elements="floating hearts, red and pink rose petals border, delicate gold filigree", font_style="elegant thin serif (Cormorant or Didot style)", mood="romantic, intimate, luxurious", ), ), ( ["8 march", "women's day", "womens day", "international women", "march 8"], OfferTheme( palette="rose gold, soft coral, mint green, cream white", background_scene="spring garden with blooming roses and peonies, soft natural light", decorative_elements="rose and peony flowers border, delicate botanical leaf sprigs, gold ribbon accents", font_style="light romantic serif with thin strokes", mood="feminine, fresh, celebratory, spring", ), ), ( ["christmas", "xmas", "new year", "new year's eve"], OfferTheme( palette="deep red, forest green, warm gold, ivory white", background_scene="festive decorated Christmas table with candles, pine branches and golden ornaments", decorative_elements="snowflakes, pine branches, gold baubles, red ribbon accents", font_style="classic bold serif with gold accents", mood="festive, warm, celebratory, luxurious", ), ), ( ["easter"], OfferTheme( palette="pastel yellow, soft lilac, mint green, blush pink", background_scene="bright spring meadow with wildflowers and soft morning light", decorative_elements="spring blossoms, Easter eggs, butterfly motifs, fresh green leaves", font_style="friendly rounded serif, light and airy", mood="fresh, cheerful, springtime, gentle", ), ), ( ["halloween"], OfferTheme( palette="deep orange, jet black, purple, burnt amber", background_scene="moody atmospheric scene with carved pumpkins and candlelight", decorative_elements="pumpkins, spider webs, bats silhouettes, gothic candelabra", font_style="gothic display serif, dramatic", mood="mysterious, spooky, dramatic, theatrical", ), ), ( ["summer", "bbq", "grill", "terrace", "garden party"], OfferTheme( palette="warm orange, golden yellow, sky blue, fresh green", background_scene="sunny outdoor terrace with lush greenery and warm afternoon light", decorative_elements="citrus slices, tropical leaves, summer flowers, sun rays", font_style="bold friendly sans-serif, casual and vibrant", mood="sunny, relaxed, fresh, outdoor", ), ), ( ["daily special", "daily menu", "lunch special", "lunch menu", "today's special"], OfferTheme( palette="clean white, warm sand, charcoal, brand accent colors", background_scene="elegant minimal restaurant interior with soft natural light", decorative_elements="simple clean dividers, subtle leaf or herb accents", font_style="modern geometric sans-serif, clean and confident", mood="clean, professional, fresh, appetizing", ), ), ( ["birthday", "anniversary", "celebration", "party"], OfferTheme( palette="champagne gold, deep navy, cream, sparkling silver", background_scene="elegant celebration setting with champagne, confetti and soft bokeh lights", decorative_elements="gold confetti, star bursts, ribbon and bow accents, champagne bubbles", font_style="celebratory display serif with gold weight", mood="joyful, luxurious, festive, special", ), ), ( ["autumn", "fall", "harvest", "october", "november"], OfferTheme( palette="warm amber, burnt sienna, deep burgundy, olive green", background_scene="cozy autumn restaurant corner with fallen leaves and warm candlelight", decorative_elements="autumn leaves, acorn and chestnut motifs, warm wood textures", font_style="warm serif with organic character", mood="cozy, warm, rustic, comforting", ), ), ( ["spring", "april", "may"], OfferTheme( palette="fresh green, blossom pink, sky blue, warm cream", background_scene="blooming spring garden with cherry blossoms and gentle sunlight", decorative_elements="cherry blossom branches, butterflies, fresh green leaves", font_style="light elegant serif, fresh and clean", mood="fresh, hopeful, light, blooming", ), ), ] _NEUTRAL_THEME = OfferTheme( palette="deep charcoal, warm gold, cream white, brand accent colors", background_scene="sophisticated restaurant interior with ambient lighting and elegant table setting", decorative_elements="subtle geometric borders, elegant dividers, minimal botanical accents", font_style="refined modern serif for headings, clean sans-serif for items", mood="elegant, premium, refined, professional", ) _MOCK_IMAGE_PATH = Path("menues/restaurant123/offer_9a8e94e6e2524de2abd34e584b315f79.png") class CreateRestaurantOfferService: def __init__(self, config: CreateRestaurantOfferServiceConfig): self.config = config # ── Theme detection ──────────────────────────────────────────────── @staticmethod def _detect_theme(occasion: str) -> OfferTheme: text = occasion.lower() for keywords, theme in _OCCASION_THEMES: if any(kw in text for kw in keywords): return theme return _NEUTRAL_THEME # ── Brand color extraction ───────────────────────────────────────── @staticmethod def _extract_brand_colors(logo_bytes: bytes, n: int = 3) -> List[str]: try: img = Image.open(io.BytesIO(logo_bytes)).convert("RGBA") img = img.resize((80, 80), Image.LANCZOS) data = np.array(img, dtype=np.uint8) r, g, b, a = data[:, :, 0], data[:, :, 1], data[:, :, 2], data[:, :, 3] mask = (a > 50) & ~((r > 230) & (g > 230) & (b > 230)) pixels = data[mask] if len(pixels) < 10: return [] sample = Image.fromarray(pixels[:, :3].reshape(1, -1, 3)) quantized = sample.quantize(colors=n, method=Image.Quantize.MEDIANCUT) palette = quantized.getpalette() or [] colors = [] for i in range(n): rv, gv, bv = palette[i * 3], palette[i * 3 + 1], palette[i * 3 + 2] if (rv > 220 and gv > 220 and bv > 220) or (rv < 30 and gv < 30 and bv < 30): continue colors.append(f"#{rv:02x}{gv:02x}{bv:02x}") return colors[:n] except Exception: return [] # ── Translation ──────────────────────────────────────────────────── async def _translate_offer_content( self, occasion: str, offer_items: str, target_language: str, city: str = "City not set", zavedenia_id: int = -1, ) -> tuple[str, str]: """Translate occasion title and offer item names into target_language. Prices (€X.XX patterns) are extracted, preserved, and re-injected after translation. """ # Extract and replace prices with placeholders so they survive translation price_pattern = re.compile(r"€\d+[.,]\d{2}") lines = offer_items.splitlines() placeholders: list[list[str]] = [] sanitized_lines: list[str] = [] for line in lines: prices = price_pattern.findall(line) placeholders.append(prices) # Replace each price with a stable token sanitized = line for i, price in enumerate(prices): sanitized = sanitized.replace(price, f"__PRICE{i}__", 1) sanitized_lines.append(sanitized) sanitized_offer = "\n".join(sanitized_lines) prompt = ( f"You are a professional restaurant content translator.\n" f"Translate the following text into {target_language}.\n" f"Rules:\n" f"- Translate dish names, drink names, and the occasion title naturally.\n" f"- Keep __PRICE0__, __PRICE1__, etc. tokens exactly as-is — do not translate or remove them.\n" f"- Keep the same line structure (one item per line).\n" f"- Do not add or remove lines.\n\n" f"Occasion title: {occasion}\n\n" f"Offer items (one per line):\n{sanitized_offer}\n\n" f"Return ONLY a JSON object:\n" f'{{"occasion": "translated occasion title", "offer_items": "translated items, one per line"}}' ) result, _ = await openai_client.json_completion( api_key=self.config.openai_api_key, model=self.config.translation_model, prompt=prompt, operation_id="translate_offer_content", service="create_restaurant_offer", timeout=30.0, analytics_context={ "city": city, "zavedenia_id": zavedenia_id, "data_input": { "app_service": "create_restaurant_offer", "occasion": occasion, "target_language": target_language, } } ) translated_occasion = result.get("occasion", occasion) translated_offer_raw = result.get("offer_items", sanitized_offer) # Re-inject prices back into their positions translated_lines = translated_offer_raw.splitlines() restored_lines: list[str] = [] for i, line in enumerate(translated_lines): if i < len(placeholders): for j, price in enumerate(placeholders[i]): line = line.replace(f"__PRICE{j}__", price, 1) restored_lines.append(line) return translated_occasion, "\n".join(restored_lines) # ── Price normalization ──────────────────────────────────────────── @staticmethod def _normalize_prices(text: str) -> str: """Ensure every numeric price in the offer items has a € prefix.""" # Match prices like: 12.90 / 12,90 / 9 / €12.90 — normalize all to €X.XX def add_euro(m: re.Match) -> str: raw = m.group(0).lstrip("€").strip() # Normalize comma decimal separator raw = raw.replace(",", ".") try: val = float(raw) return f"€{val:.2f}" except ValueError: return m.group(0) # Replace standalone numbers (with optional existing €) that look like prices return re.sub(r"€?\d+[.,]\d{1,2}(?!\d)", add_euro, text) # ── Prompt building ──────────────────────────────────────────────── def _build_prompt( self, request: CreateRestaurantOfferRequest, theme: OfferTheme, brand_colors: List[str], ) -> str: visual_prompt = (request.visual_prompt or "").strip() visual_prompt = re.sub(r"\b\d{3,4}\s*[xX]\s*\d{3,4}\b", "", visual_prompt).strip().strip(",").strip() color_instruction = ( f"Restaurant brand colors (from logo): {', '.join(brand_colors)} — " "weave these into borders, dividers, price labels, and accent elements." if brand_colors else "Use the theme palette as primary colors throughout." ) offer_lines = [self._normalize_prices(l.strip()) for l in request.offer_items.splitlines() if l.strip()] item_count = len(offer_lines) # Separate food from drinks heuristically drink_keywords = {"wine", "beer", "cocktail", "mocktail", "juice", "water", "soda", "prosecco", "champagne", "espresso", "coffee", "latte", "tea", "drink", "beverage", "spirit", "whisky", "gin", "vodka", "rum", "cider", "lemonade", "smoothie", "milkshake"} food_items = [] drink_items = [] for line in offer_lines: words = set(line.lower().split()) if words & drink_keywords: drink_items.append(line) else: food_items.append(line) food_block = "\n".join(f" • {item}" for item in food_items) if food_items else "" drink_block = "\n".join(f" • {item}" for item in drink_items) if drink_items else "" offer_block = "" if food_block: offer_block += f"DISHES:\n{food_block}\n" if drink_block: offer_block += f"\nDRINKS:\n{drink_block}\n" if not food_block and not drink_block: offer_block = "\n".join(f" • {item}" for item in offer_lines) # Build background scene — visual_prompt overrides/enriches the default theme scene if visual_prompt: background_description = ( f"{visual_prompt}. " f"The color palette and overall mood should feel: {theme.mood}, using {theme.palette}." ) else: background_description = ( f"{theme.background_scene} — rich, atmospheric, photorealistic feel. " f"Mood: {theme.mood}." ) prompt = ( f"Create a premium restaurant promotional offer poster for \"{request.occasion}\".\n\n" "═══ VISUAL THEME & ATMOSPHERE ═══\n" f"Background scene: {background_description}\n" f"Color palette dominating the entire composition: {theme.palette}.\n" f"Decorative elements woven throughout the design: {theme.decorative_elements}.\n" f"Typography style: {theme.font_style}.\n" f"{color_instruction}\n\n" "═══ TOP AREA — STRICT VERTICAL ORDER ═══\n" "The top of the poster has two clearly separated zones stacked vertically — in this exact order:\n" " ZONE 1 (topmost, ~0%–18% from top): LOGO SPACE — clean, unobstructed background. " "No text of any kind in this zone. No title, no tagline, no words. " "The background texture/scene flows naturally through this area. " "NO white boxes, NO rectangles, NO placeholder shapes.\n" " ZONE 2 (below logo space, ~18%–28% from top): OCCASION TITLE — large elegant lettering. " "This text block starts only AFTER the logo zone ends. " "The title must NEVER overlap or intrude into Zone 1. " "There must be visible clear vertical gap between the logo zone and the title.\n\n" "═══ OFFER CONTENT ═══\n" f"Present this as a curated special offer ({item_count} items) — NOT a full menu.\n" "Each item should feel like an exclusive selection, not a price list.\n\n" f"{offer_block}\n" "- Every price MUST be shown with the € symbol prefix, format €X.XX — no exceptions. " "Never show a bare number without €.\n" "- Separate DISHES from DRINKS with a themed decorative divider.\n" "- Each item gets its own breathing room — generous line spacing.\n\n" "═══ LAYOUT ═══\n" "Canvas: tall portrait 1024×1536. Strict top-to-bottom order:\n" " [0%–18%] Logo zone — background only, zero text\n" " [18%–28%] Occasion title — large elegant text, never overlapping logo zone\n" " [28%–85%] Offer items — themed, atmospheric, generous spacing\n" " [85%–100%] Decorative closing element\n" "Single column layout — premium promotional poster, not a menu grid.\n" "CRITICAL: Do NOT generate any white rectangle, blank box, or placeholder shape anywhere. " "Do NOT place any text in the top 18% of the canvas.\n" "Make it look like a real restaurant event promotion or social media announcement poster.\n\n" ) prompt += ( "CRITICAL: The result must feel like a professional restaurant marketing poster — " "not a menu, not a flyer. Rich visuals, the occasion theme must be unmistakable at a glance. " "The background is the hero — it must be vivid, immersive, and directly reflect the described scene and atmosphere." ) return prompt # ── Utilities ────────────────────────────────────────────────────── @staticmethod def _sanitize_offer_id(offer_id: str) -> str: sanitized = re.sub(r"[^A-Za-z0-9_-]", "_", offer_id.strip()) if not sanitized: raise ValidationException("offer_id must contain at least one valid character") return sanitized @staticmethod async def _download_image_bytes(url: str) -> bytes: async with httpx.AsyncClient(timeout=30.0) as client: response = await client.get(url) response.raise_for_status() return response.content @staticmethod def _remove_light_background(img: Image.Image, threshold: int = 200) -> Image.Image: """Remove light/white/near-white backgrounds aggressively using edge-aware alpha blending.""" img = img.convert("RGBA") data = np.array(img, dtype=np.float32) r, g, b, a = data[:, :, 0], data[:, :, 1], data[:, :, 2], data[:, :, 3] # Detect light pixels: high brightness AND low saturation (grey/white/beige) brightness = (r + g + b) / 3.0 max_channel = np.maximum(np.maximum(r, g), b) min_channel = np.minimum(np.minimum(r, g), b) saturation = np.where(max_channel > 0, (max_channel - min_channel) / max_channel, 0.0) # Light background: bright AND low saturation (covers white, cream, beige, light grey) light_mask = (brightness > threshold) & (saturation < 0.35) & (a > 10) # Smooth fade: fully transparent near 255, partial transparency near threshold fade = np.clip((brightness - threshold) / (255.0 - threshold), 0.0, 1.0) data[:, :, 3] = np.where(light_mask, (1.0 - fade) * a, a) return Image.fromarray(np.clip(data, 0, 255).astype(np.uint8), "RGBA") def _compose_with_logo(self, generated_image_bytes: bytes, logo_bytes: bytes) -> bytes: base_image = Image.open(io.BytesIO(generated_image_bytes)).convert("RGBA") logo_image = Image.open(io.BytesIO(logo_bytes)).convert("RGBA") logo_image = self._remove_light_background(logo_image) width = base_image.size[0] logo_max = max(160, int(width * 0.20)) margin = max(24, int(width * 0.04)) scale = min(logo_max / max(1, logo_image.width), logo_max / max(1, logo_image.height), 1.0) new_w = max(1, int(logo_image.width * scale)) new_h = max(1, int(logo_image.height * scale)) logo_resized = logo_image.resize((new_w, new_h), Image.LANCZOS) # Center horizontally at the top logo_x = (width - new_w) // 2 logo_y = margin base_image.alpha_composite(logo_resized, (logo_x, logo_y)) output = io.BytesIO() base_image.convert("RGB").save(output, format="PNG", optimize=True) return output.getvalue() # ── Main entry point ─────────────────────────────────────────────── async def create_restaurant_offer(self, request: CreateRestaurantOfferRequest) -> Dict[str, Any]: # Mock mode: skip all API calls and return pre-existing image if request.mock_data: logger.info("create_restaurant_offer_mock_mode", offer_id=request.offer_id) image_bytes = _MOCK_IMAGE_PATH.read_bytes() return { "image_bytes": image_bytes, "model": "mock", "input_tokens": 0, "output_tokens": 0, "total_tokens": 0, "price_usd": 0.0, } selected_model = ( request.model.strip() if request.model and request.model.strip() else self.config.create_restaurant_offer_model ) # Step 1: Detect occasion theme (on original language for reliable keyword matching) theme = self._detect_theme(request.occasion) # Step 2: Translate content if target_language is provided target_language = (request.target_language or "").strip() occasion = request.occasion offer_items = request.offer_items if target_language: logger.info("translate_offer_content_started", target_language=target_language) occasion, offer_items = await self._translate_offer_content( occasion=occasion, offer_items=offer_items, target_language=target_language, city=request.city, zavedenia_id=request.zavedenia_id, ) logger.info("translate_offer_content_completed", target_language=target_language) # Step 3: Download logo and extract brand colors logo_bytes = await self._download_image_bytes(request.logo_image_url) brand_colors = self._extract_brand_colors(logo_bytes) logger.info( "create_restaurant_offer_started", model=selected_model, offer_id=request.offer_id, occasion=occasion, theme_mood=theme.mood, brand_colors=brand_colors, target_language=target_language or None, has_visual_prompt=bool((request.visual_prompt or "").strip()), ) # Step 4: Build prompt using (possibly translated) content translated_request = request.model_copy(update={"occasion": occasion, "offer_items": offer_items}) prompt = self._build_prompt(translated_request, theme, brand_colors) # Step 5: Generate image client = openai_client.get_client(self.config.openai_api_key) try: response = await client.images.generate( model=selected_model, prompt=prompt, size="1024x1536", quality=self.config.image_quality, timeout=self.config.timeout, ) except APITimeoutError as exc: logger.error("create_restaurant_offer_timeout", error=str(exc), model=selected_model) raise ExternalServiceException("OpenAI", f"Image generation timed out: {exc}") except APIError as exc: logger.error("create_restaurant_offer_api_error", error=str(exc), model=selected_model) raise ExternalServiceException("OpenAI", str(exc)) if not response.data: raise ExternalServiceException("OpenAI", "Image generation returned empty result") b64 = getattr(response.data[0], "b64_json", None) if not b64: raise ExternalServiceException("OpenAI", "Image generation returned no image payload") generated_bytes = base64.b64decode(b64) # Extract usage metadata usage = getattr(response, "usage", None) input_tokens = getattr(usage, "input_tokens", 0) or 0 output_tokens = getattr(usage, "output_tokens", 0) or 0 total_tokens = getattr(usage, "total_tokens", 0) or 0 # gpt-image-1 pricing: $5/1M input tokens, $40/1M output tokens price_usd = round((input_tokens * 5 + output_tokens * 40) / 1_000_000, 6) if usage else None # Step 6: Overlay logo top-center try: composed_bytes = self._compose_with_logo(generated_bytes, logo_bytes) except Exception as exc: logger.error("create_restaurant_offer_compose_failed", offer_id=request.offer_id, error=str(exc)) raise ExternalServiceException("CreateRestaurantOfferCompose", str(exc)) logger.info( "create_restaurant_offer_completed", model=selected_model, offer_id=request.offer_id, occasion=occasion, input_tokens=input_tokens, output_tokens=output_tokens, total_tokens=total_tokens, price_usd=price_usd, ) return { "image_bytes": composed_bytes, "model": selected_model, "input_tokens": input_tokens, "output_tokens": output_tokens, "total_tokens": total_tokens, "price_usd": price_usd, }