""" ================================================================================ AKIRA — AUTONOMOUS AGENT (Motor de Decisão Autónoma) ================================================================================ Motor central que permite à Akira: 1. Tomar decisões de moderação SEM ser chamada (proativo) 2. Analisar contexto do grupo e decidir ações 3. Executar ações de infraestrutura autonomamente 4. Comunicar resultados sigilosos ao proprietário via DM Filosofia: "Kiami não espera ordens — ela age quando necessário" ================================================================================ """ import json import time from typing import Dict, Any, List, Optional from loguru import logger from datetime import datetime # Limites para moderação autónoma SPAM_MSG_COUNT = 5 # Msgs num curto período → spam SPAM_WINDOW_SECS = 10 # Janela de tempo para detetar spam FLOOD_MSG_COUNT = 3 # Msgs muito rápidas → flood FLOOD_WINDOW_SECS = 2 # Janela para flood class AutonomousAgent: """ Motor de decisão autónoma da Akira. Permite: - Analisar contexto de grupo e tomar decisões de moderação - Executar skills autonomamente com base em eventos - Reportar ações sigilosas ao proprietário """ def __init__(self): self._spam_tracker: Dict[str, List[float]] = {} # {user_jid: [timestamps]} self._action_log: List[Dict] = [] # Log de ações autónomas tomadas self._db = None self._llm_caller = None def init(self, db_instance=None, llm_caller=None): """Inicializa com dependências.""" self._db = db_instance self._llm_caller = llm_caller logger.info("🤖 [AUTONOMOUS AGENT] Motor de decisão autónoma iniciado.") # ───────────────────────────────────────────────────────────────── # ANÁLISE DE COMPORTAMENTO DE GRUPO # ───────────────────────────────────────────────────────────────── def track_message(self, user_jid: str, group_jid: str, message: str, timestamp: float = None) -> Dict[str, Any]: """ Regista uma mensagem e avalia se deve tomar acção autónoma. Retorna: Dict com ação recomendada (pode ser vazio se não há ação) """ if timestamp is None: timestamp = time.time() key = f"{group_jid}:{user_jid}" # Adiciona timestamp ao tracker if key not in self._spam_tracker: self._spam_tracker[key] = [] self._spam_tracker[key].append(timestamp) # Limpa timestamps antigos (> 60s) self._spam_tracker[key] = [ ts for ts in self._spam_tracker[key] if timestamp - ts <= 60 ] if not self._spam_tracker[key]: del self._spam_tracker[key] # ─── Detecta Flood ─── recent_flood = [ts for ts in self._spam_tracker[key] if timestamp - ts <= FLOOD_WINDOW_SECS] if len(recent_flood) >= FLOOD_MSG_COUNT: logger.warning(f"🚨 [AGENT] Flood detetado: {user_jid} em {group_jid}") return self._build_moderation_action( action="mute", target_jid=user_jid, group_jid=group_jid, reason=f"Flood: {len(recent_flood)} mensagens em {FLOOD_WINDOW_SECS}s", duration_minutes=10, severity="AVISO" ) # ─── Detecta Spam ─── recent_spam = [ts for ts in self._spam_tracker[key] if timestamp - ts <= SPAM_WINDOW_SECS] if len(recent_spam) >= SPAM_MSG_COUNT: logger.warning(f"🚨 [AGENT] Spam detetado: {user_jid} em {group_jid}") return self._build_moderation_action( action="mute", target_jid=user_jid, group_jid=group_jid, reason=f"Spam: {len(recent_spam)} mensagens em {SPAM_WINDOW_SECS}s", duration_minutes=30, severity="CRÍTICO" ) return {} # Sem ação def analyze_message_for_moderation(self, message: str, user_jid: str, group_jid: str) -> Dict[str, Any]: """ Analisa o conteúdo de uma mensagem para decidir se há necessidade de moderação. Deteta: - Links externos em grupos com anti-link → BAN + blacklist + delete - Conteúdo explicitamente ofensivo - Ameaças ou doxxing """ message_lower = message.lower() # Padrões de links externos link_patterns = ["http://", "https://", "t.me/", "bit.ly/", "wa.me/", "discord.gg/"] has_link = any(p in message_lower for p in link_patterns) if has_link: return self._build_moderation_action( action="ban", target_jid=user_jid, group_jid=group_jid, reason="Link externo detetado — ban automático + blacklist", severity="CRÍTICO", blacklist=True ) return {} def analyze_image_description_for_moderation(self, description: str, user_jid: str, group_jid: str) -> Dict[str, Any]: """ [AGENTE AUTÓNOMO VISUAL] Verifica se a descrição gerada pelo modelo de visão da imagem contém conteúdo explicitamente proibido (NSFW/Gore). """ if not description: return {} desc_lower = description.lower() # 🔧 Tenta carregar padrões NSFW do PG nsfw_keywords = None if self._db and hasattr(self._db, 'get_moderation_patterns_grouped'): try: db_p = self._db.get_moderation_patterns_grouped() if db_p.get('nsfw'): nsfw_keywords = [p['pattern'] for p in db_p['nsfw']] except Exception: pass if not nsfw_keywords: # Padrões baseados em descrições comuns de NSFW ou Gore por LLMs de visão nsfw_keywords = [ "pornografia", "nudez", "sexo explícito", "conteúdo adulto", "nsfw", "sangue extremo", "vísceras", "gore", "mutilação", "violência extrema", "nudez explícita", "ato sexual", "porn" ] if any(keyword in desc_lower for keyword in nsfw_keywords): logger.warning(f"🚨 [AGENT-VISÃO] Imagem Proibida detetada: {user_jid} em {group_jid}") return { "type": "remote_action", "action": "autonomous_action", "params": { "cmd": "delete_message", "target": user_jid, "group_jid": group_jid, "reason": "Imagem NSFW/Gore bloqueada pelo escudo visual da Akira", "notify_owner": True, "silent": True } } return {} def analyze_hostility_for_action(self, hostility_level: float, user_jid: str, group_jid: str) -> Dict[str, Any]: """ [AGENTE AUTÓNOMO] Decide ação com base no nível de hostilidade detetado. Níveis: >= 90: ban + blacklist + notificar proprietário >= 70: mute 120min + aviso >= 50: mute 30min >= 30: aviso apenas """ if hostility_level >= 90: logger.warning(f"🚨 [AGENT-HOSTIL] Hostilidade EXTREMA ({hostility_level}) de {user_jid} em {group_jid}") return self._build_moderation_action( action="ban", target_jid=user_jid, group_jid=group_jid, reason=f"Hostilidade extrema nível {hostility_level}/100 — utilizador banido e adicionado à blacklist", severity="CRÍTICO", blacklist=True ) elif hostility_level >= 70: logger.warning(f"🚨 [AGENT-HOSTIL] Hostilidade ALTA ({hostility_level}) de {user_jid} em {group_jid}") return self._build_moderation_action( action="mute", target_jid=user_jid, group_jid=group_jid, reason=f"Hostilidade alta nível {hostility_level}/100 — mute prolongado aplicado", duration_minutes=120, severity="CRÍTICO" ) elif hostility_level >= 50: logger.warning(f"⚠️ [AGENT-HOSTIL] Hostilidade MÉDIA ({hostility_level}) de {user_jid} em {group_jid}") return self._build_moderation_action( action="mute", target_jid=user_jid, group_jid=group_jid, reason=f"Hostilidade média nível {hostility_level}/100", duration_minutes=30, severity="AVISO" ) elif hostility_level >= 30: logger.info(f"⚠️ [AGENT-HOSTIL] Hostilidade BAIXA ({hostility_level}) de {user_jid} em {group_jid}") return self._build_moderation_action( action="warn", target_jid=user_jid, group_jid=group_jid, reason=f"Hostilidade detectada nível {hostility_level}/100 — aviso emitido", severity="AVISO" ) return {} def analyze_toxic_language(self, message: str, user_jid: str, group_jid: str) -> Dict[str, Any]: """ [AGENTE AUTÓNOMO] Deteta linguagem tóxica/ofensiva em português. Padrões insultuosos comuns angolanos/portugueses, ameaças de violência. Carrega padrões do PG quando disponível, fallback para hardcoded. """ if not message: return {} import re msg_lower = message.lower() # 🔧 Tenta carregar padrões do PG db_patterns = None if self._db and hasattr(self._db, 'get_moderation_patterns_grouped'): try: db_patterns = self._db.get_moderation_patterns_grouped() except Exception: pass if db_patterns: # Modo PG: padrões dinâmicos insult_patterns = [p['pattern'] for p in db_patterns.get('insults', [])] threat_patterns = [p['pattern'] for p in db_patterns.get('threats', [])] insult_patterns_boundary = ["cu"] # Sempre word-boundary else: # Fallback hardcoded insult_patterns = [ "filho da puta", "filho da p*ta", "fdp", "caralho", "puta que pariu", "cabrão", "cabrona", "otário", "otária", "idiota", "imbecil", "estúpido", "estúpida", "burro", "burra", "retardado", "retardada", "filho da mãe", "piriquito", "bosta", "merda", "fodasse", "vai te foder", "come merda", "peste", "lixo", "nojento", "nojenta", "desgraçado", "desgraçada", "maldito", "maldita", "cornos", "buceta", "piranha", "vagabunda", "vagabundo" ] insult_patterns_boundary = ["cu"] threat_patterns = [ "vou te matar", "vou matar", "matar-te", "acabar contigo", "vou te destruir", "destruir-te", "vou te arruinar", "vou dar porrada", "panhar-te", "vou te desfigurar", "morte a ti", "morte para", "vamos-te encontrar", "vais ver", "vou acabar com", "guerra", "vingança", "vou te cortar", "cortar-te", "esfaquear", "balear" ] # Conta insultos encontrados insults_found = [p for p in insult_patterns if p in msg_lower] # Padrões curtos: word boundary para evitar "cu" em "cuidar" for p in insult_patterns_boundary: if re.search(r'\b' + re.escape(p) + r'\b', msg_lower): insults_found.append(p) threats_found = [p for p in threat_patterns if p in msg_lower] if threats_found: logger.warning(f"🚨 [AGENT-TOXIC] Ameaça detetada de {user_jid} em {group_jid}") return self._build_moderation_action( action="mute", target_jid=user_jid, group_jid=group_jid, reason=f"Ameaça de violência detetada: \"{threats_found[0]}\" — mute aplicado", duration_minutes=60, severity="CRÍTICO" ) if len(insults_found) >= 3: logger.warning(f"🚨 [AGENT-TOXIC] Linguagem altamente tóxica de {user_jid} em {group_jid}") return self._build_moderation_action( action="mute", target_jid=user_jid, group_jid=group_jid, reason=f"Linguagem altamente tóxica: {len(insults_found)} termos ofensivos detetados", duration_minutes=45, severity="CRÍTICO" ) if len(insults_found) >= 1: logger.info(f"⚠️ [AGENT-TOXIC] Linguagem ofensiva de {user_jid} em {group_jid}") return self._build_moderation_action( action="warn", target_jid=user_jid, group_jid=group_jid, reason=f"Linguagem ofensiva detetada: \"{insults_found[0]}\"", severity="AVISO" ) return {} def analyze_mass_mention_abuse(self, message: str, user_jid: str, group_jid: str) -> Dict[str, Any]: """ [AGENTE AUTÓNOMO] Deteta abuso de menções (tagall spam). Se a mensagem contiver 5+ menções → aviso/mute por spam de menções. """ if not message: return {} import re # Conta menções: @numerico ou @lid: ou padrões de menção do WhatsApp mention_patterns = re.findall(r'@\d{5,15}|@\d+@s\.whatsapp\.net|@\d+@g\.us', message) mention_count = len(mention_patterns) # 🔧 Tenta carregar tagall patterns do PG tagall_patterns = None if self._db and hasattr(self._db, 'get_moderation_patterns_grouped'): try: db_p = self._db.get_moderation_patterns_grouped() if db_p.get('tagall'): tagall_patterns = [p['pattern'] for p in db_p['tagall']] except Exception: pass if not tagall_patterns: tagall_patterns = ["@everyone", "@all", "@todos", "marcar todos", "tagall", "@group"] has_tagall = any(p in message.lower() for p in tagall_patterns) total_mentions = mention_count + (1 if has_tagall else 0) if total_mentions >= 5: logger.warning(f"🚨 [AGENT-MENTION] Abuso de menções ({total_mentions}) de {user_jid} em {group_jid}") return self._build_moderation_action( action="mute", target_jid=user_jid, group_jid=group_jid, reason=f"Abuso de menções: {total_mentions} menções detetadas numa única mensagem", duration_minutes=15, severity="AVISO" ) return {} def analyze_casino_spam(self, message: str, user_jid: str, group_jid: str) -> Dict[str, Any]: """ [AGENTE AUTÓNOMO] Deteta spam de casino/apostas/PIX promotion. Padrões comuns: "vai pix", "liberação saldo", "plano dr", "caia na hora", "pacote cliente", "tdm ou saldo pix", "promo dia", "interessados". Níveis: 1 keyword = warn 2+ keywords = mute 60min 3+ keywords = ban + blacklist """ if not message: return {} msg_lower = message.lower() # Padrões de casino/betting spam (PT-AO/PT-BR) casino_patterns = { 'pix_multiplier': ['vai pix', 'faz pix', 'manda pix', 'pix no', 'faz o pix', 'me chama no pix'], 'balance_release': ['liberação saldo', 'libera saldo', 'saca saldo', 'saca teu saldo', 'saca o saldo', 'levanta saldo', 'tirar saldo'], 'casino_combo': ['caia na hora', 'pacote cliente', 'plano dr', 'plano dourado', 'promoção do dia', 'promo dia', 'tdm ou saldo pix', 'interessados chame', 'chama no pv', 'chama no privado', 'grupo vip', 'sala vip'], } all_patterns = [] for category, patterns in casino_patterns.items(): for p in patterns: if p in msg_lower: all_patterns.append((category, p)) if len(all_patterns) >= 3: logger.warning(f"🚨 [AGENT-CASINO] Spam pesado ({len(all_patterns)} padrões) de {user_jid} em {group_jid}") return self._build_moderation_action( action="ban", target_jid=user_jid, group_jid=group_jid, reason=f"Spam de casino/apostas: {len(all_patterns)} padrões detetados ({all_patterns[0][1]}, {all_patterns[1][1]}, ...) — ban + blacklist", severity="CRÍTICO", blacklist=True ) elif len(all_patterns) >= 2: logger.warning(f"⚠️ [AGENT-CASINO] Spam ({len(all_patterns)} padrões) de {user_jid} em {group_jid}") return self._build_moderation_action( action="mute", target_jid=user_jid, group_jid=group_jid, reason=f"Spam de casino: {len(all_patterns)} padrões detetados ({all_patterns[0][1]}, {all_patterns[1][1]})", duration_minutes=60, severity="CRÍTICO" ) elif len(all_patterns) == 1: logger.info(f"⚠️ [AGENT-CASINO] Possível spam de {user_jid} em {group_jid}: {all_patterns[0][1]}") return self._build_moderation_action( action="warn", target_jid=user_jid, group_jid=group_jid, reason=f"Possível spam de casino: padrão '{all_patterns[0][1]}' detetado", severity="AVISO" ) return {} def analyze_scam_phishing(self, message: str, user_jid: str, group_jid: str) -> Dict[str, Any]: """ [AGENTE AUTÓNOMO] Deteta tentativas de scam/phishing. Padrões: "falso recrutamento", "prémio", "ganhador", "fatura pendente", links suspeitos, "clica aqui", "recebe o teu", "urgente". Phishing = mute 120min + notifica dono. """ if not message: return {} import re msg_lower = message.lower() scam_patterns = [ # Recrutamento falso 'falso recrutamento', 'falsa vaga', 'oferta de emprego', # Prémios / ganhador 'ganhaste', 'fui sorteado', 'fui selecionado', 'és o vencedor', 'parabéns ganhou', 'tens um prémio', 'prémio de', 'sou vencedor', # Faturas 'fatura pendente', 'dívida ativa', 'regulariza já', 'penalização', 'multa por atraso', 'cobrança judicial', 'intimação', # Urgência artificial 'clica aqui já', 'clica no link', 'urgente', 'última chance', 'expira hoje', 'expira em 24h', 'vai expirar', # Links suspeitos (mascarados) 'bit.ly/', 'tinyurl.com/', 'cutt.ly/', 't.co/', ] # Phishing com URL mascarado link_matched = re.search(r'https?://(bit\.ly|tinyurl\.com|cutt\.ly|t\.co|ow\.ly|shorturl|is\.gd)/\S+', msg_lower) found_patterns = [p for p in scam_patterns if p in msg_lower] if link_matched or len(found_patterns) >= 2: trigger = found_patterns[0] if found_patterns else f"Link mascarado: {link_matched.group()}" logger.warning(f"🚨 [AGENT-SCAM] Phishing detetado de {user_jid} em {group_jid}: {trigger}") return self._build_moderation_action( action="mute", target_jid=user_jid, group_jid=group_jid, reason=f"Phishing/scam detetado: '{trigger}' — mute aplicado preventivamente", duration_minutes=120, severity="CRÍTICO" ) elif len(found_patterns) == 1: logger.info(f"⚠️ [AGENT-SCAM] Suspeita de phishing de {user_jid} em {group_jid}: {found_patterns[0]}") return self._build_moderation_action( action="warn", target_jid=user_jid, group_jid=group_jid, reason=f"Suspeita de phishing: '{found_patterns[0]}'", severity="AVISO" ) return {} def analyze_raid_pattern(self, recent_actions: List[Dict[str, Any]]) -> Dict[str, Any]: """ [AGENTE AUTÓNOMO] Deteta raid/coordenated attack. Recebe lista de ações recentes (warns, kicks, bans) e verifica se há aumento súbito (10+ ações em 5min = raid). Retorna ação de lockdown: aumentar fiscalização do grupo. """ if not recent_actions or len(recent_actions) < 10: return {} # Janela: 5 minutos five_min_ago = time.time() - 300 recent_window = [a for a in recent_actions if a.get('timestamp', 0) >= five_min_ago] if len(recent_window) >= 10: logger.warning(f"🚨 [AGENT-RAID] Possível raid: {len(recent_window)} ações em 5min") return { "type": "remote_action", "action": "autonomous_action", "params": { "cmd": "raid_lockdown", "group_jid": recent_window[0].get('grupo'), "reason": f"Raid detetado: {len(recent_window)} ações de moderação em 5min", "severity": "CRÍTICO", "notify_owner": True } } return {} def build_hidetag_action(self, group_jid: str, message: str) -> Dict[str, Any]: """ [AGENTE AUTÓNOMO] Constrói uma ação interna de hidetag. Transmite uma mensagem oculta para todos os membros do grupo. """ return { "type": "remote_action", "action": "autonomous_action", "params": { "cmd": "hidetag", "group_jid": group_jid, "message": message, "notify_owner": False, "silent": True } } def build_poll_action(self, group_jid: str, question: str, options: List[str]) -> Dict[str, Any]: """ [AGENTE AUTÓNOMO] Constrói uma ação interna de criação de enquete. """ if not options or len(options) < 2: return {} return { "type": "remote_action", "action": "autonomous_action", "params": { "cmd": "enquete", "group_jid": group_jid, "question": question, "options": options, "notify_owner": False, "silent": True } } # ───────────────────────────────────────────────────────────────── # CONSTRUÇÃO DE AÇÕES # ───────────────────────────────────────────────────────────────── def _build_moderation_action( self, action: str, target_jid: str, group_jid: str, reason: str, duration_minutes: int = 0, severity: str = "AVISO", blacklist: bool = False ) -> Dict[str, Any]: """Constrói uma ação de moderação autónoma.""" # Regista no log interno log_entry = { "timestamp": datetime.now().isoformat(), "tipo": "MODERAÇÃO_AUTÓNOMA", "severidade": severity, "utilizador": target_jid, "grupo": group_jid, "ação": action, "motivo": reason, "duração": f"{duration_minutes}min" if duration_minutes else "N/A" } if blacklist: log_entry["blacklist"] = True self._action_log.append(log_entry) if len(self._action_log) > 500: self._action_log = self._action_log[-500:] # Guarda no DB se disponível if self._db: try: self._db._execute_with_retry( """INSERT INTO system_events (tipo, servidor, descricao, acao_tomada, resolvido) VALUES (?, ?, ?, ?, 1)""", (severity, "railway", f"Moderação: {reason} | User: {target_jid}", action, ), commit=True ) except Exception as e: logger.debug(f"[AGENT] Não foi possível registar no DB: {e}") logger.info(f"🤖 [AGENT ACTION] {action} em {target_jid} | Motivo: {reason}") params = { "cmd": action, "target": target_jid, "group_jid": group_jid, "reason": reason, "notify_owner": True, "silent": True, "delete_message": True } if action == "ban": params["args"] = [] elif action == "blacklist": params["args"] = [] elif action in ("hidetag", "enquete"): pass else: params["args"] = [str(duration_minutes)] if duration_minutes else [] if blacklist: params["blacklist"] = True params["blacklist_reason"] = reason return { "type": "remote_action", "action": "autonomous_action", "params": params } def decide_action_from_context(self, context: str, event_type: str) -> Optional[Dict]: """ Usa o LLM para decidir uma ação com base num evento do sistema. Usado para situações mais complexas que precisam de raciocínio. """ if not self._llm_caller: return None system = ( "És a Kiami, agente autónoma de infraestrutura da Softedge. " "Analisa o evento abaixo e decide a melhor ação a tomar. " "Responde APENAS em JSON com: {\"acao\": \"string\", \"motivo\": \"string\", \"prioridade\": \"alta|media|baixa\"}. " "Ações possíveis: mute, ban, warn, blacklist, ignore." ) try: response = self._llm_caller(system, f"EVENTO: {event_type}\nCONTEXTO:\n{context}") # Extrai JSON da resposta import re json_match = re.search(r'\{[^{}]*\}', response, re.DOTALL) if json_match: return json.loads(json_match.group()) except Exception as e: logger.error(f"[AGENT] Erro ao decidir ação: {e}") return None def get_recent_actions_summary(self) -> str: """Retorna um resumo das ações autónomas recentes.""" if not self._action_log: return "Nenhuma ação autónoma registada." recent = self._action_log[-10:] # Últimas 10 ações lines = [f"• [{a['timestamp'][:16]}] {a['ação'].upper()} em {a['utilizador'].split('@')[0]} — {a['motivo']}" for a in recent] return f"📋 Últimas {len(recent)} ações autónomas:\n" + "\n".join(lines) # Instância global singleton autonomous_agent = AutonomousAgent()