Update modules/contexto.py

modules/contexto.py

@@ -1,302 +1,188 @@
-import logging
-import re
-try:
-    # sentence-transformers é uma dependência opcional; Pylance pode não encontrá-la
-    # em alguns ambientes. Silenciaremos o aviso de import faltante para manter
-    # o comportamento resiliente em tempo de execução.
-    from sentence_transformers import SentenceTransformer  # type: ignore[reportMissingImports]
-except Exception as e:
-    logging.warning(f"sentence_transformers não disponível: {e}")
-    SentenceTransformer = None
-
-from modules.database import Database
-import random
-import modules.config as config
-try:
-    import psutil  # type: ignore[reportMissingImports]
-except Exception:
-    psutil = None
-import time
-try:
-    import structlog  # type: ignore[reportMissingImports]
-except Exception:
-    structlog = None
-import sqlite3
-from typing import Optional
-from modules.treinamento import Treinamento
-
-# Configuração do logging (fallback se structlog ausente)
-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,
-        cache_logger_on_first_use=True
-    )
-    logger = structlog.get_logger(__name__)
-else:
-    logger = logging.getLogger(__name__)
-
-class Contexto:
-    """Classe para gerenciar o contexto da conversa, análise de intenções e aprendizado dinâmico de termos regionais/gírias."""
-    def __init__(self, db: Database, usuario=None):
-        self.db = db
-        self.usuario = usuario
-        self.model = None
-        self.embeddings = None
-        self._treinador: Optional[Treinamento] = None
-        self.emocao_atual = "neutra"  # Emoções: neutra, feliz, irritada, crítica
-        self.espírito_crítico = False  # Ativar espírito crítico para respostas questionadoras
-        self.base_conhecimento = {}  # Conhecimento geral aprendido
-        # Garantir que termo_contexto seja sempre um dicionário
-        termos = self.obter_aprendizado_detalhado("termos")
-        self.termo_contexto = termos if isinstance(termos, dict) else {}
-        logger.info("🟢 Inicializando Contexto (com NLP avançado, aprendizado de gírias e emoções) ...")
-
-    def get_or_create_treinador(self, interval_hours: int = 24) -> Treinamento:
-        """Retorna um treinador associado a este contexto, 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 e embeddings sob demanda."""
-        if self.model is not None:
-            return
-        start_time = time.time()
-        if psutil:
-            try:
-                process = psutil.Process()
-                mem_before = process.memory_info().rss / 1024 / 1024
-                cpu_percent = psutil.cpu_percent()
-                logger.info({"event": "Before model load", "cpu_percent": cpu_percent, "memory_mb": mem_before})
-            except Exception as e:
-                logger.warning(f"Erro ao coletar métricas de sistema: {e}")
-        else:
-            logger.debug("psutil não disponível para métricas de sistema")
-        if SentenceTransformer is None:
-            logger.warning({"event": "Modelo SentenceTransformer não será carregado", "reason": "Biblioteca não instalada"})
-            return
-        try:
-            self.model = SentenceTransformer('all-MiniLM-L6-v2')
-            logger.info({"event": "Modelo SentenceTransformer carregado com sucesso"})
-        except Exception as e:
-            logger.error({"event": "Erro ao carregar modelo", "error": str(e)})
-            self.model = None
-        self._check_embeddings()
-        duration = time.time() - start_time
-        logger.info({"event": "Modelo carregado", "duration_seconds": duration})
-
-    def _check_embeddings(self):
-        """Verifica ou cria embeddings no banco de dados SQLite, se a tabela existir."""
-        if self.model is None:
-            logger.warning({"event": "Embeddings não serão verificados", "reason": "Modelo não carregado"})
-            return
-        logger.info({"event": "Verificando embeddings no banco de dados"})
-        try:
-            with sqlite3.connect(self.db.db_path) as conn:
-                c = conn.cursor()
-                c.execute("""SELECT name FROM sqlite_master WHERE type='table' AND name='embeddings'""")
-                table_exists = c.fetchone()
-                if not table_exists:
-                    logger.warning({"event": "Tabela embeddings não encontrada", "action": "Ignorando verificação de embeddings"})
-                    return
-                c.execute("SELECT COUNT(*) FROM embeddings")
-                if c.fetchone()[0] == 0:
-                    logger.info({"event": "Criando embeddings iniciais"})
-                    sentences = ["oi", "tchau", "fixe", "puto"]
-                    embeddings = self.model.encode(sentences)
-                    for sentence, embedding in zip(sentences, embeddings):
-                        c.execute("INSERT INTO embeddings (texto, embedding) VALUES (?, ?)", (sentence, embedding.tobytes()))
-                    conn.commit()
-                    logger.info({"event": "Embeddings iniciais criados"})
-                else:
-                    logger.info({"event": "Embeddings já existem no banco de dados"})
-        except Exception as e:
-            logger.error({"event": "Erro ao verificar embeddings", "error": str(e)})
-
-    def analisar_intencao_e_normalizar(self, mensagem: str, historico: list) -> dict:
-        """Analisa a intenção, normaliza a mensagem, substitui termos aprendidos e detecta ironias e meias frases."""
-        self._load_model()  # Carrega o modelo apenas quando necessário
-        
-        # ALTERAÇÃO: Handling de encoding UTF-8 para preservar acentos e caracteres especiais
-        if not isinstance(mensagem, str):
-            mensagem = str(mensagem)
-        mensagem = mensagem.encode('utf-8', 'ignore').decode('utf-8')  # Limpa encoding ruins
-        
-        # ALTERAÇÃO: Regex Unicode-safe (permite letras acentuadas, números, etc.)
-        mensagem = re.sub(r'[^\w\s\.,!?😅👍]', '', mensagem.lower(), flags=re.UNICODE).strip()
-        
-        # Substituir termos aprendidos antes da análise
-        mensagem = self.substituir_termos_aprendidos(mensagem)
-        intencao = "neutro"
-        sentimento = "neutro"
-        ironia = False
-        meia_frase = False
-        # Detecção de meia frase (frases curtas ou incompletas)
-        if len(mensagem.split()) <= 3 or "..." in mensagem:
-            meia_frase = True
-        # Detecção de intenção
-        if any(word in mensagem for word in ["oi", "olá", "eai", "eae"]):
-            intencao = "saudacao"
-        elif any(word in mensagem for word in ["tchau", "flw", "bazar", "até"]):
-            intencao = "despedida"
-        elif any(word in mensagem for word in ["como", "tô", "tá", "bem"]):
-            intencao = "responder_bem_estar"
-        # Detecção de sentimento
-        if any(word in mensagem for word in ["fixe", "legal", "bom", "😊", "", "kkk", "rsrs"]):
-            sentimento = "positivo"
-        elif any(word in mensagem for word in ["ruim", "chato", "droga", "😡", "😢"]):
-            sentimento = "negativo"
-        # Detecção de ironia (exemplo: tom positivo com conteúdo negativo ou vice-versa)
-        if ("fixe" in mensagem or "bom" in mensagem) and ("perdi" in mensagem or "droga" in mensagem):
-            ironia = True
-            sentimento = "negativo"  # Ajustar sentimento para refletir o real
-        estilo = "informal" if any(g in mensagem for g in ['kkk', 'rsrs', 'puto']) else "normal"
-        # Analisar emoção baseada no sentimento
-        self.analisar_emocao(mensagem, sentimento)
-        contexto_ajustado = f"Mensagem: {mensagem} | Histórico: {historico[-2:] if len(historico) > 1 else historico}"
-        if ironia:
-            contexto_ajustado += " | Possível ironia detectada."
-        if meia_frase:
-            contexto_ajustado += " | Mensagem parece incompleta (meia frase)."
-        # Decidir se usar o nome do usuário em respostas (saudações, agradecimentos e despedidas)
-        # Tornar probabílistico com base na configuração para evitar sempre usar o nome completo
-        usar_nome = False
-        prob = getattr(config, 'USAR_NOME_PROBABILIDADE', 0.7)
-        if intencao in ["saudacao", "despedida"] or any(w in mensagem for w in ["obrigado", "valeu", "thanks"]):
-            try:
-                usar_nome = random.random() < float(prob)
-            except Exception:
-                usar_nome = random.random() < 0.7
-
-        return {
-            "texto_normalizado": mensagem,
-            "intencao": intencao,
-            "sentimento": sentimento,
-            "estilo": estilo,
-            "contexto_ajustado": contexto_ajustado,
-            "ironia": ironia,
-            "meia_frase": meia_frase,
-            "usar_nome": usar_nome
-        }
-
-    def balancear_contexto(self, mensagem_atual: str, nome_usuario: str, numero_usuario: str, mensagem_original: str, limite_historico: int, limite_contexto: int, is_reply: bool) -> str:
-        """Balanceia o contexto com base no histórico e mensagem atual."""
-        historico = self.db.recuperar_mensagens(nome_usuario, limite=limite_historico)
-        contexto = f"Usuário: {nome_usuario} (ID: {numero_usuario}) | Mensagem atual: {mensagem_atual}"
-        if is_reply and mensagem_original:
-            contexto += f" | Resposta a: {mensagem_original}"
-        if historico:
-            contexto += f" | Histórico recente: {historico[-limite_contexto:]}"
-        return contexto
-
-    def selecionar_resposta_predefinida(self, contexto: str) -> str:
-        """Seleciona uma resposta predefinida com base no contexto."""
-        contexto_lower = contexto.lower()
-        # Respostas muito curtas e neutras para saudações/despedidas
-        if any(w in contexto_lower for w in [" oi", "oi", "olá", "eai", "eae"]):
-            return "Oi! Tudo fixe?"
-        elif any(w in contexto_lower for w in ["tchau", "flw", "até"]):
-            return "Tchau! Fica bem."
-        return ""  # String vazia quando não há resposta predefinida
-
-    # Métodos de integração com banco e aprendizado detalhado
-    def registrar_aprendizado_detalhado(self, chave, valor):
-        if not self.usuario:
-            logger.warning("Usuário não definido para aprendizado detalhado.")
-            return
-        self.db.salvar_aprendizado_detalhado(self.usuario, chave, valor)
-
-    def obter_aprendizado_detalhado(self, chave=None):
-        if not self.usuario:
-            logger.warning("Usuário não definido para consulta de aprendizado detalhado.")
-            return None
-        return self.db.recuperar_aprendizado_detalhado(self.usuario, chave)
-
-    def obter_historico(self, limite=5):
-        if not self.usuario:
-            logger.warning("Usuário não definido para histórico.")
-            return []
-        result = self.db.recuperar_mensagens(self.usuario, limite=limite)
-        return result if result else []
-
-    def atualizar_contexto(self, mensagem, resposta):
-        """Salva a interação no banco de mensagens e aciona aprendizado de termos."""
-        if not self.usuario:
-            logger.warning("Usuário não definido para atualizar contexto; salvando como 'anonimo'.")
-            usuario = 'anonimo'
-        else:
-            usuario = self.usuario
-        try:
-            self.db.salvar_mensagem(usuario, mensagem, resposta)
-            # Aprender termos do histórico
-            historico = self.obter_historico(limite=10)  # Últimas 10 mensagens
-            self.aprender_do_historico(mensagem, resposta, historico)
-        except Exception as e:
-            logger.warning(f'Falha ao salvar mensagem no DB: {e}')
-
-    def registrar_aprendizado(self, dado, valor):
-        if not self.usuario:
-            logger.warning("Usuário não definido para aprendizado simples.")
-            return
-        self.db.salvar_aprendizado(self.usuario, dado, valor)
-
-    def obter_aprendizado(self, dado):
-        if not self.usuario:
-            logger.warning("Usuário não definido para consulta de aprendizado simples.")
-            return None
-        return self.db.recuperar_aprendizado(self.usuario, dado)
-
-    def aprender_termo_regional(self, termo, contexto, significado):
-        """Aprende um termo regional/gíria baseado no contexto."""
-        self.termo_contexto[termo] = {"contexto": contexto, "significado": significado}
-        self.registrar_aprendizado_detalhado("termos", self.termo_contexto)
-        logger.info(f"Termo '{termo}' aprendido: {significado} no contexto {contexto}")
-
-    def analisar_emocao(self, mensagem, sentimento):
-        """Analisa e atualiza a emoção da IA baseada na mensagem e sentimento."""
-        if sentimento == "positivo":
-            self.emocao_atual = "feliz"
-        elif sentimento == "negativo":
-            self.emocao_atual = "irritada"
-        else:
-            self.emocao_atual = "neutra"
-        return self.emocao_atual
-
-    def ativar_espírito_crítico(self):
-        """Ativa o espírito crítico para respostas questionadoras."""
-        self.espírito_crítico = True
-        return "Espírito crítico ativado para respostas questionadoras."
-
-    def aprender_do_historico(self, mensagem, resposta, historico):
-        """Aprende termos do histórico de conversas."""
-        if len(historico) >= 2:
-            prev_msg = historico[-2][0].lower()
-            if "como vai" in prev_msg or "tudo bem" in prev_msg:
-                if "indo" in mensagem.lower():
-                    self.aprender_termo_regional("indo", "bem_estar", "bem")
-                # Adicionar mais padrões aqui para outras gírias e contextos
-                # Ex.: if "blz" in mensagem.lower(): self.aprender_termo_regional("blz", "afirmacao", "beleza")
-
-    def substituir_termos_aprendidos(self, mensagem):
-        """Substitui termos aprendidos na mensagem."""
-        for termo, info in self.termo_contexto.items():
-            if termo in mensagem:
-                mensagem = mensagem.replace(termo, info["significado"])
-        return mensagem
-
-    def obter_aprendizados(self):
-        """Retorna os aprendizados do usuário, incluindo termos e emoções."""
-        aprendizados = {
-            "termos": self.termo_contexto,
-            "emocao_atual": self.emocao_atual,
-            "espírito_crítico": self.espírito_crítico
-        }
-        return aprendizados
+import logging
+import re
+import random
+import time
+import sqlite3
+from typing import List, Dict, Any, Optional, Tuple
+
+try:
+    from sentence_transformers import SentenceTransformer  # type: ignore
+except Exception as e:
+    logging.warning(f"sentence_transformers não disponível: {e}")
+    SentenceTransformer = None
+
+from modules.database import Database
+import modules.config as config
+
+# Logging
+logging.basicConfig(level=logging.INFO, format='%(asctime)s [%(levelname)s] %(message)s')
+logger = logging.getLogger(__name__)
+
+
+class Contexto:
+    """
+    Gerencia contexto, histórico, aprendizado dinâmico, gírias, tom, emoção.
+    Consulta o banco a cada interação → aprendizado em tempo real.
+    """
+
+    def __init__(self, db: Database, user_id: str):
+        self.db = db
+        self.user_id = user_id  # número ou nome do usuário
+        self.historico: List[Tuple[str, str]] = []
+        self.girias_aprendidas: List[Dict[str, Any]] = []
+        self.ton_predominante: str = "neutro"
+        self.emocao_atual: str = "neutra"
+        self.termo_contexto: Dict[str, Dict] = {}
+        self.model = None
+
+        # Carrega tudo do banco na primeira vez
+        self._atualizar_do_banco()
+
+    def _atualizar_do_banco(self):
+        """Consulta o banco e atualiza tudo (chamado a cada interação)"""
+        try:
+            # Histórico
+            msgs = self.db.recuperar_mensagens(self.user_id, limite=10)
+            self.historico = [(m, r) for m, r in (msgs or [])][::-1][-5:]  # últimas 5
+
+            # Gírias aprendidas
+            self.girias_aprendidas = self.db.recuperar_girias_usuario(self.user_id)
+
+            # Tom predominante
+            self.ton_predominante = self.db.obter_tom_predominante(self.user_id) or "neutro"
+
+            # Termos aprendidos (antigo termo_contexto)
+            termos_raw = self.db.recuperar_aprendizado_detalhado(self.user_id, "termos")
+            if termos_raw and isinstance(termos_raw, str):
+                try:
+                    import json
+                    self.termo_contexto = json.loads(termos_raw)
+                except:
+                    self.termo_contexto = {}
+            else:
+                self.termo_contexto = {}
+
+            # Emoção atual (baseada na última interação)
+            ultima = self.historico[-1] if self.historico else ("", "")
+            analise = self.db.analisar_emocoes_mensagem(ultima[0] + " " + ultima[1])
+            self.emocao_atual = analise["emocao"]
+
+        except Exception as e:
+            logger.error(f"Erro ao atualizar contexto do banco: {e}")
+            self.historico = []
+            self.girias_aprendidas = []
+            self.ton_predominante = "neutro"
+            self.termo_contexto = {}
+
+    def atualizar_aprendizados_do_banco(self):
+        """FORÇA ATUALIZAÇÃO DO BANCO A CADA CHAMADA (usado na api.py)"""
+        self._atualizar_do_banco()
+
+    def obter_historico(self) -> List[Tuple[str, str]]:
+        return self.historico
+
+    def atualizar_contexto(self, mensagem: str, resposta: str):
+        """Salva no banco + atualiza memória local"""
+        try:
+            self.db.salvar_mensagem(self.user_id, mensagem, resposta, numero=self.user_id)
+            self.historico.append((mensagem, resposta))
+            if len(self.historico) > 5:
+                self.historico.pop(0)
+
+            # Atualiza emoção
+            analise = self.db.analisar_emocoes_mensagem(mensagem + " " + resposta)
+            self.emocao_atual = analise["emocao"]
+
+            # Extrai gírias e salva no banco
+            self._extrair_e_salvar_girias(mensagem, resposta)
+
+        except Exception as e:
+            logger.warning(f"Erro ao atualizar contexto: {e}")
+
+    def _extrair_e_salvar_girias(self, msg: str, resp: str):
+        """Extrai gírias e salva no banco"""
+        texto = f"{msg} {resp}".lower()
+        palavras = [p for p in re.findall(r'\b\w{4,}\b', texto)
+                    if p not in {'não', 'que', 'com', 'pra', 'pro', 'uma', 'ele', 'ela'}]
+
+        contador = collections.Counter(palavras)
+        for palavra, freq in contador.most_common(5):
+            if freq > 1:
+                significado = "gíria local" if any(x in texto for x in ['puto', 'caralho', 'merda']) else "expressão comum"
+                self.db.salvar_giria_aprendida(self.user_id, palavra, significado, texto[:100])
+
+    def analisar_intencao_e_normalizar(self, mensagem: str, historico: List) -> Dict[str, Any]:
+        """Normaliza, detecta intenção, sentimento, ironia, meia frase, etc."""
+        mensagem = mensagem.strip()
+        if not mensagem:
+            return {"intencao": "vazia", "sentimento": "neutro", "usar_nome": False}
+
+        # Normalização
+        msg_lower = re.sub(r'[^\w\s\.,!?]', '', mensagem.lower())
+        msg_normalizada = self._substituir_termos_aprendidos(msg_lower)
+
+        # Detecção básica
+        saudacao = any(p in msg_lower for p in ["oi", "olá", "eai", "eae", "bom dia"])
+        despedida = any(p in msg_lower for p in ["tchau", "flw", "até logo", "bazar"])
+        pergunta = "?" in mensagem or any(p in msg_lower for p in ["como", "onde", "quem", "por que"])
+        grosseria = any(p in msg_lower for p in ["caralho", "puto", "merda", "fdp", "burro", "idiota"])
+
+        # Sentimento
+        positivo = any(p in msg_lower for p in ["fixe", "bué", "bom", "legal", "gosto", "adoro", "kkk", "rsrs"])
+        negativo = any(p in msg_lower for p in ["ruim", "chato", "droga", "ódio", "triste", "puto"])
+
+        sentimento = "positivo" if positivo else ("negativo" if negativo else "neutro")
+
+        # Ironia
+        ironia = (positivo and "perdi" in msg_lower) or (negativo and "melhor" in msg_lower)
+
+        # Meia frase
+        meia_frase = len(msg_lower.split()) <= 3 or "..." in mensagem
+
+        # Usar nome?
+        usar_nome = False
+        if saudacao or despedida or any(p in msg_lower for p in ["obrigado", "valeu"]):
+            prob = getattr(config, 'USAR_NOME_PROBABILIDADE', 0.4)
+            usar_nome = random.random() < prob
+
+        # Estilo
+        estilo = "curto" if len(mensagem) < 30 else "normal"
+        if grosseria:
+            estilo = "rude"
+
+        return {
+            "texto_normalizado": msg_normalizada,
+            "intencao": "saudacao" if saudacao else ("despedida" if despedida else ("pergunta" if pergunta else "conversa")),
+            "sentimento": sentimento,
+            "estilo": estilo,
+            "ironia": ironia,
+            "meia_frase": meia_frase,
+            "usar_nome": usar_nome,
+            "grosseria": grosseria
+        }
+
+    def _substituir_termos_aprendidos(self, texto: str) -> str:
+        """Substitui gírias aprendidas por significado (opcional, para análise)"""
+        for termo, info in self.termo_contexto.items():
+            if termo in texto:
+                texto = texto.replace(termo, info.get("significado", termo))
+        return texto
+
+    def aprender_termo_regional(self, termo: str, contexto: str, significado: str):
+        """Registra novo termo no banco e na memória"""
+        self.termo_contexto[termo] = {"contexto": contexto, "significado": significado}
+        try:
+            import json
+            self.db.salvar_aprendizado_detalhado(self.user_id, "termos", json.dumps(self.termo_contexto))
+        except:
+            pass
+
+    def obter_aprendizados(self) -> Dict[str, Any]:
+        """Retorna tudo que a IA aprendeu sobre o usuário"""
+        return {
+            "historico_recente": [f"U: {u} | A: {a}" for u, a in self.historico[-3:]],
+            "girias": [g["giria"] for g in self.girias_aprendidas[:5]],
+            "tom_predominante": self.ton_predominante,
+            "emocao_atual": self.emocao_atual,
+            "termos_aprendidos": list(self.termo_contexto.keys())[:5]
+        }