import subprocess import os import psutil import time import re import json def run_bash_command(command: str) -> str: """Executa um comando bash na VPS e retorna a saída.""" try: # Garante caminhos comuns no PATH para execução via PM2/Containers custom_env = os.environ.copy() paths = ["/usr/local/bin", "/root/.cargo/bin", "/usr/bin", "/bin"] current_path = custom_env.get("PATH", "") for p in paths: if p not in current_path: current_path = f"{p}:{current_path}" custom_env["PATH"] = current_path # Executa comando de forma segura dentro da VPS result = subprocess.run( command, shell=True, capture_output=True, text=True, timeout=120, env=custom_env ) output = result.stdout.strip() error = result.stderr.strip() # Se encontrou algo no stdout, retornamos o que achou mesmo com erro if output: return output if result.returncode != 0: if result.returncode == 127: return f"ERRO (127): Comando não encontrado. Verifique se o alias ou binário está no PATH. (Comando: {command})" return f"ERRO ({result.returncode}): {error if error else 'Nada no stderr'}" return "Sucesso (vazio)" except subprocess.TimeoutExpired: return "ERRO: O comando demorou muito e foi cancelado (timeout)." except Exception as e: return f"ERRO fatal ao rodar bash: {str(e)}" def get_system_health() -> str: """Retorna um texto detalhado da saúde do servidor para a IA.""" cpu = psutil.cpu_percent(interval=0.1) vm = psutil.virtual_memory() disk = psutil.disk_usage('/') # Uptime aproximado usando psutil uptime_seconds = time.time() - psutil.boot_time() uptime_hours = round(uptime_seconds / 3600, 1) return (f"CPU: {cpu}% | " f"RAM: {round(vm.used / (1024**3), 2)}GB usada / {round(vm.total / (1024**3), 2)}GB total ({vm.percent}%) | " f"Disco: {disk.percent}% usado | " f"Uptime: {uptime_hours}h") def read_vps_file(filepath: str) -> str: """Lê um arquivo do sistema de arquivos da VPS através do mapeamento /host_root.""" # Garante que o caminho comece com /host_root clean_path = filepath if clean_path.startswith("/host_root"): host_path = clean_path else: # Se o usuário mandou /root/vps..., vira /host_root/root/vps... host_path = os.path.join("/host_root", clean_path.lstrip("/")) try: if not os.path.exists("/host_root"): return "ERRO CRÍTICO: O diretório /host_root não existe no container! O mapeamento de volume falhou." if not os.path.exists(host_path): # Tenta listar o diretório pai para ajudar no debug parent = os.path.dirname(host_path) content = os.listdir(parent) if os.path.exists(parent) else "Pai não existe" return f"Erro: {filepath} não encontrado (Caminho real: {host_path}). Conteúdo do pai: {content}" with open(host_path, 'r') as f: return f.read(2000) except Exception as e: return f"Erro ao acessar {filepath}: {e}" def get_docker_stats() -> str: """Retorna o uso de CPU/RAM de todos os containers ativos via comando docker stats.""" return run_bash_command('docker stats --no-stream --format "table {{.Name}}\t{{.CPUPerc}}\t{{.MemUsage}}"') # ============================================================ # SISTEMA CRONOS (MEMÓRIA DE LONGO PRAZO) # ============================================================ MEMORY_ROOT = "/root/Antigravity_Memory" def cronos_log(arg: str) -> str: """ Salva memórias. Formato esperado: topic="assunto", content="texto", folder="current_week" Também aceita JSON. """ try: # Tenta como JSON primeiro try: data = json.loads(arg) topic = data.get("topic", "geral") content = data.get("content", "") folder = data.get("folder", "current_week") except: # Parser simples por regex para query="xxx" ou topic="xxx" topic_m = re.search(r'topic=["\'](.*?)["\']', arg) content_m = re.search(r'content=["\'](.*?)["\']', arg, re.S) folder_m = re.search(r'folder=["\'](.*?)["\']', arg) topic = topic_m.group(1) if topic_m else "geral" content = content_m.group(1) if content_m else arg # fallback para arg bruto folder = folder_m.group(1) if folder_m else "current_week" if not content: return "Erro: Conteúdo vazio." target_dir = os.path.join(MEMORY_ROOT, folder) if not os.path.exists(target_dir): os.makedirs(target_dir, exist_ok=True) filename = f"{topic.lower().replace(' ', '_')}.md" filepath = os.path.join(target_dir, filename) timestamp = time.strftime("%Y-%m-%d %H:%M:%S") entry = f"\n---\n### ENTRY: {timestamp}\n{content}\n" with open(filepath, "a" if os.path.exists(filepath) else "w") as f: f.write(entry) return f"Sucesso: Salvo em Cronos/{folder}/{filename}" except Exception as e: return f"Erro ao salvar em Cronos: {e}" def cronos_query(arg: str) -> str: """Busca informações. Formato: query="termo", folder="pasta" """ query_m = re.search(r'query=["\'](.*?)["\']', arg) folder_m = re.search(r'folder=["\'](.*?)["\']', arg) query = query_m.group(1) if query_m else arg folder = folder_m.group(1) if folder_m else "current_week" target_dir = os.path.join(MEMORY_ROOT, folder) return run_bash_command(f"grep -rniI '{query}' {target_dir} | head -n 20") # Mapeamento para o Agente entender quais tools ele possui (será usado no loop ReAct) AVAILABLE_TOOLS = { "run_bash_command": { "description": "Executa comandos Linux na VPS. Use para docker, git, mkdir, touch, etc.", "func": run_bash_command }, "get_system_health": { "description": "Verifica RAM, CPU e Disco globais da VPS.", "func": get_system_health }, "read_vps_file": { "description": "Lê o conteúdo de um arquivo na VPS (logs, configs).", "func": read_vps_file }, "get_docker_stats": { "description": "Retorna uma tabela com o consumo de CPU e Memória de cada container.", "func": get_docker_stats }, "cronos_log": { "description": "Salva memórias importantes (assunto, conteúdo, pasta=current_week|knowledge). Use sempre que algo for concluído.", "func": cronos_log }, "cronos_query": { "description": "Busca informações no histórico da Memória Cronos.", "func": cronos_query } }