Kit de Ferramentas para Automação Zendesk

Uma biblioteca Python projetada para acelerar o desenvolvimento de robôs de automação de processos (RPA) que interagem com a plataforma Zendesk. Ela combina a automação de interface de usuário (UI) com o Selenium e a interação via API com o Zenpy, além de fornecer componentes integrados para monitoramento (Heartbeat) e logging estruturado (BotsLogger).

Principais Funcionalidades 🚀 Gerenciamento de Driver: Crie e gerencie drivers do Selenium (Chrome, Firefox) de forma simplificada, com suporte para execução local e remota (Selenoid).

Automação UI Zendesk: Um conjunto de métodos de alto nível para interagir com a interface do Zendesk, como fazer login, preencher campos, submeter tickets, aplicar macros e muito mais.

Integração API Zendesk: Um wrapper conveniente para a biblioteca Zenpy que facilita a autenticação e a extração de dados da API do Zendesk, como buscar tickets em uma visualização.

Monitoramento e Alertas: A classe Heartbeat permite enviar atualizações de status e alertas para um painel de monitoramento centralizado via API.

Logging Estruturado: A classe BotsLogger gera eventos JSON no stdout, com identificação do bot, execução, ambiente e contexto por ticket. Dados enviados nos atributos com chaves sensíveis são mascarados automaticamente.

Runtime para RPAs: A classe BotRuntime coordena logs, Heartbeat e pulsos automáticos de vida para processos contínuos. Use-a em RPAs; para DAGs, prefira os callbacks do orquestrador.

Instalação Instale a biblioteca usando o pip:

Bash

pip install cybertron-spark==0.2.1 Como Usar A biblioteca é dividida em componentes modulares que podem ser usados em conjunto para construir um robô robusto.

1. Configurando o Driver do Selenium (Driver_Selenium) Esta classe abstrai a configuração do WebDriver.

Python

from cybertron_spark import Driver_Selenium

Configura um driver remoto (Selenoid) para o Chrome em modo headless

config_driver = Driver_Selenium( ipselenoid='http://seu-ip-selenoid:4444/wd/hub', browser='chrome', headless=True )

Cria a instância do driver

driver = config_driver.criar_driver()

print("Driver criado com sucesso!")

... seu código de automação aqui ...

driver.quit()

2. Monitorando o Robô (Heartbeat) Envie o status do seu robô para um painel de acompanhamento.

from cybertron_spark import Heartbeat

Instancia o Heartbeat com as credenciais da sua API de monitoramento

heartbeat = Heartbeat( bot_id='MEU_BOT_ZENDESK_01', endpoint='https://api.meupainel.com/status', token='SEU_TOKEN_SECRETO' )

Envia um status

heartbeat.alertas(status=execucao) print("Status 'EM EXECUÇÃO' enviado.")

Envia o ID de um ticket processado para contagem

heartbeat.alertas(ticket_id='12345') print("Registro do ticket 12345 enviado.")

Ao final, informa que o bot está ocioso

heartbeat.alertas(status=aguardando) print("Status 'AGUARDANDO CASOS' enviado.")

3. Usando a API do Zendesk (Zendesk_Zenpy) Busque tickets de uma visualização específica antes de iniciar a automação da UI.

from cybertron_spark import Zendesk_Zenpy

Autentica na API do Zendesk

zen_api = Zendesk_Zenpy( zlogin='seu_email@empresa.com/token', zpass='SEU_TOKEN_DA_API_ZENDESK', instancia='sua-instancia-zendesk' )

Pega todos os tickets da visualização com ID 9000

ID_DA_FILA = 9000 tickets_para_processar = zen_api.pegar_tickets(fila=ID_DA_FILA)

if tickets_para_processar: print(f"Encontrados {len(tickets_para_processar)} tickets: {tickets_para_processar}") else: print("Nenhum ticket encontrado na fila.") 4. Automatizando a Interface (Zendesk_Selenium) Após obter um ticket, use o Selenium para interagir com ele.

Python

from cybertron_spark import Zendesk_Selenium, Driver_Selenium

(Supondo que você já tenha um 'driver' criado)

driver = Driver_Selenium(...).criar_driver()

Instancia o controlador da UI do Zendesk

zen_ui = Zendesk_Selenium( driver=driver, usuario='seu_usuario_zendesk', senha='sua_senha_zendesk', instancia='sua-instancia-zendesk' )

Realiza o login

zen_ui.login() print("Login realizado com sucesso.")

Navega até um ticket e realiza ações

ticket_id = '12345' driver.get(f"https://sua-instancia-zendesk.zendesk.com/agent/tickets/{ticket_id}")

zen_ui.esperar_carregamento()

Aplica uma macro

zen_ui.aplicar_macro("Nome da Macro::Opção")

Adiciona uma observação interna

zen_ui.enviar_mensagem("Este ticket foi processado pelo robô.")

Submete o ticket como resolvido

zen_ui.enviar_ticket('resolvido') print(f"Ticket {ticket_id} resolvido.")

driver.quit()

5. Logging Estruturado (BotsLogger) Emita eventos JSON para o coletor de logs da plataforma. O logger não depende do Heartbeat: cada mecanismo permanece responsável por sua própria finalidade.

Python

from cybertron_spark import BotsLogger

logs = BotsLogger(bot_id='meu_bot', environment='production') logger = logs.get_logger(name)

ticket_id = '12345'

try: # Simula um erro raise ValueError("Ocorreu um problema inesperado.") except Exception as e: logger.exception( 'Falha ao processar ticket', extra={ 'event': 'ticket.processing_failed', 'ticket_id': ticket_id, }, )

logger.info('Ticket iniciado', extra={'event': 'ticket.started', 'ticket_id': ticket_id})

6. Runtime para RPAs Contínuos (BotRuntime)

from cybertron_spark import BotRuntime

runtime = BotRuntime(
    bot_id='meu_bot',
    intervalo_vida=60,
    automation={
        'display_name': 'Meu bot',
        'execution_type': 'deployment',
        'brands': ['magalu'],
    },
)
runtime.iniciar(status_inicial='EM EXECUÇÃO')

try:
    # Fluxo de negócio do bot.
    runtime.registrar_ticket(ticket_id='12345')
finally:
    runtime.encerrar()

O runtime chama registrar_vida() imediatamente e depois a cada intervalo, sem repetir status ou etapas. A API Cybertron deve disponibilizar a rota configurada em CYBERTRON_HEARTBEAT_ENDPOINT (por padrão, /transformers/bots-heartbeat/), atualizando somente last_heartbeat_at.

Quando CYBERTRON_V2_API_URL estiver injetada, o mesmo BotRuntime também registra a automação, abre e encerra uma execução, envia heartbeats, registra filas por atualizar_fila(...) e salva os itens concluídos por registrar_ticket(...). Não é necessário criar um cliente V2 em cada script. Enquanto CYBERTRON_API_URL estiver configurada, o Heartbeat legado continua sendo enviado em paralelo durante a migração.

7. Consultas BigQuery auditáveis

Para bots que tomam decisões com dados do BigQuery, use o helper do runtime. Ele captura automaticamente o identificador do job, projeto, localização, datas, volume processado, uso de cache e o hash da SQL. A SQL integral não é enviada a logs nem ao payload.

from google.cloud import bigquery
from cybertron_spark import BotRuntime

runtime = BotRuntime(bot_id="meu_bot")
consulta = runtime.consultar_bigquery(
    bigquery.Client(),
    "SELECT protocolo, status FROM operacao.protocolos",
    query_name="protocolos_pendentes_v1",
)

for linha in consulta.to_records(include_fields=["protocolo", "status"]):
    # Execute a regra de negócio e, ao publicar na API Cybertron V2, envie:
    payload = consulta.payload_do_item(linha)
    # payload["source_metadata"] identifica a consulta/job.
    # payload["input_snapshot"] preserva o retrato lido pelo bot.

Use include_fields para definir, por automação, quais colunas compõem a prova operacional. Isso evita persistir colunas que não contribuem para a auditoria. O helper é opcional: bots que não usam BigQuery não precisam instalar nem configurar nada adicional.

8. API Cybertron V2 e ambientes pelo Infisical

O endereço da V2 é definido pelo ambiente no Infisical, e não pelo código do bot. Configure a mesma chave nos dois ambientes, com valores distintos:

Mantenha CYBERTRON_API_TOKEN no mesmo caminho de secrets. A biblioteca recebe as variáveis já injetadas pelo Infisical; ela não executa login no cofre, não grava .env e não possui URL fixa de ambiente.

from cybertron_spark import CybertronV2Client

cybertron = CybertronV2Client()
cybertron.registrar_automacao({
    "id": "meu_bot",
    "execution_type": "deployment",
    "display_name": "Meu bot",
    "description": "Descrição curta",
    "is_active": True,
})

CYBERTRON_API_URL continua reservado ao Heartbeat legado. Assim, a migração para V2 pode ocorrer bot a bot, sem mudar o comportamento dos consumidores existentes.

Exemplo Completo: Estrutura de um Robô Juntando todos os componentes para criar um fluxo de automação robusto.

Python

from rpa_zendesk_toolkit import ( Heartbeat, BotsLogger, Driver_Selenium, Zendesk_Zenpy, Zendesk_Selenium, execucao, aguardando, erro ) import time

--- 1. CONFIGURAÇÃO INICIAL ---

HEARTBEAT_CONFIG = { 'bot_id': 'ZENDESK_PROCESSOR_01', 'endpoint': 'https://api.meupainel.com/status', 'token': 'SEU_TOKEN_SECRETO' }

ZENDESK_API_CREDS = { 'zlogin': 'seu_email@empresa.com/token', 'zpass': 'SEU_TOKEN_DA_API_ZENDESK', 'instancia': 'sua-instancia' }

ZENDESK_UI_CREDS = { 'usuario': 'seu_usuario_zendesk', 'senha': 'sua_senha_zendesk', 'instancia': 'sua-instancia' }

SELENOID_IP = 'http://seu-ip-selenoid:4444/wd/hub'

ID_FILA_ZENDESK = 9000

--- 2. INICIALIZAÇÃO DOS COMPONENTES ---

heartbeat = Heartbeat(**HEARTBEAT_CONFIG)

logger = BotsLogger(heartbeat_instancia=heartbeat)

zen_api = Zendesk_Zenpy(**ZENDESK_API_CREDS)

--- 3. LOOP PRINCIPAL DO ROBÔ ---

def main(): driver = None try: heartbeat.alertas(status=aguardando) tickets = zen_api.pegar_tickets(fila=ID_FILA_ZENDESK)

    if not tickets:
        print("Nenhum ticket na fila. Aguardando...")
        return

    heartbeat.alertas(status=execucao)

    # Cria o driver apenas se houver tickets para processar
    driver_manager = Driver_Selenium(ipselenoid=SELENOID_IP)
    driver = driver_manager.criar_driver()

    zen_ui = Zendesk_Selenium(driver=driver, **ZENDESK_UI_CREDS)
    zen_ui.login()

    for ticket_id in tickets:
        try:
            print(f"Processando ticket: {ticket_id}")
            driver.get(f"https://{ZENDESK_UI_CREDS['instancia']}.zendesk.com/agent/tickets/{ticket_id}")
            zen_ui.esperar_carregamento()

            # --- Lógica de negócio do seu robô ---
            zen_ui.enviar_mensagem(f"Processamento automático iniciado pelo bot {HEARTBEAT_CONFIG['bot_id']}.")
            time.sleep(2) # Simula trabalho
            zen_ui.enviar_ticket('pendente')
            zen_ui.esperar_carregamento()
            # ------------------------------------

            heartbeat.alertas(ticket_id=ticket_id) # Registra sucesso

        except Exception as e:
            logger.error(
                error=e,
                message=f"Erro ao processar ticket {ticket_id}",
                ticket_id=ticket_id
            )
            # Pular para o próximo ticket em caso de erro

except Exception as e:
    acao = logger.error(
        error=e,
        message="Erro crítico na execução principal do robô",
        ticket_id="N/A",
        error_type='CRITICAL'
    )
    if acao == 'stop':
        heartbeat.alertas(status=erro)
        # Adicione lógica para parar o robô de forma segura

finally:
    if driver:
        driver.quit()
    print("Ciclo finalizado.")

if name == 'main': while True: main() time.sleep(60) # Espera 1 minuto antes de verificar a fila novamente

Integração com Playwright e Selenwright 🚀

A biblioteca agora conta com suporte nativo ao Playwright, oferecendo automações muito mais rápidas, estáveis e resilientes (auto-wait) em comparação com o Selenium tradicional.

Foram adicionados módulos específicos para gerenciar instâncias de navegador e interagir com sistemas estratégicos (Zendesk, Sandiego e MLAdmin).

1. Configurando o Driver do Playwright

A classe Driver_Playwright gerencia a criação de contextos e páginas, com suporte a instâncias locais e remotas (Selenwright).

from playwright.sync_api import sync_playwright
from cybertron_spark.playwright_manager import Driver_Playwright

with sync_playwright() as p:
    driver_manager = Driver_Playwright(
        playwright_instance=p, 
        local=True, 
        headless=False 
    )
    browser, context, page = driver_manager.criar_browser()
  
    # ... passe o objeto 'page' para as classes de automação ...
  
    browser.close()

2. Automação Playwright: Zendesk

Realize fluxos completos na Zendesk de forma mais rápida.

from cybertron_spark.playwright_manager import Zendesk_Playwright

zen = Zendesk_Playwright(page=page, instancia="sua-instancia")
zen.login()

zen.abrir_ticket("1234567")
zen.aplicar_macro("Classificação::Teste")
zen.enviar_mensagem("<p>Nota interna automática via Playwright</p>", publica=False)
zen.enviar_ticket("aberto")
zen.fechar_ticket_atual() # Limpa a aba para não sobrecarregar a memória

3. Automação Playwright: Sandiego

Gerencie cancelamentos e trocas de pedidos diretamente no sistema Sandiego.

from cybertron_spark.playwright_manager import Sandiego_Playwright

sandiego = Sandiego_Playwright(page=page)
sandiego.login()

# Tenta realizar a troca e valida se o botão 'Abrir Ticket' está habilitado
sucesso = sandiego.gerar_troca(numero_pedido="1019386971")
if sucesso:
    print("Troca gerada com sucesso.")

4. Automação Playwright: MLAdmin

Gerencie fluxos para pedidos 1P (que possuem pacotes) no MLAdmin, incluindo suporte avançado a interceptação de alertas em tela (Dialog).

from cybertron_spark.playwright_manager import MLAdmin_Playwright

mladmin = MLAdmin_Playwright(page=page)
mladmin.login()

# Tenta gerar um vale-compras e o robô já intercepta alertas nativos do sistema
sucesso = mladmin.gerar_vale(pacote="987654321")
if sucesso:
    print("Vale gerado com sucesso.")

Licença
Este projeto está licenciado sob a licença MIT. Veja o arquivo LICENSE para mais detalhes.