beapro-state-machine 0.1.4

State machine framework for BotCity automation bots

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

Configuração de Credenciais

Chame beapro.configure() antes de instanciar ExecutionContext. Ela carrega as credenciais do Maestro a partir de um arquivo .env (por padrão) ou de parâmetros explícitos, sem sobrescrever variáveis já definidas no ambiente de processo.

import beapro
beapro.configure()  # lê do .env sem sobrescrever vars já definidas no ambiente

from beapro import BeaproStateMachine, ExecutionContext, ContextProvider

Parâmetros de configure()

Parâmetro	Padrão	Descrição
server	""	URL do servidor Maestro — sobrescreve DEFAULT_SERVER
login	""	UUID de login — sobrescreve DEFAULT_LOGIN
key	""	Chave de API — sobrescreve DEFAULT_KEY

beapro.configure(
    server="https://developers.botcity.dev",
    login="<uuid>",
    key="<chave>"
)

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

BeaproStateMachine suporta dois modos, definidos pelo parâmetro mode:

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, com o mesmo contrato do modo datapool.

Passar get_data_state explicitamente tem prioridade sobre mode.

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.log(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 no DataPool; 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. Inicialize e execute — modo datapool

import beapro
beapro.configure()

from beapro import BeaproStateMachine, ExecutionContext, ContextProvider

context = ExecutionContext(datapool_label="meu-datapool")
ContextProvider.initialize(context)

try:
    machine = BeaproStateMachine(process_state=MeuProcessState())
    machine.activate_initial_state()
finally:
    ContextProvider.reset()

3. Inicialize e execute — modo task

Ideal para bots acionados diretamente pelo Maestro com parâmetros configurados na Automação. Não requer DataPool.

import beapro
beapro.configure()

from beapro import BeaproStateMachine, ExecutionContext, ContextProvider

context = ExecutionContext()
ContextProvider.initialize(context)

try:
    machine = BeaproStateMachine(mode="task", process_state=MeuProcessState())
    machine.activate_initial_state()
finally:
    ContextProvider.reset()

ContextProvider.reset() no bloco finally limpa o singleton ao final de cada execução.

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

Pipeline completo:

1. Exceção não tratada → fail() → ErrorHandler.retry
2. Retries esgotados → fallback → log crítico + alerta enviado
3. abort → executa END para limpeza final

Credenciais de Sistema

CredentialService (acessível via self.services.credential) recupera segredos armazenados no cofre do Maestro — senhas, tokens de API, chaves, etc.

Cache de sessão: cada par (label, key) é buscado uma única vez por execução. Chamadas subsequentes retornam o valor em cache sem nova requisição ao Maestro.

class MeuProcessState(ProcessState):
    def execute(self):
        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
        )
        return "SUCCESS"

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

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

Reporte de Erros ao Maestro

ErrorReporter (acessível via self.services.error_reporter) envia exceções ao painel de erros do Maestro com anti-spam: cada combinação (tipo_de_exceção, estado) é reportada no máximo uma vez por execução. Erros repetidos em loop de DataPool são suprimidos silenciosamente.

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

Comportamento	Detalhe
task_id ausente	ignorado silenciosamente (execução sem Maestro)
StopProcessException	nunca reportada (parada controlada, não é erro)
BusinessException	marcada como type=BUSINESS no painel
demais exceções	marcadas como type=SYSTEM

Logging

O logger escreve simultaneamente em três destinos:

1. Console — INFO e acima
2. Arquivo local — output/<timestamp>.log (todos os níveis)
3. BotCity Maestro — via SDK (falha silenciosa se indisponível)

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

API Pública

import beapro
beapro.configure()

from beapro import (
    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 + configure()
├── core/
│   ├── state_machine.py      # BeaproStateMachine
│   ├── base_state.py         # BaseState (abstrata)
│   ├── context_provider.py   # ContextProvider (singleton)
│   ├── execution.py          # ExecutionContext + TransactionItem
│   ├── container.py          # ServicesContainer (DI)
│   ├── exceptions.py         # Hierarquia de exceções
│   └── settings.py           # Leitura de variáveis de ambiente
├── 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
python-dotenv	>=1.0.0
mss	>=9.0.0

Padrões de Design

Padrão	Onde
Singleton	ContextProvider
Template Method	BaseState.execute()
Dependency Injection	ServicesContainer
Iterator	DataPool
Strategy	Estados concretos substituem comportamento padrão