process_data.py

#!/usr/bin/env python
# -*- coding: utf-8 -*-

"""
Script principal para el procesamiento de datos del proyecto de análisis de comercio electrónico.

Este script realiza las siguientes acciones:
1. Carga los datos desde el archivo fuente en data/raw/
2. Realiza la limpieza y transformación de los datos
3. Guarda los datos procesados en data/processed/
"""
import os
import sys
import logging
from pathlib import Path

# Añadir el directorio src al path para poder importar los módulos
sys.path.append(str(Path(__file__).parent / 'src'))

# Importar módulos de procesamiento
from src.data.processing import process_data
from src.config import (
    RAW_DATA_PATH,
    PROCESSED_DATA_PATH,
    PROCESSING_CONFIG
)

# Configurar logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
    handlers=[
        logging.StreamHandler(),
        logging.FileHandler('data_processing.log')
    ]
)
logger = logging.getLogger(__name__)

def main():
    """Función principal que orquesta el proceso de ETL."""
    logger.info("=" * 80)
    logger.info("INICIANDO PROCESAMIENTO DE DATOS")
    logger.info("=" * 80)
    
    # Mostrar configuración
    logger.info("Configuración de procesamiento:")
    logger.info(f"- Archivo de entrada: {RAW_DATA_PATH}")
    logger.info(f"- Archivo de salida: {PROCESSED_DATA_PATH}")
    logger.info(f"- Columnas numéricas: {PROCESSING_CONFIG['numeric_columns']}")
    logger.info(f"- Columnas de texto: {PROCESSING_CONFIG['string_columns']}")
    logger.info(f"- Eliminar duplicados: {PROCESSING_CONFIG['drop_duplicates']}")
    logger.info(f"- Eliminar filas con valores faltantes en: {PROCESSING_CONFIG['drop_na']}")
    
    try:
        # Ejecutar el proceso de ETL
        df_processed = process_data()
        
        # Mostrar resumen
        logger.info("=" * 50)
        logger.info("PROCESO COMPLETADO EXITOSAMENTE")
        logger.info("=" * 50)
        logger.info(f"Total de registros procesados: {len(df_processed):,}")
        logger.info(f"Total de clientes únicos: {df_processed['CustomerID'].nunique():,}")
        logger.info(f"Total de productos únicos: {df_processed['StockCode'].nunique():,}")
        logger.info(f"Total facturado: ${df_processed['TotalAmount'].sum():,.2f}")
        logger.info(f"Datos guardados en: {PROCESSED_DATA_PATH}")
        
    except Exception as e:
        logger.error(f"Error durante el procesamiento: {str(e)}", exc_info=True)
        sys.exit(1)

if __name__ == "__main__":
    main()