import asyncio import base64 import hashlib import io import json from pathlib import Path import time from urllib.parse import quote import httpx import structlog from PIL import Image, ImageOps from ..menu_extraction.service import MenuExtractionService, ImageDocument from .adapter import CreateMenuAdapter from .config import InsertMenuItemsServiceConfig from ...shared.exceptions.base import ( MenuExtractionException, ExternalServiceException, ValidationException ) logger = structlog.get_logger("insert_menu_items.service") class InsertMenuItemsService: """Service for handling menu image uploads and menu creation""" def __init__( self, config: InsertMenuItemsServiceConfig, menu_extraction_service: MenuExtractionService ): self.config = config self.menu_extraction_service = menu_extraction_service self.adapter = CreateMenuAdapter() self._observability_scheduled_request_ids: set[str] = set() async def process_menu_upload( self, *, image_content: bytes, filename: str, mime_type: str, city: str, token: str, zavedenia_id: int = -1, vanue_name: str = "", source_language: str = "bg", target_language: str = "en", request_id: str, raw_form_fields: dict | None = None, ) -> None: """ Background task to process menu image and create menu This method runs asynchronously after returning response to client. Errors are logged but not raised. """ start_time = time.time() logger.info( "menu_upload_processing_started", request_id=request_id, city=city, filename=filename, source_language=source_language, target_language=target_language ) upload_trace = self._save_observability_upload_trace( image_content=image_content, filename=filename, mime_type=mime_type, request_id=request_id, ) self._schedule_upload_observability( image_content=image_content, filename=filename, mime_type=mime_type, city=city, zavedenia_id=zavedenia_id, vanue_name=vanue_name, token=token, source_language=source_language, target_language=target_language, request_id=request_id, upload_trace=upload_trace, raw_form_fields=raw_form_fields or {}, ) # Print upload request details for debugging print("\n" + "🔵" + "="*78 + "🔵") print("📥 BACKGROUND TASK STARTED - MENU UPLOAD PROCESSING") print("="*80) print(f"Request ID: {request_id}") print(f"City: {city}") print(f"Filename: {filename}") print(f"Source Language: {source_language}") print(f"Target Language: {target_language}") print(f"Image Size: {len(image_content)} bytes ({round(len(image_content)/1024, 2)} KB)") print(f"MIME Type: {mime_type}") print("🔵" + "="*78 + "🔵\n") try: # Step 1: Extract menu from image using menu_extraction service logger.info( "STEP_1_START: Calling menu extraction service", request_id=request_id, image_size_bytes=len(image_content), mime_type=mime_type ) try: image_doc = ImageDocument( content=image_content, mime_type=mime_type, filename=filename ) menu_response = await self.menu_extraction_service.extract_menu( images=[image_doc], source_language=source_language, target_language=target_language, api_key_override=None, analytics_context={ "city": city, "zavedenia_id": zavedenia_id, "data_input": { "filename": filename, "source_language": source_language, "target_language": target_language, } } ) # Log the complete menu extraction response logger.info( "MENU_EXTRACTION_RESPONSE", request_id=request_id, response_data=menu_response.model_dump() ) # Print to console for visibility print("\n" + "="*80) print("MENU EXTRACTION SERVICE RESPONSE") print("="*80) print(f"Request ID: {request_id}") print(f"Restaurant: {menu_response.menu.restaurant_name}") print(f"Categories: {len(menu_response.menu.categories)}") print(f"Model: {menu_response.model}") print("\nFull Response Data:") import json print(json.dumps(menu_response.model_dump(), indent=2, ensure_ascii=False)) print("="*80 + "\n") logger.info( "STEP_1_COMPLETED: Menu extraction successful", request_id=request_id, categories_count=len(menu_response.menu.categories), restaurant_name=menu_response.menu.restaurant_name, model_used=menu_response.model ) except Exception as e: logger.error( "STEP_1_FAILED: Menu extraction error", request_id=request_id, error=str(e), error_type=type(e).__name__ ) raise # Step 2: Transform response using adapter logger.info( "STEP_2_START: Transforming menu data using adapter", request_id=request_id, categories_to_transform=len(menu_response.menu.categories) ) try: create_menu_payload = self.adapter.transform( menu_response=menu_response, city=city, token=token ) # Calculate total items total_items = sum(len(category["items"]) for category in create_menu_payload["menu"]) # Log the complete transformed payload that will be sent to createMenu.php logger.info( "TRANSFORMED_PAYLOAD_FOR_CREATE_MENU", request_id=request_id, payload=create_menu_payload ) # Print to console for visibility print("\n" + "="*80) print("PAYLOAD BEING SENT TO createMenu.php") print("="*80) print(f"Request ID: {request_id}") print(f"City: {city}") print(f"URL: {self.config.create_menu_url}") print(f"Categories: {len(create_menu_payload['menu'])}") print(f"Total Items: {total_items}") print("\nFull Payload:") import json print(json.dumps(create_menu_payload, indent=2, ensure_ascii=False)) print("="*80 + "\n") logger.info( "STEP_2_COMPLETED: Menu data transformation successful", request_id=request_id, categories_count=len(create_menu_payload["menu"]), total_items=total_items, target_city=city ) except Exception as e: logger.error( "STEP_2_FAILED: Adapter transformation error", request_id=request_id, error=str(e), error_type=type(e).__name__ ) raise # Step 3: Call external createMenu.php API logger.info( "STEP_3_START: Calling external createMenu.php API", request_id=request_id, city=city, url=self.config.create_menu_url, categories_count=len(create_menu_payload["menu"]), total_items=total_items ) try: await self._call_create_menu_api(create_menu_payload, request_id) logger.info( "STEP_3_COMPLETED: External API call successful", request_id=request_id, city=city ) except Exception as e: logger.error( "STEP_3_FAILED: External API call error", request_id=request_id, error=str(e), error_type=type(e).__name__ ) raise # Calculate processing time processing_time = round(time.time() - start_time, 2) # Print final success summary to console print("\n" + "🟢" + "="*78 + "🟢") print("✅ MENU UPLOAD PROCESSING COMPLETED SUCCESSFULLY") print("="*80) print(f"Request ID: {request_id}") print(f"City: {city}") print(f"Filename: {filename}") print(f"Categories: {len(create_menu_payload['menu'])}") print(f"Total Items: {total_items}") print(f"Processing Time: {processing_time}s") print(f"Status: ✅ SUCCESS") print("🟢" + "="*78 + "🟢\n") logger.info( "========================================", ) logger.info( "MENU_UPLOAD_PROCESSING_COMPLETED_SUCCESSFULLY", request_id=request_id, city=city, filename=filename, source_language=source_language, target_language=target_language, categories_count=len(create_menu_payload["menu"]), total_items=total_items, processing_time_seconds=processing_time, status="SUCCESS" ) logger.info( "========================================", ) except MenuExtractionException as e: processing_time = round(time.time() - start_time, 2) # Print error summary to console print("\n" + "🔴" + "="*78 + "🔴") print("❌ MENU UPLOAD PROCESSING FAILED - MENU EXTRACTION ERROR") print("="*80) print(f"Request ID: {request_id}") print(f"City: {city}") print(f"Filename: {filename}") print(f"Error Stage: menu_extraction") print(f"Processing Time: {processing_time}s") print(f"Status: ❌ FAILED") print(f"\nError Details:") print(str(e)) print("🔴" + "="*78 + "🔴\n") logger.error( "========================================", ) logger.error( "MENU_UPLOAD_PROCESSING_FAILED", request_id=request_id, city=city, filename=filename, error_stage="menu_extraction", error=str(e), processing_time_seconds=processing_time, status="FAILED" ) logger.error( "========================================", ) except ExternalServiceException as e: processing_time = round(time.time() - start_time, 2) # Print error summary to console print("\n" + "🔴" + "="*78 + "🔴") print("❌ MENU UPLOAD PROCESSING FAILED - CREATE MENU API ERROR") print("="*80) print(f"Request ID: {request_id}") print(f"City: {city}") print(f"Filename: {filename}") print(f"Error Stage: create_menu_api") print(f"Processing Time: {processing_time}s") print(f"Status: ❌ FAILED") print(f"\nError Details:") print(str(e)) print("🔴" + "="*78 + "🔴\n") logger.error( "========================================", ) logger.error( "MENU_UPLOAD_PROCESSING_FAILED", request_id=request_id, city=city, filename=filename, error_stage="create_menu_api", error=str(e), processing_time_seconds=processing_time, status="FAILED" ) logger.error( "========================================", ) except Exception as e: processing_time = round(time.time() - start_time, 2) # Print error summary to console print("\n" + "🔴" + "="*78 + "🔴") print("❌ MENU UPLOAD PROCESSING FAILED - UNEXPECTED ERROR") print("="*80) print(f"Request ID: {request_id}") print(f"City: {city}") print(f"Filename: {filename}") print(f"Error Stage: unknown") print(f"Error Type: {type(e).__name__}") print(f"Processing Time: {processing_time}s") print(f"Status: ❌ FAILED") print(f"\nError Details:") print(str(e)) print("🔴" + "="*78 + "🔴\n") logger.error( "========================================", ) logger.error( "MENU_UPLOAD_PROCESSING_FAILED", request_id=request_id, city=city, filename=filename, error_stage="unknown", error=str(e), error_type=type(e).__name__, processing_time_seconds=processing_time, status="FAILED" ) logger.error( "========================================", ) async def _call_create_menu_api( self, payload: dict, request_id: str ) -> None: """Call external createMenu.php API with retry logic""" url = self.config.create_menu_url timeout = self.config.timeout max_retries = self.config.max_retries async with httpx.AsyncClient() as client: for attempt in range(1, max_retries + 1): try: logger.info( "API_REQUEST_ATTEMPT", request_id=request_id, attempt=attempt, max_retries=max_retries, url=url, payload_size=len(str(payload)), categories_count=len(payload.get("menu", [])), city=payload.get("city") ) # Log the full request payload being sent logger.info( "CREATE_MENU_REQUEST_PAYLOAD", request_id=request_id, attempt=attempt, url=url, full_payload=payload ) # Print equivalent curl command import json from pathlib import Path from datetime import datetime payload_json = json.dumps(payload, ensure_ascii=False) curl_command = ( f"curl --location '{url}' \\\n" f"--header 'Content-Type: application/json' \\\n" f"--data '{payload_json}'" ) # Print to console print("\n" + "="*80) print("EQUIVALENT CURL COMMAND FOR createMenu.php") print("="*80) print(f"Request ID: {request_id}") print(f"Attempt: {attempt}/{max_retries}") print("\nYou can run this command manually to test:\n") print(curl_command) print("\n" + "="*80 + "\n") # Write to dedicated curl log file log_dir = Path("logs") log_dir.mkdir(exist_ok=True) curl_log_file = log_dir / "curl_commands.log" timestamp = datetime.now().isoformat() with open(curl_log_file, "a", encoding="utf-8") as f: f.write("\n" + "="*80 + "\n") f.write(f"Timestamp: {timestamp}\n") f.write(f"Request ID: {request_id}\n") f.write(f"Attempt: {attempt}/{max_retries}\n") f.write(f"City: {payload.get('city')}\n") f.write("="*80 + "\n") f.write(curl_command + "\n") f.write("="*80 + "\n\n") logger.info( "CREATE_MENU_CURL_COMMAND", request_id=request_id, curl_command=curl_command, curl_log_file=str(curl_log_file) ) response = await client.post( url, json=payload, headers={"Content-Type": "application/json"}, timeout=timeout ) response.raise_for_status() # Parse and log the full response response_text = response.text logger.info( "CREATE_MENU_RESPONSE", request_id=request_id, attempt=attempt, status_code=response.status_code, full_response=response_text ) # Print to console for visibility with success indicator print("\n" + "="*80) print("✅ RESPONSE FROM createMenu.php - SUCCESS") print("="*80) print(f"Request ID: {request_id}") print(f"Status Code: {response.status_code}") print(f"Attempt: {attempt}/{max_retries}") print(f"Status: ✅ SUCCESS") print("\nFull Response:") print(response_text) # Try to parse as JSON for better logging try: import json response_json = json.loads(response_text) logger.info( "CREATE_MENU_RESPONSE_PARSED", request_id=request_id, response_data=response_json ) print("\nParsed JSON Response:") print(json.dumps(response_json, indent=2, ensure_ascii=False)) except json.JSONDecodeError: logger.warning( "CREATE_MENU_RESPONSE_NOT_JSON", request_id=request_id, response_text=response_text ) print("\n(Response is not JSON format)") print("="*80 + "\n") logger.info( "API_REQUEST_SUCCESS", request_id=request_id, attempt=attempt, status_code=response.status_code, response_length=len(response.text) ) return except httpx.HTTPStatusError as e: # Print error to console with red X print("\n" + "="*80) print("❌ RESPONSE FROM createMenu.php - HTTP ERROR") print("="*80) print(f"Request ID: {request_id}") print(f"Status Code: {e.response.status_code}") print(f"Attempt: {attempt}/{max_retries}") print(f"Status: ❌ FAILED") print("\nError Response:") print(e.response.text[:500]) print("="*80 + "\n") logger.error( "API_REQUEST_HTTP_ERROR", request_id=request_id, attempt=attempt, max_retries=max_retries, status_code=e.response.status_code, error_message=e.response.text[:300], url=url ) if attempt == max_retries: logger.error( "API_REQUEST_FAILED_ALL_RETRIES", request_id=request_id, total_attempts=max_retries, final_status_code=e.response.status_code ) raise ExternalServiceException( "createMenu.php", f"HTTP {e.response.status_code}: {e.response.text[:100]}" ) except httpx.RequestError as e: # Print network error to console with red X print("\n" + "="*80) print("❌ RESPONSE FROM createMenu.php - NETWORK ERROR") print("="*80) print(f"Request ID: {request_id}") print(f"Attempt: {attempt}/{max_retries}") print(f"Status: ❌ FAILED") print(f"Error Type: {type(e).__name__}") print(f"\nError Details:") print(str(e)) print("="*80 + "\n") logger.error( "API_REQUEST_NETWORK_ERROR", request_id=request_id, attempt=attempt, max_retries=max_retries, error=str(e), error_type=type(e).__name__, url=url ) if attempt == max_retries: logger.error( "API_REQUEST_FAILED_ALL_RETRIES", request_id=request_id, total_attempts=max_retries, error_type="NetworkError" ) raise ExternalServiceException( "createMenu.php", f"Request failed: {str(e)}" ) # Wait before retry (exponential backoff) if attempt < max_retries: wait_time = 2 ** attempt # 2, 4, 8 seconds logger.warning( "API_REQUEST_RETRY_SCHEDULED", request_id=request_id, current_attempt=attempt, next_attempt=attempt + 1, wait_seconds=wait_time ) await asyncio.sleep(wait_time) def _schedule_upload_observability( self, *, image_content: bytes, filename: str, mime_type: str, city: str, token: str, zavedenia_id: int, vanue_name: str, source_language: str, target_language: str, request_id: str, upload_trace: dict, raw_form_fields: dict ) -> None: """Fire the n8n upload observer without delaying menu processing.""" if not self.config.observability_webhook_url: logger.info( "insert_menu_items_observability_skipped", request_id=request_id, reason="webhook_url_not_configured" ) return if request_id in self._observability_scheduled_request_ids: logger.warning( "insert_menu_items_observability_duplicate_skipped", request_id=request_id, ) return self._observability_scheduled_request_ids.add(request_id) task = asyncio.create_task( self._notify_upload_observability( image_content=image_content, filename=filename, mime_type=mime_type, city=city, zavedenia_id=zavedenia_id, vanue_name=vanue_name, token=token, source_language=source_language, target_language=target_language, request_id=request_id, upload_trace=upload_trace, raw_form_fields=raw_form_fields, ) ) task.add_done_callback( lambda completed: logger.error( "insert_menu_items_observability_task_failed", request_id=request_id, error=str(completed.exception()) ) if completed.exception() else None ) async def _notify_upload_observability( self, *, image_content: bytes, filename: str, mime_type: str, city: str, token: str, zavedenia_id: int, vanue_name: str, source_language: str, target_language: str, request_id: str, upload_trace: dict, raw_form_fields: dict ) -> None: payload = { "request_id": request_id, "city": city, "zavedenia_id": zavedenia_id, "vanueName": vanue_name or "", "token": token, "source_language": source_language, "target_language": target_language, "filename": filename, "mime_type": mime_type, "image_size_bytes": len(image_content), "notify_email": self.config.observability_notify_email, "received_image_sha256": upload_trace.get("sha256"), "received_image_saved_path": upload_trace.get("saved_path"), "received_image_public_saved_path": upload_trace.get("public_saved_path"), "received_image_public_url": upload_trace.get("public_url"), "received_image_decode_ok": upload_trace.get("decode_ok"), "received_image_decode_error": upload_trace.get("decode_error"), "received_image_width": upload_trace.get("width"), "received_image_height": upload_trace.get("height"), "raw_form_fields": raw_form_fields, } data = { **{key: str(value) for key, value in payload.items()}, "image_base64": base64.b64encode(image_content).decode("ascii"), "image_preview_base64": self._build_observability_image_preview_base64( image_content=image_content, mime_type=mime_type, ), "request_payload": json.dumps(payload, ensure_ascii=False), } files = { "file": (filename, image_content, mime_type), } headers = {} if self.config.observability_bearer_token: headers["Authorization"] = f"Bearer {self.config.observability_bearer_token}" try: async with httpx.AsyncClient(timeout=self.config.observability_timeout) as client: response = await client.post( self.config.observability_webhook_url, data=data, files=files, headers=headers ) response.raise_for_status() logger.info( "insert_menu_items_observability_sent", request_id=request_id, status_code=response.status_code ) except Exception as e: logger.warning( "insert_menu_items_observability_failed", request_id=request_id, error=str(e), error_type=type(e).__name__ ) @staticmethod def _build_observability_image_preview_base64( *, image_content: bytes, mime_type: str, ) -> str: """Build a small inline-safe JPEG preview for observability emails.""" try: with Image.open(io.BytesIO(image_content)) as image: image = ImageOps.exif_transpose(image) if image.mode not in ("RGB", "L"): image = image.convert("RGB") image.thumbnail((720, 720)) output = io.BytesIO() image.save( output, format="JPEG", quality=45, optimize=False, progressive=False, ) return base64.b64encode(output.getvalue()).decode("ascii") except Exception as e: logger.warning( "insert_menu_items_observability_preview_failed", mime_type=mime_type, error=str(e), error_type=type(e).__name__, ) return "" def _save_observability_upload_trace( self, *, image_content: bytes, filename: str, mime_type: str, request_id: str, ) -> dict: """Persist the exact upload bytes before any downstream processing.""" safe_filename = Path(filename or "menu-image").name or "menu-image" trace_dir = Path(self.config.temp_directory) / "observability_uploads" / request_id trace_path = trace_dir / safe_filename public_dir = Path(self.config.observability_public_directory) / request_id public_path = public_dir / safe_filename public_base_url = self.config.public_base_url.rstrip("/") public_url_prefix = "/" + self.config.observability_public_url_prefix.strip("/") public_url = f"{public_base_url}{public_url_prefix}/{quote(request_id)}/{quote(safe_filename)}" sha256 = hashlib.sha256(image_content).hexdigest() result = { "saved_path": str(trace_path), "public_saved_path": str(public_path), "public_url": public_url, "sha256": sha256, "decode_ok": False, "decode_error": "", "width": None, "height": None, } try: trace_dir.mkdir(parents=True, exist_ok=True) trace_path.write_bytes(image_content) public_dir.mkdir(parents=True, exist_ok=True) public_path.write_bytes(image_content) with Image.open(io.BytesIO(image_content)) as image: image = ImageOps.exif_transpose(image) image.load() result.update( decode_ok=True, width=image.width, height=image.height, ) logger.info( "insert_menu_items_observability_upload_saved", request_id=request_id, saved_path=str(trace_path), public_saved_path=str(public_path), public_url=public_url, image_size_bytes=len(image_content), sha256=sha256, mime_type=mime_type, decode_ok=True, width=result["width"], height=result["height"], ) except Exception as e: result["decode_error"] = str(e) logger.warning( "insert_menu_items_observability_upload_trace_failed", request_id=request_id, saved_path=str(trace_path), image_size_bytes=len(image_content), sha256=sha256, mime_type=mime_type, error=str(e), error_type=type(e).__name__, ) return result