257 lines
No EOL
13 KiB
Python
Executable file
257 lines
No EOL
13 KiB
Python
Executable file
"""
|
|
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."""
|
|
|
|
@staticmethod
|
|
async def generate_description(focus_group_id: str, asset_filename: str) -> str:
|
|
"""
|
|
Generate a detailed AI description of a creative asset image.
|
|
|
|
Args:
|
|
focus_group_id: The focus group ID containing the asset
|
|
asset_filename: The filename of the asset to describe
|
|
|
|
Returns:
|
|
A detailed description of the image
|
|
|
|
Raises:
|
|
ImageDescriptionError: If description generation fails
|
|
"""
|
|
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}")
|
|
# List files in the directory to help debug
|
|
asset_dir = os.path.dirname(asset_path)
|
|
if os.path.exists(asset_dir):
|
|
files_in_dir = os.listdir(asset_dir)
|
|
print(f"🔍 DESCRIPTION: Files in directory {asset_dir}: {files_in_dir}")
|
|
else:
|
|
print(f"❌ DESCRIPTION: Directory does not exist: {asset_dir}")
|
|
raise ImageDescriptionError(f"Asset file not found: {asset_path}")
|
|
|
|
# Verify the image can be loaded (optional validation)
|
|
try:
|
|
image = Image.open(asset_path)
|
|
print(f"🖼️ DESCRIPTION: Validated image {asset_filename} ({image.size[0]}x{image.size[1]})")
|
|
image.close() # Close the image since we're passing the path to LLM
|
|
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 |