SlipItIn/.github/skills/pier-cloud/scripts/pier_cloud_client.py

#!/usr/bin/env python3
"""
Cliente robusto para API Pier Cloud com retry, renovacao automatica e CLI.
"""

import requests
import os
import time
import logging
import argparse
import json
from dotenv import load_dotenv

# Configurar logging
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)

class PierCloudClient:
    """Cliente robusto para API Pier Cloud"""
    
    def __init__(self):
        load_dotenv()
        self.api_base = "https://api.piercloud.io"
        self.client_id = os.getenv("PIERCLOUD_CLIENT_ID")
        self.client_secret = os.getenv("PIERCLOUD_CLIENT_SECRET")
        self.tenancy_id = os.getenv("PIERCLOUD_TENANCY_ID", os.getenv("PIERCLOUD_BUSINESS_ID"))
        
        self.token = None
        self.token_expires = None
        
        self._validate_config()
    
    def _validate_config(self):
        """Validar configuracao necessaria"""
        required = ["client_id", "client_secret", "tenancy_id"]
        missing = [k for k in required if not getattr(self, k)]
        
        if missing:
            raise ValueError(f"Configuracao faltando: {missing}")
    
    def authenticate(self):
        """Autenticar e obter token"""
        logger.info("Autenticando...")
        
        response = requests.post(
            f"{self.api_base}/auth",
            json={"client_id": self.client_id, "client_secret": self.client_secret},
            timeout=30
        )
        
        if response.status_code == 201:
            data = response.json()['data']
            self.token = data['access_token']
            # Renovar 5 minutos antes de expirar
            self.token_expires = time.time() + data['expires_in'] - 300
            logger.info("OK Autenticado com sucesso")
            return True
        else:
            logger.error(f"X Falha na autenticacao: {response.text}")
            return False
    
    def ensure_authenticated(self):
        """Garantir token valido"""
        if not self.token or time.time() >= self.token_expires:
            return self.authenticate()
        return True
    
    def make_request(self, method, endpoint, **kwargs):
        """Fazer requisicao com retry e renovacao automatica"""
        max_retries = 3
        
        for attempt in range(max_retries):
            try:
                if not self.ensure_authenticated():
                    raise Exception("Falha na autenticacao")
                
                headers = kwargs.get('headers', {})
                headers['Authorization'] = f"Bearer {self.token}"
                kwargs['headers'] = headers
                kwargs.setdefault('timeout', 30)
                
                response = requests.request(method, f"{self.api_base}{endpoint}", **kwargs)
                
                if response.status_code == 401:
                    logger.warning("Token expirado, renovando...")
                    self.authenticate()
                    continue
                
                if response.status_code == 429:
                    wait_time = 2 ** attempt
                    logger.warning(f"Rate limit. Aguardando {wait_time}s...")
                    time.sleep(wait_time)
                    continue
                
                response.raise_for_status()
                return response.json()
                
            except requests.exceptions.RequestException as e:
                logger.error(f"Erro na tentativa {attempt + 1}: {e}")
                
                if attempt == max_retries - 1:
                    raise
                
                time.sleep(2 ** attempt)
        
        return None
    
    def list_contexts(self):
        """Listar contextos"""
        endpoint = f"/lighthouse/tenancies/{self.tenancy_id}/contexts"
        return self.make_request('GET', endpoint)
    
    def list_workspaces(self, page=1, page_size=10):
        """Listar workspaces"""
        endpoint = f"/lighthouse/tenancies/{self.tenancy_id}/workspaces"
        params = {"page": page, "page_size": page_size}
        return self.make_request('GET', endpoint, params=params)
    
    def get_workspace(self, workspace_id):
        """Obter workspace especifico"""
        endpoint = f"/lighthouse/tenancies/{self.tenancy_id}/workspaces/{workspace_id}"
        return self.make_request('GET', endpoint)
    
    def get_all_workspaces(self):
        """Obter todos os workspaces com paginacao automatica"""
        all_workspaces = []
        page = 1
        page_size = 100
        
        while True:
            result = self.list_workspaces(page=page, page_size=page_size)
            workspaces = result['data']['workspaces']
            meta = result['meta']
            
            all_workspaces.extend(workspaces)
            logger.info(f"Pagina {page}: {len(workspaces)} workspaces")
            
            if page * page_size >= meta['total']:
                break
            
            page += 1
        
        return all_workspaces

def main():
    parser = argparse.ArgumentParser(description="Cliente CLI para API Pier Cloud")
    parser.add_argument("--action", required=True, 
                       choices=["list-contexts", "list-workspaces", "get-workspace", "get-all-workspaces"],
                       help="Acao a executar")
    parser.add_argument("--workspace-id", type=int, help="ID do workspace (para get-workspace)")
    parser.add_argument("--page", type=int, default=1, help="Numero da pagina")
    parser.add_argument("--page-size", type=int, default=10, help="Itens por pagina")
    parser.add_argument("--output", help="Arquivo de saida JSON")
    
    args = parser.parse_args()
    
    try:
        client = PierCloudClient()
        
        if args.action == "list-contexts":
            result = client.list_contexts()
            contexts = result['data']['contexts']
            print(f"\nContextos: {len(contexts)}")
            for ctx in contexts:
                print(f"  - {ctx['name']} ({ctx['provider']})")
        
        elif args.action == "list-workspaces":
            result = client.list_workspaces(page=args.page, page_size=args.page_size)
            workspaces = result['data']['workspaces']
            meta = result['meta']
            print(f"\nWorkspaces: {len(workspaces)} de {meta['total']}")
            for ws in workspaces:
                print(f"  - [{ws['id']}] {ws['name']}")
        
        elif args.action == "get-workspace":
            if not args.workspace_id:
                print("X Erro: --workspace-id e obrigatorio")
                exit(1)
            result = client.get_workspace(args.workspace_id)
            ws = result['data']
            print(f"\nWorkspace: {ws['name']}")
            print(f"ID: {ws['id']}")
            print(f"Visualizacoes: {len(ws.get('views', []))}")
        
        elif args.action == "get-all-workspaces":
            workspaces = client.get_all_workspaces()
            print(f"\nTotal: {len(workspaces)} workspaces")
            
            if args.output:
                with open(args.output, 'w', encoding='utf-8') as f:
                    json.dump(workspaces, f, indent=2, ensure_ascii=False)
                print(f"OK Salvo em {args.output}")
        
    except Exception as e:
        logger.error(f"X Erro: {e}")
        exit(1)

if __name__ == "__main__":
    main()