Skip to main content

Biblioteca Python modular para automação RPA

Project description

rpaflow

Biblioteca Python modular para automação RPA.

Instalação

pip install rpaflow[all]  # Todos os módulos
pip install rpaflow[sql]  # Módulo específico

Módulos

Módulo pip install Descrição
sql pip install rpaflow[sql] MySQL, PostgreSQL, SQL Server, SQLite
excel pip install rpaflow[excel] Leitura e escrita de planilhas (openpyxl)
excel_com pip install rpaflow[excel-com] Automação Excel via COM (Windows) - 60+ métodos
browser pip install rpaflow[browser] Automação de navegador - recursive iframe
desktop pip install rpaflow[desktop] Automação via reconhecimento de imagem
log pip install rpaflow[log] Logging profissional (inspirado no Serilog)
ini pip install rpaflow[ini] Leitura e escrita de arquivos .ini
json pip install rpaflow[json] Leitura e escrita de JSON com dot notation
files pip install rpaflow[files] Operações com arquivos
api pip install rpaflow[api] Requisições HTTP/REST
email pip install rpaflow[email] Envio de emails SMTP

SQL

Conexão com bancos de dados MySQL, PostgreSQL, SQL Server e SQLite. Inspirado no System.Data.SqlClient do C#.

Instalar

pip install rpaflow[sql]

Como Usar (Passo a Passo)

from rpaflow.sql import SQL

# Conectar
db = SQL(type="sqlite", database="C:/dados/app.db")
db.connect()

# CREATE
db.execute("CREATE TABLE IF NOT EXISTS usuarios (id INTEGER PRIMARY KEY, nome TEXT, email TEXT)")

# INSERT
db.execute("INSERT INTO usuarios (nome, email) VALUES (?, ?)", "João", "joao@email.com")
db.commit()

# SELECT
usuarios = db.select("usuarios")
joao = db.select("usuarios", where={"nome": "João"})
db.disconnect()

Métodos

Conexão

Método Parâmetros Descrição
connect() Abre conexão
disconnect() Fecha conexão
is_connected() True/False

Query

Método Parâmetros Descrição
execute() sql, params Executa SQL (INSERT, UPDATE, DELETE)
commit() Salva alterações
rollback() Desfaz alterações

Select

Método Parâmetros Descrição
select() table, columns, where, order_by, limit Retorna lista de dicts
count() table, where Conta registros
exists() table, where True/False
scalar() sql, params Retorna valor único

Parâmetros Comuns

Parâmetro Padrão Descrição
type mysql, postgresql, sqlite
host localhost Servidor
port 5432/3306 Porta
user Usuário
password Senha
database Nome do banco

Parâmetros de Select

Parâmetro Tipo Descrição
columns list Colunas específicas
where dict Filtros
order_by str Ordenação
limit int Limite de registros

Exemplos Práticos

CRUD Completo

from rpaflow.sql import SQL

db = SQL(type="mysql", host="localhost", user="root", password="123", database="vendas")
db.connect()

# CREATE TABLE
db.execute("""
    CREATE TABLE IF NOT EXISTS pedidos (
        id INTEGER PRIMARY KEY AUTO_INCREMENT,
        cliente TEXT,
        valor DECIMAL(10,2),
        status TEXT DEFAULT 'pendente'
    )
""")

# INSERT
db.execute("INSERT INTO pedidos (cliente, valor) VALUES (?, ?)", "Maria", 1500.00)
db.commit()

# SELECT
pendentes = db.select("pedidos", where={"status": "pendente"})
print(f"Pendentes: {len(pendentes)}")

# UPDATE
db.execute("UPDATE pedidos SET status = 'cancelado' WHERE id = ?", 1)
db.commit()

# DELETE
db.execute("DELETE FROM pedidos WHERE id = ?", 1)
db.commit()

db.disconnect()

Usando where (dict)

usuarios = db.select("usuarios", where={"ativo": 1})
print(usuarios)

usuarios = db.select("usuarios", where={"ativo": 1, "nivel": "admin"})
print(usuarios)

Parâmetros nomeados

usuarios = db.select("usuarios", where={"nome": "João"})
pedidos = db.select("pedidos", where={"usuario_id": 1, "status": "pendente"})

SELECT com joins

results = db.select(
    "pedidos",
    columns=["pedidos.id", "clientes.nome", "pedidos.valor"],
    where={"pedidos.status": "pendente"},
    order_by="pedidos.data DESC"
)

Executar SQL raw

db.execute("UPDATE usuarios SET ativo = 0 WHERE data_cadastro < '2024-01-01'")
db.commit()

usuarios = db.select("usuarios", where={"ativo": 0})

Verificar existência

if db.exists("usuarios", {"email": "admin@email.com"}):
    print("Usuário já existe")

total = db.count("pedidos", where={"status": "concluido"})
print(f"Concluídos: {total}")

transactions

db.connect()
try:
    db.execute("UPDATE contas SET saldo = saldo - 100 WHERE id = 1")
    db.execute("UPDATE contas SET saldo = saldo + 100 WHERE id = 2")
    db.commit()
except:
    db.rollback()

Diferentes bancos

# MySQL
db = SQL(type="mysql", host="localhost", user="root", password="123", database="vendas")

# PostgreSQL
db = SQL(type="postgresql", host="localhost", user="postgres", password="123", database="vendas")

# SQL Server
db = SQL(type="sqlserver", host="localhost", user="sa", password="123", database="vendas")

# SQLite
db = SQL(type="sqlite", database="C:/dados/app.db")

Excel

Leitura e escrita de planilhas Excel (.xlsx) usando openpyxl. Funciona em qualquer SO.

Instalar

pip install rpaflow[excel]

Como Usar

from rpaflow.excel import Excel

excel = Excel()
excel.open("C:/dados/planilha.xlsx")

# Ler dados
celula = excel.get_cell("A1")
valor = excel.get_value("B2")
excel.set_value("C3", "novo valor")

# Salvar
excel.save()
excel.close()

Métodos

Abertura

Método Parâmetros Descrição
open() filepath, read_only, write_only Abre planilha
save() filepath Salva planilha
close() Fecha planilha
create_sheet() name Cria nova aba

Leitura

Método Parâmetros Descrição
get_value() cell Retorna valor da célula
get_row() row Retorna linha como tupla
get_column() col Retorna coluna como tupla
get_used_range() Retorna intervalo usado
get_max_row() Última linha com dados
get_max_column() Última coluna com dados

Escrita

Método Parâmetros Descrição
set_value() cell, value Define valor
write_row() row, data Escreve linha
write_column() col, data Escreve coluna

Conversão

Método Parâmetros Descrição
to_list() sheet_name Converte para lista de dicts
to_dict() sheet_name, key_column Converte para dict
from_list() data, start_cell Escreve lista de dicts
from_dict() data, start_cell Escreve dict

Parâmetros Comuns

Parâmetro Padrão Descrição
filepath Caminho do arquivo
cell Referência (ex: "A1")
sheet_name Ativa Nome da aba
row Número da linha
col Número da coluna

ExcelCom

Automação do Excel via COM (Windows). Controle completo: 60+ métodos para ler, escrever, filtrar, ordenar, formatar, tabelas dinâmicas, proteção de planilhas e formatação condicional.

Instalar

pip install rpaflow[excel-com]

Requisitos

  • Windows
  • Microsoft Excel instalado

Como Usar (Passo a Passo)

from rpaflow.excel_com import ExcelCom

xl = ExcelCom(visible=True)
xl.open("C:/dados/vendas.xlsx")

# Ler valor
valor = xl.get_value("A1")

# Escrever
xl.set_value("B1", "Nome")

# Filtrar por valores específicos
xl.filter_column_values("Vendas", 1, ["PCD"])

# Ordenar
xl.sort_column("Vendas", 3, order="desc")

xl.save()
xl.close()
xl.quit()

Métodos

Abertura/Busca

Método Parâmetros Descrição
open() filepath Abre arquivo Excel
open_recent(days) days Abre arquivo usado recentemente
find_files(directory) directory Lista arquivos Excel
find_files_recursive(directory) directory Lista recursivamente
check_open() Verifica se Excel está aberto

Atalhos

Método Parâmetros Descrição
new_workbook() Cria nova pasta de trabalho
save() Salva arquivo
save_as() filepath Salva como novo arquivo
close() Fecha pasta de trabalho
quit() Fecha Excel
close_without_saving() Fecha sem salvar

Planilha

Método Parâmetros Descrição
get_workbook_name() Nome do arquivo
select_sheet() sheet_name Seleciona aba
rename_sheet() old_name, new_name Renomeia aba
copy_sheet() sheet_name Copia aba
delete_sheet() sheet_name Deleta aba
get_sheets_names() Lista todas as abas
get_sheets_count() Conta abas
add_worksheet() name Adiciona nova aba

Célula

Método Parâmetros Descrição
get_value() cell Retorna valor
set_value() cell, value Define valor
get_values() range Retorna intervalo
set_values() range, values Define intervalo
clear_cell() cell Limpa célula
clear_range() range Limpa intervalo
select_cell() cell Seleciona célula
activate_cell() cell Ativa e foca célula
go_to_cell() cell Navega até célula
scroll_to_cell() cell Rola até célula
get_current_cell() Célula atual

Coluna

Método Parâmetros Descrição
select_column() col_num Seleciona coluna
get_column_width() col_num Largura da coluna
set_column_width() col_num, width Define largura
auto_fit_column() col_num Auto-ajusta largura
get_column_letter() col_num Converte número para letra
get_column_number() col_letter Converte letra para número

Linha

Método Parâmetros Descrição
select_row() row_num Seleciona linha
get_row_height() row_num Altura da linha
set_row_height() row_num, height Define altura
auto_fit_row() row_num Auto-ajusta altura
insert_row() row_num, shift Insere linha
delete_row() row_num, shift Deleta linha

Intervalo

Método Parâmetros Descrição
get_used_range() Intervalo usado
get_used_rows() Número de linhas usadas
get_used_columns() Número de colunas usadas
copy_range() range_from, range_to Copia intervalo
cut_range() range_from, range_to Move intervalo
paste_range() range Cola intervalo

Tabela Dinâmica

Método Parâmetros Descrição
create_pivot_table() source, destination, name Cria tabela dinâmica
add_pivot_field() pivot_name, field, orientation, position Adiciona campo
refresh_pivot() pivot_name Atualiza tabela dinâmica

Filtro e Ordenação

Método Parâmetros Descrição
filter_column() sheet, col_num, criteria Filtra por critérios
filter_column_values() sheet, col_num, values Filtra por valores específicos
remove_column_filter() sheet, col_num Remove filtro de coluna
clear_filters() sheet Remove todos os filtros
sort_column() sheet, col_num, order Ordena por coluna
get_filtered_rows() sheet Retorna linhas visíveis
find_row() sheet, col, value, data Encontra primeira linha com valor

Tabela Excel

Método Parâmetros Descrição
list_tables() sheet Lista tabelas
create_table() sheet, range, name Cria tabela
get_table_header_row() sheet, table_name Retorna cabeçalho
read_table() sheet, table_name, columns Lê dados como lista
read_filtered_table() sheet, table_name Lê apenas visíveis
set_table_cell() sheet, table_name, row, col, value Define valor
get_table_cell() sheet, table_name, row, col Retorna valor

Copiar/Colar

Método Parâmetros Descrição
copy() Copia seleção
cut() Move seleção
paste() Cola seleção
paste_special() operation Cola especial
delete() Deleta seleção

Desfazer/Refazer

Método Parâmetros Descrição
undo() Desfaz (Ctrl+Z)
redo() Refaz (Ctrl+Y)

Proteção

Método Parâmetros Descrição
protect_worksheet() password Protege planilha
unprotect_worksheet() password Desprotege planilha
protect_workbook() password Protege pasta
unprotect_workbook() password Desprotege pasta

Formatação Condicional

Método Parâmetros Descrição
add_conditional_format() range, rule, format Adiciona regra
clear_conditional_formats() range Limpa regras

Outros

Método Parâmetros Descrição
set_visible() visible Define visibilidade
get_visible() Retorna visibilidade
alert() message Exibe alerta

Parâmetros Comuns

Parâmetro Tipo Descrição
cell str Referência (ex: "A1")
range str Intervalo (ex: "A1:C10")
sheet_name str Nome da aba
col_num int Número da coluna
row_num int Número da linha
value any Valor a ser escrito
order str "asc" ou "desc"
data list Lista de dados para cache

Exemplos Práticos

Processar planilha grande (Toyota)

from rpaflow.excel_com import ExcelCom

xl = ExcelCom(visible=False)
xl.open(r"C:\Users\sergio.ssilva\Desktop\Dívida Montadora Toyota 30.06.xlsx")

# Ler tabela como lista de dicts
registros = xl.read_filtered_table("pgtos", "Tabela1")
print(f"Registros: {len(registors)}")

# Filtrar por valores
xl.filter_column_values("Vendas", 1, ["PCD"])
visiveis = xl.get_filtered_rows("Vendas")
print(f"Filtrados: {len(visiveis)}")

# Ordenar
xl.sort_column("Vendas", 3, order="desc")

xl.save()
xl.close()
xl.quit()

Usar cache para performance

from rpaflow.excel_com import ExcelCom

xl = ExcelCom(visible=True)
xl.open("C:/dados/producao.xlsx")

# Ler dados uma vez
data = xl.read_table("Pedidos", "Tabela1")

# Reutilizar em múltiplas buscas (30x mais rápido)
for nf in notas_fiscais:
    row = xl.find_row("Pedidos", 5, nf, data=data)
    if row:
        print(f"NF {nf} encontrada na linha {row}")

xl.quit()

Ler tabela com filtros

from rpaflow.excel_com import ExcelCom

xl = ExcelCom(visible=False)
xl.open("C:/dados/vendas.xlsx")

# Aplicar filtro
xl.filter_column_values("Vendas", 1, ["PCD", "RE"])

# Ler apenas linhas visíveis
dados = xl.read_filtered_table("Vendas", "Vendas")
print(f"Registros filtrados: {len(dados)}")

for registro in dados:
    print(f"NF: {registro['NF']}, Valor: {registro['Valor']}")

xl.quit()

Browser

Automação de navegador com suporte a múltiplos frames. Inspirado no Browser do Robot Framework.

Instalar

pip install rpaflow[browser]

Como Usar (Passo a Passo)

from rpaflow.browser import Browser

browser = Browser()
browser.start("https://site.com", type="playwright")

# Normal
browser.click("#botao")
browser.type_text("#campo", "texto")

# Recursive (busca em todos os iframes)
browser.click("#botao", recursive=True)

browser.close()

Métodos

Gerenciamento

Método Parâmetros Descrição
start() url, type, headless Inicia navegador
close() Fecha navegador
go_to() url Navega para URL
get_url() URL atual
get_title() Título da página

Interação

Método Parâmetros Descrição
click() selector, recursive Clica no elemento
double_click() selector Clique duplo
right_click() selector Clique direito
type_text() selector, text Digita texto
clear_text() selector Limpa campo
select_option() selector, value Seleciona opção
check() selector Marca checkbox
uncheck() selector Desmarca checkbox
hover() selector Passa mouse por cima
press_key() key Pressiona tecla
upload_file() selector, filepath Upload de arquivo

Busca

Método Parâmetros Descrição
find_element() selector Encontra elemento
find_all() selector Encontra todos
is_visible() selector, timeout Verifica visibilidade
is_enabled() selector Verifica se habilitado
get_text() selector Retorna texto
get_attribute() selector, attr Retorna atributo

Busca Recursiva

Método Parâmetros Descrição
find_element_in_frames() selector Encontra em frames
click_in_frames() selector Clica em frames
select_in_frames() selector, value Seleciona em frames
find_all_in_frames() selector Encontra todos em frames

Execução

Método Parâmetros Descrição
execute_js() script Executa JavaScript
wait() seconds Espera fixa
wait_for_element() selector, timeout Espera elemento

Navegação

Método Parâmetros Descrição
back() Volta
forward() Avança
reload() Recarrega
maximize() Maximiza

Imagem

Método Parâmetros Descrição
screenshot() filepath Tela do elemento
full_screenshot() filepath Tela cheia

Propriedade

Propriedade Descrição
page Retorna objeto Page do Playwright

Parâmetros Comuns

Parâmetro Padrão Descrição
selector CSS selector
recursive False Busca recursiva
timeout 30000 Timeout em ms
type playwright Motor do navegador
headless False Modo headless

Exemplos Práticos

Formulário com iframe

from rpaflow.browser import Browser

browser = Browser()
browser.start("https://site.com/form", type="playwright")

# Campo normal
browser.type_text("#nome", "João")

# Campo dentro de iframe (recursive)
browser.type_text("#campo_frame", "valor", recursive=True)

# Clicar botão dentro de iframe
browser.click("#enviar", recursive=True)

browser.close()

Login

from rpaflow.browser import Browser

browser = Browser()
browser.start("https://site.com/login", type="playwright")

browser.type_text("#email", "user@email.com")
browser.type_text("#senha", "123456")
browser.click("#entrar")

# Esperar carregar
browser.wait(3)

# Verificar login
if browser.is_visible("#dashboard"):
    print("Login realizado")

browser.close()

Acessar Page do Playwright

from rpaflow.browser import Browser

browser = Browser()
browser.start("https://site.com", type="playwright")

# Acessar page diretamente
page = browser.page
page.evaluate("alert('Olá!')")
page.screenshot(path="tela.png")

browser.close()

Desktop

Automação de aplicativos desktop via reconhecimento de imagem. Inspirado no Library do Robot Framework.

Instalar

pip install rpaflow[desktop]

Como Usar

from rpaflow.desktop import Desktop

desktop = Desktop()

# Localizar e clicar
desktop.click_image("C:/imgs/botao.png")

# Com confiança
desktop.click_image("C:/imgs/botao.png", confidence=0.90)

# Encontrar todas as ocorrências
results = desktop.find_all_images("C:/imgs/icone.png")

Métodos

Localização

Método Parâmetros Descrição
find_image() image, confidence Localiza imagem
find_all_images() image, confidence Localiza todas
wait_for_image() image, timeout Espera imagem

Interação

Método Parâmetros Descrição
click_image() image, confidence Clica na imagem
double_click_image() image Clique duplo
right_click_image() image Clique direito
drag_and_drop() source, target Arrasta

Texto

Método Parâmetros Descrição
type_text() text Digita texto
press_key() key Pressiona tecla
get_clipboard() Retorna clipboard
set_clipboard() text Define clipboard

Janela

Método Parâmetros Descrição
get_active_window() Janela ativa
set_active_window() title Ativa janela
maximize_window() title Maximiza
minimize_window() title Minimiza

Espera

Método Parâmetros Descrição
wait() seconds Espera fixa
wait_until_image() image, timeout Espera até achar imagem

Parâmetros Comuns

Parâmetro Padrão Descrição
image Caminho da imagem
confidence 0.8 Confiança (0-1)
timeout 10 Timeout em segundos

Log

Logging profissional para Python, inspirado no Serilog do C#.

Instalar

pip install rpaflow[log]

Como Usar

from rpaflow.log import Log

log = Log(path="C:/logs/meu_bot.log", level="DEBUG", json=True)

log = log.bind(bot="vendas", user="admin")
log.info("Iniciando robô")
log.error("Falha na conexão", host="localhost")
log.success("Robô finalizado")

Métodos

Criação

Método Parâmetros Descrição
Log() path, level, json Cria instância

Métodos de Log

Método Parâmetros Descrição
debug() message, **kwargs Log de debug
info() message, **kwargs Log informativo
warning() message, **kwargs Aviso
error() message, **kwargs Erro
critical() message, **kwargs Crítico
success() message, **kwargs Sucesso

Bind

Método Parâmetros Descrição
bind() **kwargs Adiciona contexto

Parâmetros Comuns

Parâmetro Padrão Descrição
path Caminho do arquivo de log
level INFO Nível mínimo
json False Formato JSON

Exemplo Prático

Log com contexto

from rpaflow.log import Log

log = Log(path="C:/logs/vendas.log", level="DEBUG")
log = log.bind(bot="vendas", user="admin")

log.info("Iniciando processamento")
log.debug("Arquivo carregado", arquivo="vendas.xlsx")
log.info("Processados 150 pedidos")
log.success("Finalizado em 5min")

Log de erros

from rpaflow.log import Log

log = Log(path="C:/logs/erros.log", level="ERROR")

try:
    # código
    pass
except Exception as e:
    log.error("Falha no processamento", erro=str(e))

INI

Leitura e escrita de arquivos .ini. Inspirado no System.IO.File do C#.

Instalar

pip install rpaflow[ini]

Como Usar

from rpaflow.ini import Ini

ini = Ini("C:/config/robô.ini")

host = ini.get("database", "host")
port = ini.get_int("database", "port")
debug = ini.get_bool("app", "debug")

ini.set("database", "host", "192.168.1.100")
ini.save()

Métodos

Criação

Método Parâmetros Descrição
Ini() filepath Cria instância

Leitura

Método Parâmetros Descrição
get() section, key Lê valor como string
get_int() section, key Lê como inteiro
get_float() section, key Lê como float
get_bool() section, key Lê como booleano
get_list() section, key Lê como lista

Escrita

Método Parâmetros Descrição
set() section, key, value Define valor
remove() section, key Remove chave
remove_section() section Remove seção
save() Salva arquivo

Seções

Método Parâmetros Descrição
sections() Lista seções
has_section() section Verifica se existe
has_option() section, key Verifica se chave existe

Parâmetros Comuns

Parâmetro Padrão Descrição
filepath Caminho do arquivo .ini
section Nome da seção
key Nome da chave
value Valor a ser gravado

Formato do Arquivo .ini

[database]
host=localhost
port=5432
user=admin
password=123456
database=vendas

[app]
debug=true
log_level=INFO
timeout=30

[smtp]
host=smtp.gmail.com
port=587
user=meu@email.com
password=minha_senha

Exemplos Práticos

Ler configurações

from rpaflow.ini import Ini

ini = Ini("C:/config/robô.ini")

# Ler valores
host = ini.get("database", "host")
port = ini.get_int("database", "port")
debug = ini.get_bool("app", "debug")

print(f"Host: {host}, Port: {port}, Debug: {debug}")

Atualizar configurações

from rpaflow.ini import Ini

ini = Ini("C:/config/robô.ini")

# Atualizar
ini.set("database", "host", "192.168.1.100")
ini.set("app", "debug", "false")
ini.save()

print("Configuração atualizada!")

Criar nova configuração

from rpaflow.ini import Ini

ini = Ini("C:/config/novo.ini")

# Definir valores
ini.set("database", "host", "localhost")
ini.set("database", "port", "5432")
ini.set("database", "user", "admin")
ini.set("database", "password", "123456")
ini.set("database", "database", "vendas")

ini.set("app", "debug", "true")
ini.set("app", "log_level", "INFO")

ini.save()

print("Configuração criada!")

Iterar seções

from rpaflow.ini import Ini

ini = Ini("C:/config/robô.ini")

for section in ini.sections():
    print(f"\n[{section}]")
    for key, value in ini.items(section):
        print(f"  {key} = {value}")

Verificar se chave existe

from rpaflow.ini import Ini

ini = Ini("C:/config/robô.ini")

if ini.has_option("database", "host"):
    print("Host configurado")

if ini.has_section("smtp"):
    print("SMTP configurado")

Criar configuração a partir de dict

from rpaflow.ini import Ini

config = {
    "database": {
        "host": "localhost",
        "port": "5432",
        "user": "admin",
        "password": "123456",
        "database": "vendas"
    },
    "app": {
        "debug": "true",
        "log_level": "INFO",
        "timeout": "30"
    }
}

ini = Ini("C:/config/novo.ini")

for section, options in config.items():
    for key, value in options.items():
        ini.set(section, key, value)

ini.save()

print("Configuração criada a partir de dict!")

Backup de configuração

from rpaflow.ini import Ini
from datetime import datetime

ini = Ini("C:/config/robô.ini")

# Criar backup com data
data = datetime.now().strftime("%Y%m%d_%H%M%S")
backup = f"C:/config/robô_backup_{data}.ini"

import shutil
shutil.copy(ini.filepath, backup)

print(f"Backup criado: {backup}")

Configuração com valores padrão

from rpaflow.ini import Ini

ini = Ini("C:/config/robô.ini")

# Valores padrão caso não existam
defaults = {
    "database": {"host": "localhost", "port": "5432"},
    "app": {"debug": "false", "timeout": "30"}
}

for section, options in defaults.items():
    for key, value in options.items():
        if not ini.has_option(section, key):
            ini.set(section, key, value)

ini.save()
print("Valores padrão aplicados!")

Ler múltiplos arquivos INI

from rpaflow.ini import Ini
import glob

# Ler todos os .ini de uma pasta
arquivos = glob.glob("C:/config/*.ini")

for arquivo in arquivos:
    ini = Ini(arquivo)
    print(f"\nArquivo: {arquivo}")
    for section in ini.sections():
        print(f"  [{section}]")
        for key, value in ini.items(section):
            print(f"    {key} = {value}")

Configuração para múltiplos robôs

from rpaflow.ini import Ini

# Criar config para cada robô
robos = {
    "vendas": {"host": "localhost", "database": "vendas"},
    "estoque": {"host": "192.168.1.100", "database": "estoque"},
    "financeiro": {"host": "10.0.0.1", "database": "financeiro"}
}

for nome, config in robos.items():
    ini = Ini(f"C:/config/{nome}.ini")
    
    for key, value in config.items():
        ini.set("database", key, value)
    
    ini.set("app", "debug", "true")
    ini.save()
    
    print(f"Configuração criada: {nome}")

Merge de configurações

from rpaflow.ini import Ini

# Carregar config base
base = Ini("C:/config/base.ini")

# Carregar config específica
especifica = Ini("C:/config/vendas.ini")

# Merge (especifica sobrescreve base)
for section in especifica.sections():
    for key, value in especifica.items(section):
        base.set(section, key, value)

base.save()

print("Configurações mergeadas!")

Verificar integridade

from rpaflow.ini import Ini

ini = Ini("C:/config/robô.ini")

# Verificar se todas as seções obrigatórias existem
obrigatorias = ["database", "app", "smtp"]
for section in obrigatorias:
    if not ini.has_section(section):
        print(f"Seção obrigatória ausente: {section}")

# Verificar se chaves obrigatórias existem
chaves_obrigatorias = {
    "database": ["host", "port", "user"],
    "app": ["debug", "log_level"]
}

for section, keys in chaves_obrigatorias.items():
    for key in keys:
        if not ini.has_option(section, key):
            print(f"Chave obrigatória ausente: [{section}] {key}")

print("Verificação concluída!")

JSON

Leitura e escrita de JSON com dot notation. Inspirado no Newtonsoft.Json do C#.

Instalar

pip install rpaflow[json]

Como Usar

from rpaflow.json import Json

json = Json()
data = json.load("config.json")

# Dot notation
data.user.name                                   # "Maria"
data.user.address.city                           # "SP"

# Iterar arrays
for bot in data.bots:
    print(f"Bot: {bot.name}, Enabled: {bot.enabled}")

Métodos

Criação

Método Parâmetros Descrição
Json() Cria instância

Leitura

Método Parâmetros Descrição
load() filepath Lê arquivo JSON
loads() text Lê string JSON

Escrita

Método Parâmetros Descrição
save() filepath, data, indent Salva arquivo JSON
dumps() data, indent Serializa para string

Utilitários

Método Parâmetros Descrição
get_keys() data Lista de chaves
get_values() data Lista de valores
get_items() data Lista de tuplas (chave, valor)
has_key() data, key Verifica se chave existe
get_value() data, key, default Valor com fallback
to_dict() data Converte para dict
to_list() data Converte para list

Dot Notation

Acessar valores

data = json.load("config.json")

# Com dot notation (igual Newtonsoft.Json)
data.user.name                       # "Maria"
data.user.address.city               # "SP"
data.database.host                   # "localhost"
data.bots[0].name                    # "invoice-bot"
data.bots[0].enabled                 # True

# Com colchetes (também funciona)
data["user"]["name"]                 # "Maria"
data["bots"][0]["name"]              # "invoice-bot"

Setar valores

data.user.name = "João"
data.user.phone = "+55 11 99999-0000"
data.user.address.country = "Brazil"  # Cria automaticamente

Iterar arrays

# Array de objetos
for bot in data.bots:
    print(f"Bot: {bot.name}")
    print(f"  Enabled: {bot.enabled}")
    print(f"  Retry: {bot.retry}")

# Dicionário
for empresa, config in data.empresas.items():
    print(f"{empresa}: {config.host}:{config.port}")

Exemplos Práticos

Iterar empresas e conectar ao banco

from rpaflow.json import Json
from rpaflow.sql import SQL
from rpaflow.log import Log

json = Json()
log = Log(path="C:/logs/robô.log", level="DEBUG")

data = json.load("C:/config/empresas.json")

for empresa, config in data.empresas.items():
    log.info(f"Processando empresa: {empresa}")
    
    sql = SQL(
        type="mysql",
        host=config.host,
        user=config.user,
        password=config.password,
        database=config.database
    )
    sql.connect()
    
    pedidos = sql.select("pedidos", where={"status": "pendente"})
    log.info(f"Encontrados {len(pedidos)} pedidos")
    
    sql.disconnect()

Iterar bots e executar

from rpaflow.json import Json
from rpaflow.log import Log

json = Json()
log = Log(path="C:/logs/bots.log", level="DEBUG")

data = json.load("C:/config/bots.json")

for bot in data.bots:
    if bot.enabled:
        log.info(f"Executando bot: {bot.name}")
        # ... executar bot
        log.success(f"Bot {bot.name} finalizado")
    else:
        log.warning(f"Bot {bot.name} desabilitado")

Carregar configuração de robô

from rpaflow.json import Json
from rpaflow.sql import SQL
from rpaflow.log import Log
from rpaflow.excel_com import ExcelCom

json = Json()
data = json.load("C:/config/robô.json")

# Configurações do banco
db = data.database
sql = SQL(
    type=db.type,
    host=db.host,
    user=db.user,
    password=db.password,
    database=db.database
)

# Configurações de log
log_config = data.logging
log = Log(
    path=log_config.file,
    level=log_config.level,
    json=log_config.json
)

# Configurações do Excel
excel_config = data.excel
xl = ExcelCom(visible=excel_config.visible)
xl.open(excel_config.path)

log.info("Robô iniciado")
sql.connect()
# ... processar
log.success("Robô finalizado")

Salvar dados processados

from rpaflow.json import Json

json = Json()

# Criar dados
resultado = {
    "processados": 150,
    "erros": 3,
    "tempo_execucao": "00:05:30",
    "detalhes": [
        {"pedido": 1001, "status": "ok"},
        {"pedido": 1002, "status": "ok"},
        {"pedido": 1003, "status": "erro", "motivo": "timeout"}
    ]
}

# Salvar
json.save("C:/logs/resultado.json", resultado, indent=4)
print("Resultado salvo")

Atualizar configuração existente

from rpaflow.json import Json

json = Json()
data = json.load("C:/config/app.json")

# Atualizar valores
data.database.host = "192.168.1.100"
data.database.port = 5432

# Adicionar novo campo
data.database.timeout = 30

# Salvar
json.save("C:/config/app.json", data)
print("Configuração atualizada")

Comparação com C# Newtonsoft.Json

C# Newtonsoft.Json Python rpaflow.json
JsonConvert.DeserializeObject<T>(json) json.loads(text)
JsonConvert.SerializeObject(obj) json.dumps(data)
File.ReadAllText("config.json") json.load("config.json")
File.WriteAllText("config.json", json) json.save("config.json", data)
obj.User.Name data.user.name
obj.Bots[0].Name data.bots[0].name
foreach (var bot in obj.Bots) for bot in data.bots:
foreach (var kvp in obj.Empresas) for empresa, config in data.empresas.items():

Files

Operações com arquivos e diretórios. Inspirado nos métodos estáticos de System.IO.File e System.IO.Path do C#.

Instalar

pip install rpaflow[files]

Como Usar

from rpaflow.files import Files

files = Files()

# Leitura/Escrita
conteudo = files.read_text("arquivo.txt")
files.write_text("arquivo.txt", "conteúdo")
files.append_text("arquivo.txt", "novo conteúdo")

# Linhas
linhas = files.read_lines("arquivo.txt")
files.write_lines("arquivo.txt", ["linha1", "linha2"])

# Operações
files.exists("arquivo.txt")
files.copy("origem.txt", "destino.txt")
files.move("antigo.txt", "novo.txt")
files.delete("lixo.txt")

# Path
files.get_filename("C:/pasta/arquivo.txt")
files.get_extension("arquivo.txt")
files.combine("C:/pasta", "arquivo.txt")

Métodos

Leitura/Escrita Texto

Método Parâmetros Descrição
read_text() filepath, encoding Lê conteúdo como texto
write_text() filepath, content, encoding Escreve texto no arquivo
append_text() filepath, content, encoding Adiciona texto ao final
read_lines() filepath, encoding Lê linhas como lista
write_lines() filepath, lines, encoding Escreve lista de linhas
append_lines() filepath, lines, encoding Adiciona linhas ao final

Leitura/Escrita Bytes

Método Parâmetros Descrição
read_bytes() filepath Lê conteúdo como bytes
write_bytes() filepath, data Escreve bytes no arquivo
append_bytes() filepath, data Adiciona bytes ao final

Operações de Arquivo

Método Parâmetros Descrição
exists() filepath Verifica se arquivo existe
copy() src, dst, overwrite Copia arquivo
move() src, dst, overwrite Move arquivo
delete() filepath Deleta arquivo
replace() source, destination, backup Substitui arquivo

Informações do Arquivo

Método Parâmetros Descrição
get_creation_time() filepath Data de criação
get_last_write_time() filepath Data da última escrita
get_last_access_time() filepath Data do último acesso
get_attributes() filepath Atributos do arquivo
set_attributes() filepath, readonly, hidden Define atributos
create_symlink() link, target Cria link simbólico

Path

Método Parâmetros Descrição
get_filename() filepath Nome do arquivo com extensão
get_filename_without_ext() filepath Nome sem extensão
get_extension() filepath Extensão do arquivo
get_directory() filepath Diretório pai
combine() *paths Junta caminhos
join() *paths Junta caminhos (os.path.join)
get_full_path() filepath Caminho absoluto
get_relative_path() base, filepath Caminho relativo
is_rooted() filepath Verifica se é absoluto
is_fully_qualified() filepath Verifica se é completo
has_extension() filepath Verifica se tem extensão
change_extension() filepath, new_ext Troca extensão
trim_separator() filepath Remove separador final
ends_with_separator() filepath Verifica se termina com separador
get_temp_path() Diretório temporário
get_temp_file() Arquivo temporário
get_random_filename() Nome aleatório
get_invalid_filename_chars() Caracteres inválidos no nome
get_invalid_path_chars() Caracteres inválidos no caminho
get_path_root() filepath Raiz do caminho
get_base_directory() Diretório base (AppDomain.BaseDirectory)

Directory

Método Parâmetros Descrição
dir_create() path Cria diretório
dir_delete() path, recursive Deleta diretório
dir_exists() path Verifica se diretório existe
dir_move() src, dst Move diretório
dir_get_files() path, pattern Lista arquivos
dir_get_dirs() path Lista subdiretórios
dir_get_all_files() path, pattern Lista todos recursivamente

File Info

Método Parâmetros Descrição
get_size() filepath Tamanho em bytes
is_readonly() filepath Verifica se é somente leitura
set_readonly() filepath, readonly Define somente leitura

Exemplos Práticos

Ler e processar arquivo CSV

from rpaflow.files import Files

files = Files()

# Ler CSV como linhas
linhas = files.read_lines("dados.csv", encoding="latin-1")

for linha in linhas:
    campos = linha.split(";")
    print(f"Nome: {campos[0]}, Idade: {campos[1]}")

Backup de arquivo

from rpaflow.files import Files
from datetime import datetime

files = Files()

# Criar backup com data
data = datetime.now().strftime("%Y%m%d_%H%M%S")
backup = f"backup_{data}.txt"
files.copy("arquivo.txt", backup)

print(f"Backup criado: {backup}")

Processar todos os arquivos de uma pasta

from rpaflow.files import Files

files = Files()

# Listar todos os .txt recursivamente
arquivos = files.dir_get_all_files("C:/dados", "*.txt")

for arquivo in arquivos:
    conteudo = files.read_text(arquivo)
    print(f"{arquivo}: {len(conteudo)} caracteres")

Verificar e criar diretório

from rpaflow.files import Files

files = Files()

if not files.dir_exists("C:/logs"):
    files.dir_create("C:/logs")
    print("Diretório criado")

# Escrever log
files.append_text("C:/logs/app.log", "Nova entrada de log\n")

Trabalhar com caminhos

from rpaflow.files import Files

files = Files()

caminho = "C:/pasta/subpasta/arquivo.txt"

print(files.get_filename(caminho))              # "arquivo.txt"
print(files.get_filename_without_ext(caminho))  # "arquivo"
print(files.get_extension(caminho))             # ".txt"
print(files.get_directory(caminho))             # "C:/pasta/subpasta"
print(files.get_path_root(caminho))             # "C:\"
print(files.is_rooted(caminho))                 # True

# Combinar caminhos
novo = files.combine("C:/pasta", "subpasta", "arquivo.txt")
print(novo)  # "C:/pasta/subpasta/arquivo.txt"

# Trocar extensão
log = files.change_extension(caminho, ".log")
print(log)  # "C:/pasta/subpasta/arquivo.log"

API

Requisições HTTP/REST.

Instalar

pip install rpaflow[api]

Como Usar

from rpaflow.api import API

api = API()

# GET
response = api.get("https://api.example.com/users")

# POST
api.post("https://api.example.com/users", json={"nome": "João"})

# PUT
api.put("https://api.example.com/users/1", json={"nome": "João Silva"})

# DELETE
api.delete("https://api.example.com/users/1")

Métodos

Método Parâmetros Descrição
get() url, headers (opcional) Requisição GET
post() url, json, headers (opcional) Requisição POST
put() url, json, headers (opcional) Requisição PUT
delete() url, headers (opcional) Requisição DELETE

Email

Envio de emails via SMTP.

Instalar

pip install rpaflow[email]

Como Usar

from rpaflow.email import Email

email = Email(smtp_host="smtp.gmail.com", smtp_port=587, user="meu@email.com", password="senha")

# Conectar
email.connect()

# Enviar
email.send(
    to="dest@email.com",
    subject="Assunto",
    body="Mensagem do email",
    attachments=["arquivo.pdf"]
)

# Desconectar
email.disconnect()

Métodos

Método Parâmetros Descrição
connect() Conecta ao SMTP
send() to, subject, body, attachments Envia email
disconnect() Desconecta

Licença

MIT License - veja LICENSE para detalhes.

Project details


Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distribution

rpaflow-0.1.1.tar.gz (389.7 kB view details)

Uploaded Source

Built Distribution

If you're not sure about the file name format, learn more about wheel file names.

rpaflow-0.1.1-py3-none-any.whl (36.7 kB view details)

Uploaded Python 3

File details

Details for the file rpaflow-0.1.1.tar.gz.

File metadata

  • Download URL: rpaflow-0.1.1.tar.gz
  • Upload date:
  • Size: 389.7 kB
  • Tags: Source
  • Uploaded using Trusted Publishing? No
  • Uploaded via: twine/6.2.0 CPython/3.14.3

File hashes

Hashes for rpaflow-0.1.1.tar.gz
Algorithm Hash digest
SHA256 525108b4d27bb2062e3096010068480817a37a4638f0fa3e87b4a78355226bff
MD5 c6798835cdc7ec3be3a555ceae0e4601
BLAKE2b-256 c9d69a680f22afba1531f05b1c8994865e1486804d568811fb66953eff15edba

See more details on using hashes here.

File details

Details for the file rpaflow-0.1.1-py3-none-any.whl.

File metadata

  • Download URL: rpaflow-0.1.1-py3-none-any.whl
  • Upload date:
  • Size: 36.7 kB
  • Tags: Python 3
  • Uploaded using Trusted Publishing? No
  • Uploaded via: twine/6.2.0 CPython/3.14.3

File hashes

Hashes for rpaflow-0.1.1-py3-none-any.whl
Algorithm Hash digest
SHA256 a48ea1ba779dece183adf90682b48a1b3c34eebbfe78691318b2a8504c9c46dc
MD5 7baa4ed89d80cf875d5c0d4b79195ade
BLAKE2b-256 26a1054e5478efdd961c40ba80978260d9c55268c5267f6122b890bc40eae918

See more details on using hashes here.

Supported by

AWS Cloud computing and Security Sponsor Datadog Monitoring Depot Continuous Integration Fastly CDN Google Download Analytics Pingdom Monitoring Sentry Error logging StatusPage Status page