# modules/api.py — AKIRA V21 FINAL INTEGRADO (Dezembro 2025) - COM TRANSIÇÃO GRADUAL """ ✅ TOTALMENTE INTEGRADO com database.py corrigido (message_id sem UNIQUE) ✅ CORREÇÃO: Usa métodos corretos do Database atualizado ✅ COMPATÍVEL com index.js e reply_metadata ✅ Sistema multi-API com fallback ✅ Cache de contexto otimizado ✅ Treinamento automático integrado ✅ ADAPTADO: Sistema de transição gradual para usuários privilegiados ✅ SIMPLIFICADO: Usa apenas config.py para toda lógica de transição ✅ TRANSIÇÃO: 3 níveis para privilegiados seguindo tom do usuário ✅ INSTABILIDADE: Não mantém formal se conversa mudar para descontraído """ import time import datetime import requests import os import json import random import re from typing import Dict, Any, List, Optional from flask import Blueprint, request, jsonify, make_response from loguru import logger # Importa módulos locais - CORRETAMENTE from .contexto import Contexto, criar_contexto from .database import Database from .treinamento import Treinamento import modules.config as config # ============================================================================ # 🔥 CACHE SIMPLES COM TRANSIÇÃO # ============================================================================ class SimpleTTLCache: def __init__(self, ttl_seconds: int = 300): self.ttl = ttl_seconds self._store = {} def __contains__(self, key): if key not in self._store: return False _, expires = self._store[key] if time.time() > expires: del self._store[key] return False return True def __setitem__(self, key, value): self._store[key] = (value, time.time() + self.ttl) def __getitem__(self, key): if key not in self: raise KeyError(key) return self._store[key][0] def get(self, key, default=None): try: return self[key] except KeyError: return default # ============================================================================ # 🧠 GERENCIADOR MULTI-API (OTIMIZADO PARA CONFIG.PY) # ============================================================================ class MultiAPIManager: def __init__(self): self.timeout = config.API_TIMEOUT self.apis_disponiveis = self._verificar_apis() logger.info(f"✅ APIs disponíveis: {', '.join(self.apis_disponiveis)}") def _verificar_apis(self): """Verifica quais APIs estão disponíveis""" apis = [] if config.MISTRAL_API_KEY and len(config.MISTRAL_API_KEY) > 10: apis.append("mistral") if config.GEMINI_API_KEY and config.GEMINI_API_KEY.startswith('AIza'): apis.append("gemini") if config.GROQ_API_KEY and len(config.GROQ_API_KEY) > 10: apis.append("groq") if config.COHERE_API_KEY and len(config.COHERE_API_KEY) > 10: apis.append("cohere") return apis def _construir_prompt( self, mensagem: str, historico: List[Dict[str, str]], mensagem_citada: str, analise: Dict[str, Any], usuario: str, tipo_conversa: str, reply_info: Optional[Dict] = None ) -> str: """ Constrói prompt usando config.py - COM TRANSIÇÃO GRADUAL """ try: logger.debug(f"📝 Construindo prompt para: {usuario}") logger.debug(f"📝 Mensagem citada: {mensagem_citada[:80] if mensagem_citada else '[Nenhuma]'}...") # Usa a função do config.py prompt = config.construir_prompt_api( mensagem=mensagem, historico=historico, mensagem_citada=mensagem_citada, analise=analise, usuario=usuario, tipo_conversa=tipo_conversa, reply_info=reply_info ) logger.debug(f"✅ Prompt construído: {len(prompt)} caracteres") return prompt except Exception as e: logger.error(f"❌ Erro ao construir prompt: {e}") # Fallback simples return f"""AKIRA V21 — PERSONALIDADE REAL SEM FILTRO Você é Akira, angolana de 18 anos de Luanda. Angolana pura e patritota USUÁRIO: {usuario} MENSAGEM: {mensagem} MENSAGEM CITADA: {mensagem_citada if mensagem_citada else 'Nenhuma'} : entenda quem citou a messagem cidada, quem escreveu originalmete para vc sabe se foi vc ou outro caso seja de outro por exemplo: messagem citada de dani texto: kkk olha isso akira a messagem a ser citada de tiago: importante é que todos se fodam. nese casso vc deve entender que a messagem aser citada é a de tiago, e o dani citou essa messgem e também te mencionou ouseja que sua popinião sobre essa messagwem citada como ele dissse "olha isso " nesse caso vc deve encarar que não foii vc que disse, por isso não pode assumeir que disse nenhuma dessas coisas, no entanto pode dar sua opinião RESPONDA COMO AKIRA (IRÔNICA, SEM FILTRO, 1-2 LINHAS, mas dependo do tom que tá, não delire, tenha uma conversa focada, e racional):""" def _obter_parametros_api(self, analise: Dict[str, Any]) -> Dict[str, Any]: """Obtém parâmetros usando config.py""" emocao = analise.get("emocao_primaria", "neutral") modo = analise.get("modo_resposta", "normal_ironico") return config.obter_parametros_api(emocao, modo) def gerar_resposta( self, mensagem: str, historico: List[Dict[str, str]], mensagem_citada: str, analise: Dict[str, Any], usuario: str, tipo_conversa: str, reply_info: Optional[Dict] = None ) -> str: """Gera resposta usando multi-API fallback""" logger.info(f"🎯 Gerando resposta para {usuario}") logger.info(f"📤 Mensagem: {mensagem[:80]}...") # Construir prompt prompt = self._construir_prompt( mensagem=mensagem, historico=historico, mensagem_citada=mensagem_citada, analise=analise, usuario=usuario, tipo_conversa=tipo_conversa, reply_info=reply_info ) # Obter parâmetros parametros = self._obter_parametros_api(analise) logger.debug(f"🔧 Parâmetros: {parametros}") # Tentar APIs em ordem for api_name in config.API_FALLBACK_ORDER: if api_name not in self.apis_disponiveis: continue try: logger.debug(f"🔄 Tentando API: {api_name}") resposta = self._chamar_api(api_name, prompt, parametros) if resposta: resposta_limpa = self._limpar_resposta(resposta) logger.info(f"✅ API {api_name} respondeu: {resposta_limpa[:80]}...") return resposta_limpa except Exception as e: logger.warning(f"❌ API {api_name} falhou: {str(e)[:100]}") continue # Fallback contextual fallback = self._gerar_fallback_contextual(mensagem, mensagem_citada, reply_info) logger.info(f"🔄 Usando fallback: {fallback}") return fallback def _chamar_api(self, api_name: str, prompt: str, params: Dict[str, Any]) -> str: """Chama API específica""" try: if api_name == "mistral": return self._chamar_mistral(prompt, params) elif api_name == "gemini": return self._chamar_gemini(prompt, params) elif api_name == "groq": return self._chamar_groq(prompt, params) elif api_name == "cohere": return self._chamar_cohere(prompt, params) except Exception as e: logger.error(f"Erro ao chamar {api_name}: {e}") return "" def _chamar_mistral(self, prompt: str, params: Dict[str, Any]) -> str: """Chama Mistral API""" try: response = requests.post( "https://api.mistral.ai/v1/chat/completions", headers={"Authorization": f"Bearer {config.MISTRAL_API_KEY}"}, json={ "model": config.MISTRAL_MODEL, "messages": [{"role": "user", "content": prompt}], "max_tokens": params.get("max_tokens", config.MAX_TOKENS), "temperature": params.get("temperature", config.TEMPERATURE), "top_p": params.get("top_p", config.TOP_P), "frequency_penalty": params.get("frequency_penalty", config.FREQUENCY_PENALTY), "presence_penalty": params.get("presence_penalty", config.PRESENCE_PENALTY) }, timeout=self.timeout ) response.raise_for_status() return response.json()["choices"][0]["message"]["content"].strip() except Exception as e: logger.error(f"Mistral falhou: {e}") return "" def _chamar_gemini(self, prompt: str, params: Dict[str, Any]) -> str: """Chama Gemini API""" try: response = requests.post( f"https://generativelanguage.googleapis.com/v1beta/models/{config.GEMINI_MODEL}:generateContent?key={config.GEMINI_API_KEY}", json={ "contents": [{"parts": [{"text": prompt}]}], "generationConfig": { "temperature": params.get("temperature", config.TEMPERATURE), "maxOutputTokens": params.get("max_tokens", config.MAX_TOKENS) } }, timeout=self.timeout ) response.raise_for_status() return response.json()["candidates"][0]["content"]["parts"][0]["text"].strip() except Exception as e: logger.error(f"Gemini falhou: {e}") return "" def _chamar_groq(self, prompt: str, params: Dict[str, Any]) -> str: """Chama Groq API""" try: response = requests.post( "https://api.groq.com/openai/v1/chat/completions", headers={"Authorization": f"Bearer {config.GROQ_API_KEY}"}, json={ "model": config.GROQ_MODEL, "messages": [{"role": "user", "content": prompt}], "max_tokens": params.get("max_tokens", config.MAX_TOKENS), "temperature": params.get("temperature", config.TEMPERATURE) }, timeout=self.timeout ) response.raise_for_status() return response.json()["choices"][0]["message"]["content"].strip() except Exception as e: logger.error(f"Groq falhou: {e}") return "" def _chamar_cohere(self, prompt: str, params: Dict[str, Any]) -> str: """Chama Cohere API""" try: response = requests.post( "https://api.cohere.ai/v1/generate", headers={"Authorization": f"Bearer {config.COHERE_API_KEY}"}, json={ "prompt": prompt, "max_tokens": params.get("max_tokens", config.MAX_TOKENS), "temperature": params.get("temperature", config.TEMPERATURE) }, timeout=self.timeout ) response.raise_for_status() return response.json()["generations"][0]["text"].strip() except Exception as e: logger.error(f"Cohere falhou: {e}") return "" def _limpar_resposta(self, texto: str) -> str: """Limpa a resposta""" if not texto: return "…" # Remove markdown texto = re.sub(r'[\*`_]+', '', texto) # Remove aspas texto = texto.strip('"\'') # Remove prefixos texto = re.sub(r'^(Akira|AKIRA)[:\s\-]+', '', texto, flags=re.IGNORECASE) # Remove espaços extras texto = re.sub(r'\s+', ' ', texto) # Limita tamanho if len(texto) > 400: last_period = texto[:397].rfind('.') if last_period > 300: texto = texto[:last_period + 1] else: texto = texto[:397] + "..." return texto.strip() # ============================================================================ # 🎯 CLASSE PRINCIPAL AKIRA API (COM TRANSIÇÃO GRADUAL) # ============================================================================ class AkiraAPI: def __init__(self): """Inicializa API totalmente integrada""" self.api = Blueprint("akira_api", __name__) self.contexto_cache = SimpleTTLCache(ttl_seconds=300) # Inicializa Database CORRETAMENTE self.db = Database(config.DB_PATH) self.llm_manager = MultiAPIManager() # Configura treinamento se habilitado if config.START_PERIODIC_TRAINER: try: self.treinador = Treinamento( self.db, interval_hours=config.TRAINING_INTERVAL_HOURS ) self.treinador.start_periodic_training() logger.info("✅ Treinamento periódico iniciado") except Exception as e: logger.error(f"❌ Erro no treinador: {e}") self.treinador = None self._setup_routes() logger.success("🚀 AKIRA V21 FINAL inicializada (com transição gradual)") def _get_user_context(self, numero: str, tipo_conversa: str, grupo_id: str = "") -> Contexto: """Obtém contexto isolado""" if tipo_conversa == "grupo" and grupo_id: key = f"grupo_{grupo_id}" else: key = f"pv_{numero}" # Cache if key in self.contexto_cache: return self.contexto_cache[key] # Cria novo contexto usando a função correta ctx = criar_contexto(self.db, key, tipo_conversa) self.contexto_cache[key] = ctx return ctx def _processar_reset(self, numero: str, usuario: str, confirmacao: bool = False) -> Dict[str, Any]: """Processa comando /reset""" # Verifica permissão usando método CORRETO if not self.db.pode_usar_reset(numero): return { "error": "COMANDO RESTRITO", "resposta": "Só o boss pode usar /reset, puto." } if not confirmacao: return { "resposta": "Quer mesmo apagar tudo? Manda /reset de novo." } # Limpa cache self.contexto_cache._store.clear() # Reseta no banco resultado = self.db.resetar_contexto_usuario(numero, "completo") return { "resposta": f"Reset feito! {resultado.get('itens_apagados', 0)} itens apagados." } def _extrair_payload_indexjs(self, data: Dict[str, Any]) -> Dict[str, Any]: """Extrai dados do payload do index.js - ATUALIZADO""" payload = { 'numero': str(data.get('numero', '')).strip(), 'usuario': data.get('usuario', 'Anônimo').strip(), 'mensagem': data.get('mensagem', '').strip(), 'tipo_conversa': data.get('tipo_conversa', 'pv'), 'tipo_mensagem': data.get('tipo_mensagem', 'texto'), 'grupo_id': data.get('grupo_id', ''), 'grupo_nome': data.get('grupo_nome', ''), 'is_reset': False, 'reply_metadata': data.get('reply_metadata', {}), 'mensagem_citada': data.get('mensagem_citada', '') } # Log básico logger.debug(f"📦 Payload recebido de {payload['usuario']}") # Detecta /reset if payload['mensagem'].strip().lower() == '/reset': payload['is_reset'] = True return payload def _setup_routes(self): """Configura rotas da API""" @self.api.before_request def handle_options(): """Lida com CORS preflight""" if request.method == 'OPTIONS': resp = make_response() resp.headers['Access-Control-Allow-Origin'] = '*' resp.headers['Access-Control-Allow-Headers'] = 'Content-Type' resp.headers['Access-Control-Allow-Methods'] = 'POST,GET' return resp @self.api.after_request def add_cors(resp): """Adiciona headers CORS""" resp.headers['Access-Control-Allow-Origin'] = '*' return resp @self.api.route('/akira', methods=['POST']) def akira_endpoint(): """Endpoint principal - COM TRANSIÇÃO GRADUAL""" try: # Recebe payload data = request.get_json() or {} payload = self._extrair_payload_indexjs(data) logger.info( f"📨 [{payload['usuario']}] ({payload['numero']}): " f"{payload['mensagem'][:80]}..." ) # Validação if not payload['mensagem']: return jsonify({'error': 'Mensagem obrigatória'}), 400 # Comando /reset if payload['is_reset']: resultado = self._processar_reset( payload['numero'], payload['usuario'], confirmacao=False ) if 'error' in resultado: return jsonify(resultado), 403 return jsonify(resultado) # Obtém contexto contexto = self._get_user_context( numero=payload['numero'], tipo_conversa=payload['tipo_conversa'], grupo_id=payload['grupo_id'] ) # Atualiza informações do usuário contexto.atualizar_informacoes_usuario( nome=payload['usuario'], numero=payload['numero'], grupo_id=payload['grupo_id'], grupo_nome=payload['grupo_nome'], tipo_conversa=payload['tipo_conversa'] ) # Obtém histórico historico = contexto.obter_historico_para_llm() # VERIFICA USUÁRIO PRIVILEGIADO E TRANSIÇÃO usuario_privilegiado = config.eh_usuario_privilegiado(payload['numero']) modo_inicial = config.forcar_modo_inicial_privilegiado(payload['numero']) # Analisa a mensagem analise = contexto.analisar_intencao_e_normalizar( mensagem=payload['mensagem'], historico=historico, mensagem_citada=payload['mensagem_citada'], reply_metadata=payload['reply_metadata'] ) # ANALISA TOM DO USUÁRIO PARA TRANSIÇÃO analise_tom = config.analisar_tom_usuario(payload['mensagem'], historico) # Obtém nível atual de transição nivel_transicao_atual = analise.get('nivel_transicao', 1 if usuario_privilegiado else 0) # Histórico recente para análise de transição historico_recente = historico[-10:] if len(historico) >= 10 else historico # DETERMINA TRANSIÇÃO SE FOR PRIVILEGIADO if usuario_privilegiado: info_transicao = config.determinar_nivel_transicao( payload['numero'], analise_tom, nivel_transicao_atual, historico_recente ) # Atualiza modo baseado na transição analise['modo_resposta'] = info_transicao['modo'] analise['nivel_transicao'] = info_transicao['nivel'] analise['info_transicao'] = info_transicao logger.info(f"👑 Usuário privilegiado {payload['numero']} - Nível: {info_transicao['nivel']} ({info_transicao['desc']})") # Adiciona informações extras analise.update({ 'usuario_privilegiado': usuario_privilegiado, 'numero': payload['numero'], 'tipo_mensagem': payload['tipo_mensagem'], 'reply_metadata': payload['reply_metadata'] }) # Prepara reply_info para config.py reply_info_for_config = payload['reply_metadata'] # Gera resposta resposta = self.llm_manager.gerar_resposta( mensagem=payload['mensagem'], historico=historico, mensagem_citada=payload['mensagem_citada'], analise=analise, usuario=payload['usuario'], tipo_conversa=payload['tipo_conversa'], reply_info=reply_info_for_config ) # Determina se é reply is_reply = bool(payload['mensagem_citada']) or ( payload['reply_metadata'] and payload['reply_metadata'].get('is_reply', False) ) reply_to_bot = False if payload['reply_metadata']: reply_to_bot = payload['reply_metadata'].get('reply_to_bot', False) # Mede tempo de resposta tempo_resposta_ms = int((time.time() - request.start_time) * 1000) if hasattr(request, 'start_time') else 0 # CORREÇÃO: Prepara reply_info_json para o Database reply_info_json = None if payload['reply_metadata']: reply_info_json = json.dumps(payload['reply_metadata'], ensure_ascii=False) # Atualiza contexto usando o Database CORRIGIDO contexto.atualizar_contexto( mensagem=payload['mensagem'], resposta=resposta, numero=payload['numero'], is_reply=is_reply, mensagem_original=payload['mensagem_citada'], reply_to_bot=reply_to_bot ) # Salva mensagem diretamente no banco (backup) try: # Gera message_id único com timestamp e random timestamp = int(time.time() * 1000) random_suffix = random.randint(1000, 9999) message_id = f"{payload['numero']}_{timestamp}_{random_suffix}" # Salva nível de transição se for privilegiado nivel_salvar = analise.get('nivel_transicao', 0) desc_transicao = analise.get('info_transicao', {}).get('desc', 'N/A') self.db.salvar_mensagem( usuario=payload['usuario'], mensagem=payload['mensagem'], resposta=resposta, numero=payload['numero'], is_reply=is_reply, mensagem_original=payload['mensagem_citada'], reply_to_bot=reply_to_bot, humor=analise.get('humor_atualizado', 'normal_ironico'), modo_resposta=analise.get('modo_resposta', 'normal_ironico'), emocao_detectada=analise.get('emocao_primaria', 'neutral'), confianca_emocao=analise.get('confianca_emocao', 0.5), tipo_mensagem=payload['tipo_mensagem'], reply_info_json=reply_info_json, usuario_nome=payload['usuario'], grupo_id=payload['grupo_id'], grupo_nome=payload['grupo_nome'], tipo_conversa=payload['tipo_conversa'], message_id=message_id, bot_response_time_ms=tempo_resposta_ms, # Campos extras para transição nivel_transicao=nivel_salvar, desc_transicao=desc_transicao, usuario_privilegiado=usuario_privilegiado ) logger.debug(f"✅ Mensagem salva no banco com message_id: {message_id}") except Exception as db_error: logger.warning(f"⚠️ Erro ao salvar mensagem no banco: {db_error}") # Registra interação para treinamento if hasattr(self, 'treinador') and self.treinador: try: self.treinador.registrar_interacao( usuario=payload['usuario'], mensagem=payload['mensagem'], resposta=resposta, numero=payload['numero'], is_reply=is_reply, mensagem_original=payload['mensagem_citada'], contexto=analise, tipo_conversa=payload['tipo_conversa'], tipo_mensagem=payload['tipo_mensagem'], reply_to_bot=reply_to_bot, reply_metadata=payload['reply_metadata'], nivel_transicao=analise.get('nivel_transicao', 0) ) logger.debug("✅ Interação registrada para treinamento") except Exception as e: logger.warning(f"⚠️ Erro ao registrar interação: {e}") logger.info(f"📤 Resposta: {resposta[:80]}...") # Log da transição se for privilegiado if usuario_privilegiado: info = analise.get('info_transicao', {}) logger.info(f"🔄 Transição: Nível {info.get('nivel', 1)} - {info.get('desc', 'N/A')}") return jsonify({"resposta": resposta}) except Exception as e: logger.error(f"❌ Erro em /akira: {e}") import traceback traceback.print_exc() return jsonify({ "error": "Erro interno", "resposta": "Erro interno, puto. Tenta de novo." }), 500 @self.api.route('/health', methods=['GET']) def health(): """Health check""" agora = datetime.datetime.now() + datetime.timedelta(hours=config.TIMEZONE_OFFSET_HOURS) # Conta usuários privilegiados privilegiados_count = len(config.USUARIOS_PRIVILEGIADOS) # Verifica APIs apis_ok = self.llm_manager.apis_disponiveis return jsonify({ "status": "✅ AKIRA V21 ONLINE COM TRANSIÇÃO GRADUAL", "hora_luanda": agora.strftime("%H:%M"), "versao": config.VERSAO, "database": "Corrigido (message_id sem UNIQUE)", "apis_disponiveis": apis_ok, "cache_size": len(self.contexto_cache._store), "treinamento_ativo": hasattr(self, 'treinador') and self.treinador is not None, "transicao_gradual": { "usuarios_privilegiados": privilegiados_count, "modo_inicial_privilegiados": "filosofico_ironico", "niveis_transicao": 3, "descricao": "Privilegiados começam formal, adaptam-se gradualmente" } }) @self.api.route('/reset', methods=['POST']) def reset_endpoint(): """Endpoint de reset dedicado""" try: data = request.get_json() or {} numero = str(data.get('numero', '')).strip() if not numero: return jsonify({"error": "Número obrigatório"}), 400 resultado = self._processar_reset(numero, "admin", confirmacao=True) if 'error' in resultado: return jsonify(resultado), 403 return jsonify(resultado) except Exception as e: logger.error(f"Erro em /reset: {e}") return jsonify({"error": "Erro interno"}), 500 @self.api.route('/info', methods=['GET']) def info(): """Informações da API""" # Lista usuários privilegiados privilegiados_info = [] for numero, dados in config.USUARIOS_PRIVILEGIADOS.items(): privilegiados_info.append({ "numero": numero, "nome": dados.get("nome", "Desconhecido"), "modo_inicial": dados.get("modo_inicial", "filosofico_ironico"), "transicao_permitida": dados.get("transicao_permitida", True) }) # Informações de transição transicao_info = { "niveis": 3, "descricao_niveis": { 1: "Nível 1 - Formal Completo (filosofico_ironico)", 2: "Nível 2 - Formal Relaxado (tecnico_formal) esse tom deve ser usado por padrão para usarios priveleigiados, e para topicos academicos", 3: "Nível 3 - Normal (normal_ironico)" }, "regras": [ "Privilegiados começam no Nível 1", "Transição gradual baseada no tom do usuário", "Não mantém formal se conversa mudar para descontraída, mas isso deve ser lento e gradual", "Adaptação natural seguindo ritmo da conversa" ] } return jsonify({ "nome": "Akira V21", "descricao": "IA com personalidade brutal e irônica", "desenvolvedor": "Isaac Quarenta", "empresa": "Softedge", "versao": config.VERSAO, "database_status": "Corrigido - message_id sem UNIQUE constraint", "usuarios_privilegiados": privilegiados_info, "sistema_transicao": transicao_info, "endpoints": ["/akira", "/health", "/reset", "/info", "/teste/privilegiado", "/transicao/info"], "configuracoes": { "temperatura_padrao": config.TEMPERATURE, "max_tokens": config.MAX_TOKENS, "timezone_offset": config.TIMEZONE_OFFSET_HOURS, "treinamento_auto": config.START_PERIODIC_TRAINER, "modo_inicial_privilegiados": "filosofico_ironico", "transicao_gradual": "ativada" } }) @self.api.route('/teste/privilegiado', methods=['POST']) def teste_privilegiado(): """Endpoint para testar usuário privilegiado""" try: data = request.get_json() or {} numero = str(data.get('numero', '')).strip() if not numero: return jsonify({"error": "Número obrigatório"}), 400 # Testa usando config.py eh_privilegiado = config.eh_usuario_privilegiado(numero) dados_privilegiado = config.verificar_usuario_privilegiado(numero) modo_inicial = config.forcar_modo_inicial_privilegiado(numero) permite_transicao = config.transicao_permitida_privilegiado(numero) return jsonify({ "numero": numero, "eh_privilegiado": eh_privilegiado, "dados_privilegiado": dados_privilegiado, "modo_inicial": modo_inicial, "permite_transicao": permite_transicao, "instrucao": "Usuário privilegiado começa formal, adapta-se gradualmente" if eh_privilegiado else "Usuário normal usa modo padrão" }) except Exception as e: logger.error(f"Erro em /teste/privilegiado: {e}") return jsonify({"error": "Erro interno"}), 500 @self.api.route('/transicao/info', methods=['GET']) def transicao_info(): """Informações sobre o sistema de transição""" return jsonify({ "sistema": "Transição Gradual para Usuários Privilegiados", "descricao": "Sistema que permite a Akira adaptar-se gradualmente ao tom da conversa", "niveis": [ { "nivel": 1, "nome": "Formal Completo", "modo": "filosofico_ironico", "descricao": "Tom respeitoso, sem gírias, formal", "exemplo": "A existência é absurda por natureza." }, { "nivel": 2, "nome": "Formal Relaxado", "modo": "tecnico_formal", "descricao": "Respeitoso mas com leve ironia, algumas gírias", "exemplo": "Ya, isso faz sentido." }, { "nivel": 3, "nome": "Normal", "modo": "normal_ironico", "descricao": "Gírias normais, ironia normal (sem xingar)", "exemplo": "Puto, tá certo." } ], "regras_transicao": [ "Privilegiados sempre começam no Nível 1", "Analisa tom do usuário a cada mensagem", "2-3 mensagens descontraídas → avança um nível", "Volta a sério → retorna gradualmente", "NÃO mantém formal se conversa mudou para descontraída" ], "usuarios_privilegiados": list(config.USUARIOS_PRIVILEGIADOS.keys()) }) @self.api.route('/transicao/simular', methods=['POST']) def transicao_simular(): """Simula transição com histórico de mensagens""" try: data = request.get_json() or {} numero = str(data.get('numero', '')).strip() mensagens = data.get('mensagens', []) if not numero: return jsonify({"error": "Número obrigatório"}), 400 if not mensagens: return jsonify({"error": "Lista de mensagens obrigatória"}), 400 eh_privilegiado = config.eh_usuario_privilegiado(numero) if not eh_privilegiado: return jsonify({ "resultado": "Usuário não é privilegiado", "modo_constante": "normal_ironico" }) # Simula transição historico_simulado = [] nivel_atual = 1 resultados = [] for i, mensagem in enumerate(mensagens): # Analisa tom analise_tom = config.analisar_tom_usuario(mensagem, historico_simulado) # Determina transição info_transicao = config.determinar_nivel_transicao( numero, analise_tom, nivel_atual, historico_simulado[-5:] if len(historico_simulado) >= 5 else historico_simulado ) # Atualiza nível nivel_atual = info_transicao['nivel'] # Adiciona ao histórico simulado historico_simulado.append({"mensagem": mensagem}) resultados.append({ "mensagem_numero": i + 1, "mensagem": mensagem, "tom_detectado": analise_tom.get("tom"), "nivel_anterior": info_transicao.get("nivel_anterior", nivel_atual), "nivel_atual": nivel_atual, "modo": info_transicao["modo"], "deve_transicionar": info_transicao["deve_transicionar"], "direcao": info_transicao["direcao"] }) return jsonify({ "usuario": numero, "privilegiado": True, "simulacao": resultados, "nivel_final": nivel_atual, "modo_final": resultados[-1]["modo"] if resultados else "filosofico_ironico" }) except Exception as e: logger.error(f"Erro em /transicao/simular: {e}") return jsonify({"error": "Erro interno"}), 500 def get_blueprint(self): """Retorna o blueprint para Flask""" return self.api # ============================================================================ # 🎯 FUNÇÃO PARA USO DIRETO (COM TRANSIÇÃO) # ============================================================================ def gerar_resposta_direta( mensagem: str, usuario: str = "Anônimo", numero: str = "000000000", tipo_conversa: str = "pv", tipo_mensagem: str = "texto", mensagem_citada: str = "", reply_metadata: Optional[Dict] = None, historico: Optional[List[Dict]] = None, nivel_transicao_atual: int = 1 ) -> str: """ Função para uso direto (sem HTTP) - COM TRANSIÇÃO Args: mensagem: Mensagem do usuário usuario: Nome do usuário numero: Número do usuário tipo_conversa: Tipo da conversa tipo_mensagem: Tipo da mensagem mensagem_citada: Mensagem citada reply_metadata: Metadata do reply historico: Histórico da conversa nivel_transicao_atual: Nível atual de transição Returns: Resposta gerada """ try: # Cria instância simplificada llm_manager = MultiAPIManager() # Verifica usuário privilegiado usuario_privilegiado = config.eh_usuario_privilegiado(numero) # Histórico padrão se não fornecido historico = historico or [] # ANALISA TOM E TRANSIÇÃO SE FOR PRIVILEGIADO analise_tom = config.analisar_tom_usuario(mensagem, historico) if usuario_privilegiado: info_transicao = config.determinar_nivel_transicao( numero, analise_tom, nivel_transicao_atual, historico[-5:] if len(historico) >= 5 else historico ) modo_resposta = info_transicao['modo'] novo_nivel = info_transicao['nivel'] else: modo_resposta = 'normal_ironico' novo_nivel = 0 # Cria análise básica analise = { 'numero': numero, 'usuario_privilegiado': usuario_privilegiado, 'emocao_primaria': 'neutral', 'tipo_mensagem': tipo_mensagem, 'reply_metadata': reply_metadata, 'modo_resposta': modo_resposta, 'nivel_transicao': novo_nivel, 'info_transicao': info_transicao if usuario_privilegiado else {} } # Gera resposta resposta = llm_manager.gerar_resposta( mensagem=mensagem, historico=historico, mensagem_citada=mensagem_citada, analise=analise, usuario=usuario, tipo_conversa=tipo_conversa, reply_info=reply_metadata ) logger.info(f"✅ Resposta direta para {usuario}: {resposta[:80]}...") if usuario_privilegiado: logger.info(f"🔄 Nível transição: {novo_nivel} ({info_transicao.get('desc', 'N/A')})") return resposta except Exception as e: logger.error(f"❌ Erro na resposta direta: {e}") return "Puto, erro. Tenta de novo." # ============================================================================ # 🎯 TESTE # ============================================================================ if __name__ == "__main__": print("=" * 80) print("TESTANDO API.PY - COM SISTEMA DE TRANSIÇÃO GRADUAL") print("=" * 80) # Testes de transição test_cases = [ { "nome": "Isaac Quarenta - Início Formal", "numero": "244978787009", "mensagens": [ "Precisamos revisar o código do projeto.", "Analise a arquitetura atual.", "O sistema precisa de otimização." ], "expectativa": "Nível 1 (Formal) mantido" }, { "nome": "Isaac Quarenta - Transição para Descontraído", "numero": "244978787009", "mensagens": [ "Tá tudo fixe?", "Ya, tás a brincar hoje?", "kkk, relaxa mano", "De boa, tás tranquilo?" ], "expectativa": "Nível 1 → 2 → 3 (gradual)" }, { "nome": "Isaac Quarenta - Volta a Sério", "numero": "244937035662", "mensagens": [ "kkk, tás louco", "Brincadeira à parte...", "Precisamos falar sério do projeto.", "Analise este código: def func(): pass" ], "expectativa": "Nível 3 → 2 → 1 (gradual)" }, { "nome": "Usuário Normal", "numero": "123456789", "mensagens": [ "Ei, tudo bem?", "Você é um bot?", "Vai à merda!" ], "expectativa": "Modo normal sempre" } ] for i, test_case in enumerate(test_cases, 1): print(f"\n🔍 TESTE {i}: {test_case['nome']}") print(f" Número: {test_case['numero']}") # Verifica se é privilegiado eh_privilegiado = config.eh_usuario_privilegiado(test_case['numero']) modo_inicial = config.forcar_modo_inicial_privilegiado(test_case['numero']) permite_transicao = config.transicao_permitida_privilegiado(test_case['numero']) print(f" É privilegiado? {eh_privilegiado}") print(f" Modo inicial: {modo_inicial}") print(f" Permite transição? {permite_transicao}") # Simula conversa historico_simulado = [] nivel_atual = 1 for j, mensagem in enumerate(test_case['mensagens']): resposta = gerar_resposta_direta( mensagem=mensagem, usuario=test_case['nome'], numero=test_case['numero'], historico=historico_simulado, nivel_transicao_atual=nivel_atual ) # Atualiza nível se for privilegiado if eh_privilegiado: analise_tom = config.analisar_tom_usuario(mensagem, historico_simulado) info = config.determinar_nivel_transicao( test_case['numero'], analise_tom, nivel_atual, historico_simulado[-5:] if len(historico_simulado) >= 5 else historico_simulado ) nivel_atual = info['nivel'] historico_simulado.append({"mensagem": mensagem, "resposta": resposta}) print(f" Msg {j+1}: '{mensagem[:30]}...'") print(f" Resposta: {resposta[:50]}...") if eh_privilegiado: print(f" Nível: {nivel_atual}") print(f" Expectativa: {test_case['expectativa']}") print("\n" + "=" * 80) print("✅ API.PY - SISTEMA DE TRANSIÇÃO GRADUAL IMPLEMENTADO") print("✅ Usuários privilegiados: Formal → Relaxado → Normal") print("✅ Transição gradual baseada no tom do usuário") print("✅ Não mantém formal se conversa mudou") print("✅ Endpoints: /akira, /health, /info, /transicao/info, /transicao/simular") print("=" * 80)