Fluxo Técnico Modular para Preparação de Dados de Vendas

"""
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 encoding UTF-8).
3) Valida estrutura, padroniza tipos e checa regras de negócio (validators.run_validations).
4) Executa diagnóstico técnico (metadados e integridade).
5) Salva metadados e relatório textual consolidado.

Execução:
    python main.py

Autor: Luciano Magalhães
Novembro 2024
"""

from __future__ import annotations
import logging

from config import (
    INPUT_CSV,
    LOGS_DIR,
    OUTPUT_DIR,
    META_JSON,
    RELATORIO_TXT,
)
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) Carregar dados (com encoding UTF-8 garantido em utils.read_csv_safely)
    df = read_csv_safely(INPUT_CSV, logger=logger)

    # 3) Validar e padronizar dataset
    df = run_validations(df, logger=logger)

    # 4) Diagnóstico técnico
    meta = diagnostico_dados(df, logger=logger)
    save_json(meta, META_JSON)

    # 5) Relatório textual consolidado
    gerar_relatorio_texto(
        meta=meta,
        caminho=RELATORIO_TXT,
        logger=logger,
    )

    logger.info("Pipeline concluído com sucesso.")


if __name__ == "__main__":
    main()

"""
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 2024
"""

INPUT_CSV = "dados_vendas.csv"
LOGS_DIR = "logs"
OUTPUT_DIR = "output"

# Saídas principais do pipeline
RELATORIO_TXT = "logs/relatorio_pipeline.txt"
META_JSON = "output/pipeline_metadados.json"

"""
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);
- Checagens de consistência semântica (desconto em [0,1], não-negativos);
- Verificação aproximada de valor_total.

Autor: Luciano Magalhães
Novembro 2024
"""
from __future__ import annotations

import logging
from typing import Dict, List

import numpy as np
import pandas as pd


# 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",
}


def validate_structure(
    df: pd.DataFrame,
    required_cols: List[str],
    logger: logging.Logger,
) -> None:
    """
    Valida se colunas obrigatórias existem no DataFrame.
    """
    missing = [c for c in required_cols if c not in df.columns]
    if missing:
        logger.error(f"Colunas ausentes: {missing}")
        raise ValueError(f"Colunas obrigatórias faltando: {missing}")
    logger.info("Estrutura validada: todas as colunas obrigatórias presentes.")


def standardize_types(df: pd.DataFrame, logger: logging.Logger) -> pd.DataFrame:
    """
    Padroniza tipos de colunas conforme EXPECTED_TYPES.
    """
    for col, tipo in EXPECTED_TYPES.items():
        if col not in df.columns:
            logger.warning(f"Coluna esperada '{col}' não encontrada. (ignorada na padronização)")
            continue

        try:
            if "datetime" in tipo:
                df[col] = pd.to_datetime(df[col], errors="coerce")
            elif tipo == "category":
                df[col] = df[col].astype("category")
            else:
                df[col] = df[col].astype(tipo, errors="ignore")
            logger.info(f"Coluna '{col}' convertida para {tipo}.")
        except Exception as e:
            logger.warning(f"Falha ao converter '{col}' para {tipo}: {e}")
  
    logger.info(f"Dtypes após padronização: {df.dtypes.astype(str).to_dict()}")
    return df


def validate_business_rules(df: pd.DataFrame, logger: logging.Logger) -> None:
    """
    Valida regras de negócio básicas:
    - quantidade >= 0
    - preco_unitario >= 0
    - desconto entre 0 e 1
    - valor_total consistente (aproximação)
    """
    issues = []

    if "quantidade" in df.columns and (df["quantidade"] < 0).any():
        issues.append("Há quantidade negativa.")

    if "preco_unitario" in df.columns and (df["preco_unitario"] < 0).any():
        issues.append("Há preço unitário negativo.")

    if "desconto" in df.columns:
        desconto = df["desconto"]
        if (desconto < 0).any() or (desconto > 1).any():
            issues.append("Desconto fora do intervalo [0,1].")

    # Checagem aproximada de valor_total
    needed = {"quantidade", "preco_unitario", "desconto", "valor_total"}
    if needed.issubset(df.columns):
        calc = df["quantidade"].astype(float) * df["preco_unitario"].astype(float) * (1 - df["desconto"].astype(float))
        mask_incons = ~np.isclose(df["valor_total"].astype(float), calc, rtol=1e-03, atol=1e-02)
        if mask_incons.any():
            qtd = int(mask_incons.sum())
            issues.append(f"Valor total inconsistente em {qtd} registros (vs. quantidade, preço e desconto).")
    else:
        faltantes = list(needed - set(df.columns))
        logger.warning(f"Regra de negócio não checada: faltam colunas {faltantes}.")

    if issues:
        for msg in issues:
            logger.warning(f"Regra de negócio: {msg}")
    else:
        logger.info("Regras de negócio validadas sem inconsistências.")


def run_validations(df: pd.DataFrame, logger: logging.Logger) -> pd.DataFrame:
    """
    Pipeline de validação e padronização:
    1) Valida estrutura mínima (REQUIRED_COLS);
    2) Padroniza tipos conforme EXPECTED_TYPES;
    3) Valida regras de negócio.

    Retorno:
        pd.DataFrame: DataFrame pronto para uso no pipeline.
    """
    validate_structure(df, REQUIRED_COLS, logger)
    df = standardize_types(df, logger)
    validate_business_rules(df, logger)
    return df

"""
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
- Conversão de colunas críticas (ex.: 'mes' para datetime)

Saídas:
- Metadados (JSON)
- Relatório textual (logs/relatorio_dados.txt)

Autor: Luciano Magalhães
Novembro 2024
"""

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 básicos do dataset:
    - Dimensões
    - Valores nulos
    - Tipos de dados
    - Conversão da coluna 'mes' para datetime
    """
    logger.info("Executando diagnóstico dos dados...")

    metadados = {
        "shape": {"linhas": df.shape[0], "colunas": df.shape[1]},
        "nulos": df.isnull().sum().to_dict(),
        "dtypes": df.dtypes.astype(str).to_dict(),
        "mes_convertido_para_datetime": False,
    }

    # Conversão de 'mes' para datetime, se existir
    if "mes" in df.columns:
        try:
            df["mes"] = pd.to_datetime(df["mes"], errors="coerce")
            metadados["mes_convertido_para_datetime"] = True
            logger.info("Coluna 'mes' convertida para datetime com sucesso.")
        except Exception as e:
            logger.warning(f"Falha ao converter 'mes' para datetime: {e}")

    return metadados


# ============================================================
# Relatório textual consolidado
# ============================================================

def gerar_relatorio_texto(meta: Dict, caminho: str, logger: logging.Logger) -> None:
    """
    Gera relatório textual consolidado com:
    - Metadados
    - Valores nulos
    - Tipos de dados
    """
    logger.info("Gerando relatório textual consolidado...")

    linhas = []
    linhas.append("=== Relatório de Diagnóstico ===\n")

    linhas.append("Dimensões do dataset:")
    linhas.append(str(meta["shape"]))
    linhas.append("\nValores nulos por variável:")
    linhas.append(str(meta["nulos"]))
    linhas.append("\nTipos de dados:")
    linhas.append(str(meta["dtypes"]))
    linhas.append("\nConversão de 'mes' para datetime:")
    linhas.append(str(meta.get("mes_convertido_para_datetime", False)))
    linhas.append("\n")

    with open(caminho, "w", encoding="utf-8") as f:
        f.write("\n".join(linhas))

    logger.info(f"Relatório textual salvo em {caminho}")

"""
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 2024
"""

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, logger: Optional[logging.Logger] = None) -> pd.DataFrame:
    """
    Lê CSV com validação de existência, encoding UTF-8 e logging.
    """
    if not Path(path).exists():
        msg = f"Arquivo não encontrado: {path}"
        if logger:
            logger.error(msg)
        raise FileNotFoundError(msg)

    try:
        df = pd.read_csv(path, encoding="utf-8")
        if logger:
            logger.info(
                f"Arquivo CSV carregado com sucesso: {path} (linhas={df.shape[0]}, colunas={df.shape[1]})"
            )
        return df

    except pd.errors.EmptyDataError:
        msg = f"O arquivo {path} está vazio."
        if logger:
            logger.error(msg)
        raise ValueError(msg)

    except pd.errors.ParserError as e:
        msg = f"Erro ao ler o CSV {path}: {e}"
        if logger:
            logger.error(msg)
        raise ValueError(msg)

    except Exception as e:
        msg = f"Erro inesperado ao carregar {path}: {e}"
        if logger:
            logger.error(msg)
        raise RuntimeError(msg)