Antes de iniciar qualquer análise, é importante garantir que os dados estejam organizados e disponíveis em um formato que permita leitura e validação. Este projeto foi estruturado como um pipeline em Python para tratar dados de vendas, com foco em etapas como ingestão, verificação e exportação. A proposta é facilitar o uso dos dados em relatórios e visualizações.
2. Objetivo
Apresentar um fluxo de trabalho em Python para organizar e validar dados de vendas. O projeto é dividido em módulos que tratam cada etapa separadamente. Os resultados são gerados em formatos como CSV, JSON e HTML, e podem ser utilizados por diferentes perfis de usuários.
3. Metodologia
O projeto foi dividido em módulos independentes, cada um com uma função específica. A execução é feita por um arquivo principal que chama os demais componentes. A seguir estão os módulos utilizados:
main.py: executa o pipeline completo;
config.py: define os caminhos e parâmetros do projeto;
utils.py: contém funções auxiliares como leitura de arquivos e configuração de logs;
validators.py: aplica regras de estrutura e negócio aos dados;
diagnostics.py: gera informações técnicas sobre os dados e o processo.
A sequência de execução segue as etapas de leitura, validação, diagnóstico e exportação.
4. Preparação do Projeto
Nesta etapa inicial o objetivo é garantir que o ambiente de execução do pipeline esteja corretamente configurado e que os artefatos básicos (entradas, diretórios de saída e logging) sejam criados de forma reprodutível e rastreável. A preparação aborda:
Configuração do logging com rotação e saída no console;
Criação dos diretórios necessários (logs e output);
Leitura segura do arquivo de entrada (dados_vendas.csv) com validações de existência e encoding;
Aplicação das validações de estrutura e regras de negócio (módulo validators.py);
Geração de diagnóstico técnico (módulo diagnostics.py) e exportação dos metadados.
In [1]:
# ============================================================# 4. Preparação do Projeto# ============================================================# from__future__importannotations# Bibliotecas padrãoimportlogging# Sistema de loggingfrompathlibimportPath# Manipulação de caminhosfromtypingimportDict,Any# Tipagem de funçõesimportos# Operações de diretóriosimportuuid# Identificadores únicos (usados em modais HTML)importhtml# Escapar conteúdo de arquivos para exibição segura# Bibliotecas externasimportpandasaspd# Manipulação de DataFramesfromIPython.displayimportdisplay,HTML# Exibição em notebooks# Imports do projetofromconfigimport(# Configurações globaisINPUT_CSV,LOGS_DIR,OUTPUT_DIR,META_JSON,RELATORIO_TXT)fromutilsimport(# Funções auxiliaressetup_logging,# Configuração de loggingensure_dirs,# Criação de diretóriosread_csv_safely,# Leitura segura de CSVsave_json,# Salvamento em JSONsave_dataframe,# Salvamento de DataFrame em CSVmoldura_texto# Impressão de moldura de seção)fromvalidatorsimportrun_validations# Validações de estrutura e regras de negóciofromdiagnosticsimport(# Diagnóstico técnicodiagnostico_dados,gerar_relatorio_texto)defpreparar_ambiente(input_csv:str=INPUT_CSV,logs_dir:str=LOGS_DIR,output_dir:str=OUTPUT_DIR,)->Dict[str,Any]:""" Etapa 4 - Preparação do Projeto. - Configura logging (arquivo + console, com rotação) - Cria diretórios necessários (logs, output, metadados, resumo) - Lê o dataset CSV de vendas de forma segura - Aplica validações de estrutura e regras de negócio - Gera diagnóstico técnico e exporta metadados e relatório """# Configuração do logger (arquivo pipeline.log + console) com nível INFOlogger=setup_logging(logs_dir=logs_dir,filename="pipeline.log",level=logging.INFO)# Garantia de diretórios base e derivados usados na execuçãoensure_dirs(logs_dir,output_dir,Path(output_dir)/"metadados",Path(output_dir)/"resumo")# Registro do início da etapalogger.info(">>> Iniciando etapa 4: Preparação do Projeto")# 2) Leitura segura do dataset CSV (existência e encoding)df=read_csv_safely(input_csv,logger=logger)# 3) Validação estrutural, tipos e regras de negócioresultado_validacao=run_validations(df,logger=logger)df_validos=resultado_validacao.validdf_invalidos=resultado_validacao.invalidlogger.info(f"Validação de negócio: total={len(df)} | "f"válidos={len(df_validos)} | inválidos={len(df_invalidos)}")# 4) Diagnóstico técnico (apenas dados válidos)meta=diagnostico_dados(df_validos,logger=logger)# 5) Salvamento dos metadados (JSON) e relatório textual consolidadotry:# Salva metadados em JSON no diretório de saída/metadadossave_json(meta,str(Path(output_dir)/"metadados"/Path(META_JSON).name))# Gera relatório textual consolidado no diretório de logsgerar_relatorio_texto(meta=meta,caminho=str(Path(logs_dir)/Path(RELATORIO_TXT).name),logger=logger)# Registro de sucessologger.info("Metadados e relatório gerados com sucesso.")exceptExceptionasexc:# Aviso em caso de falha ao salvarlogger.warning(f"Falha ao salvar metadados/relatório: {exc}")# Registro de conclusãologger.info(">>> Etapa 4 concluída com sucesso.")# Retorno para etapas seguintesreturn{"df":df,"meta":meta,"logger":logger}# Execução local para validaçãoif__name__=="__main__":# Executa a preparação do ambiente e captura o resultadoresultado=preparar_ambiente()# Extrai DataFrame e metadados do retorno para inspeçãodf=resultado["df"]meta=resultado["meta"]# Cabeçalho da seção de preparação no consolemoldura="-"*25print(f"\n{moldura}>>> 4. Preparação do Projeto <<< {moldura}\n")# Exibe dimensões do dataset (linhas e colunas) usando os metadadosprint(f"Linhas: {meta['shape']['linhas']} | Colunas: {meta['shape']['colunas']}")# Exibe status da conversão da coluna 'mes' para datetime, se aplicávelprint(f"Conversão 'mes' para datetime: {meta.get('mes_convertido_para_datetime',False)}")
2024-11-15 15:46:41,920 | INFO | pipeline_dados | >>> Iniciando etapa 4: Preparação do Projeto
2024-11-15 15:46:41,927 | INFO | pipeline_dados | CSV carregado: dados_vendas.csv | encoding=utf-8 | sep=, | linhas=500 | colunas=7
2024-11-15 15:46:41,928 | INFO | pipeline_dados | Estrutura validada: todas as colunas obrigatórias presentes.
2024-11-15 15:46:41,934 | INFO | pipeline_dados | Coluna 'mes' padronizada para datetime64[ns].
2024-11-15 15:46:41,936 | INFO | pipeline_dados | Coluna 'vendedor_id' padronizada para category.
2024-11-15 15:46:41,938 | INFO | pipeline_dados | Coluna 'produto_id' padronizada para category.
2024-11-15 15:46:41,940 | INFO | pipeline_dados | Coluna 'quantidade' padronizada para int64.
2024-11-15 15:46:41,942 | INFO | pipeline_dados | Coluna 'preco_unitario' padronizada para float64.
2024-11-15 15:46:41,945 | INFO | pipeline_dados | Coluna 'desconto' padronizada para float64.
2024-11-15 15:46:41,946 | INFO | pipeline_dados | Coluna 'valor_total' padronizada para float64.
2024-11-15 15:46:41,948 | INFO | pipeline_dados | Tipos após padronização: {'mes': 'datetime64[ns]', 'vendedor_id': 'category', 'produto_id': 'category', 'quantidade': 'int64', 'preco_unitario': 'float64', 'desconto': 'float64', 'valor_total': 'float64'}
2024-11-15 15:46:41,953 | INFO | pipeline_dados | Validação de negócio: total=500 | válidos=500 | inválidos=0
2024-11-15 15:46:41,954 | INFO | pipeline_dados | Validação de negócio: total=500 | válidos=500 | inválidos=0
2024-11-15 15:46:41,955 | INFO | pipeline_dados | Executando diagnóstico dos dados...
2024-11-15 15:46:41,961 | INFO | pipeline_dados | Gerando relatório textual consolidado...
2024-11-15 15:46:41,963 | INFO | pipeline_dados | Relatório textual salvo em logs\relatorio_pipeline.txt
2024-11-15 15:46:41,964 | INFO | pipeline_dados | Metadados e relatório gerados com sucesso.
2024-11-15 15:46:41,965 | INFO | pipeline_dados | >>> Etapa 4 concluída com sucesso.
------------------------->>> 4. Preparação do Projeto <<< -------------------------
Linhas: 500 | Colunas: 7
Conversão 'mes' para datetime: True
5. Metadados do Dataset
Nesta etapa, serão apresentados os metadados técnicos do dataset de vendas validado, organizados em um
formato tabular para facilitar a visualização da estrutura dos dados, incluindo colunas, descrições
e tipos de dados.
In [2]:
# ============================================================# 5. Metadados do Dataset# ============================================================moldura_texto("5. Metadados do Dataset")# Pré-condição explícita:# meta já foi gerado na Etapa 4 (Preparação do Projeto)# Se esta célula falhar, é porque o Item 4 não foi executadodtypes=meta["dtypes"]meta_dir=os.path.join(OUTPUT_DIR,"metadados")os.makedirs(meta_dir,exist_ok=True)meta_csv_path=os.path.join(meta_dir,"metadados.csv")descricao_variaveis={"mes":"Data da venda (mês/ano)","vendedor_id":"Identificação do vendedor","produto_id":"Identificação do produto","quantidade":"Quantidade vendida","preco_unitario":"Preço unitário do produto","desconto":"Percentual de desconto aplicado","valor_total":"Valor total da venda (após desconto)",}df_meta=pd.DataFrame({"Coluna":list(dtypes.keys()),"Descrição":[descricao_variaveis.get(c,"")forcindtypes.keys()],"Tipo de Dados":list(dtypes.values()),})save_dataframe(df_meta,meta_csv_path,index=False)display(df_meta)
-------------------->>> 5. Metadados do Dataset <<<--------------------
Coluna
Descrição
Tipo de Dados
0
mes
Data da venda (mês/ano)
datetime64[ns]
1
vendedor_id
Identificação do vendedor
category
2
produto_id
Identificação do produto
category
3
quantidade
Quantidade vendida
int64
4
preco_unitario
Preço unitário do produto
float64
5
desconto
Percentual de desconto aplicado
float64
6
valor_total
Valor total da venda (após desconto)
float64
6. Qualidade e integridade dos dados
Nesta etapa apresentamos um resumo visual da qualidade e integridade dos dados, enquanto os detalhes técnicos da execução são registrados nos arquivos de log do pipeline, reforçando a confiabilidade do fluxo.
In [3]:
# Item 6 - Qualidade e integridade dos dados# Configuração do loggerlogger=logging.getLogger("diagnostics")logger.setLevel(logging.INFO)# Validar e padronizar datasetdf=run_validations(df,logger)# Exibir molduramoldura_texto("6. Qualidade e integridade dos dados")# Extrai apenas os dados válidos do resultado da validaçãodf_validos=df.valid# Executar análise de qualidade e integridademetadados=diagnostico_dados(df_validos,logger)# Exibir dimensõesprint("\nDimensões do dataset")display(pd.DataFrame([metadados["shape"]]))# Exibir valores nulosprint("\nValores nulos por variável")display(pd.DataFrame(list(metadados["nulos"].items()),columns=["Variável","Nulos"]))# Exibir tipos de dadosprint("\nTipos de dados por variável")display(pd.DataFrame(list(metadados["dtypes"].items()),columns=["Variável","Tipo de Dados"]))# Exibir status da conversão da coluna 'mes'print("\nConversão da coluna 'mes' para datetime")print(metadados["mes_convertido_para_datetime"])
-------------------->>> 6. Qualidade e integridade dos dados <<<--------------------
Dimensões do dataset
linhas
colunas
0
500
7
Valores nulos por variável
Variável
Nulos
0
mes
0
1
vendedor_id
0
2
produto_id
0
3
quantidade
0
4
preco_unitario
0
5
desconto
0
6
valor_total
0
Tipos de dados por variável
Variável
Tipo de Dados
0
mes
datetime64[ns]
1
vendedor_id
category
2
produto_id
category
3
quantidade
int64
4
preco_unitario
float64
5
desconto
float64
6
valor_total
float64
Conversão da coluna 'mes' para datetime
True
7. Qualidade do Código, Testes Automatizados e Auditoria do Pipeline
Nesta etapa apresentamos os principais artefatos técnicos do projeto, incluindo arquivos de código-fonte,
relatórios de diagnóstico, logs de execução e cobertura de testes automatizados. A visualização desses
elementos permite inspecionar a estrutura do pipeline, verificar a qualidade do código e auditar sua execução.
In [4]:
# ============================================================# 7. Qualidade do Código, Testes Automatizados e Auditoria Final# ============================================================moldura_texto("7. Qualidade do Código, Testes Automatizados e Auditoria do Pipeline")defread_text(path:str)->str:withopen(path,"r",encoding="utf-8")asf:returnhtml.escape(f.read())arquivos={# Código-fonte"main.py":"main.py","config.py":"config.py","validators.py":"validators.py","diagnostics.py":"diagnostics.py","utils.py":"utils.py",# Artefatos do pipeline"Relatório de Diagnóstico do Dataset":os.path.join(LOGS_DIR,"relatorio_pipeline.txt"),"Log de Execução do Pipeline":os.path.join(LOGS_DIR,"pipeline.log"),"Relatório de Cobertura de Testes":os.path.join(OUTPUT_DIR,"qualidade","relatorio_pipeline_testes.txt"),}blocks=[]fornome,caminhoinarquivos.items():ifnotos.path.exists(caminho):continuemodal_id=str(uuid.uuid4())conteudo=read_text(caminho)block=f""" <button onclick="document.getElementById('{modal_id}').style.display='block'" style="background-color:#006666;color:white;padding:8px 16px;border:none; border-radius:8px;cursor:pointer;margin:5px;"> 📄 {nome} </button> <div id="{modal_id}" style="display:none; position:fixed; top:8%; left:10%; width:65%; height:65%; background-color:white; border:2px solid #333; box-shadow:0 8px 24px rgba(0,0,0,0.18); padding:20px; overflow:auto; z-index:1000;"> <div style="display:flex; justify-content:space-between;"> <h3 style="margin:0;">{nome}</h3> <button onclick="document.getElementById('{modal_id}').style.display='none'" style="background:#333;color:#fff;border:none; border-radius:6px;padding:6px 12px;cursor:pointer;"> Fechar </button> </div> <pre style="white-space:pre-wrap; font-family:monospace; font-size:14px;">{conteudo} </pre> </div> """blocks.append(block)display(HTML("".join(blocks)))
-------------------->>> 7. Qualidade do Código, Testes Automatizados e Auditoria do Pipeline <<<--------------------
main.py
"""
Arquivo: main.py
Ponto de entrada do pipeline de engenharia de dados.
Fluxo:
1) Configura logging e diretórios.
2) Lê dados de vendas com leitura defensiva (encoding e separador configuráveis).
3) Valida estrutura, padroniza tipos e separa registros válidos e inválidos.
4) Executa diagnóstico técnico apenas sobre os registros válidos.
5) Salva metadados e relatório textual consolidado.
6) Mantém registros inválidos fora do fluxo principal para auditoria.
Execução:
python main.py
Autor: Luciano Magalhães
Novembro 2025
"""
from __future__ import annotations
import logging
from config import (
INPUT_CSV,
LOGS_DIR,
OUTPUT_DIR,
META_JSON,
RELATORIO_TXT,
CSV_SEPARATOR,
CSV_ENCODINGS,
)
from diagnostics import (
diagnostico_dados,
gerar_relatorio_texto,
)
from utils import (
ensure_dirs,
read_csv_safely,
save_json,
setup_logging,
)
from validators import run_validations
def main() -> None:
"""
Executa o pipeline completo de engenharia de dados para a base de vendas.
"""
# ============================================================
# 1) Logging e diretórios
# ============================================================
logger = setup_logging(
logs_dir=LOGS_DIR,
filename="pipeline.log",
level=logging.INFO,
)
ensure_dirs(LOGS_DIR, OUTPUT_DIR)
# ============================================================
# 2) Leitura defensiva do CSV
# ============================================================
df_raw = read_csv_safely(
path=INPUT_CSV,
logger=logger,
sep=CSV_SEPARATOR,
encodings=CSV_ENCODINGS,
)
# ============================================================
# 3) Validação e padronização
# ============================================================
result = run_validations(df_raw, logger=logger)
df_validos = result.valid
df_invalidos = result.invalid
# ============================================================
# 4) Diagnóstico técnico (apenas dados válidos)
# ============================================================
meta = diagnostico_dados(df_validos, logger=logger)
save_json(meta, META_JSON)
# ============================================================
# 5) Relatório textual consolidado
# ============================================================
gerar_relatorio_texto(
meta=meta,
caminho=RELATORIO_TXT,
logger=logger,
)
# ============================================================
# 6) Log final de execução
# ============================================================
logger.info(
"Pipeline concluído com sucesso | válidos=%s | inválidos=%s",
len(df_validos),
len(df_invalidos),
)
if __name__ == "__main__":
main()
config.py
"""
Arquivo config.py
Guardamos caminhos e nomes padrão para entradas e saídas
do pipeline de engenharia de dados.
Autor: Luciano Magalhães
Novembro 2025
"""
from pathlib import Path
BASE_DIR = Path(".")
INPUT_CSV = BASE_DIR / "dados_vendas.csv"
LOGS_DIR = BASE_DIR / "logs"
OUTPUT_DIR = BASE_DIR / "output"
RELATORIO_TXT = LOGS_DIR / "relatorio_pipeline.txt"
META_JSON = OUTPUT_DIR / "metadados" / "pipeline_metadados.json"
# Parâmetros de leitura do CSV
CSV_SEPARATOR = ","
CSV_ENCODINGS = ("utf-8", "latin1")
validators.py
"""
Arquivo: validators.py
Validações e padronizações de estrutura e regras de negócio
para o pipeline de engenharia de dados.
Este módulo executa:
- Validação de colunas obrigatórias;
- Padronização de tipos (datetime, numéricos, categóricos);
- Separação explícita de registros válidos e inválidos;
- Métricas de validação para auditoria e rastreabilidade.
Autor: Luciano Magalhães
Novembro 2025
"""
from __future__ import annotations
import logging
from dataclasses import dataclass
from typing import Dict, List
import numpy as np
import pandas as pd
# ============================================================
# Configurações de validação
# ============================================================
# Colunas obrigatórias (padrão do dataset atual)
REQUIRED_COLS: List[str] = [
"mes",
"vendedor_id",
"produto_id",
"quantidade",
"preco_unitario",
"desconto",
"valor_total",
]
# Tipos esperados por coluna (padrão do dataset atual)
EXPECTED_TYPES: Dict[str, str] = {
"mes": "datetime64[ns]",
"vendedor_id": "category",
"produto_id": "category",
"quantidade": "int64",
"preco_unitario": "float64",
"desconto": "float64",
"valor_total": "float64",
}
# ============================================================
# Validação estrutural
# ============================================================
def validate_structure(
df: pd.DataFrame,
required_cols: List[str],
logger: logging.Logger,
) -> None:
"""
Valida se todas as colunas obrigatórias existem no DataFrame.
"""
missing = [c for c in required_cols if c not in df.columns]
if missing:
logger.error("Colunas obrigatórias ausentes: %s", missing)
raise ValueError(f"Colunas obrigatórias faltando: {missing}")
logger.info("Estrutura validada: todas as colunas obrigatórias presentes.")
# ============================================================
# Padronização de tipos
# ============================================================
def standardize_types(df: pd.DataFrame, logger: logging.Logger) -> pd.DataFrame:
"""
Padroniza tipos de colunas conforme EXPECTED_TYPES.
Regras:
- Conversões numéricas usam pd.to_numeric(errors='coerce');
- Datas usam pd.to_datetime(errors='coerce');
- Valores inválidos são convertidos para NaN;
- Nenhuma decisão de descarte é feita aqui.
"""
for col, expected_type in EXPECTED_TYPES.items():
if col not in df.columns:
logger.warning(
"Coluna esperada '%s' não encontrada. Ignorada na padronização.",
col,
)
continue
try:
if "datetime" in expected_type:
df[col] = pd.to_datetime(df[col], errors="coerce")
elif expected_type == "category":
df[col] = df[col].astype("category")
elif expected_type in {"int64", "float64"}:
df[col] = pd.to_numeric(df[col], errors="coerce")
else:
df[col] = df[col].astype(expected_type)
logger.info("Coluna '%s' padronizada para %s.", col, expected_type)
except Exception as exc:
logger.warning(
"Falha ao padronizar coluna '%s' para %s: %s",
col,
expected_type,
exc,
)
logger.info(
"Tipos após padronização: %s",
df.dtypes.astype(str).to_dict(),
)
return df
# ============================================================
# Resultado de validação
# ============================================================
@dataclass(frozen=True)
class ValidationResult:
"""
Resultado da validação do dataset.
"""
valid: pd.DataFrame
invalid: pd.DataFrame
metrics: Dict[str, int]
# ============================================================
# Validação semântica e separação de registros
# ============================================================
def split_valid_invalid(df: pd.DataFrame, logger: logging.Logger) -> ValidationResult:
"""
Separa registros válidos e inválidos com base nas regras de negócio.
"""
mask = pd.Series(True, index=df.index)
if "quantidade" in df.columns:
mask &= df["quantidade"].notna() & (df["quantidade"] >= 0)
if "preco_unitario" in df.columns:
mask &= df["preco_unitario"].notna() & (df["preco_unitario"] >= 0)
if "desconto" in df.columns:
mask &= df["desconto"].notna() & df["desconto"].between(0, 1)
if {"quantidade", "preco_unitario", "desconto", "valor_total"}.issubset(df.columns):
esperado = (
df["quantidade"]
* df["preco_unitario"]
* (1 - df["desconto"])
)
mask &= np.isclose(
df["valor_total"],
esperado,
rtol=1e-03,
atol=1e-02,
)
validos = df.loc[mask].copy()
invalidos = df.loc[~mask].copy()
metrics = {
"total": len(df),
"validos": len(validos),
"invalidos": len(invalidos),
}
logger.info(
"Validação de negócio: total=%s | válidos=%s | inválidos=%s",
metrics["total"],
metrics["validos"],
metrics["invalidos"],
)
return ValidationResult(
valid=validos,
invalid=invalidos,
metrics=metrics,
)
# ============================================================
# Pipeline de validação
# ============================================================
def run_validations(df: pd.DataFrame, logger: logging.Logger) -> ValidationResult:
"""
Pipeline de validação e padronização:
1) Valida estrutura mínima;
2) Padroniza tipos;
3) Separa registros válidos e inválidos.
Retorna:
ValidationResult
"""
validate_structure(df, REQUIRED_COLS, logger)
df = standardize_types(df, logger)
return split_valid_invalid(df, logger)
# Fim do arquivo validators.py
diagnostics.py
"""
Arquivo: diagnostics.py
Módulo de diagnóstico e metadados do pipeline de dados de vendas.
Responsável por:
- Dimensões do dataset
- Tipos de dados
- Valores nulos
- Verificação de colunas críticas (ex.: 'mes' em datetime)
Saídas:
- Metadados (JSON)
- Relatório textual consolidado
Autor: Luciano Magalhães
Novembro 2025
"""
from __future__ import annotations
from typing import Dict
import logging
import pandas as pd
# ============================================================
# Diagnóstico básico do dataset
# ============================================================
def diagnostico_dados(df: pd.DataFrame, logger: logging.Logger) -> Dict:
"""
Retorna metadados técnicos do dataset SEM modificar os dados.
Observações:
- Nenhuma conversão ou correção é feita aqui
- O diagnóstico apenas observa o estado atual do DataFrame
"""
logger.info("Executando diagnóstico dos dados...")
metadados = {
"shape": {
"linhas": int(df.shape[0]),
"colunas": int(df.shape[1]),
},
"nulos": df.isnull().sum().to_dict(),
"dtypes": df.dtypes.astype(str).to_dict(),
"mes_convertido_para_datetime": False,
}
if "mes" in df.columns:
metadados["mes_convertido_para_datetime"] = (
pd.api.types.is_datetime64_any_dtype(df["mes"])
)
return metadados
# ============================================================
# Relatório textual consolidado
# ============================================================
def gerar_relatorio_texto(meta: Dict, caminho: str, logger: logging.Logger) -> None:
"""
Gera relatório textual consolidado a partir dos metadados.
"""
logger.info("Gerando relatório textual consolidado...")
linhas = [
"=== Relatório de Diagnóstico ===\n",
"Dimensões do dataset:",
str(meta["shape"]),
"\nValores nulos por variável:",
str(meta["nulos"]),
"\nTipos de dados:",
str(meta["dtypes"]),
"\nColuna 'mes' em datetime:",
str(meta.get("mes_convertido_para_datetime", False)),
"\n",
]
with open(caminho, "w", encoding="utf-8") as f:
f.write("\n".join(linhas))
logger.info("Relatório textual salvo em %s", caminho)
utils.py
"""
Arquivo: utils.py
Utilidades de logging, IO e moldura de texto.
Este módulo inclui funções auxiliares para:
- Configuração de logs com rotação;
- Salvar DataFrames e JSON em diretórios conhecidos;
- Impressão com moldura no console;
- Leitura segura de CSV com validação e logging.
Autor: Luciano Magalhães
Novembro 2025
"""
from __future__ import annotations
import json
import logging
from logging.handlers import RotatingFileHandler
from pathlib import Path
from typing import Any, Dict, Optional
import pandas as pd
# ============================================================
# Logging
# ============================================================
def setup_logging(logs_dir: str, filename: str, level: int = logging.INFO) -> logging.Logger:
"""
Configura logger com rotação de arquivo, evitando duplicações.
"""
Path(logs_dir).mkdir(parents=True, exist_ok=True)
logger = logging.getLogger("pipeline_dados") # nome atualizado
logger.setLevel(level)
logger.propagate = False
# Limpa handlers antigos para evitar duplicação
if logger.hasHandlers():
logger.handlers.clear()
log_path = Path(logs_dir) / filename
handler = RotatingFileHandler(
log_path, maxBytes=2_000_000, backupCount=3, encoding="utf-8"
)
formatter = logging.Formatter(
"%(asctime)s | %(levelname)s | %(name)s | %(message)s"
)
handler.setFormatter(formatter)
logger.addHandler(handler)
console = logging.StreamHandler()
console.setFormatter(formatter)
logger.addHandler(console)
return logger
# ============================================================
# Utilidades de texto e diretórios
# ============================================================
def moldura_texto(texto: str) -> None:
"""
Imprime texto com moldura simples no console.
"""
moldura = "-" * 20
print(f"\n{moldura}>>> {texto} <<<{moldura}\n")
def ensure_dirs(*dirs: str) -> None:
"""
Garante que diretórios de saída existem.
"""
for d in dirs:
Path(d).mkdir(parents=True, exist_ok=True)
# ============================================================
# IO: DataFrames, JSON e CSV
# ============================================================
def save_dataframe(df: pd.DataFrame, path: str, index: bool = False) -> None:
"""
Salva DataFrame em CSV com encoding UTF-8.
"""
Path(path).parent.mkdir(parents=True, exist_ok=True)
df.to_csv(path, index=index, encoding="utf-8")
def save_json(data: Dict[str, Any], path: str) -> None:
"""
Salva dicionário em JSON com encoding UTF-8.
"""
Path(path).parent.mkdir(parents=True, exist_ok=True)
with open(path, "w", encoding="utf-8") as f:
json.dump(data, f, ensure_ascii=False, indent=2)
def read_csv_safely(
path: str | Path,
logger: Optional[logging.Logger] = None,
*,
sep: str = ",",
encodings: tuple[str, ...] = ("utf-8", "latin1"),
) -> pd.DataFrame:
"""
Lê CSV de forma defensiva:
- valida existência
- tenta múltiplos encodings
- permite configurar separador
- registra métricas básicas no log
"""
path = Path(path)
if not path.exists():
msg = f"Arquivo não encontrado: {path}"
if logger:
logger.error(msg)
raise FileNotFoundError(msg)
last_exc: Exception | None = None
for enc in encodings:
try:
df = pd.read_csv(path, encoding=enc, sep=sep)
if logger:
logger.info(
"CSV carregado: %s | encoding=%s | sep=%s | linhas=%s | colunas=%s",
path, enc, sep, df.shape[0], df.shape[1],
)
return df
except UnicodeDecodeError as exc:
last_exc = exc
if logger:
logger.warning("Falha de encoding=%s ao ler %s: %s", enc, path, exc)
except pd.errors.EmptyDataError as exc:
msg = f"O arquivo {path} está vazio."
if logger:
logger.error(msg)
raise ValueError(msg) from exc
except pd.errors.ParserError as exc:
msg = f"Erro de parsing ao ler {path}: {exc}"
if logger:
logger.error(msg)
raise ValueError(msg) from exc
msg = f"Não foi possível ler {path} com encodings {encodings}. Último erro: {last_exc}"
if logger:
logger.error(msg)
raise RuntimeError(msg)
# Fim do arquivo utils.py