diqueran
ad83be69f8
Parte del proyecto donde extraemos los datos de los datasets descargados para tomar decisiones y visualizar las columnas sobre las que vamos a trabajar.
138 lines
4.6 KiB
Python
138 lines
4.6 KiB
Python
import os
|
|
import pandas as pd
|
|
import mysql.connector
|
|
import json
|
|
import csv
|
|
import re
|
|
|
|
|
|
# ------------------------------------------------------------
|
|
# CONFIGURACIÓN DE LA BASE DE DATOS
|
|
# ------------------------------------------------------------
|
|
DB_CONFIG = {
|
|
"host": "localhost",
|
|
"port": 3307, # Puerto diferente al 3306 para evitar conflictos
|
|
"user": "user", # Usuario definido en docker-compose
|
|
"password": "pass", # Contraseña definida en docker-compose
|
|
"database": "adicciones"
|
|
}
|
|
|
|
DATASETS_DIR = "datasets" # Carpeta donde están los CSV
|
|
OUTPUT_FILE = "columnas_info.json" # Archivo de salida con las columnas detectadas
|
|
|
|
|
|
# ------------------------------------------------------------
|
|
# FUNCIONES AUXILIARES
|
|
# ------------------------------------------------------------
|
|
|
|
def get_connection():
|
|
# Establece conexión con la base de datos MySQL
|
|
return mysql.connector.connect(**DB_CONFIG)
|
|
|
|
|
|
def sanitize_name(name: str) -> str:
|
|
# Limpia los nombres de columnas y tablas para que sean válidos en SQL
|
|
name = name.strip().lower()
|
|
name = re.sub(r"[^a-z0-9_]+", "_", name) # Solo permite letras, números y guiones bajos
|
|
return name[:50] # Limita a 50 caracteres para cumplir con el límite de MySQL
|
|
|
|
|
|
def detect_delimiter(file_path):
|
|
# Detecta automáticamente el delimitador más probable del CSV
|
|
with open(file_path, 'r', encoding='latin-1', errors='ignore') as f:
|
|
sample = f.read(2048)
|
|
try:
|
|
dialect = csv.Sniffer().sniff(sample, delimiters=";,|\t")
|
|
delim = dialect.delimiter
|
|
|
|
# Si detecta algo raro (espacios, strings largos, etc.), usamos ';' por defecto
|
|
if delim.strip() == "" or len(delim) > 1:
|
|
return ";"
|
|
return delim
|
|
|
|
except csv.Error:
|
|
return ";"
|
|
|
|
|
|
def load_csv(path, delimiter):
|
|
# Intenta leer el CSV usando varias codificaciones comunes en archivos en español
|
|
encodings = ["utf-8", "latin-1", "cp1252"]
|
|
last_error = None
|
|
|
|
for enc in encodings:
|
|
try:
|
|
return pd.read_csv(path, delimiter=delimiter, encoding=enc)
|
|
except Exception as e:
|
|
last_error = e
|
|
|
|
# Si todas las lecturas fallan, lanza el último error
|
|
raise last_error
|
|
|
|
|
|
# ------------------------------------------------------------
|
|
# PROCESO PRINCIPAL
|
|
# ------------------------------------------------------------
|
|
|
|
def main():
|
|
conn = get_connection()
|
|
cursor = conn.cursor()
|
|
columnas_info = {}
|
|
|
|
# Recorre todos los archivos CSV en la carpeta datasets/
|
|
for filename in os.listdir(DATASETS_DIR):
|
|
if not filename.endswith(".csv"):
|
|
continue
|
|
|
|
path = os.path.join(DATASETS_DIR, filename)
|
|
table_name = sanitize_name(os.path.splitext(filename)[0])
|
|
|
|
print(f"\n📊 Procesando: {filename} → tabla '{table_name}'")
|
|
|
|
# Detecta el delimitador del archivo
|
|
delimiter = detect_delimiter(path)
|
|
print(f" → delimitador detectado: '{delimiter}'")
|
|
|
|
# Carga el CSV probando codificaciones (UTF-8, Latin-1, CP1252)
|
|
try:
|
|
df = load_csv(path, delimiter)
|
|
except Exception as e:
|
|
print(f"❌ Error leyendo {filename}: {e}")
|
|
continue
|
|
|
|
# Limpia y normaliza los nombres de columnas
|
|
df.columns = [sanitize_name(c) for c in df.columns]
|
|
columnas_info[table_name] = list(df.columns)
|
|
|
|
# Crea la tabla en MySQL si no existe
|
|
columns_sql = ", ".join([f"`{c}` TEXT" for c in df.columns])
|
|
create_sql = f"CREATE TABLE IF NOT EXISTS `{table_name}` ({columns_sql});"
|
|
cursor.execute(create_sql)
|
|
|
|
# Inserta cada fila en la tabla
|
|
for _, row in df.iterrows():
|
|
placeholders = ", ".join(["%s"] * len(df.columns))
|
|
insert_sql = f"INSERT INTO `{table_name}` VALUES ({placeholders});"
|
|
try:
|
|
cursor.execute(insert_sql, tuple(row.astype(str).fillna("").values))
|
|
except Exception as e:
|
|
print(f"⚠️ Error insertando fila en {table_name}: {e}")
|
|
|
|
# Confirma los cambios en la base de datos
|
|
conn.commit()
|
|
|
|
# Cierra la conexión y guarda la información de columnas
|
|
cursor.close()
|
|
conn.close()
|
|
|
|
with open(OUTPUT_FILE, "w", encoding="utf-8") as f:
|
|
json.dump(columnas_info, f, indent=4, ensure_ascii=False)
|
|
|
|
print(f"\n✅ Proceso completado. Columnas guardadas en '{OUTPUT_FILE}'")
|
|
|
|
|
|
# ------------------------------------------------------------
|
|
# EJECUCIÓN
|
|
# ------------------------------------------------------------
|
|
if __name__ == "__main__":
|
|
main()
|