""" Sentiment analysis module using Grok 4 (xAI API) """ import re import logging from typing import List, Dict, Tuple import requests from .config import settings logger = logging.getLogger(__name__) class GrokAnalyzer: """Sentiment analyzer using Grok 4 API.""" def __init__(self): """Initialize Grok analyzer.""" self.api_key = settings.xai_api_key self.api_url = settings.xai_api_url self.model = settings.xai_model self.use_mock = not self.api_key or self.api_key == "SEU_API_KEY_AQUI" if self.use_mock: logger.warning("Grok API key not configured, using mock analysis") def _call_grok_api(self, prompt: str, max_tokens: int = 500) -> str: """ Call Grok API with prompt. Args: prompt: User prompt max_tokens: Maximum tokens to generate Returns: API response content Raises: requests.exceptions.RequestException: On API error """ headers = { 'Authorization': f'Bearer {self.api_key}', 'Content-Type': 'application/json' } data = { 'model': self.model, 'messages': [{'role': 'user', 'content': prompt}], 'max_tokens': max_tokens, 'temperature': 0.3 # More deterministic } response = requests.post( self.api_url, json=data, headers=headers, timeout=30 ) response.raise_for_status() return response.json()['choices'][0]['message']['content'] def _mock_analysis(self, items: List[Dict]) -> Tuple[Dict, List[Dict]]: """ Mock sentiment analysis using keyword matching. Args: items: List of alert items Returns: Tuple of (sentiment counts, analyzed items) """ logger.info("Using mock sentiment analysis") negative_keywords = [ 'problema', 'crise', 'reclamação', 'processo', 'fraude', 'falha', 'erro', 'bloqueio', 'bloqueado', 'golpe', 'atraso', 'insatisfação', 'dificuldade', 'cancelamento' ] positive_keywords = [ 'crescimento', 'prêmio', 'expansão', 'sucesso', 'inovação', 'liderança', 'melhoria', 'excelência' ] sentiments = {'neg': 0, 'pos': 0, 'neu': 0} analyzed_items = [] for item in items: text = (item['title'] + ' ' + item['source']).lower() neg_score = sum(1 for kw in negative_keywords if kw in text) pos_score = sum(1 for kw in positive_keywords if kw in text) # Priority for Reclame Aqui if item.get('is_reclameaqui', False): sentiment = 'neg' reason = "Reclame Aqui sempre indica problema" sentiments['neg'] += 1 elif neg_score > pos_score: sentiment = 'neg' sentiments['neg'] += 1 reason = f"Detectadas {neg_score} palavras negativas" elif pos_score > neg_score: sentiment = 'pos' sentiments['pos'] += 1 reason = f"Detectadas {pos_score} palavras positivas" else: sentiment = 'neu' sentiments['neu'] += 1 reason = "Sem indicadores claros" analyzed_items.append({ **item, 'sentiment': sentiment, 'reason': reason }) return sentiments, analyzed_items def analyze_batch(self, items: List[Dict], batch_size: int = 10) -> Tuple[Dict, List[Dict]]: """ Analyze sentiment of alert items in batches. Args: items: List of alert items batch_size: Number of items per batch Returns: Tuple of (sentiment counts, analyzed items with sentiment) """ if self.use_mock: return self._mock_analysis(items) logger.info(f"Analyzing {len(items)} items with Grok API") sentiments = {'neg': 0, 'pos': 0, 'neu': 0} analyzed_items = [] # Process in batches for i in range(0, len(items), batch_size): batch = items[i:i+batch_size] logger.info(f"Processing batch {i//batch_size + 1}/{(len(items)-1)//batch_size + 1}") # Build prompt prompts = [] for idx, item in enumerate(batch): prompts.append(f""" [ITEM {idx+1}] Título: {item['title']} Fonte: {item['source']} Link: {item['link'][:80]}... """) full_prompt = f""" Analise o sentimento de cada notícia sobre "{settings.search_query}" em relação à reputação/imagem da empresa. Critérios: - NEGATIVO: Crise, reclamações, processos, problemas, críticas, Reclame Aqui - POSITIVO: Crescimento, prêmios, melhorias, elogios - NEUTRO: Fatos, dados, informações sem viés Para cada item, responda APENAS no formato: ITEM X: NEGATIVO/POSITIVO/NEUTRO | Razão em 1 frase. {"".join(prompts)} Responda de forma objetiva e direta. """ try: result = self._call_grok_api(full_prompt, max_tokens=1000) # Parse responses for idx, line in enumerate(result.split('\n')): if not line.strip(): continue item_idx = i + idx if item_idx >= len(items): break sentiment = 'neu' reason = '' if 'NEGATIVO' in line.upper(): sentiment = 'neg' sentiments['neg'] += 1 elif 'POSITIVO' in line.upper(): sentiment = 'pos' sentiments['pos'] += 1 else: sentiments['neu'] += 1 # Extract reason if '|' in line: reason = line.split('|')[1].strip() analyzed_items.append({ **items[item_idx], 'sentiment': sentiment, 'reason': reason }) logger.info(f"Batch analyzed: {sentiments}") except requests.exceptions.Timeout: logger.error("Grok API timeout, falling back to mock") return self._mock_analysis(items) except requests.exceptions.RequestException as e: logger.error(f"Grok API error: {e}, falling back to mock") return self._mock_analysis(items) except Exception as e: logger.error(f"Unexpected error: {e}, falling back to mock") return self._mock_analysis(items) total = len(items) logger.info(f"Analysis complete: {sentiments} out of {total} items") return sentiments, analyzed_items def get_negative_items(self, analyzed_items: List[Dict], limit: int = 10) -> List[Dict]: """ Get top negative items sorted by priority. Args: analyzed_items: List of analyzed items limit: Maximum number of items to return Returns: List of negative items (Reclame Aqui first) """ negative_items = [ item for item in analyzed_items if item.get('sentiment') == 'neg' ] # Sort: Reclame Aqui first, then by date negative_items.sort( key=lambda x: ( 0 if x.get('is_reclameaqui', False) else 1, x.get('date', '') ), reverse=True ) return negative_items[:limit]