admtracksteel/BotVPS
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refatoracao

Browse Source
This commit is contained in:
admtracksteel
2026-03-23 23:38:56 +00:00
parent 8002262cf7
commit b7e6239216
16 changed files with 2290 additions and 4321 deletions
Download Patch File Download Diff File Expand all files Collapse all files

662
tools_v2.py
View File

@@ -1,664 +1,110 @@
# ============================================================
# TOOLS_V2.PY - Ferramentas Expandidas para o Orquestrador
# NÃO SUBSTITUI tools.py - É um módulo adicional
# ============================================================
import subprocess
import os
import requests
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, coolify_api
supabase_service_role_key
)
# ============================================================
# CONSTANTS
# ============================================================
DANGER_LEVELS = {
"safe": "SAFE - Executa automático",
"medium": "MEDIUM - Informa antes",
"dangerous": "DANGEROUS - Pede confirmação"
}
# ============================================================
# UTILITY FUNCTIONS
# UTILS
# ============================================================
def run_bash(command: str, timeout: int = 120) -> Dict:
"""Executa comando bash e retorna resultado estruturado."""
# 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
)
result = subprocess.run(command, shell=True, capture_output=True, text=True, timeout=timeout)
return {
"success": result.returncode == 0,
"returncode": result.returncode,
"stdout": result.stdout.strip(),
"stderr": result.stderr.strip(),
"output": result.stdout.strip() if result.stdout else result.stderr.strip()
}
except subprocess.TimeoutExpired:
return {
"success": False,
"error": "Comando expirou (timeout)"
"output": (result.stdout or result.stderr).strip() or "Sucesso"
}
except Exception as e:
return {
"success": False,
"error": str(e)
}
def format_output(result: Dict, max_length: int = 2000) -> str:
"""Formata resultado para exibição."""
if not result.get("success"):
return f"[ERROR] Erro: {result.get('error') or result.get('stderr') or 'Desconhecido'}"
output = result.get("output", "[OK] Sucesso (sem output)")
if len(output) > max_length:
output = output[:max_length] + f"\n... (truncado, {len(output)} chars total)"
return output
return {"success": False, "output": str(e)}
# ============================================================
# DOCKER TOOLS
# ============================================================
class DockerTools:
"""Ferramentas Docker."""
@staticmethod
def ps(all_containers: bool = False) -> str:
"""Lista containers Docker."""
flags = "-a" if all_containers else ""
result = run_bash("docker ps " + flags + " --format 'table {{.Names}}\t{{.Status}}\t{{.Ports}}'")
return format_output(result)
def ps() -> str:
return run_bash("docker ps --format 'table {{.Names}}\t{{.Status}}\t{{.Ports}}'")["output"]
@staticmethod
def stats() -> str:
"""Mostra estatísticas de recursos dos containers."""
result = run_bash("docker stats --no-stream --format 'table {{.Name}}\t{{.CPUPerc}}\t{{.MemUsage}}\t{{.NetIO}}'")
return format_output(result)
return run_bash("docker stats --no-stream --format 'table {{.Name}}\t{{.CPUPerc}}\t{{.MemUsage}}'")["output"]
@staticmethod
def logs(container: str, lines: int = 50, follow: bool = False) -> str:
"""Mostra logs de um container."""
follow_flag = "-f" if follow else ""
result = run_bash(f"docker logs {follow_flag} --tail {lines} {container}")
return format_output(result, max_length=5000)
def logs(container: str, lines: int = 50) -> str:
return run_bash(f"docker logs --tail {lines} {container}")["output"]
@staticmethod
def restart(container: str) -> str:
"""Reinicia um container."""
result = run_bash(f"docker restart {container}")
return format_output(result)
@staticmethod
def stop(container: str) -> str:
"""Para um container."""
result = run_bash(f"docker stop {container}")
return format_output(result)
@staticmethod
def start(container: str) -> str:
"""Inicia um container."""
result = run_bash(f"docker start {container}")
return format_output(result)
@staticmethod
def exec(container: str, command: str) -> str:
"""Executa comando dentro de um container."""
result = run_bash(f"docker exec {container} {command}")
return format_output(result)
@staticmethod
def inspect(container: str) -> str:
"""Retorna informações detalhadas de um container."""
result = run_bash(f"docker inspect {container}")
return format_output(result, max_length=3000)
@staticmethod
def system_df() -> str:
"""Mostra uso de disco do Docker."""
result = run_bash("docker system df -v")
return format_output(result, max_length=3000)
@staticmethod
def prune(dangerous: bool = False) -> str:
"""Limpa recursos não utilizados do Docker."""
if dangerous:
result = run_bash("docker system prune -af --volumes")
else:
result = run_bash("docker system prune -f")
return format_output(result)
return run_bash(f"docker restart {container}")["output"]
# ============================================================
# GIT TOOLS
# ============================================================
class GitTools:
"""Ferramentas Git."""
@staticmethod
def pull(repo_path: str = ".") -> str:
return run_bash(f"git -C {repo_path} pull")["output"]
@staticmethod
def status(repo_path: str = ".") -> str:
"""Mostra status do repositório git."""
result = run_bash(f"git -C {repo_path} status --short")
return format_output(result)
@staticmethod
def pull(repo_path: str = ".", remote: str = "origin", branch: str = "main") -> str:
"""Faz git pull."""
result = run_bash(f"git -C {repo_path} pull {remote} {branch}")
return format_output(result)
@staticmethod
def push(repo_path: str = ".", remote: str = "origin", branch: str = "main") -> str:
"""Faz git push."""
result = run_bash(f"git -C {repo_path} push {remote} {branch}")
return format_output(result)
@staticmethod
def clone(repo_url: str, target_path: str) -> str:
"""Clona um repositório."""
result = run_bash(f"git clone {repo_url} {target_path}")
return format_output(result)
@staticmethod
def branch(repo_path: str = ".", list_all: bool = False) -> str:
"""Lista branches."""
flags = "-a" if list_all else ""
result = run_bash(f"git -C {repo_path} branch {flags}")
return format_output(result)
@staticmethod
def checkout(repo_path: str, branch: str) -> str:
"""Muda para outro branch."""
result = run_bash(f"git -C {repo_path} checkout {branch}")
return format_output(result)
@staticmethod
def log(repo_path: str = ".", count: int = 10) -> str:
"""Mostra histórico de commits."""
result = run_bash(f"git -C {repo_path} log --oneline -{count}")
return format_output(result)
@staticmethod
def diff(repo_path: str = ".") -> str:
"""Mostra diferenças não commitadas."""
result = run_bash(f"git -C {repo_path} diff")
return format_output(result)
@staticmethod
def stash(repo_path: str = ".") -> str:
"""Salva alterações temporariamente."""
result = run_bash(f"git -C {repo_path} stash")
return format_output(result)
@staticmethod
def fetch(repo_path: str = ".", remote: str = "origin") -> str:
"""Busca atualizações sem aplicar."""
result = run_bash(f"git -C {repo_path} fetch {remote}")
return format_output(result)
return run_bash(f"git -C {repo_path} status --short")["output"]
# ============================================================
# DOCKER COMPOSE TOOLS
# ============================================================
class DockerComposeTools:
"""Ferramentas Docker Compose."""
@staticmethod
def up(path: str, detach: bool = True, build: bool = False) -> str:
"""Sobe serviços com docker-compose."""
flags = "-d " if detach else ""
build_flag = "--build " if build else ""
result = run_bash(f"docker-compose -f {path} up {flags}{build_flag}")
return format_output(result)
@staticmethod
def down(path: str, volumes: bool = False) -> str:
"""Para e remove containers."""
flags = "-v" if volumes else ""
result = run_bash(f"docker-compose -f {path} down {flags}")
return format_output(result)
@staticmethod
def build(path: str, no_cache: bool = False) -> str:
"""Constrói imagens."""
flags = "--no-cache" if no_cache else ""
result = run_bash(f"docker-compose -f {path} build {flags}")
return format_output(result, max_length=5000)
@staticmethod
def ps(path: str) -> str:
"""Lista serviços."""
result = run_bash(f"docker-compose -f {path} ps")
return format_output(result)
@staticmethod
def logs(path: str, service: str = None, lines: int = 100) -> str:
"""Mostra logs dos serviços."""
service_part = f"{service}" if service else ""
result = run_bash(f"docker-compose -f {path} logs --tail {lines} {service_part}")
return format_output(result, max_length=5000)
@staticmethod
def restart(path: str, service: str = None) -> str:
"""Reinicia serviços."""
service_part = f"{service}" if service else ""
result = run_bash(f"docker-compose -f {path} restart {service_part}")
return format_output(result)
# ============================================================
# GITEA API TOOLS
# API TOOLS (ASYNC)
# ============================================================
class GiteaTools:
"""Ferramentas via API do Gitea."""
@staticmethod
def _get_headers() -> Dict:
"""Retorna headers para API do Gitea."""
token = gitea_token()
return {
"Authorization": f"token {token}",
"Content-Type": "application/json"
}
@staticmethod
def list_repos() -> str:
"""Lista repositórios do usuário."""
async def list_repos() -> str:
url = f"{gitea_api_url()}/user/repos"
try:
res = requests.get(url, headers=GiteaTools._get_headers(), timeout=10)
if res.status_code == 200:
headers = {"Authorization": f"token {gitea_token()}"}
async with httpx.AsyncClient() as client:
try:
res = await client.get(url, headers=headers)
repos = res.json()
if not repos:
return "Nenhum repositório encontrado."
output = "[REPO] **Repositórios:**\n\n"
for repo in repos[:10]:
output += f"• `{repo['name']}` - {repo.get('description', 'Sem descrição')[:50]}\n"
output += f" URL: {repo['html_url']}\n\n"
return output
return f"[ERROR] Erro: {res.status_code} - {res.text}"
except Exception as e:
return f"[ERROR] Erro: {str(e)}"
@staticmethod
def get_repo(owner: str, repo: str) -> str:
"""Busca informações de um repositório."""
url = f"{gitea_api_url()}/repos/{owner}/{repo}"
try:
res = requests.get(url, headers=GiteaTools._get_headers(), timeout=10)
if res.status_code == 200:
data = res.json()
return f"""[REPO] **{data['full_name']}**
- **Descrição:** {data.get('description', 'N/A')}
- **Linguagem:** {data.get('language', 'N/A')}
- **Stars:** {data.get('stars_count', 0)}
- **Forks:** {data.get('forks_count', 0)}
- **Última atualização:** {data.get('updated_at', 'N/A')}
- **URL:** {data['html_url']}"""
return f"[ERROR] Erro: {res.status_code}"
except Exception as e:
return f"[ERROR] Erro: {str(e)}"
@staticmethod
def list_actions(owner: str, repo: str) -> str:
"""Lista workflows/actions do repositório."""
url = f"{gitea_api_url()}/repos/{owner}/{repo}/actions/workflows"
try:
res = requests.get(url, headers=GiteaTools._get_headers(), timeout=10)
if res.status_code == 200:
workflows = res.json().get("workflows", [])
if not workflows:
return "Nenhum workflow encontrado."
output = "[WF] **Workflows:**\n\n"
for wf in workflows:
output += f"• `{wf['name']}` - {wf.get('status', 'N/A')}\n"
return output
return f"[ERROR] Erro: {res.status_code}"
except Exception as e:
return f"[ERROR] Erro: {str(e)}"
@staticmethod
def trigger_workflow(owner: str, repo: str, workflow_id: str, ref: str = "main") -> str:
"""Dispara um workflow."""
url = f"{gitea_api_url()}/repos/{owner}/{repo}/actions/workflows/{workflow_id}/dispatches"
data = {"ref": ref}
try:
res = requests.post(url, headers=GiteaTools._get_headers(), json=data, timeout=10)
if res.status_code == 204:
return f"[OK] Workflow '{workflow_id}' disparado com sucesso!"
return f"[ERROR] Erro: {res.status_code} - {res.text}"
except Exception as e:
return f"[ERROR] Erro: {str(e)}"
# ============================================================
# SUPABASE API TOOLS
# ============================================================
return "\n".join([f"• {r['name']}" for r in repos[:10]])
except Exception as e: return f"Erro Gitea: {e}"
class SupabaseTools:
"""Ferramentas via API REST do Supabase."""
@staticmethod
def _get_headers(anon_key: bool = True) -> Dict:
"""Retorna headers para API do Supabase."""
key = supabase_anon_key() if anon_key else supabase_service_role_key()
role = "anon" if anon_key else "service_role"
return {
"apikey": key,
"Authorization": f"Bearer {key}",
"Content-Type": "application/json",
"Prefer": "return=representation"
}
@staticmethod
def list_tables() -> str:
"""Lista tabelas disponíveis (via introspecção)."""
async def list_tables() -> str:
url = f"{supabase_url()}/rest/v1/"
try:
res = requests.get(url, headers=SupabaseTools._get_headers(), timeout=10)
if res.status_code == 200:
tables = res.json()
if not tables:
return "Nenhuma tabela encontrada."
output = "[DATA] **Tabelas:**\n\n"
for table in tables[:20]:
output += f"• `{table.get('table_name', 'N/A')}`\n"
return output
return f"[ERROR] Erro: {res.status_code}"
except Exception as e:
return f"[ERROR] Erro: {str(e)}"
@staticmethod
def query(table: str, select: str = "*", filters: str = None, limit: int = 10) -> str:
"""Consulta dados de uma tabela."""
url = f"{supabase_url()}/rest/v1/{table}"
params = f"select={select}&limit={limit}"
if filters:
params += f"&{filters}"
try:
res = requests.get(url, headers=SupabaseTools._get_headers(), params=params, timeout=10)
if res.status_code == 200:
data = res.json()
if not data:
return f"📭 Nenhum resultado em `{table}`."
output = f"[DATA] **Resultados de `{table}`** ({len(data)} registros):\n\n"
for row in data[:5]:
output += f"```json\n{str(row)[:200]}\n```\n"
return output
return f"[ERROR] Erro: {res.status_code} - {res.text}"
except Exception as e:
return f"[ERROR] Erro: {str(e)}"
@staticmethod
def insert(table: str, data: Dict) -> str:
"""Insere dados em uma tabela."""
url = f"{supabase_url()}/rest/v1/{table}"
try:
res = requests.post(url, headers=SupabaseTools._get_headers(anon_key=False), json=data, timeout=10)
if res.status_code in [200, 201]:
return f"[OK] Registro inserido em `{table}`!"
return f"[ERROR] Erro: {res.status_code} - {res.text}"
except Exception as e:
return f"[ERROR] Erro: {str(e)}"
@staticmethod
def update(table: str, data: Dict, filters: str) -> str:
"""Atualiza dados em uma tabela."""
url = f"{supabase_url()}/rest/v1/{table}?{filters}"
try:
res = requests.patch(url, headers=SupabaseTools._get_headers(anon_key=False), json=data, timeout=10)
if res.status_code in [200, 204]:
return f"[OK] Registro atualizado em `{table}`!"
return f"[ERROR] Erro: {res.status_code} - {res.text}"
except Exception as e:
return f"[ERROR] Erro: {str(e)}"
@staticmethod
def delete(table: str, filters: str) -> str:
"""Deleta dados de uma tabela."""
url = f"{supabase_url()}/rest/v1/{table}?{filters}"
try:
res = requests.delete(url, headers=SupabaseTools._get_headers(anon_key=False), timeout=10)
if res.status_code in [200, 204]:
return f"[OK] Registro deletado de `{table}`!"
return f"[ERROR] Erro: {res.status_code} - {res.text}"
except Exception as e:
return f"[ERROR] Erro: {str(e)}"
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}"
# ============================================================
# COOLIFY API TOOLS
# ============================================================
class CoolifyTools:
"""Ferramentas via API do Coolify."""
@staticmethod
def get_status() -> str:
"""Retorna status do Coolify."""
result = coolify_api("/status")
if "error" in result:
return f"[ERROR] Erro: {result['error']}"
return f"""[COOLIFY] **Coolify Status:**
- **Status:** {result.get('status', 'N/A')}
- **Containers:** {result.get('containers', 'N/A')}
- **Deployments:** {result.get('deployments', 'N/A')}"""
@staticmethod
def list_applications() -> str:
"""Lista aplicações no Coolify."""
from credential_manager import coolify_list_applications
apps = coolify_list_applications()
if not apps:
return "[REPO] Nenhuma aplicacao encontrada."
output = "[REPO] Aplicacoes Coolify:\n\n"
for app in apps[:10]:
output += f"- {app.get('name', 'N/A')} - {app.get('status', 'N/A')}\n"
output += f" URL: {app.get('fqdn', 'N/A')}\n\n"
return output
@staticmethod
def list_deployments(limit: int = 10) -> str:
"""Lista deployments recentes."""
from credential_manager import coolify_list_deployments
deps = coolify_list_deployments()
if not deps:
return "[DEPLOY] Nenhum deployment recente."
output = "[DEPLOY] Deployments Recentes:\n\n"
for dep in deps[:limit]:
output += f"- {dep.get('application', 'N/A')} - {dep.get('status', 'N/A')}\n"
output += f" {dep.get('created_at', 'N/A')}\n\n"
return output
# ============================================================
# FILE TOOLS
# ============================================================
class FileTools:
"""Ferramentas de manipulação de arquivos."""
@staticmethod
def list(path: str) -> str:
"""Lista conteúdo de diretório."""
result = run_bash(f"ls -la {path}")
return format_output(result)
@staticmethod
def read(path: str, lines: int = 100) -> str:
"""Lê conteúdo de arquivo."""
result = run_bash(f"head -{lines} {path}")
return format_output(result, max_length=5000)
@staticmethod
def search(path: str, pattern: str) -> str:
"""Busca texto em arquivos."""
result = run_bash(f"grep -rn '{pattern}' {path} 2>/dev/null | head -50")
return format_output(result, max_length=5000)
@staticmethod
def write(path: str, content: str) -> str:
"""Escreve conteúdo em arquivo."""
# Escapa o conteúdo para evitar injection
import shlex
safe_content = shlex.quote(content)
result = run_bash(f"echo {safe_content} > {path}")
return format_output(result)
@staticmethod
def exists(path: str) -> str:
"""Verifica se arquivo existe."""
exists = os.path.exists(path)
return f"{'[OK]' if exists else '[ERROR]'} {'Existe' if exists else 'Não existe'}: {path}"
@staticmethod
def size(path: str) -> str:
"""Retorna tamanho de arquivo."""
result = run_bash(f"du -sh {path} 2>/dev/null || ls -lh {path}")
return format_output(result)
# ============================================================
# SYSTEM TOOLS
# ============================================================
class SystemTools:
"""Ferramentas de sistema."""
@staticmethod
def df() -> str:
"""Mostra uso de disco."""
result = run_bash("df -h")
return format_output(result)
@staticmethod
def free() -> str:
"""Mostra uso de memória."""
result = run_bash("free -h")
return format_output(result)
@staticmethod
def top(limit: int = 10) -> str:
"""Mostra processos mais pesados."""
result = run_bash(f"ps aux --sort=-%cpu | head -{limit + 1}")
return format_output(result)
@staticmethod
def uptime() -> str:
"""Mostra uptime do sistema."""
result = run_bash("uptime")
return format_output(result)
@staticmethod
def services() -> str:
"""Lista serviços ativos."""
result = run_bash("systemctl list-units --type=service --state=running | head -20")
return format_output(result)
@staticmethod
def ports() -> str:
"""Lista portas em uso."""
result = run_bash("netstat -tlnp 2>/dev/null || ss -tlnp")
return format_output(result, max_length=3000)
# ============================================================
# TOOLKIT REGISTRY
# REGISTRY
# ============================================================
TOOLS_V2 = {
# DOCKER
"docker_ps": {"desc": "Lista containers Docker", "func": DockerTools.ps, "danger": "safe"},
"docker_stats": {"desc": "Estatísticas de containers", "func": DockerTools.stats, "danger": "safe"},
"docker_logs": {"desc": "Logs de container (use: docker_logs <nome> <linhas>)", "func": lambda n="app", l=50: DockerTools.log(n, int(l)), "danger": "safe"},
"docker_restart": {"desc": "Reinicia container (use: docker_restart <nome>)", "func": DockerTools.restart, "danger": "dangerous"},
"docker_stop": {"desc": "Para container", "func": DockerTools.stop, "danger": "dangerous"},
"docker_start": {"desc": "Inicia container", "func": DockerTools.start, "danger": "medium"},
"docker_exec": {"desc": "Executa comando no container", "func": DockerTools.exec, "danger": "dangerous"},
"docker_system_df": {"desc": "Uso de disco Docker", "func": DockerTools.system_df, "danger": "safe"},
"docker_prune": {"desc": "Limpa recursos Docker não usados", "func": lambda: DockerTools.prune(True), "danger": "dangerous"},
# GIT
"git_status": {"desc": "Status do repositório git", "func": GitTools.status, "danger": "safe"},
"git_pull": {"desc": "Pull do git", "func": GitTools.pull, "danger": "medium"},
"git_push": {"desc": "Push do git", "func": GitTools.push, "danger": "dangerous"},
"git_clone": {"desc": "Clona repositório", "func": GitTools.clone, "danger": "medium"},
"git_branch": {"desc": "Lista branches", "func": GitTools.branch, "danger": "safe"},
"git_log": {"desc": "Histórico de commits", "func": GitTools.log, "danger": "safe"},
"git_diff": {"desc": "Diferenças não commitadas", "func": GitTools.diff, "danger": "safe"},
"git_fetch": {"desc": "Busca atualizações", "func": GitTools.fetch, "danger": "safe"},
# DOCKER COMPOSE
"dc_up": {"desc": "Sobe serviços (use: dc_up <path>)", "func": DockerComposeTools.up, "danger": "dangerous"},
"dc_down": {"desc": "Para serviços", "func": DockerComposeTools.down, "danger": "dangerous"},
"dc_build": {"desc": "Constrói imagens", "func": DockerComposeTools.build, "danger": "medium"},
"dc_ps": {"desc": "Lista serviços", "func": DockerComposeTools.ps, "danger": "safe"},
"dc_logs": {"desc": "Logs de serviços", "func": DockerComposeTools.logs, "danger": "safe"},
"dc_restart": {"desc": "Reinicia serviços", "func": DockerComposeTools.restart, "danger": "dangerous"},
# GITEA
"gitea_list_repos": {"desc": "Lista repositórios Gitea", "func": GiteaTools.list_repos, "danger": "safe"},
"gitea_get_repo": {"desc": "Info de repositório (use: gitea_get_repo <owner/repo>)", "func": GiteaTools.get_repo, "danger": "safe"},
"gitea_list_actions": {"desc": "Lista workflows do repositório", "func": GiteaTools.list_actions, "danger": "safe"},
"gitea_trigger": {"desc": "Dispara workflow", "func": GiteaTools.trigger_workflow, "danger": "dangerous"},
# SUPABASE
"supabase_list_tables": {"desc": "Lista tabelas do Supabase", "func": SupabaseTools.list_tables, "danger": "safe"},
"supabase_query": {"desc": "Consulta tabela", "func": SupabaseTools.query, "danger": "safe"},
"supabase_insert": {"desc": "Insere dados", "func": SupabaseTools.insert, "danger": "dangerous"},
"supabase_update": {"desc": "Atualiza dados", "func": SupabaseTools.update, "danger": "dangerous"},
# COOLIFY
"coolify_status": {"desc": "Status do Coolify", "func": CoolifyTools.get_status, "danger": "safe"},
"coolify_apps": {"desc": "Lista aplicações Coolify", "func": CoolifyTools.list_applications, "danger": "safe"},
"coolify_deployments": {"desc": "Lista deployments recentes", "func": CoolifyTools.list_deployments, "danger": "safe"},
# FILES
"file_list": {"desc": "Lista diretório", "func": FileTools.list, "danger": "safe"},
"file_read": {"desc": "Lê arquivo", "func": FileTools.read, "danger": "safe"},
"file_search": {"desc": "Busca em arquivos", "func": FileTools.search, "danger": "safe"},
"file_exists": {"desc": "Verifica se arquivo existe", "func": FileTools.exists, "danger": "safe"},
"file_size": {"desc": "Tamanho de arquivo", "func": FileTools.size, "danger": "safe"},
# SYSTEM
"sys_df": {"desc": "Uso de disco", "func": SystemTools.df, "danger": "safe"},
"sys_free": {"desc": "Uso de memória", "func": SystemTools.free, "danger": "safe"},
"sys_top": {"desc": "Processos mais pesados", "func": SystemTools.top, "danger": "safe"},
"sys_uptime": {"desc": "Uptime do sistema", "func": SystemTools.uptime, "danger": "safe"},
"sys_ports": {"desc": "Portas em uso", "func": SystemTools.ports, "danger": "safe"},
"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_tools_by_danger(level: str) -> List[Dict]:
"""Retorna ferramentas por nível de perigo."""
return [
{"name": k, **v}
for k, v in TOOLS_V2.items()
if v["danger"] == level
]
def get_all_tools_formatted() -> str:
"""Retorna lista formatada de todas as ferramentas."""
output = "[TOOLS] Ferramentas Disponiveis:\n\n"
for level in ["safe", "medium", "dangerous"]:
tools = get_tools_by_danger(level)
if tools:
icon = {"safe": "[SAFE]", "medium": "[MEDIUM]", "dangerous": "[CRITICAL]"}[level]
output += f"\n{icon} **{level.upper()}**:\n"
for t in tools:
output += f" - `{t['name']}` - {t['desc']}\n"
return output
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]