Existem várias opções para armazenar os dados. O Receita Data exporta em dois: .csv ou .xlsx. Então, caso não sejam muitos dados, use .xlsx. Caso sejam mais dados, .csv é melhor pois não tem limites de linhas como o xlsx. No entanto, caso sejam muitos dados, o .csv fica muito grande e o Power BI demora para ler os arquivos. Para atualização local, nenhum problema mas em nuvem isso não funciona pois ocorre erro de "Timeout".
As possíveis soluções:
Caso a base de dados seja muito grande e a atualização seja feita na nuvem, a demora em ler os arquivos resulta em erro de Timeout. Nesse caso, há duas soluções possíveis
pyarrow do Python e não consegui instalar por falta de permissão. Se a conversão for manual, no ContAgil, é possível fazer com um simples conversor, mais abaixo.PythonQualquer uma das opções é boa. Ambos os formatos são muito mais leves. Parquet funciona como uma tabela mais organizada e mais de 10 vezes menor e a compressão em Gzip é muito eficiente e deixa o arquivo bem pequeno. Mas não há uma ferramenta (pelo menos não encontrei) no ContAgil para fazer essas conversões. Então, para fazer qualquer uma das conversões, é preciso um script em Python.
Este são os códigos que fazem a conversão/compressão. Por limitação das bibliotecas Python, atualmente só da para usar o compressor mas achei melhor deixar o conversor aqui para futura referência ou caso queira fazer manualmente.
Para conversão de .csv para .parquet
import os
import pandas as pd
# 1. Defina os caminhos das pastas onde o ContÁgil salva os dados
pasta_dados = r"C:\Users\70924597968\OneDrive - Receita Federal do Brasil\Ceint\Dashboard\Dados"
def converter_pasta(caminho_pasta):
if not os.path.exists(caminho_pasta):
print(f"Pasta não encontrada: {caminho_pasta}")
return
arquivos_csv = [f for f in os.listdir(caminho_pasta) if f.lower().endswith('.csv')]
if not arquivos_csv:
print(f"Nenhum CSV encontrado em: {caminho_pasta}")
return
print(f"\nIniciando conversão na pasta: {caminho_pasta}")
for nome_arquivo in arquivos_csv:
caminho_csv = os.path.join(caminho_pasta, nome_arquivo)
caminho_parquet = os.path.join(caminho_pasta, nome_arquivo.replace('.csv', '.parquet'))
try:
print(f" -> Convertendo {nome_arquivo}...")
# Lê o CSV
df = pd.read_csv(caminho_csv, sep=';', encoding='cp1252', low_memory=False, dtype=str)
# Salva como Parquet compactado
df.to_parquet(caminho_parquet, index=False)
# Deleta o original
os.remove(caminho_csv)
print(f" Sucesso! {nome_arquivo} substituído por Parquet.")
except Exception as e:
print(f" Erro ao converter {nome_arquivo}: {e}")
if __name__ == '__main__':
print("=== OTIMIZADOR DE DADOS PARA POWER BI ===")
converter_pasta(pasta_dados)
print("\nProcesso finalizado!")
Este é o código que comprime os arquivos .csv
import os
import pandas as pd
import glob
# Defina a pasta onde estão os arquivos .csv
pasta_dados = r'C:\Users\70924597968\OneDrive - Receita Federal do Brasil\Ceint\Dashboard\dados'
def converter_para_gz():
# Lista todos os CSVs na pasta
arquivos = glob.glob(os.path.join(pasta_dados, "*.csv"))
for arquivo in arquivos:
print(f"Compactando: {os.path.basename(arquivo)}")
try:
# Lê o CSV
df = pd.read_csv(arquivo, sep=';', encoding='cp1252', low_memory=False)
# Salva como Gzip (o Pandas usa a biblioteca 'gzip' nativa do Python automaticamente)
# O nome do arquivo será o mesmo, mas terminando em .csv.gz
nome_gz = arquivo + ".gz"
df.to_csv(nome_gz, sep=';', index=False, encoding='utf-8', compression='gzip')
# Remove o CSV original após confirmar a criação do .gz
if os.path.exists(nome_gz):
os.remove(arquivo)
print(f"Sucesso: {os.path.basename(nome_gz)}")
except Exception as e:
print(f"Erro no arquivo {arquivo}: {e}")
if __name__ == '__main__':
converter_para_gz()
Para que a conversão seja feita automaticamente e diariamente, use o código abaixo. Além de ser bem pequeno, ele roda a cada hora apenas para verificar se já foi feita a conversão no dia. Se já foi feita, ele não faz nada. Então não tem nenhum perigo de sobrecarregar a rede. Salve-o em uma pasta da sua preferência. Lembre-se de que o endereço em script_conversor = ... tem que ser o local onde você salvou o conversor_csv_gzip.py.
import os
import sys
import time
import tempfile
import subprocess
from datetime import datetime
pasta_local = os.path.join(os.path.expanduser("~"), "AppData", "Local", "AutomacaoDashboardCeint")
os.makedirs(pasta_local, exist_ok=True)
lock_file = os.path.join(tempfile.gettempdir(), 'meu_conversor.lock')
arquivo_controle = os.path.join(pasta_local, "ultima_execucao.txt")
arquivo_log = os.path.join(pasta_local, "schedule_log.txt")
python_exe = r'C:\Users\70924597968\AppData\Local\Programs\Python\Python311\python.exe'
script_conversor = r'C:\Users\70924597968\OneDrive - Receita Federal do Brasil\Ceint\Dashboard\conversor_csv_gzip.py'
def log(msg):
if os.path.exists(arquivo_log) and os.path.getsize(arquivo_log) > 2_000_000: # ~2MB
os.remove(arquivo_log)
with open(arquivo_log, 'a', encoding='utf-8') as f:
f.write(f"{datetime.now()} - {msg}\n")
def processo_existe(pid):
resultado = subprocess.run(
['tasklist', '/FI', f'PID eq {pid}'],
capture_output=True, text=True,
creationflags=subprocess.CREATE_NO_WINDOW
)
return str(pid) in resultado.stdout
def lock_esta_travado():
if not os.path.exists(lock_file):
return False
# Camada 1: idade do lock (mais confiável que checagem de PID)
idade_segundos = time.time() - os.path.getmtime(lock_file)
if idade_segundos > 7200: # mais de 2 horas = certamente travado
log(f"Lock com {idade_segundos/3600:.1f}h de idade — considerado órfão.")
return False
# Camada 2: checagem de PID (para locks recentes)
try:
with open(lock_file) as f:
pid_antigo = int(f.read().strip())
if processo_existe(pid_antigo):
return True
except (OSError, ValueError):
pass
return False
if lock_esta_travado():
sys.exit(0)
with open(lock_file, 'w') as f:
f.write(str(os.getpid()))
log("Script iniciado.")
def rodar_conversor():
resultado = subprocess.run(
[python_exe, script_conversor],
capture_output=True, text=True,
creationflags=subprocess.CREATE_NO_WINDOW
)
log(f"Conversor executado. Código: {resultado.returncode}")
if resultado.stderr:
log(f"Erro: {resultado.stderr}")
try:
while True:
# Renova o lock a cada ciclo, provando que o processo está ativo
with open(lock_file, 'w') as f:
f.write(str(os.getpid()))
agora = datetime.now()
hora_atual = agora.hour
hoje = agora.strftime("%Y-%m-%d")
if 8 <= hora_atual < 17:
ultima_data = ""
if os.path.exists(arquivo_controle):
with open(arquivo_controle, "r") as f:
ultima_data = f.read()
if ultima_data != hoje:
log(f"Executando conversor pela primeira vez hoje ({hoje}).")
rodar_conversor()
with open(arquivo_controle, "w") as f:
f.write(hoje)
else:
log("Já rodou hoje, aguardando.")
else:
log(f"Fora do horário ({hora_atual}h). Aguardando.")
time.sleep(3600)
finally:
log("Script finalizado, removendo lock.")
if os.path.exists(lock_file):
os.remove(lock_file)
Como definimos esta pasta para salvar os logs: pasta_local = os.path.join(os.path.expanduser("~"), "AppData", "Local", "AutomacaoDashboardCeint") você tem que criá-la em AppData, Local. Lembre-se que AppData é pasta oculta então tem que mudar as configurações de exibição de pastas para mostrá-la.
Para rodar diariamente, é preciso criar um arquivo .bat e colocá-lo na inicialização do sistema. Use o código abaixo e salve com a extensão .bat e coloque o arquivo em shell:startup
Faça isso da seguinte forma:
Windows + R no seu teclado. Isso abrirá a janelinha "Executar".shell:startup e dê Enter.C:\Users\SEU_USUARIO\AppData\Roaming\Microsoft\Windows\Start Menu\Programs\Startup.schedule.py
@echo off
start /min "" "C:\Users\70924597968\AppData\Local\Programs\Python\Python311\pythonw.exe" "C:\Users\70924597968\OneDrive - Receita Federal do Brasil\Ceint\Dashboard\schedule.py"
O OneDrive for Business (corporativo) roda sobre a mesma estrutura do SharePoint. Portanto, usaremos o conector do SharePoint para ler a sua pasta.
https://rfbgov-my.sharepoint.com/personal/marcos_rinaldi_rfb_gov_br/= SharePoint.Files("https://rfbgov-my.sharepoint.com/personal/marcos_rinaldi_rfb_gov_br/", [ApiVersion = 15])
.Files para .Contents.
= SharePoint.Contents("https://rfbgov-my.sharepoint.com/personal/marcos_rinaldi_rfb_gov_br/", [ApiVersion = 15])
A tela vai mudar. Em vez de uma lista infinita de arquivos, você verá apenas as pastas raízes do seu OneDrive (como Documents, Documentos Compartilhados, etc.).
Lembre-se do truque de nomenclatura que a Microsoft faz entre o navegador e o Power BI.No seu navegador, a pasta principal aparece com o nome amigável de "Meus arquivos". Mas "por baixo dos panos" — que é o modo "cru" como o Power Query enxerga o servidor —, essa pasta raiz tem o nome de sistema Documents (Documentos).
// 1) A sequência correta para chegar na pasta:
FonteNuvem = SharePoint.Contents("https://rfbgov-my.sharepoint.com/personal/marcos_rinaldi_rfb_gov_br/", [ApiVersion = 15]),
Documents = FonteNuvem{[Name="Documents"]}[Content],
Ceint = Documents{[Name="Ceint"]}[Content],
Dashboard = Ceint{[Name="Dashboard"]}[Content],
PastaDados = Dashboard{[Name="dados"]}[Content],
SomenteGzip = Table.SelectRows(PastaDados, each Text.EndsWith([Name], ".gz")),
LerGzip = Table.AddColumn(SomenteGzip, "Dados", each
let
Descompactado = Binary.Decompress([Content], Compression.GZip),
// Alterado para 65001 (UTF-8) para evitar erro de acentuação. O padrão seria 1252
csv = Csv.Document(Descompactado, [Delimiter=";", Encoding=65001, QuoteStyle=QuoteStyle.Csv]),
cab = Table.PromoteHeaders(csv, [PromoteAllScalars=true]),
norm = Table.TransformColumnNames(cab, each Text.Trim(Text.Lower(_)))
in
norm
),
Combinado = Table.Combine(LerGzip[Dados]),
Essa pode ser uma alternativa caso ainda tenha problemas na atualização em nuvem. O Power BI é muito "fominha": quando você manda ele combinar uma pasta, ele tenta abrir e baixar dezenas de arquivos CSVs ao mesmo tempo para ser mais rápido. O servidor do SharePoint entende esse pico de requisições simultâneas como um excesso de carga e simplesmente "corta" a conexão de um dos arquivos. Se isso acontecer:
Arquivo > Opções e configurações > Opções.ARQUIVO ATUAL (Current File) e clique em Carregamento de Dados (Data Load).Tabelas Paralelas, desmarque a opção: Habilitar o carregamento paralelo de tabelas.Como você já conectou o Power BI diretamente ao SharePoint/OneDrive na nuvem, você não precisa mais abrir o Power BI Desktop para atualizar os dados!
Se você quer que a atualzaçao aconteça em horário fixo e não quando há alguma alteração nos arquivos a forma mais blindada contra erros é usar o agendamento nativo do Power BI
Modelo Semântico (Dataset).> Configurações.Conexões de gateway e nuvem.Método de Autenticação OAuth2, defina o Nível de Privacidade como Organizacional e clique no botão para Entrar (fazer o login com a sua conta da Receita).Pronto, o Power BI já tem autorização para ler o seu OneDrive sozinho.
Configurações > Atualizar: ative a opção Desativado... irá mudar para Ativado.Pronto! Agora o Power BI vai atualizar os dados sozinho, sem precisar abrir o Desktop.
Lembre-se de confirmar na página inicial do Meu workspace o horário da atualização por conta da possível diferença de fuso entre o SharePoint e o horário local