# ============================================================ # TOOLS_V2.PY - Ferramentas Expandidas para o Orquestrador # NÃO SUBSTITUI tools.py - É um módulo adicional # ============================================================ import subprocess import os import requests 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 ) # ============================================================ # CONSTANTS # ============================================================ DANGER_LEVELS = { "safe": "SAFE - Executa automático", "medium": "MEDIUM - Informa antes", "dangerous": "DANGEROUS - Pede confirmação" } # ============================================================ # UTILITY FUNCTIONS # ============================================================ def run_bash(command: str, timeout: int = 120) -> Dict: """Executa comando bash e retorna resultado estruturado.""" try: 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)" } 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 # ============================================================ # 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) @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) @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) @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) # ============================================================ # GIT TOOLS # ============================================================ class GitTools: """Ferramentas Git.""" @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) # ============================================================ # 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 # ============================================================ 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.""" url = f"{gitea_api_url()}/user/repos" try: res = requests.get(url, headers=GiteaTools._get_headers(), timeout=10) if res.status_code == 200: 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 # ============================================================ 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).""" 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)}" # ============================================================ # 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 # ============================================================ 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 )", "func": lambda n="app", l=50: DockerTools.log(n, int(l)), "danger": "safe"}, "docker_restart": {"desc": "Reinicia container (use: docker_restart )", "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 )", "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 )", "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"}, } 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