Update modules/contexto.py

modules/contexto.py

@@ -1,189 +1,81 @@
-import logging
-import re
-import random
-import time
-import collections
-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
+"""
+Contexto de conversa por usuário
+================================
+- Memória afetiva
+- Análise de emoção (retorna str)
+- Gírias aprendidas
+- Integração com banco
+"""
+import json
+from typing import List, Tuple, Dict, Any
+from loguru import logger
 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):
+    def __init__(self, db, user_key: str):
         self.db = db
-        self.user_id = user_id  # número ou nome do usuário
+        self.user_key = user_key
         self.historico: List[Tuple[str, str]] = []
-        self.girias_aprendidas: List[Dict[str, Any]] = []
+        self.emocao_atual: str = config.HUMOR_INICIAL  # str
+        self.termo_contexto: List[str] = []
+        self.girias_aprendidas: List[Dict[str, str]] = []
         self.ton_predominante: str = "neutro"
-        self.emocao_atual: str = "neutra"
-        self.termo_contexto: Dict[str, Dict] = {}
-        self.model = None
+        self._carregar_do_banco()
 
-        # 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)"""
+    def _carregar_do_banco(self):
         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"]
+            dados = self.db.carregar_contexto(self.user_key)
+            if dados:
+                self.historico = json.loads(dados.get("historico", "[]"))
+                self.emocao_atual = dados.get("emocao_atual", config.HUMOR_INICIAL)  # str
+                self.termo_contexto = json.loads(dados.get("termos", "[]"))
+                self.girias_aprendidas = json.loads(dados.get("girias", "[]"))
+                self.ton_predominante = dados.get("tom", "neutro")
+        except Exception as e:
+            logger.warning(f"Erro ao carregar contexto: {e}")
 
+    def salvar_no_banco(self):
+        try:
+            self.db.salvar_contexto(
+                self.user_key,
+                json.dumps(self.historico[-config.MEMORIA_MAX:]),
+                self.emocao_atual,  # str
+                json.dumps(self.termo_contexto),
+                json.dumps(self.girias_aprendidas),
+                self.ton_predominante
+            )
         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 = {}
+            logger.error(f"Erro ao salvar contexto: {e}")
 
     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)
-
+            girias = self.db.listar_girias_aprendidas(self.user_key)
+            self.girias_aprendidas = [{"giria": g[0], "significado": g[1]} for g in girias]
         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)
+            logger.warning(f"Erro ao atualizar gírias: {e}")
+
+    def analisar_emocoes_mensagem(self, mensagem: str) -> str:
+        """Retorna apenas a emoção como string"""
+        msg = mensagem.lower()
+        if any(p in msg for p in ["feliz", "fixe", "bué bom", "adoro", "rsrs", "kkk"]):
+            return "feliz"
+        if any(p in msg for p in ["triste", "chateado", "merda", "puto", "caralho"]):
+            return "triste"
+        if any(p in msg for p in ["irritada", "foda-se", "vsf", "burro"]):
+            return "irritada"
+        if any(p in msg for p in ["curioso", "o que", "como", "porquê"]):
+            return "curiosa"
+        return "neutra"
 
-        # 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_historico(self) -> List[Tuple[str, str]]:
+        return self.historico[-3:]
 
-    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]
-        }
+    def atualizar_contexto(self, mensagem: str, resposta: str):
+        self.historico.append((mensagem, resposta))
+        if len(self.historico) > config.MEMORIA_MAX:
+            self.historico.pop(0)
+        
+        # Atualiza emoção com base na mensagem do usuário
+        self.emocao_atual = self.analisar_emocoes_mensagem(mensagem)
+        
+        self.salvar_no_banco()