| 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 |
|
|
| HF_TOKEN = os.getenv("HF_TOKEN") |
| MODEL_ID = "Qwen/Qwen2.5-Coder-7B-Instruct" |
| MAX_AUTOFIX = 3 |
|
|
| 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 |
|
|
| @spaces.GPU |
| def run_chat(message, history, user_context, max_tokens, temperature, top_p, token, file_content, file_name, memoria): |
| 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}] |
| 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 |
| if not (token or HF_TOKEN): |
| blocked=history+new_turns+[{"role": "assistant", "content": "⚠️ Falta HF_TOKEN"}] |
| yield blocked, None, None, "Falta token", memoria |
| return |
| try: |
| client=InferenceClient(token=token or HF_TOKEN, model=MODEL_ID) |
| 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="" |
| for chunk in client.chat_completion(messages=messages, max_tokens=int(max_tokens), stream=True, temperature=float(temperature), top_p=float(top_p)): |
| 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" |
| 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" |
| intentos=0 |
| while intentos<MAX_AUTOFIX: |
| ok,res=ejecutar_codigo_sandbox(main_code, f"test.{ext}") |
| ultima=res |
| if ok: break |
| intentos+=1 |
| code_file, txt_file=make_download_files(resp_text) |
| yield history+new_turns+[{"role": "assistant", "content": resp_text}], 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 |
|
|
| with gr.Blocks(title="Aura Programadora") as demo: |
| gr.Markdown("# 💻 Aura Programadora") |
| 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="HF Token (opcional)", 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) |
