86 lines
2.9 KiB
Python
86 lines
2.9 KiB
Python
"""Utilitários de entrada, saída e log dos scripts.
|
|
|
|
Entradas:
|
|
Caminhos definidos em 01_config.py.
|
|
|
|
Saídas:
|
|
Validações de arquivos, diretórios criados, logs em logs/execucao.log e
|
|
funcoes pequenas para exportacao.
|
|
|
|
Papel no pipeline:
|
|
Padroniza mensagens de erro e evita repetição de código de filesystem.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import importlib
|
|
import logging
|
|
import sys
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
from typing import Iterable
|
|
|
|
import pandas as pd
|
|
|
|
config = importlib.import_module("01_config")
|
|
|
|
|
|
def configurar_ambiente() -> None:
|
|
"""Garante diretorios e deixa src/ importavel para execucoes pelo VS Code."""
|
|
config.garantir_diretorios()
|
|
project_root = str(config.PROJECT_ROOT)
|
|
if project_root not in sys.path:
|
|
sys.path.insert(0, project_root)
|
|
|
|
|
|
def configurar_logger(nome: str = "planejador_missao") -> logging.Logger:
|
|
"""Cria logger simples em arquivo e terminal."""
|
|
configurar_ambiente()
|
|
logger = logging.getLogger(nome)
|
|
logger.setLevel(logging.INFO)
|
|
logger.handlers.clear()
|
|
|
|
formato = logging.Formatter("%(asctime)s | %(levelname)s | %(message)s")
|
|
file_handler = logging.FileHandler(config.LOG_EXECUCAO, encoding="utf-8")
|
|
file_handler.setFormatter(formato)
|
|
stream_handler = logging.StreamHandler()
|
|
stream_handler.setFormatter(logging.Formatter("%(message)s"))
|
|
logger.addHandler(file_handler)
|
|
logger.addHandler(stream_handler)
|
|
logger.info("Execucao iniciada em %s", datetime.now().strftime("%Y-%m-%d %H:%M:%S"))
|
|
return logger
|
|
|
|
|
|
def exigir_arquivos(paths: Iterable[Path]) -> None:
|
|
"""Interrompe a execução com mensagem clara se algum arquivo faltar."""
|
|
ausentes = [path for path in paths if not path.exists()]
|
|
if ausentes:
|
|
lista = "\n".join(f"- {path}" for path in ausentes)
|
|
raise FileNotFoundError(f"Erro: arquivos obrigatorios ausentes:\n{lista}")
|
|
|
|
|
|
def exigir_colunas(df: pd.DataFrame, colunas: Iterable[str], origem: str) -> None:
|
|
"""Valida colunas obrigatorias de uma tabela carregada."""
|
|
faltantes = [col for col in colunas if col not in df.columns]
|
|
if faltantes:
|
|
raise ValueError(
|
|
f"Erro: coluna obrigatória '{faltantes[0]}' não encontrada em {origem}. "
|
|
"Verifique o arquivo de entrada."
|
|
)
|
|
|
|
|
|
def exportar_excel(path: Path, abas: dict[str, pd.DataFrame]) -> Path:
|
|
"""Exporta um workbook Excel com uma aba por DataFrame."""
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
with pd.ExcelWriter(path, engine="openpyxl") as writer:
|
|
for nome, df in abas.items():
|
|
df.to_excel(writer, sheet_name=nome[:31], index=False)
|
|
return path
|
|
|
|
|
|
def exportar_csv(path: Path, df: pd.DataFrame) -> Path:
|
|
"""Exporta CSV garantindo existencia do diretorio."""
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
df.to_csv(path, index=False)
|
|
return path
|