Update app.py

app.py

@@ -1,13 +1,10 @@
 import os
 import uuid
-import sqlite3
 import io
-import csv
 import zipfile
 import re
 import difflib
 import tempfile
-import shutil
 from typing import List, Optional, Dict, Any
 
 from fastapi import FastAPI, UploadFile, File, HTTPException, Form
@@ -20,6 +17,9 @@ from langdetect import detect
 from transformers import MarianMTModel, MarianTokenizer
 from openai import OpenAI
 
+import psycopg2
+import mysql.connector
+
 # ======================================================
 # 0) Configuración general
 # ======================================================
@@ -28,12 +28,24 @@ from openai import OpenAI
 MODEL_DIR = os.getenv("MODEL_DIR", "stvnnnnnn/t5-large-nl2sql-spider")
 DEVICE = torch.device("cpu")  # inferencia en CPU
 
-# Directorio donde se guardan las BDs convertidas a SQLite
-UPLOAD_DIR = os.getenv("UPLOAD_DIR", "uploaded_dbs")
-os.makedirs(UPLOAD_DIR, exist_ok=True)
-
-# Registro en memoria de conexiones (todas terminan siendo SQLite)
-# { conn_id: { "db_path": str, "label": str } }
+# === Motores reales: variables de entorno ===
+# PostgreSQL: usaremos UN solo DB (POSTGRES_DB) y un schema por conexión lógica
+POSTGRES_HOST = os.getenv("POSTGRES_HOST", "localhost")
+POSTGRES_PORT = int(os.getenv("POSTGRES_PORT", "5432"))
+POSTGRES_USER = os.getenv("POSTGRES_USER", "postgres")
+POSTGRES_PASSWORD = os.getenv("POSTGRES_PASSWORD", "postgres")
+POSTGRES_DB = os.getenv("POSTGRES_DB", "postgres")
+
+# MySQL: crearemos una base de datos por conexión lógica
+MYSQL_HOST = os.getenv("MYSQL_HOST", "localhost")
+MYSQL_PORT = int(os.getenv("MYSQL_PORT", "3306"))
+MYSQL_USER = os.getenv("MYSQL_USER", "root")
+MYSQL_PASSWORD = os.getenv("MYSQL_PASSWORD", "root")
+
+# Registro en memoria de conexiones:
+# { conn_id: { "engine": "postgresql"|"mysql", "namespace": str, "label": str } }
+#   - engine = motor real
+#   - namespace = schema (Postgres) o database (MySQL) donde vive esa BD
 DB_REGISTRY: Dict[str, Dict[str, Any]] = {}
 
 # Cliente OpenAI para transcripción de audio (Whisper / gpt-4o-transcribe)
@@ -47,13 +59,13 @@ openai_client = OpenAI(api_key=OPENAI_API_KEY) if OPENAI_API_KEY else None
 # ======================================================
 
 app = FastAPI(
-    title="NL2SQL T5-large Backend Universal (single-file)",
+    title="NL2SQL T5-large Backend (MySQL/PostgreSQL)",
     description=(
         "Intérprete NL→SQL (T5-large Spider) para usuarios no expertos. "
-        "El usuario solo sube su BD (SQLite / dump .sql / CSV / ZIP de datos) "
-        "y todo se convierte internamente a SQLite."
+        "El usuario sube dumps .sql / .zip y se cargan en motores reales "
+        "(MySQL/PostgreSQL)."
     ),
-    version="1.0.0",
+    version="2.0.0",
 )
 
 app.add_middleware(
@@ -125,342 +137,271 @@ def translate_es_to_en(text: str) -> str:
 
 
 # ======================================================
-# 3) Utilidades de BDs: creación/ingesta a SQLite
+# 3) Conexiones a motores reales y helpers
 # ======================================================
 
-def _sanitize_identifier(name: str) -> str:
-    """Hace un nombre de tabla/columna seguro para SQLite."""
-    base = name.strip().replace(" ", "_")
-    base = re.sub(r"[^0-9a-zA-Z_]", "_", base)
-    if not base:
-        base = "table"
-    if base[0].isdigit():
-        base = "_" + base
-    return base
+def get_pg_conn():
+    return psycopg2.connect(
+        host=POSTGRES_HOST,
+        port=POSTGRES_PORT,
+        user=POSTGRES_USER,
+        password=POSTGRES_PASSWORD,
+        dbname=POSTGRES_DB,
+    )
 
 
-def create_empty_sqlite_db(label: str) -> str:
-    """Crea un archivo .sqlite vacío y lo devuelve."""
-    conn_id = f"db_{uuid.uuid4().hex[:8]}"
-    db_filename = f"{conn_id}.sqlite"
-    db_path = os.path.join(UPLOAD_DIR, db_filename)
-    conn = sqlite3.connect(db_path)
-    conn.close()
-    DB_REGISTRY[conn_id] = {"db_path": db_path, "label": label}
-    return conn_id
+def get_mysql_conn(db_name: Optional[str] = None):
+    params = dict(
+        host=MYSQL_HOST,
+        port=MYSQL_PORT,
+        user=MYSQL_USER,
+        password=MYSQL_PASSWORD,
+    )
+    if db_name:
+        params["database"] = db_name
+    return mysql.connector.connect(**params)
 
 
-def import_sql_dump_to_sqlite(db_path: str, sql_text: str) -> None:
+def detect_sql_dialect(sql_text: str) -> str:
     """
-    Convertidor avanzado MySQL → SQLite.
-    Limpia, reordena y ejecuta el schema de forma segura en SQLite.
-    Si al final no se crea ninguna tabla, lanza ValueError.
+    Heurística simple:
+    - Si ve ENGINE, AUTO_INCREMENT, backticks, etc. → MySQL
+    - Si ve SERIAL, search_path, ::, PL/pgSQL, etc. → PostgreSQL
     """
+    text = sql_text.lower()
+    if any(kw in text for kw in ["engine=", "auto_increment", "unsigned", " collate ", " character set ", "`"]):
+        return "mysql"
+    if any(kw in text for kw in ["serial", " set search_path", "copy ", "::", "language plpgsql"]):
+        return "postgresql"
+    return "unknown"
 
-    # ======================================================
-    # 1) Limpieza inicial del dump
-    # ======================================================
-
-    # Remover comentarios estilo MySQL/PostgreSQL
-    sql_text = re.sub(r"/\*![\s\S]*?\*/;", "", sql_text)
-    sql_text = re.sub(r"/\*[\s\S]*?\*/", "", sql_text)
-    sql_text = re.sub(r"--.*?\n", "", sql_text)
-
-    # Remover DELIMITER y statements SET típicos
-    sql_text = re.sub(r"DELIMITER\s+.+", "", sql_text)
-    sql_text = re.sub(r"SET\s+[^;]+;", "", sql_text)
-
-    # Quitar ENGINE, ROW_FORMAT, AUTO_INCREMENT
-    sql_text = re.sub(r"ENGINE=\w+", "", sql_text)
-    sql_text = re.sub(r"ROW_FORMAT=\w+", "", sql_text)
-    sql_text = re.sub(r"AUTO_INCREMENT=\d+", "", sql_text)
-
-    # Quitar COLLATE y CHARSET
-    sql_text = re.sub(r"DEFAULT CHARSET=\w+", "", sql_text)
-    sql_text = re.sub(r"CHARACTER SET \w+", "", sql_text)
-    sql_text = re.sub(r"COLLATE \w+", "", sql_text)
-
-    # Quitar dominios y tipos personalizados (PostgreSQL)
-    sql_text = re.sub(r"CREATE\s+DOMAIN[\s\S]+?;", "", sql_text, flags=re.IGNORECASE)
-    sql_text = re.sub(r"CREATE\s+TYPE[\s\S]+?;", "", sql_text, flags=re.IGNORECASE)
-
-    # Reemplazar backticks por nada (MySQL → SQLite)
-    sql_text = sql_text.replace("`", "")
 
-    # ======================================================
-    # 2) Dividir en statements individuales
-    # ======================================================
-    raw_statements = sql_text.split(";")
-
-    # Tablas para ejecutar CREATE TABLE sin foreign keys primero
-    create_tables = []
-    foreign_keys = []
-    inserts = []
-    others = []
-
-    for st in raw_statements:
-        stmt = st.strip()
-        if not stmt:
-            continue
-
-        upper = stmt.upper()
-
-        if upper.startswith("CREATE TABLE"):
-            # separar claves foráneas
-            if "FOREIGN KEY" in upper:
-                fixed = []
-                fk_lines = []
-
-                for line in stmt.split("\n"):
-                    if "FOREIGN KEY" in line.upper():
-                        fk_lines.append(line.strip().rstrip(","))
-                    else:
-                        fixed.append(line)
-
-                table_sql = "\n".join(fixed)
-                create_tables.append(table_sql)
-                foreign_keys.append((extract_table_name(stmt), fk_lines))
-
-            else:
-                create_tables.append(stmt)
-
-        elif upper.startswith("INSERT INTO"):
-            inserts.append(stmt)
+def create_logical_db(engine: str, label: str) -> str:
+    """
+    Crea una "BD lógica":
+      - En PostgreSQL: un SCHEMA nuevo dentro de POSTGRES_DB
+      - En MySQL: una DATABASE nueva
+    """
+    conn_id = f"db_{uuid.uuid4().hex[:8]}"
 
-        else:
-            others.append(stmt)
-
-    # ======================================================
-    # 3) Convertir tipos MySQL → SQLite
-    # ======================================================
-
-    def convert_types(sql: str) -> str:
-        sql = re.sub(r"\bTINYINT\(1\)\b", "INTEGER", sql)
-        sql = re.sub(r"\bINT\b|\bINTEGER\b", "INTEGER", sql)
-        sql = re.sub(r"\bBIGINT\b", "INTEGER", sql)
-        sql = re.sub(r"\bDECIMAL\([0-9,]+\)", "REAL", sql)
-        sql = re.sub(r"\bDOUBLE\b|\bFLOAT\b", "REAL", sql)
-        sql = re.sub(r"\bDATETIME\b|\bTIMESTAMP\b", "TEXT", sql)
-        sql = re.sub(r"\bVARCHAR\([0-9]+\)", "TEXT", sql)
-        sql = re.sub(r"\bCHAR\([0-9]+\)", "TEXT", sql)
-        sql = re.sub(r"\bTEXT\b", "TEXT", sql)
-        sql = re.sub(r"\bUNSIGNED\b", "", sql)
-        return sql
-
-    create_tables = [convert_types(c) for c in create_tables]
-    inserts = [convert_types(i) for i in inserts]
-
-    # ======================================================
-    # 4) Ejecutar en orden
-    # ======================================================
-    conn = sqlite3.connect(db_path)
-    cur = conn.cursor()
-
-    cur.execute("PRAGMA foreign_keys = OFF;")
-
-    for ct in create_tables:
+    if engine == "postgresql":
+        schema_name = conn_id
+        conn = get_pg_conn()
         try:
-            cur.executescript(ct + ";")
-        except Exception as e:
-            print("Error CREATE TABLE:", e)
-            print("SQL:", ct)
-
-    for ins in inserts:
+            conn.autocommit = True
+            cur = conn.cursor()
+            cur.execute(f'CREATE SCHEMA IF NOT EXISTS "{schema_name}";')
+            cur.close()
+        finally:
+            conn.close()
+        DB_REGISTRY[conn_id] = {"engine": "postgresql", "namespace": schema_name, "label": label}
+
+    elif engine == "mysql":
+        db_name = conn_id
+        conn = get_mysql_conn()
         try:
-            cur.executescript(ins + ";")
-        except Exception as e:
-            print("Error INSERT:", e)
-            print("SQL:", ins)
-
-    # ======================================================
-    # 5) Reconstruir claves foráneas manualmente
-    # ======================================================
-
-    for table, fks in foreign_keys:
-        for fk in fks:
-            try:
-                add_foreign_key_sqlite(conn, table, fk)
-            except Exception as e:
-                print("Error agregando FK:", e, " → ", fk)
-
-    cur.execute("PRAGMA foreign_keys = ON;")
-    conn.commit()
-
-    # ======================================================
-    # 6) Validar que realmente se crearon tablas
-    # ======================================================
-    cur.execute("SELECT COUNT(*) FROM sqlite_master WHERE type='table';")
-    n_tables = cur.fetchone()[0]
-    conn.close()
-
-    if n_tables == 0:
-        raise ValueError(
-            "No se creó ninguna tabla desde el dump SQL. "
-            "Probablemente es un dump de otro motor (por ejemplo PostgreSQL) "
-            "o usa sintaxis no compatible con SQLite."
-        )
-
+            cur = conn.cursor()
+            cur.execute(f"CREATE DATABASE IF NOT EXISTS `{db_name}`;")
+            conn.commit()
+            cur.close()
+        finally:
+            conn.close()
+        DB_REGISTRY[conn_id] = {"engine": "mysql", "namespace": db_name, "label": label}
 
-def extract_table_name(create_stmt: str) -> str:
-    m = re.search(r"CREATE TABLE\s+(\w+)", create_stmt, re.IGNORECASE)
-    return m.group(1) if m else "unknown"
+    else:
+        raise ValueError(f"Engine no soportado: {engine}")
 
+    return conn_id
 
-def add_foreign_key_sqlite(conn, table: str, fk_line: str):
-    """
-    Reconstrucción automática:
-    - Lee schema actual
-    - Añade FK en nueva versión
-    - Copia datos
-    """
-    cur = conn.cursor()
 
-    cur.execute(f"SELECT sql FROM sqlite_master WHERE type='table' AND name='{table}';")
-    result = cur.fetchone()
-    if not result:
-        return
+def ensure_connection(conn_id: str) -> Dict[str, Any]:
+    if conn_id not in DB_REGISTRY:
+        raise HTTPException(status_code=404, detail=f"connection_id '{conn_id}' no registrado")
+    return DB_REGISTRY[conn_id]
 
-    original_sql = result[0]
-    new_sql = original_sql.rstrip(")") + f", {fk_line} )"
 
-    cur.execute(f"ALTER TABLE {table} RENAME TO _old_{table};")
-    cur.execute(new_sql)
-    cur.execute(f"INSERT INTO {table} SELECT * FROM _old_{table};")
-    cur.execute(f"DROP TABLE _old_{table};")
+# ======================================================
+# 4) Carga de scripts SQL (schema.sql / data.sql / ZIP)
+# ======================================================
 
-    conn.commit()
+def _execute_sql_script_postgres(schema: str, sql_text: str) -> None:
+    conn = get_pg_conn()
+    try:
+        conn.autocommit = False
+        cur = conn.cursor()
+        # Trabajamos siempre dentro del schema lógico
+        cur.execute(f'SET search_path TO "{schema}";')
+        # Ejecución simple por ';' (suficiente para Sakila/Pagila)
+        parts = sql_text.split(";")
+        for stmt in parts:
+            s = stmt.strip()
+            if not s:
+                continue
+            cur.execute(s + ";")
+        conn.commit()
+        cur.close()
+    finally:
+        conn.close()
 
 
-def import_csv_to_sqlite(db_path: str, csv_bytes: bytes, table_name: str) -> None:
-    """
-    Crea una tabla en SQLite con columnas TEXT y carga datos desde un CSV.
-    """
-    table = _sanitize_identifier(table_name or "data")
-    conn = sqlite3.connect(db_path)
+def _execute_sql_script_mysql(db_name: str, sql_text: str) -> None:
+    conn = get_mysql_conn()
     try:
-        f = io.StringIO(csv_bytes.decode("utf-8", errors="ignore"))
-        reader = csv.reader(f)
-        rows = list(reader)
-
-        if not rows:
-            return
-
-        header = rows[0]
-        cols = [_sanitize_identifier(c or f"col_{i}") for i, c in enumerate(header)]
-
-        col_defs = ", ".join(f'"{c}" TEXT' for c in cols)
-        conn.execute(f'CREATE TABLE IF NOT EXISTS "{table}" ({col_defs});')
-
-        placeholders = ", ".join(["?"] * len(cols))
-        for row in rows[1:]:
-            row = list(row) + [""] * (len(cols) - len(row))
-            row = row[:len(cols)]
-            conn.execute(
-                f'INSERT INTO "{table}" ({", ".join(cols)}) VALUES ({placeholders})',
-                row,
-            )
-
+        cur = conn.cursor()
+        cur.execute(f"USE `{db_name}`;")
+        # mysql-connector permite multi=True
+        for _ in cur.execute(sql_text, multi=True):
+            pass
         conn.commit()
+        cur.close()
     finally:
         conn.close()
 
 
-def import_zip_of_csvs_to_sqlite(db_path: str, zip_bytes: bytes) -> None:
+def load_sql_into_connection(sql_text: str, engine: str, conn_id: Optional[str], label: str) -> str:
     """
-    Para un ZIP con múltiples CSV: cada CSV se vuelve una tabla.
-    (Se mantiene por compatibilidad, aunque ahora manejamos ZIPs
-     más generales en /upload.)
+    - Si conn_id es None → crea nueva BD lógica y carga el script.
+    - Si conn_id existe → ejecuta el script encima (ej: primero schema.sql, luego data.sql).
     """
-    conn = sqlite3.connect(db_path)
-    conn.close()
+    if not engine or engine not in ("mysql", "postgresql"):
+        raise HTTPException(status_code=400, detail="Engine inválido o no soportado (usa 'mysql' o 'postgresql').")
 
-    with zipfile.ZipFile(io.BytesIO(zip_bytes)) as zf:
-        for name in zf.namelist():
-            if not name.lower().endswith(".csv"):
-                continue
-            with zf.open(name) as f:
-                csv_bytes = f.read()
-            base_name = os.path.basename(name)
-            table_name = os.path.splitext(base_name)[0]
-            import_csv_to_sqlite(db_path, csv_bytes, table_name)
+    if conn_id is None:
+        conn_id = create_logical_db(engine, label)
+    else:
+        ensure_connection(conn_id)
 
+    info = DB_REGISTRY[conn_id]
+    namespace = info["namespace"]
 
-# ======================================================
-# 4) Introspección de esquema y ejecución (sobre SQLite)
-# ======================================================
+    if engine == "postgresql":
+        _execute_sql_script_postgres(namespace, sql_text)
+    else:
+        _execute_sql_script_mysql(namespace, sql_text)
 
-def introspect_sqlite_schema(db_path: str) -> Dict[str, Any]:
-    if not os.path.exists(db_path):
-        raise FileNotFoundError(f"SQLite no encontrado: {db_path}")
+    return conn_id
 
-    conn = sqlite3.connect(db_path)
-    cur = conn.cursor()
 
-    cur.execute("SELECT name FROM sqlite_master WHERE type='table';")
-    tables = [row[0] for row in cur.fetchall()]
+# ======================================================
+# 5) Introspección de esquema y ejecución (sobre motores reales)
+# ======================================================
 
-    tables_info = {}
-    foreign_keys = []
-    parts = []
+def introspect_schema(conn_id: str) -> Dict[str, Any]:
+    info = ensure_connection(conn_id)
+    engine = info["engine"]
+    ns = info["namespace"]
 
-    for t in tables:
-        cur.execute(f"PRAGMA table_info('{t}');")
-        rows = cur.fetchall()
-        cols = [r[1] for r in rows]
-        tables_info[t] = {"columns": cols}
+    tables_info: Dict[str, Dict[str, List[str]]] = {}
+    parts: List[str] = []
 
-        cur.execute(f"PRAGMA foreign_key_list('{t}');")
-        fks = cur.fetchall()
-        for (id, seq, table, from_col, to_col, on_update, on_delete, match) in fks:
-            foreign_keys.append({
-                "from_table": t,
-                "from_column": from_col,
-                "to_table": table,
-                "to_column": to_col
-            })
+    if engine == "postgresql":
+        conn = get_pg_conn()
+        try:
+            cur = conn.cursor()
+            cur.execute(
+                """
+                SELECT table_name, column_name
+                FROM information_schema.columns
+                WHERE table_schema = %s
+                ORDER BY table_name, ordinal_position;
+                """,
+                (ns,),
+            )
+            rows = cur.fetchall()
+            cur.close()
+        finally:
+            conn.close()
+        for table, col in rows:
+            tables_info.setdefault(table, {"columns": []})
+            tables_info[table]["columns"].append(col)
+
+    else:  # MySQL
+        conn = get_mysql_conn(ns)
+        try:
+            cur = conn.cursor()
+            cur.execute(
+                """
+                SELECT table_name, column_name
+                FROM information_schema.columns
+                WHERE table_schema = %s
+                ORDER BY table_name, ordinal_position;
+                """,
+                (ns,),
+            )
+            rows = cur.fetchall()
+            cur.close()
+        finally:
+            conn.close()
+        for table, col in rows:
+            tables_info.setdefault(table, {"columns": []})
+            tables_info[table]["columns"].append(col)
 
-        parts.append(f"{t}(" + ", ".join(cols) + ")")
+    for t, info_t in tables_info.items():
+        parts.append(f"{t}(" + ", ".join(info_t["columns"]) + ")")
 
-    conn.close()
     schema_str = " ; ".join(parts) if parts else "(empty_schema)"
 
     return {
         "tables": tables_info,
-        "foreign_keys": foreign_keys,
-        "schema_str": schema_str
+        "foreign_keys": [],    # podrías enriquecer esto luego
+        "schema_str": schema_str,
     }
 
 
-def execute_sqlite(db_path: str, sql: str) -> Dict[str, Any]:
-    forbidden = ["drop ", "delete ", "update ", "insert ", "alter ", "replace "]
+def execute_sql(conn_id: str, sql: str) -> Dict[str, Any]:
+    """
+    Ejecuta SOLO consultas de lectura (SELECT).
+    Bloquea operaciones destructivas por seguridad en el demo.
+    """
+    info = ensure_connection(conn_id)
+    engine = info["engine"]
+    ns = info["namespace"]
+
+    forbidden = ["drop ", "delete ", "update ", "insert ", "alter ", "replace ", "truncate ", "create "]
     sql_low = sql.lower()
     if any(f in sql_low for f in forbidden):
         return {
             "ok": False,
-            "error": "Query bloqueada por seguridad (operación destructiva).",
+            "error": "Query bloqueada por seguridad (operación potencialmente destructiva).",
             "rows": None,
-            "columns": []
+            "columns": [],
         }
 
     try:
-        conn = sqlite3.connect(db_path)
-        cur = conn.cursor()
-        cur.execute(sql)
-        rows = cur.fetchall()
-        col_names = [desc[0] for desc in cur.description] if cur.description else []
-        conn.close()
-        return {"ok": True, "error": None, "rows": rows, "columns": col_names}
+        if engine == "postgresql":
+            conn = get_pg_conn()
+            try:
+                cur = conn.cursor()
+                cur.execute(f'SET search_path TO "{ns}";')
+                cur.execute(sql)
+                rows = cur.fetchall()
+                cols = [desc[0] for desc in cur.description] if cur.description else []
+                cur.close()
+            finally:
+                conn.close()
+        else:
+            conn = get_mysql_conn(ns)
+            try:
+                cur = conn.cursor()
+                cur.execute(sql)
+                rows = cur.fetchall()
+                cols = [desc[0] for desc in cur.description] if cur.description else []
+                cur.close()
+            finally:
+                conn.close()
+
+        return {"ok": True, "error": None, "rows": rows, "columns": cols}
     except Exception as e:
         return {"ok": False, "error": str(e), "rows": None, "columns": []}
 
 
 # ======================================================
-# 4.1) SQL REPAIR LAYER (avanzado)
+# 6) SQL REPAIR LAYER (igual que antes, pero agnóstico de motor)
 # ======================================================
 
 def _normalize_name_for_match(name: str) -> str:
     s = name.lower()
-    s = s.replace('"', '').replace("`", "")
+    s = s.replace('"', "").replace("`", "")
     s = s.replace("_", "")
     if s.endswith("s") and len(s) > 3:
         s = s[:-1]
@@ -489,11 +430,9 @@ def _build_schema_indexes(tables_info: Dict[str, Dict[str, List[str]]]) -> Dict[
 def _best_match_name(missing: str, index: Dict[str, List[str]]) -> Optional[str]:
     if not index:
         return None
-
     key = _normalize_name_for_match(missing)
     if key in index and index[key]:
         return index[key][0]
-
     candidates = difflib.get_close_matches(key, list(index.keys()), n=1, cutoff=0.7)
     if not candidates:
         return None
@@ -537,11 +476,11 @@ def try_repair_sql(sql: str, error: str, schema_meta: Dict[str, Any]) -> Optiona
     missing_table = None
     missing_column = None
 
-    m_t = re.search(r"no such table: ([\w\.]+)", error)
+    m_t = re.search(r"no such table: ([\w\.]+)", error or "", re.IGNORECASE)
     if m_t:
         missing_table = m_t.group(1)
 
-    m_c = re.search(r"no such column: ([\w\.]+)", error)
+    m_c = re.search(r"no such column: ([\w\.]+)", error or "", re.IGNORECASE)
     if m_c:
         missing_column = m_c.group(1)
 
@@ -583,7 +522,7 @@ def try_repair_sql(sql: str, error: str, schema_meta: Dict[str, Any]) -> Optiona
 
 
 # ======================================================
-# 5) Construcción de prompt y NL→SQL + re-ranking
+# 7) Construcción de prompt y NL→SQL + re-ranking
 # ======================================================
 
 def build_prompt(question_en: str, db_id: str, schema_str: str) -> str:
@@ -595,11 +534,8 @@ def build_prompt(question_en: str, db_id: str, schema_str: str) -> str:
 
 
 def nl2sql_with_rerank(question: str, conn_id: str) -> Dict[str, Any]:
-    if conn_id not in DB_REGISTRY:
-        raise HTTPException(status_code=404, detail=f"connection_id '{conn_id}' no registrado")
-
-    db_path = DB_REGISTRY[conn_id]["db_path"]
-    meta = introspect_sqlite_schema(db_path)
+    ensure_connection(conn_id)
+    meta = introspect_schema(conn_id)
     schema_str = meta["schema_str"]
 
     detected = detect_language(question)
@@ -646,19 +582,19 @@ def nl2sql_with_rerank(question: str, conn_id: str) -> Dict[str, Any]:
             "raw_sql_model": raw_sql,
         }
 
-        exec_info = execute_sqlite(db_path, raw_sql)
+        exec_info = execute_sql(conn_id, raw_sql)
 
         if (not exec_info["ok"]) and (
-            "no such table" in (exec_info["error"] or "")
-            or "no such column" in (exec_info["error"] or "")
+            "no such table" in (exec_info["error"] or "").lower()
+            or "no such column" in (exec_info["error"] or "").lower()
         ):
             current_sql = raw_sql
             last_error = exec_info["error"]
             for step in range(1, 4):
-                repaired_sql = try_repair_sql(current_sql, last_error, meta)
+                repaired_sql = try_repair_sql(current_sql, last_error or "", meta)
                 if not repaired_sql or repaired_sql == current_sql:
                     break
-                exec_info2 = execute_sqlite(db_path, repaired_sql)
+                exec_info2 = execute_sql(conn_id, repaired_sql)
                 cand["repaired_from"] = current_sql if cand["repaired_from"] is None else cand["repaired_from"]
                 cand["repair_note"] = f"auto-repair (table/column name, step {step})"
                 cand["sql"] = repaired_sql
@@ -705,29 +641,33 @@ def nl2sql_with_rerank(question: str, conn_id: str) -> Dict[str, Any]:
 
 
 # ======================================================
-# 6) Schemas Pydantic
+# 8) Schemas Pydantic
 # ======================================================
 
 class UploadResponse(BaseModel):
     connection_id: str
     label: str
-    db_path: str
+    engine: str
+    namespace: str  # schema (PG) o database (MySQL)
     note: Optional[str] = None
 
 
 class ConnectionInfo(BaseModel):
     connection_id: str
     label: str
+    engine: str
 
 
 class SchemaResponse(BaseModel):
     connection_id: str
+    engine: str
     schema_summary: str
     tables: Dict[str, Dict[str, List[str]]]
 
 
 class PreviewResponse(BaseModel):
     connection_id: str
+    engine: str
     table: str
     columns: List[str]
     rows: List[List[Any]]
@@ -758,24 +698,27 @@ class SpeechInferResponse(BaseModel):
 
 
 # ======================================================
-# 7) Endpoints FastAPI
+# 9) Endpoints FastAPI
 # ======================================================
 
 @app.on_event("startup")
 async def startup_event():
     load_nl2sql_model()
-    print(f"✅ Backend NL2SQL inicializado. MODEL_DIR={MODEL_DIR}, UPLOAD_DIR={UPLOAD_DIR}")
+    print("✅ Backend NL2SQL inicializado (motores MySQL/PostgreSQL).")
 
 
-@app.post("/upload", response_model=UploadResponse)
-async def upload_database(db_file: UploadFile = File(...)):
+@app.post("/upload-sql", response_model=UploadResponse)
+async def upload_sql(
+    db_file: UploadFile = File(...),
+    engine: Optional[str] = Form(None),        # "mysql" | "postgresql" (opcional)
+    connection_id: Optional[str] = Form(None), # para el caso schema.sql + data.sql
+):
     """
-    Subida universal de BD.
-    El usuario puede subir:
-      - .sqlite / .db → se usa tal cual
-      - .sql → dump MySQL/SQLite (best effort → convertido a SQLite)
-      - .csv → se crea una BD SQLite y una tabla
-      - .zip → puede contener .sqlite/.db, .sql o .csv (se detecta automáticamente)
+    Subida de dumps SQL.
+    - Acepta .sql directo o .zip con varios .sql
+    - Detecta motor automáticamente (MySQL/PostgreSQL) si engine no se especifica
+    - Si connection_id es None, crea una nueva BD lógica
+    - Si connection_id existe, ejecuta el SQL encima (ej: schema.sql luego data.sql)
     """
     filename = db_file.filename
     if not filename:
@@ -784,168 +727,98 @@ async def upload_database(db_file: UploadFile = File(...)):
     fname_lower = filename.lower()
     contents = await db_file.read()
 
-    note: Optional[str] = None
-    conn_id: Optional[str] = None
-
-    # Caso 1: SQLite nativa
-    if fname_lower.endswith(".sqlite") or fname_lower.endswith(".db"):
-        conn_id = f"db_{uuid.uuid4().hex[:8]}"
-        dst_path = os.path.join(UPLOAD_DIR, f"{conn_id}.sqlite")
-        with open(dst_path, "wb") as f:
-            f.write(contents)
-        DB_REGISTRY[conn_id] = {"db_path": dst_path, "label": filename}
-        note = "SQLite file stored as-is."
-
-    # Caso 2: dump .sql
-    elif fname_lower.endswith(".sql"):
-        conn_id = create_empty_sqlite_db(label=filename)
-        db_path = DB_REGISTRY[conn_id]["db_path"]
-        sql_text = contents.decode("utf-8", errors="ignore")
+    if not (fname_lower.endswith(".sql") or fname_lower.endswith(".zip")):
+        raise HTTPException(
+            status_code=400,
+            detail="Formato no soportado. Usa: .sql o .zip (con archivos .sql).",
+        )
 
-        try:
-            import_sql_dump_to_sqlite(db_path, sql_text)
-            note = "SQL dump imported into SQLite (best effort)."
-        except Exception as e:
-            # Limpiar registro porque la BD quedó inútil
-            try:
-                if os.path.exists(db_path):
-                    os.remove(db_path)
-            except Exception:
-                pass
-            DB_REGISTRY.pop(conn_id, None)
+    sql_text = ""
+    note = None
 
+    # Caso: archivo .sql único
+    if fname_lower.endswith(".sql"):
+        sql_text = contents.decode("utf-8", errors="ignore")
+        detected = detect_sql_dialect(sql_text)
+        final_engine = engine or detected
+        if final_engine == "unknown":
             raise HTTPException(
                 status_code=400,
                 detail=(
-                    "No se pudo convertir este archivo .sql a una base de datos SQLite utilizable. "
-                    "Es probable que sea un dump de otro motor (por ejemplo PostgreSQL) "
-                    "o que contenga sintaxis avanzada no compatible con SQLite.\n"
-                    f"Detalle técnico: {e}"
+                    "No se pudo detectar el motor SQL (MySQL/PostgreSQL). "
+                    "Vuelve a subir el archivo indicando engine='mysql' o 'postgresql'."
                 ),
             )
 
-    # Caso 3: CSV simple
-    elif fname_lower.endswith(".csv"):
-        conn_id = create_empty_sqlite_db(label=filename)
-        db_path = DB_REGISTRY[conn_id]["db_path"]
-        table_name = os.path.splitext(os.path.basename(filename))[0]
-        import_csv_to_sqlite(db_path, contents, table_name)
-        note = "CSV imported into a single SQLite table."
-
-    # Caso 4: ZIP universal
-    elif fname_lower.endswith(".zip"):
-        try:
-            with zipfile.ZipFile(io.BytesIO(contents)) as zf:
-                names = [info.filename for info in zf.infolist() if not info.is_dir()]
-
-                sqlite_names = [n for n in names if n.lower().endswith((".sqlite", ".db"))]
-                sql_names = [n for n in names if n.lower().endswith(".sql")]
-                csv_names = [n for n in names if n.lower().endswith(".csv")]
-
-                # 4.1: si el ZIP trae una BD SQLite nativa
-                if sqlite_names:
-                    inner = sqlite_names[0]
-                    conn_id = f"db_{uuid.uuid4().hex[:8]}"
-                    dst_path = os.path.join(UPLOAD_DIR, f"{conn_id}.sqlite")
-                    with zf.open(inner) as src, open(dst_path, "wb") as dst:
-                        shutil.copyfileobj(src, dst)
-                    DB_REGISTRY[conn_id] = {
-                        "db_path": dst_path,
-                        "label": f"{filename}::{os.path.basename(inner)}",
-                    }
-                    note = "SQLite database extracted from ZIP."
-
-                # 4.2: dumps SQL (uno o varios)
-                elif sql_names:
-                    conn_id = create_empty_sqlite_db(label=filename)
-                    db_path = DB_REGISTRY[conn_id]["db_path"]
-
-                    if len(sql_names) == 1:
-                        with zf.open(sql_names[0]) as f:
-                            sql_text = f.read().decode("utf-8", errors="ignore")
-                    else:
-                        parts = []
-                        for n in sorted(sql_names):
-                            with zf.open(n) as f:
-                                parts.append(f"-- FILE: {n}\n")
-                                parts.append(f.read().decode("utf-8", errors="ignore"))
-                        sql_text = "\n\n".join(parts)
-
-                    try:
-                        import_sql_dump_to_sqlite(db_path, sql_text)
-                        note = "SQL dump(s) from ZIP imported into SQLite."
-                    except Exception as e:
-                        try:
-                            if os.path.exists(db_path):
-                                os.remove(db_path)
-                        except Exception:
-                            pass
-                        DB_REGISTRY.pop(conn_id, None)
-
-                        raise HTTPException(
-                            status_code=400,
-                            detail=(
-                                "No se pudo convertir los archivos .sql dentro del ZIP "
-                                "a una base de datos SQLite utilizable. "
-                                "Es probable que sean dumps de otro motor (por ejemplo PostgreSQL) "
-                                "o que contengan sintaxis avanzada no compatible con SQLite.\n"
-                                f"Detalle técnico: {e}"
-                            ),
-                        )
-
-                # 4.3: solo CSVs
-                elif csv_names:
-                    conn_id = create_empty_sqlite_db(label=filename)
-                    db_path = DB_REGISTRY[conn_id]["db_path"]
-
-                    for name in csv_names:
-                        with zf.open(name) as f:
-                            csv_bytes = f.read()
-                        table_name = os.path.splitext(os.path.basename(name))[0]
-                        import_csv_to_sqlite(db_path, csv_bytes, table_name)
-
-                    note = "CSV files from ZIP imported into SQLite (one table per CSV)."
-
-                else:
-                    raise HTTPException(
-                        status_code=400,
-                        detail="El ZIP no contiene archivos .sqlite/.db/.sql/.csv utilizables.",
-                    )
-
-        except zipfile.BadZipFile:
-            raise HTTPException(status_code=400, detail="Archivo ZIP inválido o corrupto.")
+        conn_id = load_sql_into_connection(sql_text, final_engine, connection_id, filename)
+        info = DB_REGISTRY[conn_id]
+        note = f"SQL ejecutado sobre motor {final_engine}."
+        return UploadResponse(
+            connection_id=conn_id,
+            label=info["label"],
+            engine=info["engine"],
+            namespace=info["namespace"],
+            note=note,
+        )
 
-    else:
+    # Caso: ZIP con varios .sql (ej: schema.sql + data.sql, o muchos scripts)
+    try:
+        with zipfile.ZipFile(io.BytesIO(contents)) as zf:
+            sql_names = [n for n in zf.namelist() if n.lower().endswith(".sql")]
+            if not sql_names:
+                raise HTTPException(
+                    status_code=400,
+                    detail="El ZIP no contiene archivos .sql utilizables.",
+                )
+
+            combined_sql_parts = []
+            for name in sorted(sql_names):
+                with zf.open(name) as f:
+                    combined_sql_parts.append(f"-- FILE: {name}\n")
+                    combined_sql_parts.append(f.read().decode("utf-8", errors="ignore"))
+            sql_text = "\n\n".join(combined_sql_parts)
+
+    except zipfile.BadZipFile:
+        raise HTTPException(status_code=400, detail="Archivo ZIP inválido o corrupto.")
+
+    detected = detect_sql_dialect(sql_text)
+    final_engine = engine or detected
+    if final_engine == "unknown":
         raise HTTPException(
             status_code=400,
-            detail="Formato no soportado. Usa: .sqlite, .db, .sql, .csv o .zip",
+            detail=(
+                "No se pudo detectar el motor SQL (MySQL/PostgreSQL) en el ZIP. "
+                "Vuelve a subir indicando engine='mysql' o 'postgresql'."
+            ),
         )
 
+    conn_id = load_sql_into_connection(sql_text, final_engine, connection_id, filename)
+    info = DB_REGISTRY[conn_id]
+    note = f"ZIP con scripts SQL ejecutado sobre motor {final_engine}."
     return UploadResponse(
         connection_id=conn_id,
-        label=DB_REGISTRY[conn_id]["label"],
-        db_path=DB_REGISTRY[conn_id]["db_path"],
+        label=info["label"],
+        engine=info["engine"],
+        namespace=info["namespace"],
         note=note,
     )
 
 
 @app.get("/connections", response_model=List[ConnectionInfo])
 async def list_connections():
-    out = []
+    out: List[ConnectionInfo] = []
     for cid, info in DB_REGISTRY.items():
-        out.append(ConnectionInfo(connection_id=cid, label=info["label"]))
+        out.append(ConnectionInfo(connection_id=cid, label=info["label"], engine=info["engine"]))
     return out
 
 
 @app.get("/schema/{connection_id}", response_model=SchemaResponse)
 async def get_schema(connection_id: str):
-    if connection_id not in DB_REGISTRY:
-        raise HTTPException(status_code=404, detail="connection_id no encontrado")
-
-    db_path = DB_REGISTRY[connection_id]["db_path"]
-    meta = introspect_sqlite_schema(db_path)
+    info = ensure_connection(connection_id)
+    meta = introspect_schema(connection_id)
     return SchemaResponse(
         connection_id=connection_id,
+        engine=info["engine"],
         schema_summary=meta["schema_str"],
         tables=meta["tables"],
     )
@@ -953,22 +826,38 @@ async def get_schema(connection_id: str):
 
 @app.get("/preview/{connection_id}/{table}", response_model=PreviewResponse)
 async def preview_table(connection_id: str, table: str, limit: int = 20):
-    if connection_id not in DB_REGISTRY:
-        raise HTTPException(status_code=404, detail="connection_id no encontrado")
+    info = ensure_connection(connection_id)
+    engine = info["engine"]
+    ns = info["namespace"]
 
-    db_path = DB_REGISTRY[connection_id]["db_path"]
     try:
-        conn = sqlite3.connect(db_path)
-        cur = conn.cursor()
-        cur.execute(f'SELECT * FROM "{table}" LIMIT {int(limit)};')
-        rows = cur.fetchall()
-        cols = [d[0] for d in cur.description] if cur.description else []
-        conn.close()
+        if engine == "postgresql":
+            conn = get_pg_conn()
+            try:
+                cur = conn.cursor()
+                cur.execute(f'SET search_path TO "{ns}";')
+                cur.execute(f'SELECT * FROM "{table}" LIMIT %s;', (int(limit),))
+                rows = cur.fetchall()
+                cols = [d[0] for d in cur.description] if cur.description else []
+                cur.close()
+            finally:
+                conn.close()
+        else:
+            conn = get_mysql_conn(ns)
+            try:
+                cur = conn.cursor()
+                cur.execute(f"SELECT * FROM `{table}` LIMIT %s;", (int(limit),))
+                rows = cur.fetchall()
+                cols = [d[0] for d in cur.description] if cur.description else []
+                cur.close()
+            finally:
+                conn.close()
     except Exception as e:
         raise HTTPException(status_code=400, detail=f"Error al leer tabla '{table}': {e}")
 
     return PreviewResponse(
         connection_id=connection_id,
+        engine=engine,
         table=table,
         columns=cols,
         rows=[list(r) for r in rows],
@@ -984,12 +873,12 @@ async def infer_sql(req: InferRequest):
 @app.post("/speech-infer", response_model=SpeechInferResponse)
 async def speech_infer(
     connection_id: str = Form(...),
-    audio: UploadFile = File(...)
+    audio: UploadFile = File(...),
 ):
     if openai_client is None:
         raise HTTPException(
             status_code=500,
-            detail="OPENAI_API_KEY no está configurado en el backend."
+            detail="OPENAI_API_KEY no está configurado en el backend.",
         )
 
     if audio.content_type is None:
@@ -1027,6 +916,7 @@ async def health():
         "status": "ok",
         "model_loaded": t5_model is not None,
         "connections": len(DB_REGISTRY),
+        "engines_in_use": list({info["engine"] for info in DB_REGISTRY.values()}),
         "device": str(DEVICE),
     }
 
@@ -1034,13 +924,13 @@ async def health():
 @app.get("/")
 async def root():
     return {
-        "message": "NL2SQL T5-large universal backend is running (single-file SQLite engine).",
+        "message": "NL2SQL T5-large backend is running (MySQL/PostgreSQL engines).",
         "endpoints": [
-            "POST /upload            (subir .sqlite / .db / .sql / .csv / .zip)",
-            "GET  /connections       (listar BDs subidas)",
-            "GET  /schema/{id}       (esquema resumido)",
+            "POST /upload-sql        (subir .sql o .zip con .sql, engine auto o manual)",
+            "GET  /connections       (listar BDs lógicas)",
+            "GET  /schema/{id}       (esquema resumido desde motor real)",
             "GET  /preview/{id}/{t}  (preview de tabla)",
-            "POST /infer             (NL→SQL + ejecución)",
+            "POST /infer             (NL→SQL + ejecución segura en el motor)",
             "POST /speech-infer      (NL por voz → SQL + ejecución)",
             "GET  /health            (estado del backend)",
             "GET  /docs              (OpenAPI UI)",