beapro-state-machine

BeaPro (BotCity Enterprise Automation Process) é uma biblioteca Python de automação baseada em máquina de estados para construir bots de processamento de dados com integração nativa à plataforma BotCity Maestro.

Fornece uma arquitetura declarativa com quatro estados padrão e um pipeline de tratamento de erros embutido, deixando ao desenvolvedor apenas a responsabilidade de implementar a lógica de negócio no estado PROCESS.

Instalação

pip install beapro-state-machine

Fluxo de Estados

INIT → GET_DATA → PROCESS → END
  ↓        ↓          ↓
  └──────→ ErrorHandler (retry → fallback → abort)

Estado	Responsabilidade
INIT	Valida ambiente e prepara serviços
GET_DATA	Fornece o próximo item para processamento
PROCESS	Lógica de negócio — ponto de extensão principal
END	Reporta contagens ao Maestro e finaliza a task
ErrorHandler	Captura exceções, executa retries e envia alertas em fallback

Modos de Operação

Modo	mode=	Fonte dos dados	Iteração
datapool (padrão)	"datapool"	Itens do DataPool do Maestro	Loop até esgotar todos os itens
task	"task"	Parâmetros da task atual	Processa uma única vez

No modo task não é necessário definir datapool_label. Os parâmetros configurados na Automação do Maestro ficam disponíveis em context.item.data dentro do ProcessState.

Uso Básico

1. Implemente o estado PROCESS

from beapro.state_templates import ProcessState

class MeuProcessState(ProcessState):
    def execute(self):
        item = self.context.item
        self.services.logger.info(f"Processando: {item.data}")
        # lógica de negócio aqui
        return "SUCCESS"

Valores de retorno aceitos por execute():

Retorno	Efeito
"SUCCESS"	Item marcado como sucesso; avança para o próximo
"BUSINESS_ERROR"	Item marcado como erro de negócio; avança para o próximo
qualquer outro / exceção	Aciona pipeline de retry → fallback → abort

2. Execute com beapro.Bot

Bot é o ponto de entrada principal. Gerencia credenciais, contexto e ciclo de vida internamente.

Modo datapool:

import beapro
from beapro.state_templates import ProcessState

class MeuProcessState(ProcessState):
    def execute(self):
        self.services.logger.info(f"Processando: {self.context.item.data}")
        return "SUCCESS"

beapro.Bot(datapool_label="meu-datapool").set_process(MeuProcessState()).run()

Modo task:

import beapro
from beapro.state_templates import ProcessState

class MeuProcessState(ProcessState):
    def execute(self):
        params = self.context.item.data
        self.services.logger.info(f"Parâmetros: {params}")
        return "SUCCESS"

beapro.Bot(mode="task").set_process(MeuProcessState()).run()

Com credenciais explícitas (execução local):

beapro.Bot(
    datapool_label="meu-datapool",
    server="https://developers.botcity.dev",
    login="<uuid>",
    key="<chave-api>",
    retry_limit=5,
).set_process(MeuProcessState()).run()

Com Runner do Maestro: deixe server, login e key nos valores padrão — o Runner injeta as credenciais automaticamente via sys.argv.

beapro.Bot(datapool_label="meu-datapool").set_process(MeuProcessState()).run()

Parâmetros do Bot

Parâmetro	Padrão	Descrição
datapool_label	""	Label do DataPool no Maestro (modo datapool)
mode	"datapool"	"datapool" ou "task"
server	"https://developers.botcity.dev"	URL do servidor Maestro
login	""	UUID de login
key	""	Chave de API
retry_limit	3	Número máximo de retries por estado
logs_label	""	Label para filtrar logs no Maestro

Tratamento de Erros

Lance as exceções da lib para acionar comportamentos específicos do pipeline:

from beapro import BusinessException, SystemException, StopProcessException
from beapro.state_templates import ProcessState

class MeuProcessState(ProcessState):
    def execute(self):
        item = self.context.item

        if not item.data.get("cpf"):
            raise BusinessException("CPF ausente — item inválido")

        if not self._conectar_erp():
            raise SystemException("Timeout ao conectar ao ERP")

        return "SUCCESS"

Exceção	Efeito
BusinessException	Item marcado como erro de negócio; sem retry
SystemException	Aciona retry; em abort, reinicia a task automaticamente
StopProcessException	Parada controlada sem registrar erro
InterruptRequested	Interrupção via Maestro; relançada sem retry

Serviços Disponíveis em execute()

Dentro de qualquer estado, self.services dá acesso aos serviços da execução.

Logging

self.services.logger.log("Mensagem", level="INFO")  # genérico
self.services.logger.info("Mensagem")
self.services.logger.warn("Aviso")
self.services.logger.error("Erro")
self.services.logger.debug("Debug")  # apenas arquivo local, não enviado ao Maestro

BEAPRO_LOG_DIR=/caminho/para/logs   # padrão: output/

Alertas

# Anti-spam: cada title é enviado no máximo uma vez por execução
self.services.alert.info(title="Início", message="Bot iniciado")
self.services.alert.warn(title="Lentidão", message="Tempo de resposta elevado")
self.services.alert.error(title="Falha", message="Conexão ao ERP perdida")

Screenshots

# Salva localmente — retorna o caminho do arquivo ou None
path = self.services.screenshot.capture(label="erro")

# Salva localmente e envia ao Maestro como artefato — anti-spam por label
path = self.services.screenshot.capture_and_upload(label="erro")

Credenciais

CredentialService recupera segredos do cofre do Maestro. Cache de sessão: cada par (label, key) é buscado uma única vez por execução.

senha = self.services.credential.get(label="sistema-erp", key="senha")
usuario = self.services.credential.get(
    label="sistema-erp",
    key="usuario",
    fallback="admin_local",  # retornado se Maestro estiver indisponível
)

# Invalidar cache (força nova busca no próximo get)
self.services.credential.invalidate(label="cofre", key="chave")
self.services.credential.invalidate_all()

Situação	Resultado
Maestro disponível	valor do cofre (cacheado)
Maestro indisponível + fallback fornecido	fallback retornado silenciosamente
Maestro indisponível + sem fallback	exceção original relançada

Reporte de item do DataPool

self.services.datapool.report_success("Processado com sucesso")
self.services.datapool.report_error("BUSINESS EXCEPTION", "CPF inválido")
# error_type: 'SYSTEM EXCEPTION' | 'BUSINESS EXCEPTION' | 'INTERRUPTION REQUESTED'

Reporte de erros ao Maestro

Anti-spam: cada combinação (tipo_de_exceção, estado) é reportada no máximo uma vez por execução.

try:
    # lógica de negócio
    return "SUCCESS"
except Exception as e:
    screenshot = self.services.screenshot.capture(label="erro")
    self.services.error_reporter.report(
        exception=e,
        state="PROCESS",
        screenshot=screenshot,
    )
    raise

Uso Avançado — API de Baixo Nível

Para cenários que exijam estados INIT ou END customizados, screenshot_handler próprio, ou controle explícito do ciclo de vida, use a API diretamente:

import beapro
beapro.configure()  # deve ser chamado antes de instanciar ExecutionContext

from beapro import BeaproStateMachine, ExecutionContext, ContextProvider
from beapro.state_templates import ProcessState, InitState

class MeuInit(InitState):
    def execute(self):
        self.services.logger.info("Setup customizado")
        return "SUCCESS"

class MeuProcess(ProcessState):
    def execute(self):
        return "SUCCESS"

context = ExecutionContext(datapool_label="meu-datapool")
ContextProvider.initialize(context)
try:
    BeaproStateMachine(
        init_state=MeuInit(),
        process_state=MeuProcess(),
    ).activate_initial_state()
finally:
    ContextProvider.reset()

Parâmetros de configure()

Parâmetro	Padrão	Descrição
server	"https://developers.botcity.dev"	URL do servidor Maestro
login	""	UUID de login
key	""	Chave de API

API Pública

import beapro

# Ponto de entrada principal
beapro.Bot(...)

# Configuração manual (uso avançado)
beapro.configure()

from beapro import (
    Bot,
    BeaproStateMachine,   # orquestrador principal
    ExecutionContext,     # estado compartilhado da execução
    TransactionItem,      # item de dados atual
    ContextProvider,      # singleton de contexto global
    BusinessException,
    SystemException,
    StopProcessException,
    InterruptRequested,
)

from beapro.state_templates import ProcessState  # base para extensão

Estrutura da Biblioteca

beapro/
├── __init__.py               # API pública + Bot + configure()
├── bot.py                    # Bot — ponto de entrada unificado
├── core/
│   ├── state_machine.py      # BeaproStateMachine
│   ├── base_state.py         # BaseState (abstrata)
│   ├── context_provider.py   # ContextProvider (singleton)
│   ├── execution.py          # ExecutionContext + TransactionItem + gateway do SDK
│   ├── container.py          # ServicesContainer (DI)
│   └── exceptions.py         # Hierarquia de exceções
├── services/
│   ├── logger.py             # MaestroLogger
│   ├── datapool.py           # DataPool (lazy init)
│   ├── alert.py              # Alert
│   ├── screenshot.py         # ScreenshotService
│   ├── credential.py         # CredentialService (cofre do Maestro + cache)
│   └── error_reporter.py     # ErrorReporter (anti-spam para painel do Maestro)
└── state_templates/
    ├── init.py               # InitState
    ├── get_data.py           # GetDataState (modo datapool)
    ├── get_task.py           # GetTaskState (modo task)
    ├── process.py            # ProcessState (ponto de extensão)
    └── end.py                # EndState

Dependências

Pacote	Versão
botcity-maestro-sdk	==0.9.0
python-statemachine	>=3.0.0,<4.0.0
mss	>=9.0.0

🔗 Documentação

• BotCity Documentação
• BotCity Maestro SDK
• BotCity Plugins
• BotCity Community