Initial project import

scripts/02_io_utils.py

@@ -0,0 +1,85 @@
+"""Utilitarios de entrada, saida e log dos scripts.
+
+Entradas:
+    Caminhos definidos em 01_config.py.
+
+Saidas:
+    Validacoes de arquivos, diretorios criados, logs em logs/execucao.log e
+    funcoes pequenas para exportacao.
+
+Papel no pipeline:
+    Padroniza mensagens de erro e evita repeticao de codigo 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 execucao 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 obrigatoria '{faltantes[0]}' nao 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