215 lines
8.8 KiB
Python
215 lines
8.8 KiB
Python
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() -> str:
|
|
"""Consulta os últimos 5 deploies registrados no Coolify via banco de dados."""
|
|
cmd = 'docker exec coolify psql -U coolify -d coolify -c "SELECT p.name as project, a.name as application, d.status, d.updated_at FROM deployments d JOIN applications a ON d.application_id = a.id JOIN projects p ON a.project_id = p.id ORDER BY d.updated_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]
|