Spaces:
Running
Running
| """ | |
| Bot Registry - Detecta e gerencia interações bot-to-bot. | |
| Previne loops infinitos e responde apropriadamente. | |
| Criado como parte da Fase 3: Correções de Bot-to-Bot | |
| Data: 2026-05-15 | |
| """ | |
| import json | |
| import os | |
| from typing import Dict, List, Tuple | |
| from loguru import logger | |
| from datetime import datetime, timedelta | |
| class BotRegistry: | |
| """Gerencia lista de bots conhecidos e detecta comunicação bot-to-bot.""" | |
| def __init__(self, config_file: str = "bots.json"): | |
| self.logger = logger | |
| self.config_file = config_file | |
| self.bots: Dict[str, Dict] = {} | |
| self.interaction_history: Dict[str, List] = {} | |
| self.load_bot_list() | |
| def load_bot_list(self): | |
| """Carrega lista de bots conhecidos de arquivo JSON.""" | |
| try: | |
| if os.path.exists(self.config_file): | |
| with open(self.config_file, 'r', encoding='utf-8') as f: | |
| self.bots = json.load(f) | |
| self.logger.info(f"✅ Bot registry carregado: {len(self.bots)} bots") | |
| else: | |
| self.bots = self._get_default_bots() | |
| self.save_bot_list() | |
| self.logger.info(f"⚠️ Bot registry criado com padrões: {len(self.bots)} bots") | |
| except Exception as e: | |
| self.logger.error(f"❌ Erro ao carregar bot registry: {e}") | |
| self.bots = self._get_default_bots() | |
| def save_bot_list(self): | |
| """Salva lista de bots em arquivo.""" | |
| try: | |
| with open(self.config_file, 'w', encoding='utf-8') as f: | |
| json.dump(self.bots, f, indent=2, ensure_ascii=False) | |
| self.logger.debug(f"💾 Bot registry salvo") | |
| except Exception as e: | |
| self.logger.error(f"❌ Erro ao salvar bot registry: {e}") | |
| def _get_default_bots(self) -> Dict: | |
| """Retorna lista padrão de bots conhecidos.""" | |
| return { | |
| "83692085067963": { | |
| "name": "Isa-IA", | |
| "platform": "whatsapp", | |
| "type": "ai_assistant", | |
| "behavior": "crítica e debate", | |
| "action": "ignore", | |
| "added": datetime.now().isoformat() | |
| } | |
| } | |
| def is_bot(self, sender_id: str) -> Tuple[bool, Dict]: | |
| """ | |
| Verifica se sender_id é de um bot conhecido. | |
| Returns: | |
| (is_bot, bot_info) | |
| """ | |
| normalized_id = self._normalize_id(sender_id) | |
| if normalized_id in self.bots: | |
| return True, self.bots[normalized_id] | |
| for bot_id, bot_info in self.bots.items(): | |
| if normalized_id.startswith(bot_id) or bot_id.startswith(normalized_id): | |
| return True, bot_info | |
| return False, {} | |
| def should_respond_to_bot(self, bot_id: str) -> bool: | |
| """ | |
| Verifica se AKIRA deve responder a este bot. | |
| Retorna False se: | |
| - Bot está em lista de ignore | |
| - Há muitas interações recentes (loop detection) | |
| """ | |
| is_bot, bot_info = self.is_bot(bot_id) | |
| if not is_bot: | |
| return True | |
| if bot_info.get('action') == 'ignore': | |
| self.logger.info(f"🚫 [BOT-IGNORE] {bot_info.get('name')} está em ignore") | |
| return False | |
| if self._detect_loop(bot_id): | |
| self.logger.warning(f"⚠️ [LOOP-DETECTION] Loop detectado com {bot_info.get('name')}") | |
| return False | |
| return True | |
| def record_interaction(self, bot_id: str, direction: str = "out"): | |
| """ | |
| Registra interação com bot para detecção de loop. | |
| Args: | |
| bot_id: ID do bot | |
| direction: "in" (mensagem recebida) ou "out" (resposta enviada) | |
| """ | |
| normalized_id = self._normalize_id(bot_id) | |
| if normalized_id not in self.interaction_history: | |
| self.interaction_history[normalized_id] = [] | |
| self.interaction_history[normalized_id].append({ | |
| "timestamp": datetime.now().isoformat(), | |
| "direction": direction | |
| }) | |
| def _detect_loop(self, bot_id: str, window_minutes: int = 5) -> bool: | |
| """ | |
| Detecta se há muitas interações recentes (possível loop). | |
| Returns True se há 5+ troca em 5 minutos. | |
| """ | |
| normalized_id = self._normalize_id(bot_id) | |
| if normalized_id not in self.interaction_history: | |
| return False | |
| history = self.interaction_history[normalized_id][-10:] | |
| cutoff = datetime.now() - timedelta(minutes=window_minutes) | |
| recent = [ | |
| h for h in history | |
| if datetime.fromisoformat(h["timestamp"]) > cutoff | |
| ] | |
| if len(recent) >= 5: | |
| self.logger.warning(f"🔄 [LOOP] {len(recent)} interações em {window_minutes} min") | |
| return True | |
| return False | |
| def register_bot(self, bot_id: str, name: str, bot_type: str = "unknown", action: str = "ignore"): | |
| """Registra novo bot na lista.""" | |
| normalized_id = self._normalize_id(bot_id) | |
| self.bots[normalized_id] = { | |
| "name": name, | |
| "platform": "whatsapp", | |
| "type": bot_type, | |
| "action": action, | |
| "added": datetime.now().isoformat() | |
| } | |
| self.save_bot_list() | |
| self.logger.info(f"✅ Bot registrado: {name} ({normalized_id})") | |
| def _normalize_id(self, sender_id: str) -> str: | |
| """Normaliza sender_id removendo extensões.""" | |
| return sender_id.split('@')[0] if '@' in sender_id else sender_id | |
| def get_bot_response(self, sender_name: str, bot_info: Dict) -> str: | |
| """Retorna resposta apropriada quando detecta bot.""" | |
| return ( | |
| f"🤖 Detecto comunicação com outro bot ({bot_info.get('name', 'desconhecido')}). " | |
| f"Ignorando para evitar loops infinitos. " | |
| f"Prefiro conversa com humanos reais." | |
| ) | |
| # Instância global | |
| bot_registry = BotRegistry() | |