import os import json import logging import time import requests from typing import Dict, Any, Optional, List from datetime import datetime, timezone from openai import AzureOpenAI from sqlalchemy.orm import Session from PIL import Image from io import BytesIO from src.marketing.apps.hwGpt.model import ChatThread, ChatMessage from src.marketing.apps.persona.model import StorePersona logger = logging.getLogger(__name__) class ChatbotLLMService: """LLM service for the hwGpt chatbot with store persona integration and image generation.""" def __init__(self): self.client = AzureOpenAI( api_key=os.getenv("AZURE_OPENAI_API_KEY"), api_version=os.getenv("AZURE_OPENAI_API_VERSION", "2024-02-15-preview"), azure_endpoint=os.getenv("AZURE_OPENAI_ENDPOINT") ) self.deployment_name = os.getenv("AZURE_OPENAI_DEPLOYMENT", "gpt-4") # Image generation settings self.IDEOGRAM_URL = os.getenv("IDEOGRAM_URL") self.IDEOGRAM_API_KEY = os.getenv("IDEOGRAM_API_KEY") self.BASE_URL = os.getenv("BASE_URL", "https://hubwallet-dev.dreamztesting.com/") # Get the root directory of your project dynamically (same as menu design) BASE_DIR = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..", "..")) # Use the same static path as menu design module self.STATIC_PATH = os.path.join(BASE_DIR, "menu_design", "designer", "static", "generated_images") # Ensure the directory exists os.makedirs(self.STATIC_PATH, exist_ok=True) def get_store_persona_context(self, db: Session, store_id: int, branch_id: Optional[int] = None) -> str: """Get store persona context for LLM prompts.""" try: # Validate store_id parameter if not isinstance(store_id, int) or store_id <= 0: logger.error(f"Invalid store_id: {store_id}, type: {type(store_id)}") return "No specific store persona available. Please provide general assistance." # Validate branch_id parameter if provided if branch_id is not None and (not isinstance(branch_id, int) or branch_id <= 0): logger.error(f"Invalid branch_id: {branch_id}, type: {type(branch_id)}") branch_id = None # Reset to None if invalid # Query for store persona query = db.query(StorePersona).filter( StorePersona.store_id == store_id, StorePersona.is_active == True ) if branch_id: query = query.filter(StorePersona.branch_id == branch_id) persona = query.first() if not persona: return "No specific store persona available. Please provide general assistance." # Build persona context persona_context = f""" Store Brand Persona: - Industry: {persona.industry or 'Not specified'} - Target Audience: {persona.audience_type or 'Not specified'} - Brand Voice: {persona.brand_voice or 'Not specified'} - Content Style: {persona.content_style or 'Not specified'} - Analysis Summary: {persona.analysis_summary or 'Not specified'} - Confidence Score: {persona.confidence_score or 'Not specified'} Based on this persona, you should: 1. Match the brand voice and tone 2. Use industry-specific terminology when appropriate 3. Tailor responses to the target audience 4. Maintain the established content style 5. Provide contextually relevant information """ return persona_context.strip() except Exception as e: logger.error(f"Error getting store persona: {str(e)}") return "Unable to retrieve store persona. Please provide general assistance." def build_conversation_context( self, db: Session, thread_id: int, max_messages: int = 10 ) -> List[Dict[str, str]]: """Build conversation context from thread history.""" try: # Validate thread_id is a valid integer if not isinstance(thread_id, int) or thread_id <= 0: logger.warning(f"Invalid thread_id: {thread_id}, returning empty context") return [] # Get recent messages from the thread messages = db.query(ChatMessage).filter( ChatMessage.thread_id == thread_id, ChatMessage.is_deleted == False ).order_by(ChatMessage.created_at.desc()).limit(max_messages).all() # Reverse to get chronological order messages.reverse() conversation_context = [] for message in messages: conversation_context.append({ "role": message.role, "content": message.content }) return conversation_context except Exception as e: logger.error(f"Error building conversation context: {str(e)}") return [] def create_system_prompt(self, store_persona: str, thread_context: Optional[str] = None) -> str: """Create system prompt with store persona and marketing focus.""" system_prompt = f"""You are a marketing-focused AI assistant for a business store. {store_persona} Your role is to: 1. **Help with marketing strategies and campaigns** 2. **Provide content creation assistance for social media, emails, and ads** 3. **Analyze customer engagement and suggest improvements** 4. **Help with brand messaging and positioning** 5. **Assist with marketing analytics interpretation** 6. **Provide creative ideas for promotional activities** 7. **Help optimize marketing performance and ROI** Marketing Guidelines: - Focus on **customer acquisition and retention** - Suggest **promotional strategies and campaigns** - Help with **content marketing and social media** - Provide **marketing analytics insights** - Assist with **brand development and messaging** - Suggest **lead generation and conversion tactics** - Help with **marketing automation and workflows** Remember: You are a **marketing assistant** - focus on helping with marketing activities, campaigns, content creation, and strategy. Keep responses marketing-focused and business-oriented. {thread_context if thread_context else ""} Stay within marketing scope: social media, advertising, content creation, customer engagement, brand development, and marketing analytics.""" return system_prompt.strip() def generate_response( self, user_message: str, store_persona: str, conversation_history: List[Dict[str, str]], thread_context: Optional[str] = None ) -> Dict[str, Any]: """Generate AI response using Azure OpenAI.""" try: # Create system prompt system_prompt = self.create_system_prompt(store_persona, thread_context) # Prepare messages for OpenAI messages = [{"role": "system", "content": system_prompt}] # Add conversation history (limit to last 10 messages to avoid token limits) for msg in conversation_history[-10:]: messages.append({"role": msg["role"], "content": msg["content"]}) # Add current user message messages.append({"role": "user", "content": user_message}) logger.info(f"Generating response with {len(messages)} context messages") # Call Azure OpenAI response = self.client.chat.completions.create( model=self.deployment_name, messages=messages, max_tokens=1000, temperature=0.7, top_p=0.9, frequency_penalty=0.1, presence_penalty=0.1 ) # Extract response ai_response = response.choices[0].message.content usage = response.usage logger.info(f"Generated response: {len(ai_response)} characters") logger.info(f"Tokens used: {usage.total_tokens}") return { "success": True, "response": ai_response, "tokens_used": usage.total_tokens, "model_used": self.deployment_name } except Exception as e: logger.error(f"Error generating AI response: {str(e)}") return { "success": False, "error": str(e), "response": "I apologize, but I'm experiencing technical difficulties. Please try again in a moment." } async def process_chat_message( self, db: Session, user_id: int, user_message: str, store_id: int, branch_id: Optional[int] = None, thread_id: Optional[int] = None, ) -> Dict[str, Any]: """Process a chat message and generate AI response.""" try: # Get store persona store_persona = self.get_store_persona_context(db, store_id, branch_id) print("store_persona", store_persona) # Get conversation history (only if thread_id is valid) conversation_history = [] thread_context = "General marketing assistance" if isinstance(thread_id, int) and thread_id > 0: try: conversation_history = self.build_conversation_context(db, thread_id) # Get thread context thread = db.query(ChatThread).filter(ChatThread.id == thread_id).first() if thread: thread_context = f"Current conversation topic: {thread.title or 'General chat'}" except Exception as e: logger.warning(f"Could not load thread context for thread_id {thread_id}: {str(e)}") conversation_history = [] thread_context = "General marketing assistance" print("conversation_history", conversation_history) print("thread_context", thread_context) # Generate AI response ai_response = self.generate_response( user_message, store_persona, conversation_history, thread_context ) if not ai_response["success"]: return ai_response # Save user message to database user_msg = ChatMessage( thread_id=thread_id, user_id=user_id, role="user", content=user_message, message_type="text" ) db.add(user_msg) db.flush() # Get the ID without committing # Save AI response to database ai_msg = ChatMessage( thread_id=thread_id, user_id=user_id, # Same user for now, could be system user ID role="assistant", content=ai_response["response"], message_type="text", tokens_used=ai_response.get("tokens_used"), model_used=ai_response.get("model_used") ) db.add(ai_msg) # Update thread timestamp thread.updated_at = datetime.now(timezone.utc) # Commit all changes db.commit() # Refresh objects to get IDs db.refresh(user_msg) db.refresh(ai_msg) logger.info(f"Processed chat message: User message ID {user_msg.id}, AI response ID {ai_msg.id}") return { "success": True, "user_message": { "id": user_msg.id, "content": user_msg.content, "role": user_msg.role, "created_at": user_msg.created_at.isoformat() }, "ai_response": { "id": ai_msg.id, "content": ai_msg.content, "role": ai_msg.role, "created_at": ai_msg.created_at.isoformat(), "tokens_used": ai_msg.tokens_used, "model_used": ai_msg.model_used }, "thread_id": thread_id, "store_persona_summary": store_persona[:200] + "..." if len(store_persona) > 200 else store_persona } except Exception as e: logger.error(f"Error processing chat message: {str(e)}") db.rollback() return { "success": False, "error": str(e), "response": "I apologize, but I encountered an error processing your message. Please try again." } def get_chat_suggestions( self, store_persona: str, context: str = "general" ) -> List[str]: """Generate marketing-focused chat suggestions.""" try: suggestions_prompt = f"""Based on this store persona: {store_persona} Generate 5 marketing-focused conversation starters that marketing teams or business owners might ask about. Context: {context} Focus on: - Marketing strategy questions - Content creation help - Campaign planning - Social media management - Customer engagement tactics - Brand development - Marketing analytics Return only the suggestions, one per line, without numbering or extra text.""" response = self.client.chat.completions.create( model=self.deployment_name, messages=[{"role": "user", "content": suggestions_prompt}], max_tokens=200, temperature=0.7 ) suggestions = response.choices[0].message.content.strip().split('\n') # Clean up suggestions suggestions = [s.strip() for s in suggestions if s.strip() and not s.startswith(('1.', '2.', '3.', '4.', '5.'))] return suggestions[:5] # Return max 5 suggestions except Exception as e: logger.error(f"Error generating chat suggestions: {str(e)}") return [ "How can I help you with your marketing strategy?", "What type of content should I create for your target audience?", "How can I improve my social media engagement?", "What marketing campaigns should I plan for this quarter?", "How can I optimize my marketing ROI?" ] def generate_marketing_image_prompt( self, user_message: str, store_persona: str, conversation_history: List[Dict[str, str]] ) -> str: """Generate a detailed image prompt for marketing content based on user input and store persona.""" try: # Build conversation context conversation_context = "" if conversation_history: recent_messages = conversation_history[-5:] # Last 5 messages conversation_context = "\n".join([ f"{msg['role']}: {msg['content']}" for msg in recent_messages ]) prompt = f""" Create a detailed image prompt for marketing content based on the following information: Store Persona: {store_persona} User Request: {user_message} Recent Conversation Context: {conversation_context} Requirements: - Create a professional marketing image suitable for social media, ads, or promotional materials - The image should align with the store's brand persona and target audience - Focus on visual appeal and marketing effectiveness - Include relevant product/service elements if mentioned - Use appropriate colors, style, and composition for the brand - Make it suitable for digital marketing use (social media, website, ads) IMPORTANT: - Do NOT include any text, logos, or written content in the image - Focus purely on visual elements and composition - Make it high-quality and professional - Ensure it's appropriate for marketing use Generate a detailed prompt (max 300 characters) that can be used to create this marketing image. """ response = self.client.chat.completions.create( model=self.deployment_name, messages=[ {"role": "system", "content": "You are an expert at creating detailed image prompts for marketing content. Focus on visual elements, composition, and brand alignment."}, {"role": "user", "content": prompt} ], max_tokens=300, temperature=0.7 ) image_prompt = response.choices[0].message.content.strip() # Add marketing-specific constraints image_prompt += " Professional marketing image, no text, no logos, high quality, suitable for social media and advertising." logger.info(f"Generated marketing image prompt: {image_prompt}") return image_prompt except Exception as e: logger.error(f"Error generating image prompt: {str(e)}") # Fallback prompt return f"Professional marketing image for {user_message}, no text, high quality, suitable for social media advertising" def generate_image_with_ideogram(self, image_prompt: str, session_id: str = "default") -> Dict[str, Any]: """Generate an image using Ideogram API and save it locally.""" try: # Generate unique filename with timestamp and session ID timestamp = int(time.time()) image_filename = f"hwgpt_marketing_{session_id}_{timestamp}.png" image_filepath = os.path.join(self.STATIC_PATH, image_filename) # Check if Ideogram API is configured if not self.IDEOGRAM_URL or not self.IDEOGRAM_API_KEY: return { "success": False, "error": "Ideogram API not configured", "image_url": None } # Step 1: Call Ideogram API response = requests.post( self.IDEOGRAM_URL, headers={ "Api-Key": self.IDEOGRAM_API_KEY, }, files={ "prompt": (None, image_prompt), 'aspect_ratio': (None, "1x1"), 'magic_prompt': (None, "ON"), 'negative_prompt': (None, "Text, text overlay, logos, brands, written content, words, letters, typography. No text allowed in the image. Strictly no text. No text in the image. No text in the image. No text in the image. No text in the image. No text in the image."), } ) logger.info(f"Received response from Ideogram API: {response.status_code}") # Step 2: Check response if response.status_code != 200: return { "success": False, "error": f"Ideogram API error: {response.text}", "image_url": None } response_data = response.json() if not response_data.get('data') or len(response_data['data']) == 0: return { "success": False, "error": "No image data returned by Ideogram API", "image_url": None } generation_url = response_data['data'][0].get('url') if not generation_url: return { "success": False, "error": "No image URL returned by Ideogram API", "image_url": None } # Step 3: Download image image_response = requests.get(generation_url, timeout=120) if image_response.status_code != 200: return { "success": False, "error": f"Failed to download image: {image_response.status_code}", "image_url": None } # Step 4: Save image locally img = Image.open(BytesIO(image_response.content)) img.save(image_filepath) logger.info(f"Image saved locally at: {image_filepath}") # Step 5: Create the public image URL (same as menu design) base_url = self.BASE_URL if self.BASE_URL.endswith('/') else self.BASE_URL + '/' image_url = f"{base_url}static/generated_images/{image_filename}" logger.info(f"Image URL: {image_url}") return { "success": True, "image_url": image_url, "local_path": image_filepath, "filename": image_filename, "error": None } except Exception as e: logger.error(f"Error generating image: {str(e)}") return { "success": False, "error": str(e), "image_url": None } async def process_image_generation_request( self, db: Session, user_id: int, user_message: str, store_id: int, branch_id: Optional[int] = None, thread_id: Optional[int] = None, session_id: str = "default" ) -> Dict[str, Any]: """Process an image generation request and return both text response and image.""" try: # Get store persona store_persona = self.get_store_persona_context(db, store_id, branch_id) # Get conversation history conversation_history = [] if isinstance(thread_id, int) and thread_id > 0: conversation_history = self.build_conversation_context(db, thread_id) # Generate image prompt image_prompt = self.generate_marketing_image_prompt( user_message, store_persona, conversation_history ) # Generate image image_result = self.generate_image_with_ideogram(image_prompt, session_id) # Generate text response about the image if image_result["success"]: text_response = f"I've generated a marketing image based on your request: '{user_message}'. The image has been created to align with your store's brand persona and is suitable for your marketing campaigns. You can use this image for social media posts, advertisements, or other promotional materials." else: text_response = f"I understand you'd like an image for: '{user_message}'. However, I encountered an issue generating the image: {image_result['error']}. Let me help you with other marketing strategies or try again later." # Save user message to database user_msg = ChatMessage( thread_id=thread_id, user_id=user_id, role="user", content=user_message, message_type="image_request" ) db.add(user_msg) db.flush() # Save AI response to database ai_msg = ChatMessage( thread_id=thread_id, user_id=user_id, role="assistant", content=( f"{text_response}\n\n" f"{'![Generated Image](' + image_result['image_url'] + ')' if image_result.get('image_url') else ''}" ), message_type="image_response", image_url=image_result.get('image_url') if image_result.get('success') else None ) db.add(ai_msg) # Update thread timestamp if thread exists if thread_id: thread = db.query(ChatThread).filter(ChatThread.id == thread_id).first() if thread: thread.updated_at = datetime.now(timezone.utc) # Commit all changes db.commit() # Refresh objects to get IDs db.refresh(user_msg) db.refresh(ai_msg) logger.info(f"Processed image generation request: User message ID {user_msg.id}, AI response ID {ai_msg.id}") return { "success": True, "user_message": { "id": user_msg.id, "content": user_msg.content, "role": user_msg.role, "message_type": user_msg.message_type, "created_at": user_msg.created_at.isoformat() }, "ai_response": { "id": ai_msg.id, "content": ai_msg.content, "role": ai_msg.role, "message_type": ai_msg.message_type, "created_at": ai_msg.created_at.isoformat() }, "image_generation": image_result, "image_prompt": image_prompt, "thread_id": thread_id, "store_persona_summary": store_persona[:200] + "..." if len(store_persona) > 200 else store_persona } except Exception as e: logger.error(f"Error processing image generation request: {str(e)}") db.rollback() return { "success": False, "error": str(e), "response": "I apologize, but I encountered an error processing your image generation request. Please try again." }