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.1.33 Como Usar A biblioteca é dividida em componentes modulares que podem ser usados em conjunto para construir um robô robusto.
- 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()
- 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.")
- 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()
- 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})
- Runtime para RPAs Contínuos (BotRuntime)
from cybertron_spark import BotRuntime
runtime = BotRuntime(bot_id='meu_bot', intervalo_vida=60)
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.
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.
Metadata
Release files for cybertron-spark 0.1.33
For a detailed explanation of source distributions (sdists) and built distributions (wheels), please see the package formats documentation.
Source distribution (sdist)
| File | Size | Uploaded | |
|---|---|---|---|
| cybertron_spark-0.1.33.tar.gz | 30.0 kB | Details |
Built distribution (wheel)
| File | Interpreter | ABI | Platform | Reset |
|---|---|---|---|---|
| cybertron_spark-0.1.33-py3-none-any.whl | Python 3 | none | any | Details |
Total release size: 58.1 kB
Release files / cybertron_spark-0.1.33.tar.gz
| Download URL | cybertron_spark-0.1.33.tar.gz |
|---|---|
| Size | 30.0 kB |
| Tags | Source |
|
SHA-256 checksum How to use checksums |
95a2eea7e7e5b994c6a9eef6d29f158f10a3b141e63683bd5bca2c0536bfe7ac
|
|
BLAKE2b-256 checksum How to use checksums |
01be41d8b3ea3ab2662efbbac09151973f6a13b7e26fcef3b5ac7771a52d2c17
|
| Upload date | |
|
Uploaded using Trusted Publishing? What is trusted publishing? |
No |
| Uploaded via |
twine/7.0.0 CPython/3.10.12
|
Release files / cybertron_spark-0.1.33-py3-none-any.whl
| Download URL | cybertron_spark-0.1.33-py3-none-any.whl |
|---|---|
| Size | 28.1 kB |
| Tags | Python 3 |
|
SHA-256 checksum How to use checksums |
963bb380c1ffa5b1d1e70377ee3bcc08bd1a1f748330975e76dd3665867178a3
|
|
BLAKE2b-256 checksum How to use checksums |
f266c79347aa8e46ccda2420f0d1a59647e0c5353f6ac2697870cbc7a6f519e8
|
| Upload date | |
|
Uploaded using Trusted Publishing? What is trusted publishing? |
No |
| Uploaded via |
twine/7.0.0 CPython/3.10.12
|