import os import json import logging from typing import Dict, Any, List, Optional from openai import AzureOpenAI from sqlalchemy.orm import Session from sqlalchemy import func, and_ from datetime import datetime, timezone from src.marketing.apps.post.model import CalendarPost, CalendarPostType from src.marketing.apps.persona.model import StorePersona from src.marketing.apps.persona import schema logger = logging.getLogger(__name__) class AzureLLMPersonaService: """Service for generating store personas using Azure OpenAI.""" 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", "gpt0") def generate_persona_for_store( self, db: Session, store_id: int, branch_id: Optional[int] = None ) -> schema.PersonaGenerationResponse: """Generate persona for a store based on post data.""" try: # Check if persona already exists existing_persona = self._get_existing_persona(db, store_id, branch_id) if existing_persona: return schema.PersonaGenerationResponse( success=True, message="Persona already exists", data=self._build_persona_response(existing_persona) ) # Get post data for analysis post_data = self._get_store_post_data(db, store_id, branch_id) if not post_data: return schema.PersonaGenerationResponse( success=False, message="No post data found for analysis" ) # Check if we have content to analyze if not post_data.get('post_content'): return schema.PersonaGenerationResponse( success=False, message="No post content found for analysis. Please ensure posts have content in CalendarPostType." ) # Generate persona using LLM persona_data = self._analyze_with_llm(post_data) # Create persona record persona = self._create_persona(db, store_id, branch_id, persona_data) return schema.PersonaGenerationResponse( success=True, message="Persona generated successfully", data=self._build_persona_response(persona), analysis_metadata=persona_data ) except Exception as e: logger.error(f"Error generating persona for store {store_id}: {str(e)}") return schema.PersonaGenerationResponse( success=False, message=f"Error generating persona: {str(e)}" ) def _get_existing_persona( self, db: Session, store_id: int, branch_id: Optional[int] = None ) -> Optional[StorePersona]: """Get existing persona for store/branch.""" try: query = db.query(StorePersona).filter(StorePersona.store_id == store_id) if branch_id: query = query.filter(StorePersona.branch_id == branch_id) else: query = query.filter(StorePersona.branch_id.is_(None)) return query.filter(StorePersona.is_active == True).first() except Exception as e: logger.warning(f"Error querying existing persona (schema issue?): {str(e)}") # Try without the is_active filter as fallback try: query = db.query(StorePersona).filter(StorePersona.store_id == store_id) if branch_id: query = query.filter(StorePersona.branch_id == branch_id) else: query = query.filter(StorePersona.branch_id.is_(None)) return query.first() except Exception as e2: logger.error(f"Fallback query also failed: {str(e2)}") return None def _get_store_post_data(self, db: Session, store_id: int, branch_id: Optional[int] = None) -> Dict[str, Any]: """Get post data for LLM analysis.""" query = db.query(CalendarPost, CalendarPostType).join(CalendarPostType).filter( CalendarPost.store_id == store_id, CalendarPost.is_deleted == False, CalendarPostType.is_deleted == False ) if branch_id: query = query.filter(CalendarPost.branch_id == branch_id) posts = query.limit(50).all() # Limit to recent 50 posts if not posts: return {} # Check if we have any content to analyze has_content = any(post_type.content for _, post_type in posts) if not has_content: logger.warning(f"No content found in posts for store {store_id}") return { 'post_content': [], 'post_types': [], 'engagement_data': [], 'total_posts': len(posts), 'store_id': store_id, 'branch_id': branch_id, 'post_titles': [], 'post_dates': [] } logger.info(f"Found {len(posts)} posts for store {store_id}, {len([p for _, p in posts if p.content])} with content") # Extract relevant data post_content = [] post_types = [] engagement_data = [] for post, post_type in posts: if post_type.content: post_content.append(post_type.content[:200]) # Limit content length if post_type.post_type: post_types.append(post_type.post_type) # Basic engagement metrics - CalendarPost doesn't have likes, so we'll use other metrics if post_type.content: engagement_data.append({ 'content': post_type.content[:100], 'post_type': post_type.post_type, 'title': post.title if post.title else 'No Title' }) result_data = { 'post_content': post_content[:20], # Limit to 20 posts for analysis 'post_types': list(set(post_types)), # Unique post types 'engagement_data': engagement_data[:10], # Top 10 engaging posts 'total_posts': len(posts), 'store_id': store_id, 'branch_id': branch_id, 'post_titles': [post.title for post, _ in posts if post.title][:10], # Post titles 'post_dates': [post.schedule_date for post, _ in posts if post.schedule_date][:10] # Post dates } logger.info(f"Post data summary: {len(post_content)} content items, {len(set(post_types))} post types") logger.debug(f"Sample content: {post_content[:2] if post_content else 'None'}") return result_data def _analyze_with_llm(self, post_data: Dict[str, Any]) -> Dict[str, Any]: """Analyze post data using Azure OpenAI to generate persona.""" try: # Create analysis prompt prompt = self._create_analysis_prompt(post_data) # Call Azure OpenAI response = self.client.chat.completions.create( model=self.deployment_name, messages=[ { "role": "system", "content": "You are an expert brand analyst. Analyze the provided social media post data and identify the store's brand persona. Be concise and specific." }, { "role": "user", "content": prompt } ], temperature=0.3, max_tokens=500 ) # Parse response analysis_text = response.choices[0].message.content persona_data = self._parse_llm_response(analysis_text) return { 'analysis_text': analysis_text, 'persona_data': persona_data, 'confidence_score': 0.85 # Default confidence } except Exception as e: logger.error(f"Error in LLM analysis: {str(e)}") # Return default persona data return { 'analysis_text': "Analysis failed, using default persona", 'persona_data': { 'industry': 'General Retail', 'audience_type': 'General Consumers', 'brand_voice': 'Professional', 'content_style': 'Informative' }, 'confidence_score': 0.5 } def _create_analysis_prompt(self, post_data: Dict[str, Any]) -> str: """Create the analysis prompt for the LLM.""" prompt = f""" Analyze this social media data and identify the store's brand persona: Store ID: {post_data['store_id']} Total Posts: {post_data['total_posts']} Post Types: {', '.join(post_data['post_types'])} Sample Post Content: {chr(10).join(post_data['post_content'][:5])} Post Titles (if available): {chr(10).join(post_data.get('post_titles', [])[:5])} Post Schedule Dates: {chr(10).join([str(date) for date in post_data.get('post_dates', [])[:5]])} Based on this data, identify: 1. Industry: What industry does this store operate in? 2. Audience Type: Who is their target audience? 3. Brand Voice: What is their communication style? 4. Content Style: How do they present their content? Respond in this exact JSON format: {{ "industry": "industry name", "audience_type": "audience description", "brand_voice": "voice description", "content_style": "style description" }} """ return prompt def _parse_llm_response(self, response_text: str) -> Dict[str, str]: """Parse LLM response to extract persona data.""" try: # Try to extract JSON from response if '{' in response_text and '}' in response_text: start = response_text.find('{') end = response_text.rfind('}') + 1 json_str = response_text[start:end] data = json.loads(json_str) return { 'industry': data.get('industry', 'General'), 'audience_type': data.get('audience_type', 'General Consumers'), 'brand_voice': data.get('brand_voice', 'Professional'), 'content_style': data.get('content_style', 'Informative') } except Exception as e: logger.error(f"Error parsing LLM response: {str(e)}") # Return default values if parsing fails return { 'industry': 'General Retail', 'audience_type': 'General Consumers', 'brand_voice': 'Professional', 'content_style': 'Informative' } def _create_persona( self, db: Session, store_id: int, branch_id: Optional[int], persona_data: Dict[str, Any] ) -> StorePersona: """Create new persona record.""" persona = StorePersona( store_id=store_id, branch_id=branch_id, industry=persona_data['persona_data']['industry'], audience_type=persona_data['persona_data']['audience_type'], brand_voice=persona_data['persona_data']['brand_voice'], content_style=persona_data['persona_data']['content_style'], analysis_summary=persona_data['analysis_text'], confidence_score=persona_data['confidence_score'], last_analyzed=datetime.now(timezone.utc) ) db.add(persona) db.commit() db.refresh(persona) return persona def _build_persona_response(self, persona: StorePersona) -> schema.StorePersonaResponse: """Build response from database model.""" return schema.StorePersonaResponse( id=persona.id, store_id=persona.store_id, branch_id=persona.branch_id, industry=persona.industry, audience_type=persona.audience_type, brand_voice=persona.brand_voice, content_style=persona.content_style, analysis_summary=persona.analysis_summary, confidence_score=persona.confidence_score, last_analyzed=persona.last_analyzed, created_at=persona.created_at, updated_at=persona.updated_at, is_active=persona.is_active )