BotVPS/tools_v2.py

import subprocess
import os
import httpx
import asyncio
from typing import Dict, List, Optional
from credential_manager import (
    gitea_api_url, gitea_token, supabase_url, supabase_anon_key,
    supabase_service_role_key
)
from deploy_manager import DeployManager

# ============================================================
# UTILS
# ============================================================

def run_bash(command: str, timeout: int = 120) -> Dict:
    # Auto-moderniza docker-compose
    command = command.replace("docker-compose", "docker compose")
    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

        result = subprocess.run(command, shell=True, capture_output=True, text=True, timeout=timeout, env=custom_env)
        return {
            "success": result.returncode == 0,
            "output": (result.stdout or result.stderr).strip() or "Sucesso"
        }
    except Exception as e:
        return {"success": False, "output": str(e)}

# ============================================================
# DOCKER TOOLS
# ============================================================

class DockerTools:
    @staticmethod
    def ps() -> str:
        return run_bash("docker ps --format 'table {{.Names}}\t{{.Status}}\t{{.Ports}}'")["output"]

    @staticmethod
    def stats() -> str:
        return run_bash("docker stats --no-stream --format 'table {{.Name}}\t{{.CPUPerc}}\t{{.MemUsage}}'")["output"]

    @staticmethod
    def logs(container: str, lines: int = 50) -> str:
        return run_bash(f"docker logs --tail {lines} {container}")["output"]

    @staticmethod
    def restart(container: str) -> str:
        return run_bash(f"docker restart {container}")["output"]

# ============================================================
# GIT TOOLS
# ============================================================

class GitTools:
    @staticmethod
    def pull(repo_path: str = ".") -> str:
        return run_bash(f"git -C {repo_path} pull")["output"]

    @staticmethod
    def status(repo_path: str = ".") -> str:
        return run_bash(f"git -C {repo_path} status --short")["output"]

# ============================================================
# API TOOLS (ASYNC)
# ============================================================

class GiteaTools:
    @staticmethod
    async def list_repos() -> str:
        url = f"{gitea_api_url()}/user/repos"
        headers = {"Authorization": f"token {gitea_token()}"}
        async with httpx.AsyncClient() as client:
            try:
                res = await client.get(url, headers=headers)
                repos = res.json()
                return "\n".join([f"• {r['name']}" for r in repos[:10]])
            except Exception as e: return f"Erro Gitea: {e}"

class SupabaseTools:
    @staticmethod
    async def list_tables() -> str:
        url = f"{supabase_url()}/rest/v1/"
        headers = {"apikey": supabase_anon_key(), "Authorization": f"Bearer {supabase_anon_key()}"}
        async with httpx.AsyncClient() as client:
            try:
                res = await client.get(url, headers=headers)
                return str(res.json())
            except Exception as e: return f"Erro Supabase: {e}"

# ============================================================
# SYSTEM TOOLS (MASTER SKILL)
# ============================================================

class SystemTools:
    @staticmethod
    def execute_bash(command: str) -> str:
        """Executa um comando bash arbitrário na VPS."""
        return run_bash(command)["output"]

    @staticmethod
    def read_file(path: str) -> str:
        """Lê o conteúdo total de um arquivo na VPS."""
        try:
            with open(path, 'r') as f:
                return f.read()
        except Exception as e:
            return f"Erro ao ler arquivo: {e}"

    @staticmethod
    def write_file(path_content: str) -> str:
        """Escreve conteúdo em um arquivo. Formato esperado: 'caminho|conteúdo'"""
        try:
            if "|" not in path_content:
                return "Erro: Use o formato 'caminho|conteúdo'"
            path, content = path_content.split("|", 1)
            path = path.strip()
            # Garante que o diretório existe
            os.makedirs(os.path.dirname(os.path.abspath(path)), exist_ok=True)
            with open(path, 'w') as f:
                f.write(content)
            return f"Sucesso: {path} atualizado."
        except Exception as e:
            return f"Erro ao escrever: {e}"

    @staticmethod
    def list_dir(path: str = ".") -> str:
        """Lista arquivos e pastas de um diretório."""
        try:
            items = os.listdir(path)
            return "\n".join(items)
        except Exception as e:
            return f"Erro ao listar {path}: {e}"

    @staticmethod
    def pm2_status() -> str:
        """Lista todos os processos gerenciados pelo PM2."""
        return run_bash("pm2 jlist")["output"]

    @staticmethod
    def pm2_restart(name: str) -> str:
        """Reinicia um processo no PM2 pelo nome ou ID."""
        return run_bash(f"pm2 restart {name}")["output"]

# ============================================================
# WORKSPACE TOOLS (GWS CLI)
# ============================================================

class WorkspaceTools:
    @staticmethod
    def gws_command(cmd: str) -> str:
        """Executa um comando gws completo (ex: gws-mr gmail +send ...)."""
        return run_bash(cmd)["output"]

    @staticmethod
    def magic_deploy(git_url: str) -> str:
        """Clona um repositório git e tenta realizar o deploy automático (Docker/Node/Python)."""
        dm = DeployManager()
        return dm.magic_deploy(git_url)

    @staticmethod
    def coolify_deploy_status(arg: str = None) -> str:
        """Consulta os últimos 5 deploies registrados no Coolify via banco de dados."""
        cmd = 'docker exec coolify-db psql -U coolify -d coolify -c "SELECT application_name as application, status, created_at FROM application_deployment_queues ORDER BY created_at DESC LIMIT 5;"'
        return run_bash(cmd)["output"]

# ============================================================
# REGISTRY
# ============================================================

TOOLS_V2 = {
    # Docker
    "docker_ps": {"desc": "Lista containers ativos", "func": DockerTools.ps, "danger": "safe"},
    "docker_stats": {"desc": "Uso de CPU/RAM por container", "func": DockerTools.stats, "danger": "safe"},
    "docker_logs": {"desc": "Ver logs de um container", "func": DockerTools.logs, "danger": "safe"},
    "docker_restart": {"desc": "Reiniciar container", "func": DockerTools.restart, "danger": "dangerous"},
    
    # Git
    "git_pull": {"desc": "Faz pull no repositório atual", "func": GitTools.pull, "danger": "medium"},
    "git_status": {"desc": "Verifica status do git", "func": GitTools.status, "danger": "safe"},
    "gitea_repos": {"desc": "Lista repositórios no Gitea", "func": GiteaTools.list_repos, "danger": "safe"},
    
    # Supabase
    "supabase_tables": {"desc": "Lista tabelas no Supabase", "func": SupabaseTools.list_tables, "danger": "safe"},
    
    # System (Master Skill)
    "bash": {"desc": "Executa comando shell direto", "func": SystemTools.execute_bash, "danger": "dangerous"},
    "read_file": {"desc": "Lê conteúdo de um arquivo", "func": SystemTools.read_file, "danger": "safe"},
    "write_file": {"desc": "Cria ou edita arquivo (caminho|conteúdo)", "func": SystemTools.write_file, "danger": "dangerous"},
    "ls": {"desc": "Lista arquivos num diretório", "func": SystemTools.list_dir, "danger": "safe"},
    "pm2_status": {"desc": "Status dos processos PM2", "func": SystemTools.pm2_status, "danger": "safe"},
    "pm2_restart": {"desc": "Reiniciar processo PM2", "func": SystemTools.pm2_restart, "danger": "medium"},
    "magic_deploy": {"desc": "Deploy automático via URL Git", "func": WorkspaceTools.magic_deploy, "danger": "dangerous"},
    "coolify_status": {"desc": "Status dos últimos deploies no Coolify", "func": WorkspaceTools.coolify_deploy_status, "danger": "safe"},
    
    # Google Workspace
    "gws": {"desc": "Executa comando GWS CLI (ex: gws-mr drive files list)", "func": WorkspaceTools.gws_command, "danger": "medium"},
}

def get_all_tools_formatted() -> str:
    res = "🛠️ **Ferramentas Antigravity Ativas (V2)**:\n\n"
    for name, info in TOOLS_V2.items():
        res += f"- `{name}`: {info['desc']} [{info['danger'].upper()}]\n"
    return res

def get_tools_by_danger(level: str) -> List:
    return [{"name": k, **v} for k, v in TOOLS_V2.items() if v["danger"] == level]