# modules/contexto.py import logging import re import random import time import sqlite3 import json from typing import Optional, List, Dict, Tuple, Any import modules.config as config from .database import Database from .treinamento import Treinamento try: from sentence_transformers import SentenceTransformer except Exception as e: logging.warning(f"sentence_transformers não disponível: {e}") SentenceTransformer = None try: import psutil except Exception: psutil = None try: import structlog except Exception: structlog = None logger = logging.getLogger(__name__) logging.basicConfig(level=logging.INFO, format='%(asctime)s [%(levelname)s] %(message)s') if 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 PALAVRAS_POSITIVAS = ['bom', 'ótimo', 'incrível', 'feliz', 'adorei', 'top', 'fixe', 'bué', 'show', 'legal', 'bacana'] PALAVRAS_NEGATIVAS = ['ruim', 'péssimo', 'triste', 'ódio', 'raiva', 'chateado', 'merda', 'porra', 'odeio'] 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. """ def __init__(self, db: Database, usuario: Optional[str] = None): self.db = db self.usuario = usuario self.model: Optional[SentenceTransformer] = None self.embeddings: Optional[Dict[str, Any]] = None self._treinador: Optional[Treinamento] = None # Estado de conversa self.emocao_atual = "neutra" self.espírito_crítico = False self.base_conhecimento = {} # Garante que termo_contexto seja sempre um dicionário self.termo_contexto: Dict[str, Dict] = {} self.atualizar_aprendizados_do_banco() logger.info("Inicializando Contexto (com NLP avançado, aprendizado de gírias e emoções) ...") # Cache para termos regionais e gírias self.cache_girias: Dict[str, Any] = {} def atualizar_aprendizados_do_banco(self): """Carrega todos os dados de aprendizado persistentes do banco.""" try: termos_aprendidos = self.db.recuperar_girias_usuario(self.usuario) if self.usuario else [] self.termo_contexto = { termo['giria']: {"significado": termo['significado'], "frequencia": termo['frequencia']} for termo in termos_aprendidos } except Exception as e: logger.warning(f"Falha ao carregar termos/gírias do DB: {e}") self.termo_contexto = {} try: emocao_salva = self.db.recuperar_aprendizado_detalhado(self.usuario, "emocao_atual") if self.usuario else None if emocao_salva: emocao_dict = json.loads(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 Exception as e: logger.warning(f"Falha ao carregar emoção do DB: {e}") logger.info(f"Aprendizados carregados para {self.usuario}.") @property def ton_predominante(self) -> Optional[str]: """Retorna o tom predominante do usuário (acessa o DB).""" if self.usuario: return self.db.obter_tom_predominante(self.usuario) return None def get_or_create_treinador(self, interval_hours: int = 24) -> Treinamento: """Retorna um treinador associado, criando se necessário.""" if self._treinador is None: self._treinador = Treinamento(self.db, contexto=self, interval_hours=interval_hours) return self._treinador def _load_model(self): """Carrega o modelo SentenceTransformer sob demanda.""" if self.model is not None: return if SentenceTransformer is None: logger.warning("SentenceTransformer não instalado") return try: self.model = SentenceTransformer('all-MiniLM-L6-v2') logger.info("Modelo SentenceTransformer carregado") except Exception as e: logger.error(f"Erro ao carregar modelo: {e}") self.model = None self._check_embeddings() def _check_embeddings(self): """Verifica ou cria embeddings no banco.""" if self.model and not self.embeddings: self.embeddings = {"conhecimento_base": "placeholder"} def analisar_emocoes_mensagem(self, mensagem: str) -> Dict[str, Any]: """Analisa sentimento e emoção da mensagem (heurística).""" mensagem_lower = mensagem.strip().lower() 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" emocao_predominante = "alegria" if sentimento == "positivo" else "frustração" if sentimento == "negativo" else "neutra" 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 intenção, normaliza e detecta estilo.""" self._load_model() if not isinstance(mensagem, str): mensagem = str(mensagem) mensagem_lower = mensagem.strip().lower() # Intenção intencao = "pergunta" if '?' not in mensagem_lower and 'porquê' not in mensagem_lower and '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" # Sentimento analise_emocional = self.analisar_emocoes_mensagem(mensagem_lower) # 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" usar_nome = random.random() < getattr(config, 'USAR_NOME_PROBABILIDADE', 0.7) return { "texto_normalizado": mensagem_lower, "intencao": intencao, "sentimento": analise_emocional['sentimento_detectado'], "estilo": estilo, "contexto_ajustado": self.substituir_termos_aprendidos(mensagem_lower), "ironia": False, "meia_frase": False, "usar_nome": usar_nome, "emocao": self.emocao_atual } def obter_historico(self, limite: int = 5) -> List[Tuple[str, str]]: """Recupera histórico do banco.""" if not self.usuario: return [] raw = self.db.recuperar_mensagens(self.usuario, limite=limite) return raw if raw else [] def obter_historico_para_llm(self) -> List[Dict]: """Formato esperado pelo LLMManager.generate()""" raw = self.obter_historico(limite=10) history = [] for user_msg, bot_msg in raw: history.append({"role": "user", "content": user_msg}) history.append({"role": "assistant", "content": bot_msg}) return history def atualizar_contexto(self, mensagem: str, resposta: str, numero: Optional[str] = None): """Salva interação e aprende.""" usuario = self.usuario or 'anonimo' final_numero = numero or self.usuario try: self.db.salvar_mensagem(usuario, mensagem, resposta, numero=final_numero) historico = self.obter_historico(limite=10) self.aprender_do_historico(mensagem, resposta, historico) self.salvar_estado_contexto_no_db(final_numero) except Exception as e: logger.warning(f'Falha ao salvar: {e}') def salvar_estado_contexto_no_db(self, user_key: str): """Persiste estado no DB.""" termos_json = json.dumps(self.termo_contexto) try: self.db.salvar_aprendizado_detalhado(user_key, "emocao_atual", json.dumps({"emocao": self.emocao_atual})) self.db.salvar_contexto( user_key=user_key, historico="[]", emocao_atual=self.emocao_atual, termos=termos_json, girias=termos_json, tom=self.emocao_atual ) except Exception as e: logger.error(f"Falha ao salvar contexto: {e}") def aprender_do_historico(self, mensagem: str, resposta: str, historico: List[Tuple[str, str]]): """Aprende gírias do histórico.""" if not self.usuario: return mensagem_lower = mensagem.lower() girias_angolanas_simples = ['ya', 'bué', 'fixe', 'puto', 'kota', 'mwangolé'] for giria in girias_angolanas_simples: if giria in mensagem_lower: try: significado = f'termo regional para {giria}' self.db.salvar_giria_aprendida(self.usuario, giria, significado, mensagem[:50]) self.termo_contexto[giria] = { "significado": significado, "frequencia": self.termo_contexto.get(giria, {}).get("frequencia", 0) + 1 } except Exception as e: logger.warning(f"Erro ao salvar gíria: {e}") def substituir_termos_aprendidos(self, mensagem: str) -> str: """Substitui termos aprendidos.""" for termo, info in self.termo_contexto.items(): if isinstance(info, dict) and "significado" in info: 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]: """Recupera aprendizado detalhado.""" try: raw = self.db.recuperar_aprendizado_detalhado(self.usuario, chave) return json.loads(raw) if raw else None except Exception as e: logger.warning(f"Erro ao obter aprendizado: {e}") return None def obter_emocao_atual(self) -> str: return self.emocao_atual def ativar_espírito_crítico(self): self.espírito_crítico = True def obter_aprendizados(self) -> Dict[str, Any]: """Retorna todos os aprendizados.""" return { "termos": self.termo_contexto, "emocao_preferida": self.emocao_atual, "ton_predominante": self.ton_predominante } def salvar_conhecimento_base(self, chave: str, valor: Any): self.base_conhecimento[chave] = valor def obter_conhecimento_base(self, chave: str) -> Optional[Any]: return self.base_conhecimento.get(chave)