""" Image Description Service This service generates detailed AI-powered descriptions of creative assets for focus group research. It helps distinguish between multiple images in the conversation context. """ import os import logging from typing import Optional from PIL import Image from app.services.llm_service import LLMService, LLMServiceError from app.services.conversation_context_service import ConversationContextService from app.utils.prompt_loader import load_prompt, PromptLoaderError logger = logging.getLogger(__name__) class ImageDescriptionError(Exception): """Exception raised for errors in image description generation.""" pass class ImageDescriptionService: """Service for generating AI-powered descriptions of creative assets.""" IMAGE_EXTENSIONS = {'jpg', 'jpeg', 'png', 'gif', 'webp', 'bmp', 'tiff'} TEXT_EXTENSIONS = {'txt', 'md', 'csv', 'rtf'} @staticmethod async def generate_description(focus_group_id: str, asset_filename: str) -> str: """ Generate a detailed AI description of a creative asset (image or document). For images: uses multimodal LLM. For text documents: reads content and summarizes with LLM. For other documents (pdf, docx, xlsx): returns a descriptive label. """ try: print(f"🎨 DESCRIPTION: Generating AI description for {asset_filename}") # Resolve the full path to the asset asset_path = ConversationContextService._resolve_asset_path(focus_group_id, asset_filename) print(f"🔍 DESCRIPTION: Resolved asset path: {asset_path}") # Check if file exists if not os.path.exists(asset_path): print(f"❌ DESCRIPTION: File does not exist at path: {asset_path}") raise ImageDescriptionError(f"Asset file not found: {asset_path}") ext = asset_filename.rsplit('.', 1)[-1].lower() if '.' in asset_filename else '' # ── Non-image: text files ────────────────────────────────────────────── if ext in ImageDescriptionService.TEXT_EXTENSIONS: try: with open(asset_path, 'r', encoding='utf-8', errors='replace') as f: content = f.read(8000) # first 8k chars from app.models.focus_group import FocusGroup focus_group = await FocusGroup.find_by_id(focus_group_id) llm_model = focus_group.get('llm_model') if focus_group else None summary = await LLMService.generate_content( prompt=f"Summarize this document in 2-3 sentences for focus group context:\n\n{content}", temperature=0.3, model_name=llm_model ) return summary.strip() except Exception as e: return f"Text document: {asset_filename}" # ── Non-image: binary documents (pdf, docx, xlsx, etc.) ─────────────── if ext not in ImageDescriptionService.IMAGE_EXTENSIONS: ext_label = ext.upper() if ext else 'Document' original = asset_filename.split('-')[-1] if '-' in asset_filename else asset_filename return f"{ext_label} document: {original}" # ── Image: validate with PIL ────────────────────────────────────────── try: image = Image.open(asset_path) print(f"🖼️ DESCRIPTION: Validated image {asset_filename} ({image.size[0]}x{image.size[1]})") image.close() except Exception as e: raise ImageDescriptionError(f"Failed to validate image {asset_filename}: {str(e)}") # Load the description prompt try: prompt = load_prompt('image-description', {}) print(f"📝 DESCRIPTION: Loaded description prompt ({len(prompt)} chars)") except PromptLoaderError as e: raise ImageDescriptionError(f"Failed to load description prompt: {str(e)}") # Generate description using multimodal LLM try: print(f"🚀 DESCRIPTION: Calling LLM service with image: {asset_path}") # Get LLM model for this focus group from app.models.focus_group import FocusGroup focus_group = await FocusGroup.find_by_id(focus_group_id) llm_model = focus_group.get('llm_model') if focus_group else None description = await LLMService.generate_multimodal_content( prompt=prompt, image_paths=[asset_path], temperature=0.7, model_name=llm_model ) print(f"✅ DESCRIPTION: LLM returned description ({len(description)} chars): {description[:100]}...") return description.strip() except LLMServiceError as e: raise ImageDescriptionError(f"LLM description generation failed: {str(e)}") except Exception as e: print(f"⚠️ DESCRIPTION: LLM service failed, generating fallback description: {str(e)}") # Generate fallback description based on image info try: # We already validated the image exists earlier, so we can safely open it image = Image.open(asset_path) width, height = image.size image.close() # Create a simple but useful fallback description fallback_description = f"a {width}x{height} pixel marketing advertisement image" print(f"✅ DESCRIPTION: Generated fallback description: '{fallback_description}'") return fallback_description except Exception as fallback_error: error_msg = f"Failed to generate description for {asset_filename}: LLM failed ({str(e)}) and fallback failed ({str(fallback_error)})" print(f"❌ DESCRIPTION: {error_msg}") raise ImageDescriptionError(error_msg) @staticmethod def enhance_creative_review_question(original_question: str, asset_filename: str, description: str) -> str: """ Enhance a creative review question by incorporating AI-generated image description. Args: original_question: The original question text asset_filename: The asset filename being referenced description: The AI-generated description of the image Returns: Enhanced question text with detailed visual description """ try: import re print(f"🔧 ENHANCEMENT: Enhancing question for {asset_filename}") print(f"🔧 Original: {original_question[:100]}...") print(f"🔧 Description: {description[:100]}...") # Use regex patterns to handle punctuation and variations # Escape the filename for regex use escaped_filename = re.escape(asset_filename) # Define comprehensive patterns that handle punctuation after filenames regex_patterns = [ # Quoted filenames with optional punctuation (rf"('{escaped_filename}')([.,;!?]*)", rf"\1 - {description}\2"), (rf'("{escaped_filename}")([.,;!?]*)', rf'\1 - {description}\2'), # Titled/labeled patterns (rf"(titled\s+['\"]?{escaped_filename}['\"]?)([.,;!?]*)", rf"\1 - {description}\2"), (rf"(asset\s+['\"]?{escaped_filename}['\"]?)([.,;!?]*)", rf"\1 - {description}\2"), (rf"(image\s+['\"]?{escaped_filename}['\"]?)([.,;!?]*)", rf"\1 - {description}\2"), # Unquoted filename with word boundaries (rf"\b({escaped_filename})\b([.,;!?]*)", rf"\1 - {description}\2") ] enhanced_question = original_question enhancement_applied = False # Try each regex pattern for pattern, replacement in regex_patterns: if re.search(pattern, enhanced_question, re.IGNORECASE): enhanced_question = re.sub(pattern, replacement, enhanced_question, flags=re.IGNORECASE) print(f"✅ ENHANCEMENT: Enhanced with regex pattern: {enhanced_question[:150]}...") enhancement_applied = True break # If no regex patterns worked, try simple string replacement as fallback if not enhancement_applied and asset_filename in original_question: # Simple replacement that adds description after any occurrence of filename enhanced_question = original_question.replace( asset_filename, f"{asset_filename} - {description}" ) print(f"✅ ENHANCEMENT: Enhanced with simple replacement: {enhanced_question[:150]}...") enhancement_applied = True # Final fallback: append description if no enhancements worked if not enhancement_applied: enhanced_question = f"{original_question} The image shows {description}." print(f"⚠️ ENHANCEMENT: Appended description to end: {enhanced_question[:150]}...") return enhanced_question except Exception as e: error_msg = f"Failed to enhance question for {asset_filename}: {str(e)}" print(f"❌ ENHANCEMENT: {error_msg}") # Return original question if enhancement fails return original_question @staticmethod def enhance_creative_review_question_with_display_reference(original_question: str, display_reference: str, description: str) -> str: """ Enhance a creative review question by incorporating AI-generated image description using display reference. This is the new metadata-driven approach that doesn't rely on filename parsing. Args: original_question: The original question text display_reference: The display reference (e.g., "Asset 1", "My Campaign Ad") description: The AI-generated description of the image Returns: Enhanced question text with detailed visual description """ try: import re print(f"🔧 ENHANCEMENT: Enhancing question for display reference: {display_reference}") print(f"🔧 Original: {original_question[:100]}...") print(f"🔧 Description: {description[:100]}...") # Use regex patterns to handle punctuation and variations # Escape the display reference for regex use escaped_reference = re.escape(display_reference) # Define patterns that match display references in various contexts regex_patterns = [ # Quoted display references with optional punctuation (rf"('{escaped_reference}')([.,;!?]*)", rf"\1 - {description}\2"), (rf'("{escaped_reference}")([.,;!?]*)', rf'\1 - {description}\2'), # Common phrases with display references (rf"(review\s+{escaped_reference})([.,;!?]*)", rf"\1 - {description}\2"), (rf"(look at\s+{escaped_reference})([.,;!?]*)", rf"\1 - {description}\2"), (rf"(consider\s+{escaped_reference})([.,;!?]*)", rf"\1 - {description}\2"), (rf"(examine\s+{escaped_reference})([.,;!?]*)", rf"\1 - {description}\2"), # Direct display reference with word boundaries (rf"\b({escaped_reference})\b([.,;!?]*)", rf"\1 - {description}\2") ] enhanced_question = original_question enhancement_applied = False # Try each regex pattern for pattern, replacement in regex_patterns: if re.search(pattern, enhanced_question, re.IGNORECASE): enhanced_question = re.sub(pattern, replacement, enhanced_question, flags=re.IGNORECASE) print(f"✅ ENHANCEMENT: Enhanced with regex pattern: {enhanced_question[:150]}...") enhancement_applied = True break # If no regex patterns worked, try simple string replacement as fallback if not enhancement_applied and display_reference in original_question: enhanced_question = enhanced_question.replace(display_reference, f"{display_reference} - {description}") print(f"✅ ENHANCEMENT: Enhanced with simple replacement: {enhanced_question[:150]}...") enhancement_applied = True # Final fallback: append description if no enhancements worked if not enhancement_applied: enhanced_question = f"{original_question} The {display_reference.lower()} shows {description}." print(f"⚠️ ENHANCEMENT: Appended description to end: {enhanced_question[:150]}...") return enhanced_question except Exception as e: error_msg = f"Failed to enhance question for {display_reference}: {str(e)}" print(f"❌ ENHANCEMENT: {error_msg}") # Return original question if enhancement fails return original_question