111 lines
4.3 KiB
Python
111 lines
4.3 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
|
|
)
|
|
|
|
# ============================================================
|
|
# UTILS
|
|
# ============================================================
|
|
|
|
def run_bash(command: str, timeout: int = 120) -> Dict:
|
|
# Auto-moderniza docker-compose
|
|
command = command.replace("docker-compose", "docker compose")
|
|
try:
|
|
result = subprocess.run(command, shell=True, capture_output=True, text=True, timeout=timeout)
|
|
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}"
|
|
|
|
# ============================================================
|
|
# REGISTRY
|
|
# ============================================================
|
|
|
|
TOOLS_V2 = {
|
|
"docker_ps": {"desc": "Lista containers", "func": DockerTools.ps, "danger": "safe"},
|
|
"docker_stats": {"desc": "Uso de recursos", "func": DockerTools.stats, "danger": "safe"},
|
|
"docker_logs": {"desc": "Ver logs", "func": DockerTools.logs, "danger": "safe"},
|
|
"docker_restart": {"desc": "Reiniciar container", "func": DockerTools.restart, "danger": "dangerous"},
|
|
"git_pull": {"desc": "Atualizar código", "func": GitTools.pull, "danger": "medium"},
|
|
"git_status": {"desc": "Ver status git", "func": GitTools.status, "danger": "safe"},
|
|
"gitea_repos": {"desc": "Listar repos no Gitea", "func": GiteaTools.list_repos, "danger": "safe"},
|
|
"supabase_tables": {"desc": "Listar tabelas Supabase", "func": SupabaseTools.list_tables, "danger": "safe"},
|
|
}
|
|
|
|
def get_all_tools_formatted() -> str:
|
|
res = "🛠️ **Ferramentas Disponíveis**:\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]
|