import asyncio import base64 import json from dataclasses import dataclass from typing import Optional, List, Dict, Any import structlog from ...shared.clients.openai_client import openai_client from ...shared.exceptions.base import MenuExtractionException, ValidationException, RateLimitException from .config import MenuExtractionServiceConfig from .models.dto import MenuExtractionResponse, MenuPayload, MenuCategory from .models.schemas import MENU_JSON_SCHEMA logger = structlog.get_logger("menu_extraction.service") DEFAULT_TOKEN_LIMIT = 95000 DEFAULT_IMAGE_TOKEN_ESTIMATE = 4000 @dataclass class ImageDocument: content: bytes mime_type: str filename: Optional[str] = None data_url: Optional[str] = None class MenuExtractionService: def __init__(self, config: MenuExtractionServiceConfig): self.config = config self._api_key_lock = asyncio.Lock() self._api_key_index = 0 async def extract_menu( self, *, images: List[ImageDocument], source_language: str, target_language: str, api_key_override: Optional[str] = None, analytics_context: Optional[Dict[str, Any]] = None, ) -> MenuExtractionResponse: """Extract menu data from images""" if not images: raise ValidationException("At least one menu image is required") normalized_source = self._normalize_language(source_language) normalized_target = self._normalize_language(target_language) # Build initial prompt for token estimation initial_prompt = self._build_prompt(normalized_source, normalized_target, is_continuation=False) batches = self._chunk_images(images, initial_prompt) logger.info( "menu_extraction_started", image_count=len(images), batches_count=len(batches), source_language=normalized_source, target_language=normalized_target ) if not batches: combined_payload = MenuPayload(restaurant_name=None, categories=[]) else: tasks = [ self._process_batch( batch=batch, source_language=normalized_source, target_language=normalized_target, api_key_override=api_key_override, analytics_context=analytics_context, batch_index=batch_index, total_batches=len(batches) ) for batch_index, batch in enumerate(batches, start=1) ] batch_payloads = await asyncio.gather(*tasks) combined_payload = self._merge_payloads(batch_payloads) if normalized_target != normalized_source: combined_payload = await self._ensure_target_language( payload=combined_payload, source_language=normalized_source, target_language=normalized_target, api_key_override=api_key_override, analytics_context=analytics_context ) logger.info( "menu_extraction_completed", image_count=len(images), batches_count=len(batches), categories_count=len(combined_payload.categories), source_language=normalized_source, target_language=normalized_target ) return MenuExtractionResponse( menu=combined_payload, source_language=normalized_source, target_language=normalized_target, model=self.config.menu_extraction_model, image_count=len(images), ) async def _process_batch( self, *, batch: List[ImageDocument], source_language: str, target_language: str, api_key_override: Optional[str], analytics_context: Optional[Dict[str, Any]], batch_index: int, total_batches: int ) -> MenuPayload: logger.info( "menu_extraction_batch_started", batch_index=batch_index, total_batches=total_batches, batch_size=len(batch) ) # Use continuation prompt for batches after the first one is_continuation = batch_index > 1 prompt = self._build_prompt(source_language, target_language, is_continuation=is_continuation) content_blocks = self._build_content_blocks(prompt, batch) try: raw_menu = await self._invoke_with_key_failover( content_blocks=content_blocks, api_key_override=api_key_override, analytics_context=self._merge_analytics_context( analytics_context, { "data_input": { "app_service": "menu_extraction", "source_language": source_language, "target_language": target_language, "batch_index": batch_index, "total_batches": total_batches, } } ), batch_index=batch_index, total_batches=total_batches ) except RateLimitException: raise except Exception as exc: logger.error( "menu_extraction_failed", error=str(exc), batch_index=batch_index, total_batches=total_batches ) raise MenuExtractionException(f"Menu extraction failed: {exc}") try: payload = MenuPayload.model_validate(raw_menu) except Exception as exc: logger.error( "menu_validation_failed", error=str(exc), batch_index=batch_index, total_batches=total_batches ) raise MenuExtractionException(f"Menu validation failed for batch {batch_index}: {exc}") logger.info( "menu_extraction_batch_completed", batch_index=batch_index, total_batches=total_batches, categories_count=len(payload.categories) ) return payload @staticmethod def _normalize_language(value: Optional[str], fallback: str = "bg") -> str: if not value: return fallback sanitized = value.strip().split("-")[0] return sanitized.lower() or fallback def _chunk_images(self, images: List[ImageDocument], prompt: str) -> List[List[ImageDocument]]: token_limit = self.config.max_tokens_per_batch if not token_limit or token_limit <= 0: token_limit = DEFAULT_TOKEN_LIMIT image_token_estimate = ( self.config.image_tokens_estimate if getattr(self.config, "image_tokens_estimate", None) else DEFAULT_IMAGE_TOKEN_ESTIMATE ) base_tokens = self._estimate_tokens(prompt) + max(0, self.config.schema_tokens_estimate) if token_limit <= base_tokens: raise ValidationException( "Prompt configuration exceeds model token budget. Reduce prompt length or increase max_tokens_per_batch." ) per_image_tokens = base_tokens + image_token_estimate if per_image_tokens > token_limit: raise ValidationException( "Single image exceeds model token budget. Reduce prompt length or increase max_tokens_per_batch." ) return [[document] for document in images] @staticmethod def _image_to_data_url(document: ImageDocument) -> str: if document.data_url: return document.data_url if not document.content: raise ValidationException(f"Uploaded image is empty: {document.filename}") encoded = base64.b64encode(document.content).decode("utf-8") mime_type = document.mime_type or "image/jpeg" document.data_url = f"data:{mime_type};base64,{encoded}" return document.data_url @staticmethod def _estimate_tokens(text: str) -> int: if not text: return 0 return max(1, len(text) // 4) def _build_content_blocks(self, prompt: str, batch: List[ImageDocument]) -> List[Dict[str, Any]]: blocks: List[Dict[str, Any]] = [{"type": "text", "text": prompt}] for doc in batch: blocks.append({ "type": "image_url", "image_url": {"url": self._image_to_data_url(doc)} }) return blocks @staticmethod def _merge_analytics_context( base_context: Optional[Dict[str, Any]], extra_context: Dict[str, Any] ) -> Dict[str, Any]: merged = dict(base_context or {}) merged_data_input = dict(merged.get("data_input", {})) merged_data_input.update(extra_context.get("data_input", {})) if merged_data_input: merged["data_input"] = merged_data_input for key, value in extra_context.items(): if key != "data_input": merged[key] = value return merged async def _invoke_with_key_failover( self, *, content_blocks: List[Dict[str, Any]], api_key_override: Optional[str], analytics_context: Optional[Dict[str, Any]], batch_index: Optional[int], total_batches: Optional[int] ) -> Dict[str, Any]: available_keys = self._available_api_keys(api_key_override) attempted: set[str] = set() last_exception: Optional[Exception] = None while len(attempted) < len(available_keys): api_key = await self._select_api_key(api_key_override, attempted) if not api_key: break try: raw_menu, _ = await openai_client.json_completion( api_key=api_key, model=self.config.menu_extraction_model, messages=[{ "role": "user", "content": content_blocks }], operation_id="menu_extraction", service="menu_extraction", max_tokens=self.config.max_tokens, temperature=self.config.temperature, timeout=self.config.timeout, analytics_context=analytics_context, json_schema={ "name": "menu_schema", "schema": MENU_JSON_SCHEMA, "strict": True, } ) return raw_menu except RateLimitException as exc: attempted.add(api_key) last_exception = exc log_payload: Dict[str, Any] = { "api_key_suffix": self._mask_api_key(api_key), "attempted_keys": len(attempted), "total_keys": len(available_keys) } if batch_index is not None: log_payload["batch_index"] = batch_index if total_batches is not None: log_payload["total_batches"] = total_batches logger.warning("menu_extraction_rate_limited_api_key", **log_payload) continue if last_exception: raise last_exception raise MenuExtractionException("Menu extraction failed: No OpenAI API key available") def _available_api_keys(self, override: Optional[str] = None) -> List[str]: if override: return [override] return self.config.openai_api_keys or [self.config.openai_api_key] async def _select_api_key( self, override: Optional[str], excluded: Optional[set[str]] = None ) -> Optional[str]: excluded = excluded or set() keys = self._available_api_keys(override) if not keys: return None if override: return None if override in excluded else override for _ in range(len(keys)): api_key = await self._next_api_key(keys) if api_key not in excluded: return api_key return None async def _next_api_key(self, keys: List[str]) -> str: async with self._api_key_lock: api_key = keys[self._api_key_index] self._api_key_index = (self._api_key_index + 1) % len(keys) return api_key @staticmethod def _mask_api_key(api_key: str) -> str: if len(api_key) <= 8: return "****" return f"...{api_key[-4:]}" @staticmethod def _build_prompt(source_language: str, target_language: str, is_continuation: bool = False) -> str: if is_continuation: return ( "You are an expert at extracting structured menu data from restaurant menu images. " "This is a CONTINUATION PAGE of a multi-page menu.\n\n" f"The menu text is in '{source_language}'. " f"Produce the JSON response in '{target_language}'. " "Translate every textual field (names, descriptions, prices, currencies, serving sizes, notes) into the target language " "while keeping proper nouns accurate and preserving numeric values.\n\n" "EXTRACTION RULES:\n" "1. ITEM NAMES: Extract the exact dish/item name. Preserve capitalization and spelling.\n" "2. DESCRIPTIONS: Extract any descriptions, ingredients lists, or preparation notes. If none exist, use null.\n" " - Ingredients may be separated by pipes (|), commas, or line breaks\n" " - Combine multi-line descriptions into a single field\n" " - Preserve separator structure (keep pipes or commas as shown)\n" " - IMPORTANT: Provide TWO description fields:\n" " * description: Translated to target language\n" " * description_uk: ALWAYS translated to English (regardless of target language)\n" "3. PRICES: Extract ONLY numeric values with decimal precision, no currency symbols:\n" " - Common format: '300 г | 19.99 лв. | 10.22 €' (serving | price_bgn | price_eur)\n" " - CRITICAL: When BOTH BGN (лв) and EUR (€) prices are present, ALWAYS extract the EUR (€) price\n" " - If multiple prices exist: '300 г | 19.99 лв. | 10.22 €' → extract '10.22' (EUR price)\n" " - If only one price exists: extract that price regardless of currency\n" " - Include handwritten prices\n" " - Use decimal notation with 2 decimal places when applicable (e.g., '12.50' not '12.5')\n" " - Remove all currency symbols (лв, €, EUR, LEV, etc.)\n" " - Example: '300 г | 19.99 лв. | 10.22 €' → extract '10.22' (NOT 19.99)\n" " - Example: '250 г | 24.99 лв. | 12.78 €' → extract '12.78' (NOT 24.99)\n" " - Example: '7.50€' (single price) → extract '7.50'\n" "4. SERVING SIZE: Extract any portion information with the unit (e.g., '300 ml', '2 pieces', '150g', '280 г').\n" " - In format '300 г | 19.99 лв. | 10.22 €', extract '300 г' as serving_size\n" "5. MEASURE: Analyze serving_size and return an INTEGER representing the unit type. Must be EXACTLY:\n" " - 0 - if serving_size contains weight units (g, gr, gram, гр, грам, kg, кг, etc.)\n" " - 1 - if serving_size contains volume units (ml, мл, milliliter, l, л, liter, etc.)\n" " - 2 - if serving_size contains count/piece units (pc, pcs, piece, pieces, кол, парче, part, бр, брой, шт, штука, etc.)\n" " - 3 - if serving_size contains length units (cm, см, centimeter, etc.)\n" " - null - if no serving_size or cannot determine measure type\n" "6. RESTAURANT NAME: Set to null (this is a continuation page).\n" "7. CATEGORY NAMES: Use empty string \"\" for all category names (this is a continuation page).\n" "8. NOTES: Capture any allergen info, spice levels, dietary markers (ВЕГАН, ПИКАНТЕН), or special qualifiers.\n\n" "MULTI-COLUMN LAYOUT HANDLING:\n" "- Menus may have 2-4 columns. Read TOP-TO-BOTTOM within each column, then LEFT-TO-RIGHT across columns\n" "- Category headers are bold/uppercase text without prices (e.g., АНТИПАСТИ, САЛАТИ, МЕСО, РИБА)\n" "- A category may continue across columns - track context carefully\n" "- Each item typically has: Name (line 1) → Description/ingredients (lines 2-N) → Serving|Price1|Price2 (last line)\n\n" "ITEM VARIATIONS:\n" "- Some items have variations listed below (e.g., Caesar salad with chicken, with shrimp)\n" "- Create SEPARATE items for each variation with full name\n" "- Example: 'Caesar Salad' (base), 'Caesar Salad with Grilled Chicken' (variation 1), 'Caesar Salad with Shrimp' (variation 2)\n\n" "IMPORTANT:\n" "- Extract EVERY item visible in the images\n" "- Handle Cyrillic, Latin, and mixed scripts\n" "- Handle multi-column layouts (2-4 columns)\n" "- Recognize both printed and handwritten prices\n" "- Always format prices with 2 decimal places (e.g., '15.00' not '15')\n" "- CRITICAL: Extract EUR (€) price when both BGN and EUR are present\n" "- Remove ALL currency symbols from price values\n" "- Preserve pipe-separated ingredient lists\n" "- If information is unclear or missing, use null\n" "- Since this is a continuation page: restaurant_name must be null and category names must be empty strings" ) else: return ( "You are an expert at extracting structured menu data from restaurant menu images. " "Your task is to carefully analyze the provided menu images and extract ALL menu items with complete details.\n\n" f"The menu text is in '{source_language}'. " f"Produce the JSON response in '{target_language}'. " "Translate every textual field (names, descriptions, prices, currencies, serving sizes, notes, category titles) into the target language " "while keeping proper nouns accurate and preserving numeric values.\n\n" "EXTRACTION RULES:\n" "1. ITEM NAMES: Extract the exact dish/item name. Preserve capitalization and spelling.\n" "2. DESCRIPTIONS: Extract any descriptions, ingredients lists, or preparation notes. If none exist, use null.\n" " - Ingredients may be separated by pipes (|), commas, or line breaks\n" " - Combine multi-line descriptions into a single field\n" " - Preserve separator structure (keep pipes or commas as shown)\n" " - Example: 'Авокадо | Чери Домати | Годжи Бери' → 'Avocado | Cherry Tomatoes | Goji Berries'\n" " - IMPORTANT: Provide TWO description fields:\n" " * description: Translated to target language\n" " * description_uk: ALWAYS translated to English (regardless of target language)\n" " - Example: If target is 'bg' and source description is 'Fresh salad with tomatoes':\n" " * description: 'Свежа салата с домати' (in target language)\n" " * description_uk: 'Fresh salad with tomatoes' (always in English)\n" "3. PRICES: Extract ONLY numeric values with decimal precision, no currency symbols:\n" " - Common format: '300 г | 19.99 лв. | 10.22 €' (serving | price_bgn | price_eur)\n" " - CRITICAL: When BOTH BGN (лв) and EUR (€) prices are present, ALWAYS extract the EUR (€) price\n" " - If multiple prices exist: '300 г | 19.99 лв. | 10.22 €' → extract '10.22' (EUR price)\n" " - If only one price exists: extract that price regardless of currency\n" " - Include handwritten prices\n" " - Use decimal notation with 2 decimal places when applicable (e.g., '12.50' not '12.5')\n" " - Remove all currency symbols (лв, €, EUR, LEV, etc.)\n" " - Example: '300 г | 19.99 лв. | 10.22 €' → extract '10.22' (NOT 19.99)\n" " - Example: '250 г | 24.99 лв. | 12.78 €' → extract '12.78' (NOT 24.99)\n" " - Example: '400 г | 28.99 лв. | 14.82 €' → extract '14.82' (NOT 28.99)\n" " - Example: '7.50€' (single price) → extract '7.50'\n" "4. SERVING SIZE: Extract any portion information with the unit (e.g., '300 ml', '2 pieces', '150g', '280 г').\n" " - In format '300 г | 19.99 лв. | 10.22 €', extract '300 г' as serving_size\n" " - Keep the unit exactly as shown (г, g, ml, бр, etc.)\n" "5. MEASURE: Analyze serving_size and return an INTEGER representing the unit type. Must be EXACTLY:\n" " - 0 - if serving_size contains weight units (g, gr, gram, гр, грам, kg, кг, etc.)\n" " - 1 - if serving_size contains volume units (ml, мл, milliliter, l, л, liter, etc.)\n" " - 2 - if serving_size contains count/piece units (pc, pcs, piece, pieces, кол, парче, part, бр, брой, шт, штука, etc.)\n" " - 3 - if serving_size contains length units (cm, см, centimeter, etc.)\n" " - null - if no serving_size or cannot determine measure type\n" "6. CATEGORIES: Group items under their menu section headings.\n" " - Category headers are typically bold, uppercase text without prices\n" " - Common categories: АНТИПАСТИ, САЛАТИ, МЕСО, РИБА, PIZZA, ПАСТА, ДЕСЕРТИ, etc.\n" " - Preserve category organization as shown in the menu\n" " - IMPORTANT: Provide TWO description fields for categories:\n" " * description: Translated to target language\n" " * description_uk: ALWAYS translated to English (regardless of target language)\n" "7. NOTES: Capture any allergen info, spice levels, dietary markers (ВЕГАН, БЕЗ ГЛУТЕН, ПИКАНТЕН), or special qualifiers.\n\n" "MULTI-COLUMN LAYOUT HANDLING:\n" "- Menus may have 2-4 columns. Read TOP-TO-BOTTOM within each column, then LEFT-TO-RIGHT across columns\n" "- Category headers appear at the start of sections and may span partial columns\n" "- A category may continue across multiple columns or split across the layout\n" "- Each item typically has: Name (line 1) → Description/ingredients (lines 2-N) → Serving|Price1|Price2 (last line)\n" "- Typical item structure:\n" " Line 1: Item Name\n" " Line 2-N: Ingredients | separated | by | pipes\n" " Last line: 300 г | 19.99 лв. | 10.22 €\n\n" "ITEM VARIATIONS:\n" "- Some items have variations listed below the main item (e.g., Caesar salad, with chicken, with shrimp)\n" "- Create SEPARATE items for each variation with descriptive names\n" "- Example input:\n" " 'Салата Цезар 300 г | 17.99 лв. | 9.20 €'\n" " 'с пиле 300 г | 19.99 лв. | 10.22 €'\n" " 'със скариди 300 г | 21.99 лв. | 11.24 €'\n" "- Extract as:\n" " Item 1: name='Caesar Salad', price='17.99'\n" " Item 2: name='Caesar Salad with Chicken', price='19.99'\n" " Item 3: name='Caesar Salad with Shrimp', price='21.99'\n\n" "IMPORTANT:\n" "- Extract EVERY item visible in the images\n" "- Handle Cyrillic, Latin, and mixed scripts\n" "- Handle multi-column layouts (2-4 columns) with proper reading order\n" "- Recognize both printed and handwritten prices\n" "- Always format prices with 2 decimal places (e.g., '15.00' not '15')\n" "- CRITICAL: Extract EUR (€) price when both BGN and EUR prices are present\n" "- Remove ALL currency symbols from price values\n" "- Preserve pipe-separated ingredient lists in descriptions\n" "- Create separate items for variations (don't combine into one item)\n" "- If information is unclear or missing, use null\n" "- Preserve the original menu structure and categories as they appear" ) @staticmethod def _build_translation_prompt( payload: MenuPayload, source_language: str, target_language: str ) -> str: payload_json = json.dumps(payload.model_dump(), ensure_ascii=False) return ( "You are a professional culinary translator. " f"The following JSON represents menu data currently in '{source_language}'. " f"Translate EVERY textual field into '{target_language}' while keeping the JSON structure identical. " "Translate restaurant_name, category names/descriptions, and each menu item's name, description, serving_size, and notes. " "Keep price values as numeric strings without modification (no currency symbols). " "Keep measure values unchanged (must remain as integers: 0, 1, 2, 3, or null)." "If a text field is missing, keep it null.\n\n" "MENU_JSON:\n" f"{payload_json}" ) @staticmethod def _merge_payloads(payloads: List[MenuPayload]) -> MenuPayload: if not payloads: return MenuPayload(restaurant_name=None, categories=[]) combined_name: Optional[str] = None categories: List[MenuCategory] = [] category_index: Dict[str, int] = {} for payload in payloads: if not combined_name and payload.restaurant_name: combined_name = payload.restaurant_name for category in payload.categories: key = category.name.strip().lower() existing_idx = category_index.get(key) if existing_idx is not None: existing = categories[existing_idx] if not existing.description and category.description: existing.description = category.description existing.items.extend(item.model_copy(deep=True) for item in category.items) else: category_copy = category.model_copy(deep=True) categories.append(category_copy) category_index[key] = len(categories) - 1 return MenuPayload(restaurant_name=combined_name, categories=categories) async def _ensure_target_language( self, *, payload: MenuPayload, source_language: str, target_language: str, api_key_override: Optional[str], analytics_context: Optional[Dict[str, Any]] ) -> MenuPayload: prompt = self._build_translation_prompt(payload, source_language, target_language) content_blocks = [{"type": "text", "text": prompt}] logger.info( "menu_translation_enforcement_started", source_language=source_language, target_language=target_language ) try: raw_menu = await self._invoke_with_key_failover( content_blocks=content_blocks, api_key_override=api_key_override, analytics_context=self._merge_analytics_context( analytics_context, { "data_input": { "app_service": "menu_extraction", "stage": "translation_enforcement", "source_language": source_language, "target_language": target_language, } } ), batch_index=None, total_batches=None ) except RateLimitException: raise except Exception as exc: logger.error("menu_translation_enforcement_failed", error=str(exc)) raise MenuExtractionException(f"Menu translation enforcement failed: {exc}") translated_payload = MenuPayload.model_validate(raw_menu) logger.info( "menu_translation_enforcement_completed", source_language=source_language, target_language=target_language ) return translated_payload