""" Voice-to-SQL — pregúntale a una base de datos SaaS de ejemplo en lenguaje natural. Un LLM (Llama 3.3 70B vía Groq) traduce la pregunta a SQL, se valida que sea de SOLO LECTURA y se ejecuta sobre una BD SQLite generada en memoria. Demo completa (con voz, Web Speech API) en https://adrianmoreno-dev.com/demo/voice-to-sql-dashboard Por Adrián Moreno · https://adrianmoreno-dev.com """ import os, re, sqlite3, random from datetime import datetime, timedelta import requests import pandas as pd import gradio as gr GROQ_KEY = os.environ.get("GROQ_API_KEY", "") MODELS = ["llama-3.3-70b-versatile", "llama-3.1-8b-instant"] SCHEMA_DESC = """TABLAS DISPONIBLES (esquema exacto, NO inventar columnas): clientes (200 filas) - id INTEGER PRIMARY KEY - nombre TEXT - pais TEXT -- "España"|"Francia"|"Italia"|"Alemania"|"Reino Unido"|"Portugal"|"México"|"Argentina"|"Chile"|"Colombia" - sector TEXT -- "E-commerce"|"SaaS"|"FinTech"|"EdTech"|"HealthTech"|"MarTech"|"PropTech"|"AgriTech" - plan TEXT -- "free"|"starter"|"pro"|"enterprise" - fecha_alta DATE -- ISO YYYY-MM-DD - activo INTEGER -- 1=activo, 0=churn productos (8 filas) - id INTEGER PRIMARY KEY - nombre TEXT - tier TEXT -- "addon"|"core"|"premium" - precio_mensual REAL -- en EUR ventas (5000 filas, último año) - id INTEGER PRIMARY KEY - cliente_id INTEGER → clientes.id - producto_id INTEGER → productos.id - fecha DATE -- ISO YYYY-MM-DD - importe REAL -- EUR pagados - mrr_aporte REAL -- contribución MRR REGLAS: - SQLite dialect (NOT PostgreSQL/MySQL). - Para "últimos N meses" usa: date('now', '-N months') - Para agrupar por mes: strftime('%Y-%m', fecha) - Si la pregunta es ambigua, asume la interpretación más útil para un dashboard analítico.""" SYSTEM_PROMPT = ( "Eres un experto en SQLite que convierte preguntas en lenguaje natural en " "consultas SQL ejecutables. Responde ÚNICAMENTE con la consulta SQL, sin " "explicación, sin markdown, sin punto y coma final, sin prefijos.\n\n" + SCHEMA_DESC + "\n\nREGLA CRÍTICA: solo SELECT. Nunca UPDATE/INSERT/DELETE/DROP." ) def build_db(): con = sqlite3.connect(":memory:", check_same_thread=False) con.executescript(""" CREATE TABLE clientes (id INTEGER PRIMARY KEY, nombre TEXT, pais TEXT, sector TEXT, plan TEXT, fecha_alta DATE, activo INTEGER DEFAULT 1); CREATE TABLE productos (id INTEGER PRIMARY KEY, nombre TEXT, tier TEXT, precio_mensual REAL); CREATE TABLE ventas (id INTEGER PRIMARY KEY, cliente_id INTEGER, producto_id INTEGER, fecha DATE, importe REAL, mrr_aporte REAL); """) paises = ["España", "Francia", "Italia", "Alemania", "Reino Unido", "Portugal", "México", "Argentina", "Chile", "Colombia"] sectores = ["E-commerce", "SaaS", "FinTech", "EdTech", "HealthTech", "MarTech", "PropTech", "AgriTech"] planes = ["free", "starter", "pro", "enterprise"] productos = [("Plan Free", "core", 0), ("Plan Starter", "core", 29), ("Plan Pro", "core", 99), ("Plan Enterprise", "core", 499), ("Add-on Analytics", "addon", 19), ("Add-on API Plus", "addon", 39), ("Add-on White Label", "premium", 149), ("Add-on Priority Support", "premium", 79)] con.executemany("INSERT INTO productos (nombre,tier,precio_mensual) VALUES (?,?,?)", productos) empresas = ["Acme", "Globex", "Initech", "Umbrella", "Stark", "Wayne", "Pied Piper", "Hooli", "Soylent", "Cyberdyne", "Tyrell", "Weyland", "Aperture", "Oscorp", "LexCorp", "Wonka"] sufijos = ["Tech", "Labs", "Group", "Solutions", "Systems", "Digital", "Cloud", "Studio"] hoy = datetime.now().date() for i in range(1, 201): con.execute("INSERT INTO clientes (nombre,pais,sector,plan,fecha_alta,activo) VALUES (?,?,?,?,?,?)", (f"{random.choice(empresas)} {random.choice(sufijos)} {i}", random.choice(paises), random.choice(sectores), random.choices(planes, weights=[0.4, 0.3, 0.2, 0.1])[0], (hoy - timedelta(days=random.randint(7, 730))).isoformat(), 1 if random.random() > 0.12 else 0)) precios = [p[2] for p in productos] ventas = [] for _ in range(5000): pid = random.randint(1, 8) importe = round(precios[pid - 1] * random.uniform(0.85, 1.0), 2) ventas.append((random.randint(1, 200), pid, (hoy - timedelta(days=random.randint(1, 365))).isoformat(), importe, importe)) con.executemany("INSERT INTO ventas (cliente_id,producto_id,fecha,importe,mrr_aporte) VALUES (?,?,?,?,?)", ventas) con.commit() return con CON = build_db() def generate_sql(question: str) -> str: if not GROQ_KEY: raise RuntimeError("Falta GROQ_API_KEY (secreto del Space).") last = "" for model in MODELS: try: r = requests.post( "https://api.groq.com/openai/v1/chat/completions", headers={"Authorization": f"Bearer {GROQ_KEY}"}, json={"model": model, "temperature": 0, "max_tokens": 400, "messages": [{"role": "system", "content": SYSTEM_PROMPT}, {"role": "user", "content": question}]}, timeout=30) if r.status_code == 200: sql = r.json()["choices"][0]["message"]["content"].strip() sql = re.sub(r"^```\w*", "", sql).replace("```", "").strip().rstrip(";").strip() return sql last = f"HTTP {r.status_code}" except Exception as e: last = str(e) raise RuntimeError(f"Groq no disponible ({last})") def sanitize(sql: str) -> str: low = sql.strip().lower() if not (low.startswith("select") or low.startswith("with")): raise ValueError("Solo se permiten consultas SELECT.") if re.search(r"\b(insert|update|delete|drop|alter|create|pragma|attach|replace|vacuum)\b", low): raise ValueError("Solo lectura: palabra prohibida detectada.") if ";" in sql.strip().rstrip(";"): raise ValueError("Solo una sentencia.") return sql def ask(question: str): if not question or len(question.strip()) < 3: return "-- Escribe una pregunta", None try: sql = generate_sql(question) sanitize(sql) cur = CON.execute(sql) cols = [d[0] for d in cur.description] df = pd.DataFrame(cur.fetchmany(100), columns=cols) return sql, df except Exception as e: return f"-- Error: {e}", None with gr.Blocks(title="Voice-to-SQL", theme=gr.themes.Soft(primary_hue="teal")) as demo: gr.Markdown( "# 🗣️→🗄️ Voice-to-SQL — pregúntale a tu base de datos\n" "Escribe una pregunta de negocio en **lenguaje natural** y un LLM (Llama 3.3 70B) la traduce " "a SQL, la valida (solo lectura) y la ejecuta sobre una base de datos **SaaS de ejemplo** " "(200 clientes · 8 planes · 5 000 ventas).\n\n" "👉 Versión con **voz** (Web Speech API) y más en " "[adrianmoreno-dev.com](https://adrianmoreno-dev.com/demo/voice-to-sql-dashboard?utm_source=huggingface)" ) q = gr.Textbox(label="Tu pregunta", placeholder="¿Cuáles son los 5 productos más vendidos?", lines=1) btn = gr.Button("Consultar", variant="primary") sql_out = gr.Code(label="SQL generado", language="sql") df_out = gr.Dataframe(label="Resultados", wrap=True) gr.Examples( ["¿Cuáles son los 5 productos más vendidos?", "Ingresos totales por país", "MRR medio por plan", "¿Cuántos clientes han hecho churn?", "Top 10 clientes por importe total gastado"], inputs=q) btn.click(ask, q, [sql_out, df_out]) q.submit(ask, q, [sql_out, df_out]) gr.Markdown("Proyecto open source de **Adrián Moreno** · [portfolio](https://adrianmoreno-dev.com?utm_source=huggingface) · solo SELECT, base de datos sintética") if __name__ == "__main__": demo.launch()