app.py

import os
import re
import ast
import tempfile
import sys
import subprocess
from html.parser import HTMLParser
import gradio as gr
from huggingface_hub import InferenceClient
import spaces
from groq import Groq

GROQ_API_KEY = os.getenv("GROQ_API_KEY")
MODEL_ID = "openai/gpt-oss-120b"   
REASONING_EFFORT = "low"           
MAX_AUTOFIX = 3
DEMO_LIMIT = 200                  

# Contador global (se reinicia al restartear el Space)
_USAGES = {"count": 0, "blocked": False}

BLOQUEOS_JAILBREAK = [
    "ignora tus reglas", "olvida instrucciones", "actua como un",
    "eres ahora", "system prompt", "reveala tus instrucciones",
    "dime tu system prompt", "hackear", "malware", "virus", "exploit"
]



SYSTEM_PROMPT_BASE = """Eres Aura, asistente senior que REVISA, CORRIGE Y AUTO-REPARA código hasta que ejecute.
Tu ÚNICA función es programar, revisar, debuggear y corregir código.
Si te doy un error de ejecución, corrige el código COMPLETO inmediatamente.
FLUJO OBLIGATORIO:
1. Analiza ERRORES DETECTADOS que te paso.
2. Responde con:
   🔍 Errores encontrados:
   - lista
   ✅ Código corregido: en bloque de código completo
   📝 Qué corregiste: explicación breve
Si NO es programación, responde: Soy Aura, asistente especializado exclusivamente en programación."""

def extract_file_text(file):
    if not file: return ""
    try:
        if isinstance(file, (list, tuple)): file = file[0]
        path = None
        for attr in ("path", "name", "file_path", "orig_name"):
            val = getattr(file, attr, None)
            if isinstance(val, str) and val: path = val; break
        if path is None and isinstance(file, dict):
            for k in ("path", "name", "file_path"):
                v = file.get(k)
                if isinstance(v, str) and v: path = v; break
        if path is None and isinstance(file, str): path = file
        if not path or os.path.isdir(path) or not os.path.isfile(path): return ""
        with open(path, 'r', encoding='utf-8', errors='ignore') as f: return f.read()[:20000]
    except: return ""


def get_ext(n):
    n = (n or "").lower()
    if n.endswith((".html", ".htm")): return "html"
    if n.endswith((".js", ".jsx", ".mjs")): return "js"
    if n.endswith((".ts", ".tsx")): return "ts"
    if n.endswith(".css"): return "css"
    if n.endswith(".sql"): return "sql"
    if n.endswith(".py"): return "py"
    return "py"


def detectar_errores_python(codigo):
    try:
        ast.parse(codigo)
        return []
    except SyntaxError as err:
        return [f"🔴 SyntaxError línea {err.lineno}: {err.msg}"]
    except Exception as e:
        return [f"🔴 Error: {e}"]


def analizar_todo(codigo, nombre):
    ext = get_ext(nombre)
    if ext == "py":
        return detectar_errores_python(codigo)
    return []


def ejecutar_codigo_sandbox(codigo, nombre):
    ext = get_ext(nombre)
    if ext in ["html", "css"]:
        return True, f"✅ {ext.upper()} validado."
    if ext == "sql":
        return True, "✅ SQL analizada."
    if ext == "py":
        with tempfile.NamedTemporaryFile(mode='w', suffix='.py', delete=False, encoding='utf-8') as f:
            f.write(codigo)
            temp_path = f.name
        try:
            result = subprocess.run(
                [sys.executable, temp_path],
                capture_output=True, text=True, timeout=10
            )
            out = result.stdout + (f"\n{result.stderr}" if result.stderr else "")
            if result.returncode != 0:
                return False, f"❌ ERROR:\n{out}"
            return True, f"✅ OK:\n{out or '(sin salida)'}"
        except subprocess.TimeoutExpired:
            return False, "❌ TIMEOUT"
        except Exception as e:
            return False, str(e)
        finally:
            try:
                os.unlink(temp_path)
            except:
                pass
    return True, "✅ Código recibido."


def to_messages(history):
    msgs = []
    if not isinstance(history, list):
        return msgs
    for item in history:
        if isinstance(item, dict) and "role" in item and "content" in item:
            c = item["content"]
            if isinstance(c, list):
                c = "".join(b.get("text", "") for b in c if isinstance(b, dict))
            if item["role"] in ("user", "assistant"):
                msgs.append({"role": item["role"], "content": str(c)})
    return msgs


def make_download_files(text):
    code_path, txt_path = None, None
    try:
        blocks = re.findall(r'```(?:python|javascript|js|html|sql)?\n(.*?)\n```', text, re.DOTALL)
        if blocks:
            main_code = max(blocks, key=len)
            with tempfile.NamedTemporaryFile(mode='w', suffix='.py', delete=False, encoding='utf-8') as f:
                f.write(main_code)
                code_path = f.name
        with tempfile.NamedTemporaryFile(mode='w', suffix='.txt', delete=False, encoding='utf-8') as f:
            f.write(text)
            txt_path = f.name
    except:
        pass
    return code_path, txt_path


def autofix_loop(client, messages, main_code, ext, max_tokens, temperature, top_p):
    """Reintenta enviando el error al modelo hasta MAX_AUTOFIX veces."""
    ultima = "Listo"
    codigo_actual = main_code
    historial_fix = list(messages)

    for intento in range(1, MAX_AUTOFIX + 1):
        ok, res = ejecutar_codigo_sandbox(codigo_actual, f"test.{ext}")
        ultima = res
        if ok:
            return codigo_actual, ultima, True

        # Le mandamos el error al modelo para que lo corrijas
        yield_msg = (
            f"El código que me diste FALLÓ al ejecutarse.\n\n"
            f"--- SALIDA/ERROR ---\n{res}\n--- FIN ERROR ---\n\n"
            f"Corrige el problema y responde SOLO con el código completo corregido "
            f"en un único bloque ```{ext}```. No expliques nada más."
        )
        historial_fix.append({"role": "assistant", "content": f"```{ext}\n{codigo_actual}\n```"})
        historial_fix.append({"role": "user", "content": yield_msg})

        r = client.chat.completions.create(
            model=MODEL_ID,
            messages=historial_fix,
            max_tokens=int(max_tokens),
            temperature=float(temperature),
            top_p=float(top_p)
        )
        texto_fix = r.choices[0].message.content or ""
        blocks = re.findall(r'```(?:python|javascript|js|html|sql)?\n(.*?)\n```', texto_fix, re.DOTALL)
        if not blocks:
            ultima = f"⚠️ Intento {intento}: el modelo no devolvió código.\n{res}"
            break
        codigo_actual = max(blocks, key=len)

    return codigo_actual, ultima, False


# ─── SISTEMA DE LÍMITES ───

def decrement_usage():
    global _USAGES
    _USAGES["count"] += 1
    return DEMO_LIMIT - _USAGES["count"]


def check_usage(token):
    """Retorna (ok, restante, footer_extra)"""
    if token:
        return True, -1, ""  # -1 = ilimitado (usa su propia key)
    if _USAGES["blocked"]:
        return False, 0, "\n\n⛔ **Demo agotado.** Usa tu propia API key de Groq para continuar."
    restante = decrement_usage()
    if restante <= 0:
        _USAGES["blocked"] = True
        return True, 0, "\n\n⛔ **Límite de 200 demos alcanzado.** Obtén tu key en [console.groq.com/keys](https://console.groq.com/keys) para uso ilimitado."
    return True, restante, None


@spaces.GPU
def run_chat(message, history, user_context, max_tokens, temperature, top_p, token, file_content, file_name, memoria):
    global _USAGES

    if not isinstance(history, list):
        history = []
    if not message or not message.strip():
        yield history, None, None, "⚠️ Mensaje vacío.", memoria
        return

    new_turns = [{"role": "user", "content": message}]

    # Jailbreak check
    for b in BLOQUEOS_JAILBREAK:
        if b in message.lower():
            blocked = history + new_turns + [{"role": "assistant", "content": "Soy Aura, asistente especializado exclusivamente en programación."}]
            yield blocked, None, None, "⛔ Bloqueado", memoria
            return

    # Verificar límites de demo
    ok, restante, footer_extra = check_usage(token)
    if not ok:
        yield history + new_turns + [{"role": "assistant", "content": "⛔ Demo agotado. Usa tu propia API key."}], None, None, "⛔ Demo agotado", memoria
        return

    api_key = token or GROQ_API_KEY
    if not api_key:
        blocked = history + new_turns + [{"role": "assistant", "content": "⚠️ Falta GROQ_API_KEY"}]
        yield blocked, None, None, "Falta token", memoria
        return

    try:
        client = Groq(api_key=api_key)

        contexto_archivo = ""
        if file_content and file_name:
            errores = analizar_todo(file_content, file_name)
            contexto_archivo = f"\n--- ARCHIVO ({file_name}) ---\n{file_content}\n"
            if errores:
                contexto_archivo += "\n".join(errores) + "\n"

        final_system = SYSTEM_PROMPT_BASE
        if user_context:
            final_system += f"\n--- CONTEXTO ---\n{user_context}\n"
        if memoria:
            final_system += f"\n--- MEMORIA ---\n{memoria}\n"

        messages = [{"role": "system", "content": final_system}]
        messages.extend(to_messages(history)[-6:])
        messages.append({"role": "user", "content": message + contexto_archivo})

        resp_text = ""
        stream = client.chat.completions.create(
            model=MODEL_ID,
            messages=messages,
            max_tokens=int(max_tokens),
            stream=True,
            temperature=float(temperature),
            top_p=float(top_p)
        )
        for chunk in stream:
            if chunk.choices and chunk.choices[0].delta.content:
                resp_text += chunk.choices[0].delta.content
                yield history + new_turns + [{"role": "assistant", "content": resp_text}], None, None, "Generando...", memoria

        ultima = "Listo"

        # Auto-fix loop
        blocks = re.findall(r'```(?:python|javascript|js|html|sql)?\n(.*?)\n```', resp_text, re.DOTALL)
        if blocks:
            main_code = max(blocks, key=len)
            ext = get_ext(file_name) if file_name else "py"
            yield history + new_turns + [{"role": "assistant", "content": resp_text + "\n\n🔁 *Verificando ejecución...*"}], None, None, "Ejecutando...", memoria

            codigo_final, ultima, ok_fix = autofix_loop(
                client, messages, main_code, ext, max_tokens, temperature, top_p
            )
            if not ok_fix and codigo_final != main_code:
                resp_text = resp_text + f"\n\n---\n🔧 **Código auto-corregido:**\n```{ext}\n{codigo_final}\n```"
                yield history + new_turns + [{"role": "assistant", "content": resp_text}], None, None, ultima, memoria

        code_file, txt_file = make_download_files(resp_text)

        # Agregar footer de límite
        final_resp = resp_text
        if footer_extra:
            final_resp += footer_extra
        elif restante > 0:
            final_resp += f"\n\n*🔹 Quedan {restante} demos. Usa tu propia API key para uso ilimitado.*"

        yield history + new_turns + [{"role": "assistant", "content": final_resp}], code_file, txt_file, ultima, memoria

    except Exception as e:
        import traceback
        print(traceback.format_exc())
        yield history + [{"role": "user", "content": message}, {"role": "assistant", "content": f"💥 Error: {e}"}], None, None, str(e), memoria


# ─── UI ───

with gr.Blocks(title="Aura Programadora") as demo:
    gr.Markdown("# 💻 Aura Programadora (Groq)")
    with gr.Row():
        with gr.Column(scale=3):
            chatbot = gr.Chatbot(label="Chat", height=500)
            chat_input = gr.Textbox(placeholder="Escribe tu código o error...", lines=2)
            with gr.Row():
                send_btn = gr.Button("🚀 Enviar", variant="primary")
                clear_btn = gr.Button("🗑️ Limpiar")
            with gr.Accordion("📎 Adjuntar Archivo", open=False):
                file_input = gr.File(
                    label="Sube archivo (.py,.js,.html,.sql)",
                    file_types=[".py", ".js", ".html", ".sql", ".jsx", ".ts"]
                )
                file_name_state = gr.Textbox(visible=False)
                file_content_state = gr.Textbox(visible=False)
        with gr.Column(scale=2):
            with gr.Accordion("⚙️ Configuración IA", open=False):
                user_context_input = gr.Textbox(
                    label="Tus Instrucciones", lines=3,
                    placeholder="Ej: explica paso a paso"
                )
                max_tokens_input = gr.Slider(1, 4096, 2048, label="Max Tokens")
                temp_input = gr.Slider(0.1, 1.0, 0.2, label="Temperature")
                top_p_input = gr.Slider(0.1, 1.0, 0.9, label="Top-p")
                token_input = gr.Textbox(
                    label="Tu API Key Groq (opcional, usamos la demo)",
                    type="password"
                )
            exec_output = gr.Textbox(label="Resultado Ejecución", lines=4)
            download_code = gr.File(label="Código Corregido")
            download_txt = gr.File(label="Chat Completo")
            gr.Markdown("### 🧠 Memoria")
            visible_memoria = gr.Textbox(
                label="Notas del Usuario", lines=2,
                placeholder="Ej: Mi proyecto usa FastAPI..."
            )
            hidden_memoria = gr.Textbox(visible=False)

    def on_file_upload(file):
        if not file:
            return "", ""
        if isinstance(file, (list, tuple)):
            file = file[0]
        fname = getattr(file, "orig_name", None) or getattr(file, "name", "") or ""
        if isinstance(file, dict):
            fname = file.get("orig_name") or file.get("name") or ""
        return extract_file_text(file), os.path.basename(str(fname))

    file_input.upload(
        fn=on_file_upload,
        inputs=[file_input],
        outputs=[file_content_state, file_name_state]
    )

    event_inputs = [
        chat_input, chatbot, user_context_input,
        max_tokens_input, temp_input, top_p_input,
        token_input, file_content_state, file_name_state,
        visible_memoria
    ]
    event_outputs = [chatbot, download_code, download_txt, exec_output, hidden_memoria]

    send_btn.click(fn=run_chat, inputs=event_inputs, outputs=event_outputs)
    chat_input.submit(fn=run_chat, inputs=event_inputs, outputs=event_outputs)
    clear_btn.click(fn=lambda: ([], None, None, "", ""), inputs=[], outputs=event_outputs)


if __name__ == "__main__":
    demo.queue().launch(ssr_mode=False)