etl_soporte.py

# =====================
# IMPORTAR LIBRERÍAS
# =====================

# Conexión a MySQL
import mysql.connector
from mysql.connector import errorcode
import pymysql
from sqlalchemy import create_engine

# Manejo de datos
import pandas as pd
import numpy as np

# Imputación de nulos
from sklearn.impute import SimpleImputer
from sklearn.experimental import enable_iterative_imputer
from sklearn.impute import IterativeImputer
from sklearn.impute import KNNImputer

# Visualización
import seaborn as sns
import matplotlib.pyplot as plt

# Evaluar distribuciones estadísticas
from scipy.stats import shapiro, kstest, chi2_contingency
import scipy.stats as stats

# Gestión de los warnings
import warnings
warnings.filterwarnings("ignore")

# Configuración
pd.set_option('display.max_columns', None)

# =====================
# # CONFIGURACIÓN: Crear la conexión con el gestor de BBDD
# =====================

host = 'localhost'
user = 'root'
password = 'AdalabAlumnas'
database = 'BBDD_TO'


# =====================
# EXTRACCIÓN Y EXPLORACIÓN: Cargar y explorar datos
# =====================
def explorar_dataset(ruta, index_col=0, n=5):
    """Carga y explora un dataset desde un archivo CSV."""
    df = pd.read_csv(ruta, index_col=index_col)

    print("\n--- Información general del DataFrame ---")
    df.info()
    
    print("\nPrimeras filas del DataFrame:")
    display(df.head(n))
    
    print("\nFilas aleatorias del DataFrame:")
    display(df.sample(n))
    
    print(f"\nTamaño del Dataframe: {df.size} elementos")
    print(f"Forma del dataset: {df.shape[0]} filas, {df.shape[1]} columnas")
    print("\n--- Información sobre las columnas ---")
    print(f"Columnas del dataset: {df.columns.tolist()}")
    print("\nConteo de valores no nulos por columna:")
    print(df.count())

    return df

def mostrar_valores_frecuencias(df, select_dtypes=None):
    """Muestra valores únicos y su frecuencia en columnas seleccionadas."""
    columnas = df.select_dtypes(include=select_dtypes).columns
    for columna in columnas:
        print(f"\n----------- Análisis de la columna: '{columna}' -----------")
        print(f'Valores únicos:\n {df[columna].unique()}')
        print(f'Frecuencias:\n {df[columna].value_counts().reset_index()}')

def analizar_nulos(df):
    """Analiza valores nulos en el DataFrame."""
    resumen_nulos = pd.DataFrame({
        'Valores Nulos': df.isnull().sum(),
        'Porcentaje Nulos (%)': (df.isnull().sum() / len(df)) * 100,
        'Tipo de Dato': df.dtypes
    })
    return resumen_nulos

# =====================
# TRANSFORM: Transformar datos
# =====================
def estandarizar_columnas(df):
    """Estandariza los nombres de las columnas."""
    df.columns = df.columns.str.strip().str.lower().str.replace(" ", "_").str.replace(r'[^\w\s]', '', regex=True)
    print(f"Nombres de columnas estandarizados: {df.columns.tolist()}")
    return df

def transformar_datos(df):
    # Cambiamos GENDER con valores 0 y 1 a valores Male y Female
    dataset['gender'] = dataset['gender'].astype(str)  
    dataset['gender'] = dataset['gender'].replace({'0': 'Male', '1': 'Female'})
    dataset.sample(5)

    # Corregimos valores de la columna 'jobrole'  y 'roledepartment, homegeneizándolos.
    dataset['jobrole'] = dataset['jobrole'].str.title()
    dataset['roledepartament'] = dataset['roledepartament'].str.title()
    dataset.sample(5)

    # Corrección de 'Marreid' por 'Married'
    dataset['maritalstatus'] = dataset['maritalstatus'].replace({'Marreid': 'Married'})
    dataset.sample(10)

    # Reemplazamos valores de remotework a dos valores: 'No' y 'Yes'
    dataset['remotework'] = dataset['remotework'].replace({'0': 'No', '1': 'Yes'})
    dataset['remotework'] = dataset['remotework'].replace({'False': 'No', 'True': 'Yes'})
    dataset.sample(5)

    # Reemplazamos los valores que no son correctos
    cambiar_edad = {"forty-seven": "47", "fifty-eight": "58","thirty-six": "36", "fifty-five": "55", "fifty-two": "52", "thirty-one": "31",
                    "thirty": "30", "twenty-six": "26", "thirty-seven": "37", "thirty-two": "32", "twenty-four": "24" }
    # Reemplazar los valores en la columna "numeros" del DataFrame
    dataset["age"] = dataset["age"].replace(cambiar_edad)


    # Transformamos valores negativos a absolutos
    dataset['distancefromhome'] = dataset['distancefromhome'].abs()

    # Transformamos variables categóricas a numéricas

    # Reemplazamos 'Not Available' por nulo 'nan', y cambiamos a numérico:
    dataset['hourlyrate'] = dataset['hourlyrate'].replace('Not Available', np.nan)
    dataset['hourlyrate'] = dataset['hourlyrate'].astype('float')
    dataset['hourlyrate'].dtype

    # Reemplazamos '$' por '', y la coma por punto en columna 'daylerate' 
    dataset['dailyrate'] = dataset['dailyrate'].str.replace('$', '', regex=False)
    dataset['dailyrate'] = dataset['dailyrate'].str.replace(',', '.', regex=False)
    dataset['dailyrate'] = dataset['dailyrate'].str.replace('nan', 'NaN', regex=False)
    dataset['dailyrate'] = dataset['dailyrate'].replace('NaN', 0).astype(float).astype(int)
    dataset['dailyrate'] = dataset['dailyrate'].astype(float).astype(int)
    dataset['dailyrate'].value_counts()

    # Convertimos columna 'age' a variable numérica
    dataset['age'] = pd.to_numeric(dataset['age'])

    # Reemplazar NaN con 0 en la columna 'employeenumber' 
    dataset['employeenumber'] = dataset['employeenumber'].fillna(0) # Verificar los valores únicos 
    print(dataset['employeenumber'].value_counts())

    return df

def eliminar_columnas(df, columnas_a_eliminar):
    # Filtramos las columnas que realmente existen en el DataFrame
    columnas_existentes = [col for col in columnas_a_eliminar if col in df.columns]

    if columnas_existentes:
        df.drop(columns=columnas_existentes, inplace=True)
        print(f" Columnas eliminadas: {columnas_existentes}")
    else:
        print(" No se encontraron columnas para eliminar.")
    
    return df

# Función para cambiar comas por puntos.
def cambiar_comas(valor):
    """Convierte valores con comas en números con punto decimal."""
    if isinstance(valor, str):
        return float(valor.replace(",", "."))
    return valor

# Función gestionar nulos variables categóricas
def imputar_nulos_moda(df, columnas):
    # Se crea una lista de las columnas
    columnas = [col for col in columnas if col in df.columns]
    # Itermoas por la columnas de la lista creada
    for columna in columnas:
        if df[columna].isnull().sum() > 0:
            moda = df[columna].mode()[0]
            df[columna].fillna(moda, inplace=True)

    print("\nDespués del reemplazo usando 'fillna', quedan los siguientes nulos:")
    print(df[columnas].isnull().sum())

    return df

# Gestión de nulos con variable numéricas
def imputacion_nulos(df):
    """Imputa valores nulos usando la media."""
    print("Valores nulos antes de la imputación:")
    print(df.isnull().sum())

    imputer = SimpleImputer(strategy='mean')
    df.iloc[:, :] = imputer.fit_transform(df)

    print("Valores nulos después de la imputación:")
    print(df.isnull().sum())

    plt.figure(figsize=(10, 6))
    sns.histplot(df.select_dtypes(include=['number']), kde=True)
    plt.title("Distribución de variables numéricas")
    plt.show()

    return df

# =====================
# LOAD: Cargar datos a MySQL
# =====================

# =====================
# CONEXIÓN A MySQL
# =====================

# Función para crear conexión con BBDD
def create_connection():
    try:
        cnx = mysql.connector.connect(user='root', password='AlumnaAdalab',
                                      host='127.0.0.1'
                                      )
        return cnx
    # en caso de que no lo consigas por que hay algún error entonces ...
    except mysql.connector.Error as err:
        # si es un error con la contraseña devuelveme un mensaje de acceso denegado ya que tenemos problemas con la contraseña
        if err.errno == errorcode.ER_ACCESS_DENIED_ERROR:
            print('Something is wrong with your user name or password')
        # si el error no tiene que ver con la contraseña, puede ser porque la base de datos no exista, devuelveme un mensaje de que la base de datos no existe
        elif err.errno == errorcode.ER_BAD_DB_ERROR:
            print('Database does not exist')
        # si no es por ninguno de los errores anteriores, printeame cual es el error que estoy teniendo en mi conexión
        else:
            print(err)
        return None
    else:
        cnx.close()

# Función para crear BBDD
def crear_bbdd(user, password, host, database):
    """Crea una base de datos en MySQL si no existe."""
    try:
        conexion = pymysql.connect(host=host, user=user, password=password)
        cursor = conexion.cursor()
        cursor.execute(f"CREATE DATABASE IF NOT EXISTS {database}")
        print(f"Base de Datos '{database}' creada exitosamente.")
    except pymysql.MySQLError as e:
        print(f"Error MySQL: {e}")
    except Exception as e:
        print(f"Error inesperado: {e}")
    finally:
        if 'conexion' in locals() and conexion.open:
            cursor.close()
            conexion.close()
            print("Conexión cerrada.")

# Función para crear tabla de empleados
def crear_tabla_empleados(user, password, host, database):
    """Crea la tabla employees en la base de datos MySQL."""
    try:
        cnx = mysql.connector.connect(user=user, password=password, host=host, database=database)
        mycursor = cnx.cursor()
        
        query_employees = """
        CREATE TABLE IF NOT EXISTS employees (
            employeenumber INTEGER AUTO_INCREMENT PRIMARY KEY,
            age INT NOT NULL,
            gender VARCHAR(255),
            maritalstatus VARCHAR(255),
            educationfield VARCHAR(255),
            datebirth INT,
            department VARCHAR(255),
            jobrole VARCHAR(255),
            joblevel VARCHAR(255),
            attrition VARCHAR(255),
            dailyrate VARCHAR(255),
            environmentsatisfaction VARCHAR(255),
            percentsalaryhike INT,
            performancerating FLOAT,
            totalworkingyears INT,
            yearsatcompany INT
        )"""
        
        mycursor.execute(query_employees)
        cnx.commit()
        print("Tabla employees creada correctamente.")
        
    except mysql.connector.Error as err:
        print(f"Error Code {err.errno}")
        print(f"SQLSTATE {err.sqlstate}")
        print(f"MESSAGE {err.msg}")
    
    finally:
        if 'mycursor' in locals() and mycursor:
            mycursor.close()
        if 'cnx' in locals() and cnx and cnx.is_connected():
            cnx.close()
            print("Conexión cerrada.")

# Función para cargar datos
def cargar_datos_a_tabla(user, password, host, database, table_name, ruta):
    """Carga datos desde un archivo CSV en una tabla MySQL."""
    try:
        cnx = mysql.connector.connect(user=user, password=password, host=host, database=database)
        mycursor = cnx.cursor()
        df = pd.read_csv(ruta)
        columns = ", ".join(df.columns)
        values_placeholder = ", ".join(["%s"] * len(df.columns))
        query_insert = f"INSERT INTO {table_name} ({columns}) VALUES ({values_placeholder})"
        for _, row in df.iterrows():
            mycursor.execute(query_insert, tuple(row))
        cnx.commit()
        print(f"Datos cargados en la tabla {table_name} exitosamente.")
    except mysql.connector.Error as err:
        print(f"Error Code {err.errno}")
        print(f"SQLSTATE {err.sqlstate}")
        print(f"MESSAGE {err.msg}")
    finally:
        if cnx.is_connected():
            cnx.close()
            print("Conexión cerrada.")            


# =====================
# PROCESO ETL COMPLETO COMO FUNCIÓN
# =====================

def proceso_etl(user, password, host, database, table_name, ruta):
    
    # EXTRACCIÓN: Cargar datos desde CSV
    print("\n Cargando el dataset...")
    df = explorar_dataset(ruta)

    # ANÁLISIS DE NULOS
    print("\n Análisis de valores nulos antes de la transformación:")
    print(analizar_nulos(df))

    # ELIMINACIÓN DE COLUMNAS IRRELEVANTES
    columnas_a_eliminar = ['sameasmonthlyincome', 'roledepartament', 'over18', 
                           'numberchildren', 'salary', 'employeecount', 
                           'standardhours', 'yearsincurrentrole']
    df = eliminar_columnas(df, columnas_a_eliminar)

    # IMPUTACIÓN DE NULOS EN VARIABLES CATEGÓRICAS
    columnas_moda = ["businesstravel", "department", "overtime"]
    df = imputar_nulos_moda(df, columnas_moda)

    # ESTANDARIZACIÓN DE NOMBRES DE COLUMNAS
    df = estandarizar_columnas(df)

    # TRANSFORMACIONES ESPECÍFICAS
    df = transformar_datos(df)

    # IMPUTACIÓN DE NULOS EN VARIABLES NUMÉRICAS
    df = imputacion_nulos(df)

    # CREACIÓN Y CARGA EN MYSQL
    crear_bbdd(user, password, host, database)
    crear_tabla_empleados(user, password, host, database)
    cargar_datos_a_tabla(user, password, host, database, table_name, ruta)

    print("\n Proceso ETL finalizado con éxito.")
    
    return df  # Devuelve el DataFrame transformado para verificación