# type: ignore """ ================================================================================ AKIRA V21 ULTIMATE - CONTEXTO MODULE ================================================================================ Gerenciador de contexto de conversa com NLP avançado, análise emocional, aprendizado dinâmico de gírias e adaptação de tom por usuário. Features: - Análise de intenção e normalização de texto - Detecção de emoções com fallback heurístico - Aprendizado de gírias regionais (Angola) - Histórico de conversa persistente - Tom adaptativo por usuário - Integração com EmotionAnalyzer do config - Sistema de embeddings para similaridade - Cache inteligente - Logging detalhado ================================================================================ """ import logging import re import random import time import sqlite3 import json from typing import Optional, List, Dict, Tuple, Any, Union from datetime import datetime # Imports robustos com fallback - CORRIGIDO para usar modules. try: import modules.config as config from .database import Database from .treinamento import Treinamento CONTEXTO_AVAILABLE = True except ImportError: try: from . import config from .database import Database from .treinamento import Treinamento CONTEXTO_AVAILABLE = True except ImportError: CONTEXTO_AVAILABLE = False import modules.config as config try: from modules.database import Database from modules.treinamento import Treinamento except ImportError: Database = None Treinamento = None # Imports opcionais com fallbacks try: from sentence_transformers import SentenceTransformer # type: ignore SENTENCE_TRANSFORMER_AVAILABLE = True except Exception as e: logging.warning(f"sentence_transformers não disponível: {e}") SentenceTransformer = None # type: ignore SENTENCE_TRANSFORMER_AVAILABLE = False try: import psutil # type: ignore PSUTIL_AVAILABLE = True except Exception: psutil = None # type: ignore PSUTIL_AVAILABLE = False try: import structlog # type: ignore STRUCTLOG_AVAILABLE = True except Exception: structlog = None # type: ignore STRUCTLOG_AVAILABLE = False logger = logging.getLogger(__name__) # Configuração do logging logging.basicConfig(level=logging.INFO, format='%(asctime)s [%(levelname)s] %(message)s') if STRUCTLOG_AVAILABLE and structlog: structlog.configure( processors=[ structlog.processors.TimeStamper(fmt="iso"), structlog.stdlib.add_log_level, structlog.processors.JSONRenderer() ], context_class=dict, logger_factory=structlog.stdlib.LoggerFactory(), wrapper_class=structlog.stdlib.BoundLogger, ) # Palavras para análise de sentimento heurística (fallback) PALAVRAS_POSITIVAS = [ 'bom', 'ótimo', 'incrível', 'feliz', 'adorei', 'top', 'fixe', 'bué', 'show', 'legal', 'bacana', 'excelente', 'maravilhoso', 'perfeito' ] PALAVRAS_NEGATIVAS = [ 'ruim', 'péssimo', 'triste', 'ódio', 'raiva', 'chateado', 'merda', 'porra', 'odeio', 'horrível', 'terrible', 'p不佳' ] # Cache global para emotion analyzer - usa singleton do config _emotion_analyzer: Any = None def _get_emotion_analyzer() -> Any: """ Obtém instância do EmotionAnalyzer do config.py. Usa fallback heurístico interno se necessário. """ global _emotion_analyzer if _emotion_analyzer is None: try: # Usa o singleton do config.py _emotion_analyzer = config.get_emotion_analyzer() if _emotion_analyzer is not None: logger.info("EmotionAnalyzer obtido do config") else: # Fallback: cria analyzer dummy com heurística interna class FallbackAnalyzer: """Fallback analyzer com heurística básica.""" def analisar(self, texto, historico=None, nivel=None): return self._heuristica(texto) def _heuristica(self, texto): lower = texto.lower() # Detecção de emoção simples if any(w in lower for w in ['feliz', 'fixe', 'bom', 'top', 'adorei', 'amo', 'kkk', 'haha']): return {'emocao': 'joy', 'confianca': 0.8, 'nivel_analise': 'heuristica_fallback'} elif any(w in lower for w in ['triste', 'chateado', 'mal', 'péssimo', 'pessimo', '🥺', '💔']): return {'emocao': 'sadness', 'confianca': 0.8, 'nivel_analise': 'heuristica_fallback'} elif any(w in lower for w in ['raiva', 'odio', 'puta', 'caralho', 'merda', 'vsf', 'fdp', '🔥']): return {'emocao': 'anger', 'confianca': 0.8, 'nivel_analise': 'heuristica_fallback'} elif any(w in lower for w in ['medo', 'assustado', 'preocupado', '😨', '🥶']): return {'emocao': 'fear', 'confianca': 0.8, 'nivel_analise': 'heuristica_fallback'} elif any(w in lower for w in ['surpresa', 'nossa', 'eita', 'uau', '😮', '🤯']): return {'emocao': 'surprise', 'confianca': 0.8, 'nivel_analise': 'heuristica_fallback'} elif any(w in lower for w in ['amo', 'te amo', 'paixão', 'paixao', 'coração', 'coracao', '🥰', '❤️']): return {'emocao': 'love', 'confianca': 0.8, 'nivel_analise': 'heuristica_fallback'} else: return {'emocao': 'neutral', 'confianca': 0.5, 'nivel_analise': 'heuristica_fallback'} _emotion_analyzer = FallbackAnalyzer() logger.info("Usando FallbackAnalyzer para análise emocional") except Exception as e: logger.warning(f"Erro ao obter EmotionAnalyzer: {e}") _emotion_analyzer = None return _emotion_analyzer class Contexto: """ Classe para gerenciar o contexto da conversa, análise de intenções e aprendizado dinâmico de termos regionais/gírias para cada usuário. Attributes: db: Instância do banco de dados usuario: Identificador do usuário model: Modelo SentenceTransformer (carregado sob demanda) embeddings: Cache de embeddings emocao_atual: Emoção atual do usuário espirito_critico: Modo de espírito crítico ativado base_conhecimento: Base de conhecimento persistente termo_contexto: Dicionário de termos/gírias aprendidos cache_girias: Cache de gírias por usuário primeira_mensagem: Flag para detectar primeira interação tom_anterior: Tom da última mensagem para transição lenta contagem_mensagens_tom: Contador para transição gradual """ def __init__(self, db: Optional[Database] = None, usuario: Optional[str] = None, conversation_id: Optional[str] = None): """ Inicializa o contexto de conversa. Args: db: Instância do banco de dados Database usuario: Identificador do usuário (número de telefone ou nome) conversation_id: ID único da conversa para isolamento (opcional) """ self.usuario: Optional[str] = usuario self.conversation_id: Optional[str] = conversation_id # Inicializa db corretamente - se None, cria nova instância if db is not None: self.db = db else: try: from .database import Database self.db = Database() except ImportError: try: from modules.database import Database self.db = Database() except ImportError: self.db = None logger.warning("Database não disponível no contexto") self.model: Optional[Any] = None self.embeddings: Optional[Dict[str, Any]] = None self._treinador: Optional[Treinamento] = None # Estado de conversa self.emocao_atual: str = "neutra" self.espirito_critico: bool = False self.base_conhecimento: Dict[str, Any] = {} # Garante que termo_contexto seja sempre um dicionário self.termo_contexto: Dict[str, Dict[str, Any]] = {} self.cache_girias: Dict[str, Any] = {} # Novas flags para primeira mensagem e transição lenta de tom self.primeira_mensagem: bool = True self.tom_anterior: str = "neutro" self.contagem_mensagens_tom: int = 0 self.tom_atual: str = "neutro" # Carrega aprendizados do banco self.atualizar_aprendizados_do_banco() logger.info(f"🟢 Contexto inicializado para usuário: {usuario} (Isolation ID: {conversation_id})") # Carrega modelo sob demanda self._load_model() def atualizar_aprendizados_do_banco(self): """Carrega todos os dados de aprendizado persistentes do banco.""" try: if self.usuario and self.db is not None: termos_aprendidos = self.db.recuperar_girias_usuario(self.usuario) self.termo_contexto = { termo['giria']: { "significado": termo['significado'], "frequencia": termo['frequencia'] } for termo in termos_aprendidos } else: self.termo_contexto = {} except Exception as e: logger.warning(f"Falha ao carregar termos/gírias do DB: {e}") self.termo_contexto = {} try: if self.usuario and self.db is not None: emocao_salva = self.db.recuperar_aprendizado_detalhado(self.usuario, "emocao_atual") if emocao_salva: # Tenta parsear como JSON primeiro try: if isinstance(emocao_salva, str): emocao_dict = json.loads(emocao_salva) else: emocao_dict = emocao_salva if isinstance(emocao_dict, dict) and 'emocao' in emocao_dict: self.emocao_atual = emocao_dict['emocao'] elif isinstance(emocao_salva, str): self.emocao_atual = emocao_salva except (json.JSONDecodeError, TypeError): # Se não for JSON válido, usa como string direta if isinstance(emocao_salva, str): self.emocao_atual = emocao_salva except Exception as e: logger.warning(f"Falha ao carregar emoção do DB: {e}") @property def ton_predominante(self) -> Optional[str]: """ Retorna o tom predominante do usuário, acessando o DB. Returns: Tom predominante ou None se não disponível """ if self.usuario and self.db is not None: return self.db.obter_tom_predominante(self.usuario) return None def get_or_create_treinador(self, interval_hours: int = 24) -> Treinamento: """Retorna um entrenador associado a este contexto.""" if self._treinador is None: db_param: Database = self.db if self.db is not None else Database() self._treinador = Treinamento(db_param, contexto=self, interval_hours=interval_hours) return self._treinador def _load_model(self): """Carrega o modelo SentenceTransformer e embeddings sob demanda.""" if self.model is not None: return if not SENTENCE_TRANSFORMER_AVAILABLE: logger.warning("SentenceTransformer não disponível") return start_time = time.time() try: # Tenta carregar normalmente self.model = SentenceTransformer('all-MiniLM-L6-v2') logger.info("Modelo SentenceTransformer carregado com sucesso") except Exception as e: logger.warning(f"Erro inicial ao carregar modelo: {e}") logger.warning("Tentando recuperar de cache corrompido/download incompleto...") try: # Tenta limpar cache específico se possível ou carregar forçando # Como SentenceTransformer não expõe force_download fácil no init, # tentamos carregar ignorando cache ou reinstanciando # Na verdade, a melhor forma é tentar baixar via snapshot_download from huggingface_hub import snapshot_download logger.info("Forçando download do snapshot...") snapshot_download(repo_id="sentence-transformers/all-MiniLM-L6-v2", force_download=True) # Tenta carregar novamente após forçar download self.model = SentenceTransformer('all-MiniLM-L6-v2') logger.info("Modelo recuperado e carregado com sucesso!") except Exception as e2: logger.error(f"Falha crítica ao carregar modelo: {e2}") self.model = None self._check_embeddings() duration = time.time() - start_time logger.info(f"Modelo carregado em {duration:.2f}s") def _check_embeddings(self): """Verifica ou cria embeddings no banco de dados.""" if self.model and not self.embeddings: try: self.embeddings = {"conhecimento_base": "placeholder_embedding_data"} except Exception as e: logger.warning(f"Não foi possível carregar embeddings: {e}") def analisar_emocoes_mensagem(self, mensagem: str) -> Dict[str, Any]: """ Analisa o sentimento e emoção da mensagem (Heurística simples). Args: mensagem: Texto da mensagem para análise Returns: Dicionário com análise emocional """ mensagem_lower = mensagem.strip().lower() # Análise de Sentimento pos_count = sum(mensagem_lower.count(w) for w in PALAVRAS_POSITIVAS) neg_count = sum(mensagem_lower.count(w) for w in PALAVRAS_NEGATIVAS) sentimento = "neutro" if pos_count > neg_count: sentimento = "positivo" elif neg_count > pos_count: sentimento = "negativo" # Determinar Emoção Predominante if sentimento == "positivo": emocao_predominante = "alegria" elif sentimento == "negativo": emocao_predominante = "frustração" else: emocao_predominante = "neutra" # Atualiza o estado self.emocao_atual = emocao_predominante return { "sentimento_detectado": sentimento, "emocao_predominante": emocao_predominante, "intensidade_positiva": pos_count, "intensidade_negativa": neg_count, "tom_sugerido": "casual" if sentimento != "neutro" else "neutro" } def analisar_intencao_e_normalizar( self, mensagem: str, historico: List[Tuple[str, str]] ) -> Dict[str, Any]: """ Analisa a intenção, normaliza a mensagem e detecta sentimentos/estilo. Args: mensagem: Mensagem do usuário historico: Histórico de conversas Returns: Dicionário com análise completa """ self._load_model() if not isinstance(mensagem, str): mensagem = str(mensagem) mensagem_lower = mensagem.strip().lower() # 1. Análise de Intenção intencao = "pergunta" if '?' not in mensagem_lower and ('porquê' not in mensagem_lower or 'porque' not in mensagem_lower): intencao = "afirmacao" if any(w in mensagem_lower for w in ['ola', 'oi', 'bom dia', 'boa tarde', 'boa noite', 'como vai']): intencao = "saudacao" if any(w in mensagem_lower for w in ['tchau', 'ate mais', 'adeus', 'fim', 'parar']): intencao = "despedida" # 2. Análise de Sentimento/Emoção try: emotion_analyzer = _get_emotion_analyzer() # type: ignore[call-overload] nlp_config = getattr(config, 'NLP_CONFIG', None) nivel = getattr(nlp_config, 'level', 'advanced') if nlp_config else 'advanced' # Converte histórico para formato esperado historico_dict: List[Dict[str, str]] = [] for h in historico: if isinstance(h, tuple) and len(h) >= 2: historico_dict.append({"mensagem": h[0], "resposta": h[1]}) # Verificação robusta para evitar "Object of type None cannot be called" if emotion_analyzer is not None and hasattr(emotion_analyzer, 'analisar'): analise_emocional = emotion_analyzer.analisar( mensagem_lower, historico=historico_dict, nivel=nivel ) self.emocao_atual = analise_emocional.get('emocao', 'neutra') else: raise ValueError("EmotionAnalyzer não possui método 'analisar'") except Exception as e: logger.warning(f"EmotionAnalyzer falhou, usando fallback heurístico: {e}") analise_emocional = self.analisar_emocoes_mensagem(mensagem_lower) # 3. Análise de Estilo estilo = "informal" if len(re.findall(r'[A-ZÀ-Ÿ]{3,}', mensagem)) >= 2 or re.search(r'\b(Senhor|Doutor|Atenciosamente)\b', mensagem, re.IGNORECASE): estilo = "formal" # 4. Outras bandeiras ironia = False meia_frase = False usar_nome = random.random() < getattr(config, 'USAR_NOME_PROBABILIDADE', 0.7) return { "texto_normalizado": mensagem_lower, "intencao": intencao, "sentimento": analise_emocional.get('sentimento_detectado', analise_emocional.get('emocao', 'neutral')), "estilo": estilo, "contexto_ajustado": self.substituir_termos_aprendidos(mensagem_lower), "ironia": ironia, "meia_frase": meia_frase, "usar_nome": usar_nome, "emocao": self.emocao_atual, "confianca_emocao": analise_emocional.get('confianca', 0.5), "nivel_analise": analise_emocional.get('nivel_analise', 'heuristica') } def obter_historico(self, limite: int = 5) -> List[Tuple[str, str]]: """ Recupera o histórico de mensagens do banco de dados. Args: limite: Número máximo de mensagens a recuperar Returns: Lista de tuplas (mensagem, resposta) """ if not self.usuario: return [] if self.db is None: return [] try: # 🔥 CONTEXT ISOLATION: Usa conversation_id se disponível raw_result = self.db.recuperar_historico( self.usuario, limite=limite, conversation_id=self.conversation_id ) return raw_result if raw_result else [] except Exception as e: logger.warning(f"Erro ao recuperar histórico: {e}") return [] def obter_historico_expandido(self, limite: int = 30) -> List[Tuple[str, str]]: """ Recupera histórico expandido (últimas 30 mensagens) para contexto completo. Args: limite: Número máximo de mensagens (padrão 30) Returns: Lista de tuplas (mensagem, resposta) """ return self.obter_historico(limite=limite) def criar_resumo_topicos_conversa(self, historico: List[Tuple[str, str]]) -> Dict[str, Any]: """ Cria resumo inteligente de tópicos da conversa em tempo real. USADO APENAS PARA CONTEXTO INTERNO DA API - NÃO VAI NAS RESPOSTAS! Args: historico: Histórico de conversas Returns: Dicionário com resumo de tópicos (para prompt interno apenas) """ if not historico: return {"topicos": [], "resumo": "Conversa vazia"} # Análise de tópicos baseada em palavras-chave topicos_detectados = [] mensagens_concat = " ".join([msg for msg, _ in historico]).lower() # Palavras-chave por categoria categorias = { "tecnologia": ["computador", "programa", "código", "app", "site", "internet", "ai", "bot"], "pessoal": ["eu", "minha", "meu", "vida", "família", "amigo", "trabalho"], "entretenimento": ["música", "filme", "jogo", "esporte", "notícia", "youtube"], "ajuda": ["ajuda", "como", "explicar", "ensinar", "dúvida", "problema"], "conversa": ["oi", "ola", "bom", "tudo", "bem", "como vai"] } for categoria, palavras in categorias.items(): if any(palavra in mensagens_concat for palavra in palavras): topicos_detectados.append(categoria) # Resumo baseado no histórico num_mensagens = len(historico) resumo = f"Conversa com {num_mensagens} mensagens sobre: {', '.join(topicos_detectados[:3])}" return { "topicos": topicos_detectados, "resumo": resumo, "num_mensagens": num_mensagens, "timestamp": datetime.now().isoformat(), "nota": "ESTE RESUMO É APENAS PARA CONTEXTO INTERNO DA API - NÃO INCLUIR NAS RESPOSTAS!" } def processar_contexto_reply( self, mensagem: str, reply_metadata: Dict[str, Any], historico_geral: List[Tuple[str, str]] ) -> Dict[str, Any]: """ Processa contexto específico de reply, mantendo histórico geral. Enhanced to properly search and analyze quoted message content. Args: mensagem: Mensagem atual reply_metadata: Metadados do reply historico_geral: Histórico geral da conversa Returns: Contexto enriquecido com reply """ contexto_reply = { "is_reply": reply_metadata.get('is_reply', False), "reply_to_bot": reply_metadata.get('reply_to_bot', False), "quoted_author": reply_metadata.get('quoted_author_name', ''), "quoted_text": reply_metadata.get('quoted_text_original', ''), "context_hint": reply_metadata.get('context_hint', ''), "historico_geral": historico_geral, "resumo_topicos": self.criar_resumo_topicos_conversa(historico_geral) } # Enhanced reply processing if contexto_reply["is_reply"]: # Get the most complete quoted content quoted_content = self._extract_full_quoted_content(reply_metadata) contexto_reply["quoted_content_full"] = quoted_content # Analyze quoted content for context clues content_analysis = self._analyze_quoted_content_for_reply(quoted_content, mensagem) contexto_reply["content_analysis"] = content_analysis # Search for related context in history related_context = self._find_related_context_in_history(quoted_content, historico_geral) contexto_reply["related_context"] = related_context # Determine reply priority and type reply_priority = self._calculate_reply_priority( reply_metadata, quoted_content, mensagem ) contexto_reply["reply_priority"] = reply_priority # Extract key topics from quoted content topics = self._extract_topics_from_quoted_content(quoted_content) contexto_reply["topics_identified"] = topics return contexto_reply def _extract_full_quoted_content(self, reply_metadata: Dict[str, Any]) -> str: """ Extrai o conteúdo completo citado de vários campos de metadados. Args: reply_metadata: Metadados do reply Returns: Conteúdo completo citado """ # Prioriza campos em ordem de completude fields_to_check = [ 'mensagem_citada', # Campo completo se disponível 'quoted_text_original', # Texto original 'quoted_text', # Texto citado genérico 'reply_content', # Conteúdo do reply 'full_message', # Mensagem completa ] for field in fields_to_check: if field in reply_metadata and reply_metadata[field]: content = str(reply_metadata[field]).strip() if len(content) > 5: #过滤太短的内容 return content # Fallback: tenta extrair de campos genéricos for key, value in reply_metadata.items(): if isinstance(value, str) and len(value) > 10: # 检查是否是消息内容 if any(word in value.lower() for word in ['eu', 'você', 'tu', 'mim', 'nosso', 'teu']): return value.strip() return "" def _analyze_quoted_content_for_reply( self, quoted_content: str, current_message: str ) -> Dict[str, Any]: """ Analisa o conteúdo citado para encontrar pistas de contexto. Args: quoted_content: Conteúdo citado current_message: Mensagem atual Returns: Análise do conteúdo citado """ if not quoted_content: return {"empty": True} quoted_lower = quoted_content.lower() current_lower = current_message.lower() # 检测内容类型 content_type = "general" if any(w in quoted_lower for w in ['?', 'qual', 'quando', 'onde', 'como', 'por que']): content_type = "question" elif any(w in quoted_lower for w in ['eu', 'mim', 'meu', 'minha', 'eu sou']): content_type = "personal" elif any(w in quoted_lower for w in ['akira', 'bot', 'você', 'vc']): content_type = "about_bot" # 检测主题关键词 keywords = [] keyword_mapping = { "tempo": ["tempo", "clima", "chover", "sol", "temperatura"], "musica": ["música", "musica", "youtube", "yt", "radiohead", "vampire weekend"], "traducao": ["traduz", "letra", "ingles", "english", "tradução"], "pesquisa": ["pesquisa", "web", "google", "busca", "buscar"], "emocao": ["triste", "feliz", "raiva", "amor", "medo", "alegria"], } for category, words in keyword_mapping.items(): if any(w in quoted_lower for w in words): keywords.append(category) # 检测语气 tone = "neutral" if any(w in quoted_lower for w in ['kkk', 'haha', '😂', '🤣']): tone = "humorous" elif any(w in quoted_lower for w in ['!!!', '???', 'nossa', 'eita']): tone = "excited" elif any(w in quoted_lower for w in ['.', '..', '...']): tone = "thoughtful" return { "content_type": content_type, "keywords": keywords, "tone": tone, "length": len(quoted_content), "has_question": '?' in quoted_content, "is_about_bot": "about_bot" in keywords, "has_emotion_keywords": len([k for k in keywords if k == "emocao"]) > 0 } def _find_related_context_in_history( self, quoted_content: str, historico: List[Tuple[str, str]] ) -> List[Dict[str, Any]]: """ Busca contexto relacionado no histórico baseado no conteúdo citado. Args: quoted_content: Conteúdo citado historico: Histórico de conversas Returns: Lista de contextos relacionados encontrados """ if not quoted_content or not historico: return [] related_contexts = [] quoted_words = set(quoted_content.lower().split()) for i, (msg_user, msg_bot) in enumerate(historico): if not msg_user or not msg_bot: continue # 计算相似度 msg_words = set((msg_user + " " + msg_bot).lower().split()) intersection = quoted_words.intersection(msg_words) if intersection: # 找到相关上下文 similarity = len(intersection) / len(quoted_words.union(msg_words)) if similarity > 0.1: # 只返回相关度较高的 related_contexts.append({ "index": i, "similarity": round(similarity, 3), "user_message": msg_user[:100] if len(msg_user) > 100 else msg_user, "bot_response": msg_bot[:100] if len(msg_bot) > 100 else msg_bot, "common_words": list(intersection)[:5] }) # 按相似度排序 related_contexts.sort(key=lambda x: x["similarity"], reverse=True) return related_contexts[:5] # 返回前5个最相关的 def _calculate_reply_priority( self, reply_metadata: Dict[str, Any], quoted_content: str, current_message: str ) -> Dict[str, Any]: """ Calcula a prioridade e tipo do reply. Args: reply_metadata: Metadados do reply quoted_content: Conteúdo citado current_message: Mensagem atual Returns: Dicionário com prioridade calculada """ priority = 1 # 默认优先级 priority_type = "normal" should_prioritize = False # 检测是否是回复给bot is_reply_to_bot = reply_metadata.get('reply_to_bot', False) # 检测是否是简短问题 current_words = current_message.split() is_short_question = ( len(current_words) <= 5 and any(w in current_message.lower() for w in ['?', 'qual', 'quando', 'onde', 'como', 'oq']) ) # 检测引用内容是否完整 has_quoted_content = len(quoted_content) > 10 # 计算优先级 if is_reply_to_bot and is_short_question: priority = 4 # 最高优先级 priority_type = "critical_short_question" should_prioritize = True elif is_reply_to_bot: priority = 3 priority_type = "reply_to_bot" should_prioritize = True elif is_short_question: priority = 2 priority_type = "short_question" should_prioritize = True elif has_quoted_content: priority = 1.5 priority_type = "has_content" return { "priority": priority, "type": priority_type, "should_prioritize": should_prioritize, "is_reply_to_bot": is_reply_to_bot, "is_short_question": is_short_question, "has_quoted_content": has_quoted_content, "multiplier": min(priority / 2, 1.0) # 优先级倍数 } def _extract_topics_from_quoted_content(self, quoted_content: str) -> List[str]: """ Extrai temas principais do conteúdo citado. Args: quoted_content: Conteúdo citado Returns: Lista de temas identificados """ if not quoted_content: return [] topics = [] quoted_lower = quoted_content.lower() # 定义主题关键词 topic_keywords = { "tempo_clima": ["tempo", "clima", "chover", "sol", "chuva", "temperatura", "amanhã", "hoje"], "musica": ["música", "musica", "youtube", "yt", "radiohead", "vampire weekend", "banda", "cantor", "link"], "traducao": ["traduz", "letra", "ingles", "english", "português", "portugues", "significado"], "pesquisa": ["pesquisa", "web", "google", "busca", "buscar", "encontrar", "procurar"], "emocoes": ["triste", "feliz", "raiva", "amor", "medo", "alegria", "sentimento", "sinto"], "tecnologia": ["programa", "código", "app", "site", "internet", "bot", "akira"], "vida": ["vida", "trabalho", "família", "familia", "amigo", "pessoa", "situação", "situacao"], "tradicao": ["tradição", "tradição", "cultura", "angola", "costume", "costumes"] } for topic, keywords in topic_keywords.items(): if any(kw in quoted_lower for kw in keywords): topics.append(topic) # 如果没有找到主题，返回general if not topics: topics.append("general") return topics def _analisar_contexto_mensagem_citada( self, mensagem_citada: str, historico: List[Tuple[str, str]] ) -> Dict[str, Any]: """ Analisa o contexto da mensagem citada no reply. Args: mensagem_citada: Texto da mensagem citada historico: Histórico geral Returns: Análise do contexto da mensagem citada """ # Buscar similaridade com histórico similaridade_max = 0 contexto_relacionado = "" for msg_user, msg_bot in historico: # Similaridade simples baseada em palavras comuns palavras_citada = set(mensagem_citada.lower().split()) palavras_historico = set((msg_user + " " + msg_bot).lower().split()) intersecao = palavras_citada.intersection(palavras_historico) if intersecao: similaridade = len(intersecao) / len(palavras_citada.union(palavras_historico)) if similaridade > similaridade_max: similaridade_max = similaridade contexto_relacionado = msg_bot if msg_bot else msg_user return { "similaridade": similaridade_max, "contexto_relacionado": contexto_relacionado, "palavras_comuns": list(intersecao) if 'intersecao' in locals() else [] } def atualizar_contexto( self, mensagem: str, resposta: str, numero: Optional[str] = None ): """ Salva a interação no banco e aciona aprendizado de termos. Args: mensagem: Mensagem do usuário resposta: Resposta gerada numero: Número de telefone """ if not self.usuario: usuario = 'anonimo' else: usuario = self.usuario final_numero = numero if numero else self.usuario try: if self.db is not None: self.db.salvar_mensagem(usuario, mensagem, resposta, numero=final_numero) historico = self.obter_historico(limite=10) self.aprender_do_historico(mensagem, resposta, historico) if final_numero: self.salvar_estado_contexto_no_db(str(final_numero) if final_numero else "desconhecido") except Exception as e: logger.warning(f'Falha ao salvar mensagem no DB: {e}') def salvar_estado_contexto_no_db(self, user_key: str): """ Persiste o estado atual da classe Contexto no banco de dados. Args: user_key: Chave do usuário (deve ser string) """ if self.db is None: return termos_json = json.dumps(self.termo_contexto) emocao_str = self.emocao_atual try: self.db.salvar_aprendizado_detalhado(user_key, "emocao_atual", json.dumps({"emocao": emocao_str})) self.db.salvar_contexto( user_key=user_key, historico="[]", emocao_atual=emocao_str, termos=termos_json, girias=termos_json, tom=emocao_str ) logger.debug(f"Contexto do usuário {user_key} salvo no DB.") except Exception as e: logger.error(f"Falha ao salvar estado do contexto no DB: {e}") def aprender_do_historico( self, mensagem: str, resposta: str, historico: List[Tuple[str, str]] ): """ Aprende termos do histórico de conversas. Args: mensagem: Mensagem do usuário resposta: Resposta gerada historico: Histórico de conversas """ if not self.usuario: return if self.db is None: return mensagem_lower = mensagem.lower() # Gírias angolanas comuns girias_angolanas = ['ya', 'bué', 'fixe', 'puto', 'kapa', 'muxima', 'kalai'] for giria in girias_angolanas: if giria in mensagem_lower: try: significado_placeholder = f'termo regional para {giria}' self.db.salvar_giria_aprendida( self.usuario, giria, significado_placeholder, mensagem[:50] ) freq_atual = self.termo_contexto.get(giria, {}).get("frequencia", 0) self.termo_contexto[giria] = { "significado": significado_placeholder, "frequencia": freq_atual + 1 } except Exception as e: logger.warning(f"Erro ao salvar gíria no DB: {e}") def substituir_termos_aprendidos(self, mensagem: str) -> str: """ Substitui termos aprendidos na mensagem. Args: mensagem: Mensagem original Returns: Mensagem com termos substituídos """ for termo, info in self.termo_contexto.items(): if isinstance(info, dict) and "significado" in info: # Substitui apenas a palavra inteira (case insensitive) mensagem = re.sub( r'\b' + re.escape(termo) + r'\b', info["significado"], mensagem, flags=re.IGNORECASE ) return mensagem def obter_aprendizado_detalhado(self, chave: str) -> Optional[Dict[str, Any]]: """ Recupera aprendizados detalhados do usuário. Args: chave: Chave do aprendizado Returns: Dicionário com o aprendizado ou None """ if not self.usuario: return None if self.db is None: return None try: raw_data = self.db.recuperar_aprendizado_detalhado(self.usuario, chave) if raw_data: if isinstance(raw_data, str): return json.loads(raw_data) return raw_data return None except Exception as e: logger.warning(f"Erro ao obter aprendizado detalhado: {e}") return None def obter_emocao_atual(self) -> str: """Recupera a emoção atual do usuário.""" return self.emocao_atual def ativar_espirito_critico(self): """Ativa o espírito crítico para respostas questionadoras.""" self.espirito_critico = True def obter_aprendizados(self) -> Dict[str, Any]: """ Retorna os aprendizados do usuário. Returns: Dicionário com termos, emoção e tom """ aprendizados = { "termos": self.termo_contexto, "emocao_preferida": self.emocao_atual, "ton_predominante": self.ton_predominante } return aprendizados def salvar_conhecimento_base(self, chave: str, valor: Any): """Salva uma informação na base de conhecimento.""" self.base_conhecimento[chave] = valor def obter_conhecimento_base(self, chave: str) -> Optional[Any]: """Obtém uma informação da base de conhecimento.""" return self.base_conhecimento.get(chave) def obter_historico_para_llm(self) -> List[Dict[str, str]]: """ Retorna o histórico no formato esperado pelos LLMs. Returns: Lista de dicionários com role e content """ historico = self.obter_historico() if historico and len(historico) > 0: return [ {"role": "user", "content": h[0]} if isinstance(h, tuple) and len(h) >= 2 else h for h in historico ] return [] # ================================================================ # FUNÇÕES AUXILIARES (para compatibilidade com testar_correcoes.py) # ================================================================ def criar_contexto(db: Optional[Database], identificador: str) -> Contexto: """ Factory function para criar contexto. Args: db: Instância do banco de dados identificador: Identificador do usuário Returns: Instância de Contexto """ return Contexto(db=db, usuario=identificador) # Funções auxiliares para config.py def eh_usuario_privilegiado(numero: str) -> bool: """ Verifica se um número é de usuário privilegiado. Args: numero: Número de telefone Returns: True se for privilegiado """ try: from .database import Database db = Database() return db.eh_privilegiado(numero) except Exception as e: logger.error(f"Erro ao verificar privilégios: {e}") return False def forcar_modo_inicial_privilegiado(numero: str) -> str: """ Retorna o modo de fala forçado para usuário privilegiado. Args: numero: Número de telefone Returns: Modo de fala """ try: from .database import Database db = Database() modo = db.obter_modo_fala_privilegiado(numero) return modo if modo else "tecnico_formal" except Exception as e: logger.error(f"Erro ao obter modo de fala: {e}") return "tecnico_formal" def analisar_tom_usuario(mensagem: str) -> str: """ Analisa o tom de uma mensagem. Args: mensagem: Texto da mensagem Returns: Tom detectado """ contexto = Contexto(db=None, usuario=None) analise = contexto.analisar_emocoes_mensagem(mensagem) return analise.get("tom_sugerido", "neutro") def determinar_nivel_transicao( numero: str, tom: str, nivel_atual: int ) -> int: """ Determina o nível de transição baseado no tom. Usa transição LENTA e gradual conforme configurações do config. Args: numero: Número do usuário tom: Tom detectado nivel_atual: Nível atual Returns: Novo nível de transição (mudança muito gradual) """ # threshold configurado no config.py (atual: 0.9) threshold = getattr(config, 'TRANSICAO_HUMOR_THRESHOLD', 0.9) nivel_max = getattr(config, 'NIVEL_TRANSICAO_MAX', 1) # Com threshold de 0.9, só muda se tiver 90% de certeza # Com nivel_max = 1, só pode mudar 1 nível por vez (muito lento) if tom in ["formal", "tecnico_formal"]: return min(nivel_atual + 1, nivel_max) elif tom in ["casual", "informal"]: return max(nivel_atual - 1, 1) return nivel_atual