import json import threading import re from loguru import logger from typing import List, Dict, Any, Optional # Imports robustos com fallback try: from .database import Database from . import config except ImportError: try: from modules.database import Database import modules.config as config except ImportError: Database = None config = None class PersonaTracker: """ Rastreador de Persona em Background (Character.AI style LTM). Analisa as conversas recentes do usuário silenciosamente e extrai seus traços de personalidade, gostos e emoções no banco de dados. """ def __init__(self, db: Database, llm_client: Any): """ Args: db (Database): Instância do banco de dados (database.py) llm_client (Any): Instância do cliente LLM (ex: MultiLLMClient) """ self.db = db self.llm_client = llm_client self.processing_users = set() def track_background(self, numero_usuario: str, historico_recente: List[Dict[str, str]]) -> None: """ Dispara a análise de persona em background para não bloquear a resposta do bot. Args: numero_usuario: ID ou número do usuário. historico_recente: Lista de dicionários {'role': '...', 'content': '...'} com as últimas mensagens do usuário. """ if numero_usuario in self.processing_users: return # Já está a ser analisado neste momento if not historico_recente or len(historico_recente) < 3: return # Muito pouco contexto para extrair algo útil self.processing_users.add(numero_usuario) thread = threading.Thread( target=self._analyze_and_save, args=(numero_usuario, historico_recente), daemon=True ) thread.start() @staticmethod def _safe_serialize(data: Any) -> Any: """Converte datetime objects para strings para serialização JSON segura.""" if isinstance(data, dict): return {k: PersonaTracker._safe_serialize(v) for k, v in data.items()} elif isinstance(data, list): return [PersonaTracker._safe_serialize(item) for item in data] elif hasattr(data, 'isoformat'): # datetime, date, etc. return data.isoformat() return data def _analyze_and_save(self, numero_usuario: str, historico: List[Dict[str, str]]) -> None: """Método interno que roda na Thread.""" try: # Recupera a persona atual para o LLM saber o que já sabemos persona_atual = self.db.recuperar_persona(numero_usuario) or {} # Serializa datetime objects para strings antes do json.dumps persona_atual = self._safe_serialize(persona_atual) # Formata histórico apenas com as falas do usuário user_messages = [] for msg in historico: if msg.get('role') != 'user': continue c = msg.get('content', '') if msg.get('listen'): c = f"[grupo listen] {c}" user_messages.append(c) if not user_messages: return historico_texto = "\n".join([f"User: {msg}" for msg in user_messages[-10:]]) perfil_atual_str = json.dumps(persona_atual, ensure_ascii=False) if persona_atual else "Ainda não definido." prompt = f"""Você é um analista comportamental focado em rastreamento de persona (Long-Term Memory). Analise as mensagens recentes deste usuário e atualize/extraia o seu perfil. [PERFIL ATUAL NO BANCO DE DADOS] {perfil_atual_str} [MENSAGENS RECENTES] {historico_texto} EXTRAIA/ATUALIZE os seguintes traços com base APENAS nas mensagens recentes e no perfil atual. Mantenha os traços do perfil atual que não foram contraditórios. Seja CONCISO. Use bullet points curtos na sua mente e preencha os campos em formato JSON estrito. Retorne APENAS um JSON válido. É OBRIGATÓRIO USAR ASPAS DUPLAS NAS CHAVES E NOS VALORES ("chave": "valor"): {{ "personalidade": "Resumo calmo, agressivo, divertido, direto, etc.", "vicios_linguagem": "Expressões ou gírias que ele usa muito.", "gostos": "O que ele demonstrou gostar ou tópicos de interesse.", "desgostos": "O que o irrita, o que ele odeia.", "emocional": "Traços emocionais, forças ou gatilhos/fraquezas." }} """ # Chama o LLM (garante formato json) # Agora retorna (resposta, modelo_usado) ou apenas resposta response_raw = self.llm_client.generate(prompt, []) modelo_usado = "desconhecido" if isinstance(response_raw, tuple): response_json_str = response_raw[0] modelo_usado = response_raw[1] if len(response_raw) > 1 else "desconhecido" else: response_json_str = response_raw if not response_json_str: return # Extrai o JSON (Robusto contra texto extra, markdown e quebras parciais) response_clean = response_json_str.strip() # 1. Localiza o início do JSON, permitindo quebras (truncado) if '{' in response_clean: start_pts = response_clean.find('{') # Procura o fechamento de chave mais à direita que seja válido # ou limita se o LLM alucinou e não fechou end_pts = response_clean.rfind('}') if end_pts > start_pts: response_clean = response_clean[start_pts:end_pts+1] else: # Se não encontrou o '}', tenta fechar artificialmente response_clean = response_clean[start_pts:] + '}' else: # Se não tem '{', tenta criar um wrapper se parecer um objeto response_clean = "{" + response_clean.strip(" \n\r,") + "}" # 2. Normalização agressiva de caracteres response_clean = response_clean.replace('\r', '').replace('\n', ' ') response_clean = re.sub(r'\s+', ' ', response_clean) # Remove múltiplos espaços response_clean = re.sub(r'\\+', r'\\', response_clean) # Tenta converter aspas simples em duplas para chaves/valores response_clean = re.sub(r"(?"\g<2>":', response_clean) dados_extraidos = json.loads(rc_temp) parsed_success = True except Exception: pass # Fallback extremo 2: Modo de extração de emergência (Regex por Campo) # Ideal para '{ personalidade: Direto, ..., vicios_linguagem: x, ... }' if not parsed_success or not isinstance(dados_extraidos, dict): logger.warning(f"Iniciando MODO DE EMERGÊNCIA (Regex) para Persona de {numero_usuario}...") dados_extraidos = {} # Regex para pegar chave: valor mesmo sem aspas, parando em vírgula ou fim de objeto patterns = { "personalidade": r"personalidade[\"']?\s*[:=]\s*([^,}]+)", "vicios_linguagem": r"vicios_?linguagem[\"']?\s*[:=]\s*([^,}]+)", "gostos": r"gostos[\"']?\s*[:=]\s*([^,}]+)", "desgostos": r"desgostos[\"']?\s*[:=]\s*([^,}]+)", "emocional": r"emocional[\"']?\s*[:=]\s*([^,}]+)" } for chave, pattern in patterns.items(): match = re.search(pattern, response_clean, re.IGNORECASE) if match: val = match.group(1).strip() if (val.startswith('"') and val.endswith('"')) or (val.startswith("'") and val.endswith("'")): val = val[1:-1].strip() dados_extraidos[chave] = val if dados_extraidos: parsed_success = True if not parsed_success: logger.warning(f"Iniciando MODO DE EMERGÊNCIA (Fatiamento) para Persona de {numero_usuario}...") dados_extraidos = {} chaves_possiveis = ["personalidade", "vicios_linguagem", "vicioslinguagem", "gostos", "desgostos", "emocional"] # 1. Encontra todas as ocorrências de todas as chaves e suas posições posicoes = [] for chave in chaves_possiveis: # Procura "chave:" ou "chave =" ou "'chave':" etc. for m in re.finditer(rf"['\"]?{chave}['\"]?\s*[:=]", response_clean, re.IGNORECASE): posicoes.append({ "chave": chave, "inicio_valor": m.end(), "pos_chave": m.start() }) # Ordena por posição no texto posicoes.sort(key=lambda x: x["pos_chave"]) # 2. Extrai o conteúdo entre as chaves for i in range(len(posicoes)): p_atual = posicoes[i] chave_original = p_atual["chave"] fim_valor = posicoes[i+1]["pos_chave"] if i + 1 < len(posicoes) else len(response_clean) valor = response_clean[p_atual["inicio_valor"]:fim_valor].strip() # Limpeza agressiva do valor valor = re.sub(r'^[\s\'"{\[:]+|[\s\'"}\],:]+$', '', valor).strip() if valor and len(valor) > 2: real_key = "vicios_linguagem" if chave_original == "vicioslinguagem" else chave_original dados_extraidos[str(real_key)] = valor if dados_extraidos: parsed_success = True if not dados_extraidos: # Se falhou tudo, mas temos a string, tentamos salvar a string bruta como nota logger.warning(f"Falha total no Parser JSON do Persona Tracker para {numero_usuario}. Salvando payload bruto como nota.") dados_extraidos = {"personalidade": response_json_str[:250]} parsed_success = True # Limpa chaves inválidas chaves_validas = ["personalidade", "vicios_linguagem", "gostos", "desgostos", "emocional"] campos_atualizar = {k: str(v) for k, v in dados_extraidos.items() if k in chaves_validas} if campos_atualizar: sucesso = self.db.atualizar_persona(numero_usuario, campos_atualizar) if sucesso: logger.info(f"✅ Persona LTM atualizada para o usuário {numero_usuario} em background via [{modelo_usado}].") else: logger.warning(f"Falha ao salvar a persona no banco para {numero_usuario}.") except json.JSONDecodeError: logger.warning(f"Falha no Parser JSON do Persona Tracker para {numero_usuario}.") except Exception as e: logger.error(f"Erro no Persona Tracker background: {e}") finally: if numero_usuario in self.processing_users: self.processing_users.remove(numero_usuario)