Update app.py

app.py

@@ -1,176 +1,82 @@
-import io
-import sys
-import time
-import ast
-import traceback
-import importlib
-import subprocess
-from multiprocessing import Process, Queue
-
-try:
-    import resource  # Linux only (Hugging Face Spaces)
-except Exception:
-    resource = None
-
+# app.py
 import gradio as gr
+import sys, io, traceback, re, subprocess, importlib, multiprocessing, resource, signal
 
-# -----------------------------
-# Sandbox utilities
-# -----------------------------
-SAFE_BUILTINS = {
-    "abs": abs, "all": all, "any": any, "bool": bool, "bytes": bytes, "callable": callable,
-    "chr": chr, "complex": complex, "dict": dict, "dir": dir, "divmod": divmod,
-    "enumerate": enumerate, "filter": filter, "float": float, "format": format,
-    "hash": hash, "help": help, "hex": hex, "int": int, "isinstance": isinstance,
-    "issubclass": issubclass, "iter": iter, "len": len, "list": list, "map": map,
-    "max": max, "min": min, "next": next, "object": object, "oct": oct, "ord": ord,
-    "pow": pow, "print": print, "range": range, "repr": repr, "reversed": reversed,
-    "round": round, "set": set, "slice": slice, "sorted": sorted, "str": str,
-    "sum": sum, "tuple": tuple, "type": type, "zip": zip,
-    "__import__": __import__,   # ✅ لازم برای import
-}
-
-
-def _parse_imports(code: str) -> list:
-    """Parse AST and return list of imports found in code."""
-    try:
-        tree = ast.parse(code, mode="exec")
-    except SyntaxError as e:
-        raise ValueError(f"SyntaxError: {e}")
-
-    imports = []
-    for node in ast.walk(tree):
-        if isinstance(node, (ast.Import, ast.ImportFrom)):
-            for alias in node.names:
-                pkg = alias.name.split(".")[0]
-                imports.append(pkg)
-    return list(set(imports))
-
-
+# ----------------------
+# Helper: Install missing packages safely
+# ----------------------
 def _install_missing(packages):
-    """Install packages if not already installed."""
+    """Install packages if not already installed. Fail gracefully if not found."""
     for pkg in packages:
         try:
             importlib.import_module(pkg)
         except ImportError:
-            subprocess.check_call([sys.executable, "-m", "pip", "install", pkg])
+            try:
+                subprocess.check_call(
+                    [sys.executable, "-m", "pip", "install", "--disable-pip-version-check", "-q", pkg]
+                )
+            except subprocess.CalledProcessError:
+                # فقط هشدار می‌ده، برنامه نمی‌ترکه
+                print(f"[WARN] Package '{pkg}' not found on PyPI or failed to install.", file=sys.stderr)
+
+# ----------------------
+# Worker process
+# ----------------------
+def _worker(code, stdin_input, conn):
+    sys.stdin = io.StringIO(stdin_input)
+    sys.stdout = io.StringIO()
+    sys.stderr = io.StringIO()
+
+    # محدودیت حافظه و زمان
+    resource.setrlimit(resource.RLIMIT_AS, (512 * 1024 * 1024, 512 * 1024 * 1024))  # 512MB
+    signal.alarm(10)  # 10 ثانیه تایم‌اوت
 
-
-def _apply_limits(mem_mb: int, cpu_seconds: int):
-    if resource is None:
-        return
-    if mem_mb and mem_mb > 0:
-        bytes_limit = int(mem_mb) * 1024 * 1024
-        resource.setrlimit(resource.RLIMIT_AS, (bytes_limit, bytes_limit))
-    if cpu_seconds and cpu_seconds > 0:
-        resource.setrlimit(resource.RLIMIT_CPU, (cpu_seconds, cpu_seconds))
-
-
-def _worker(code: str, stdin_data: str, mem_limit_mb: int, cpu_limit_s: int, q: Queue):
-    stdout_buf, stderr_buf = io.StringIO(), io.StringIO()
-    start = time.time()
     try:
-        _apply_limits(mem_limit_mb, cpu_limit_s)
-
-        # Redirect streams
-        orig_stdout, orig_stderr, orig_stdin = sys.stdout, sys.stderr, sys.stdin
-        sys.stdout, sys.stderr, sys.stdin = stdout_buf, stderr_buf, io.StringIO(stdin_data or "")
-
-        # Install missing packages if needed
-        imports = _parse_imports(code)
+        # پیدا کردن importها
+        imports = re.findall(r'^\s*import (\w+)|^\s*from (\w+)', code, flags=re.MULTILINE)
+        imports = {i[0] or i[1] for i in imports}
         if imports:
             _install_missing(imports)
 
-        safe_globals = {"__builtins__": SAFE_BUILTINS}
-
-        exec(compile(code, filename="<user_code>", mode="exec"), safe_globals, None)
-
+        safe_globals = {"__builtins__": __builtins__}
+        exec(compile(code, "<user_code>", "exec"), safe_globals, None)
+        out = sys.stdout.getvalue()
+        err = sys.stderr.getvalue()
+        conn.send((out, err))
     except Exception:
-        traceback.print_exc(file=stderr_buf)
+        conn.send(("", traceback.format_exc()))
     finally:
-        try:
-            sys.stdout, sys.stderr, sys.stdin = orig_stdout, orig_stderr, orig_stdin
-        except Exception:
-            pass
-        elapsed = time.time() - start
-        q.put({
-            "stdout": stdout_buf.getvalue(),
-            "stderr": stderr_buf.getvalue(),
-            "elapsed": elapsed,
-        })
-
-
-def run_code(code: str, stdin_data: str = "", timeout_s: float = 5.0, mem_limit_mb: int = 512):
-    q: Queue = Queue()
-    p = Process(target=_worker, args=(code, stdin_data, int(mem_limit_mb), int(max(1, timeout_s)), q))
+        conn.close()
+
+# ----------------------
+# Runner
+# ----------------------
+def run_code(code, stdin_input=""):
+    parent_conn, child_conn = multiprocessing.Pipe()
+    p = multiprocessing.Process(target=_worker, args=(code, stdin_input, child_conn))
     p.start()
-    p.join(timeout=timeout_s)
-
+    p.join(12)  # یک مقدار بیشتر از تایم‌اوت داخلی
     if p.is_alive():
         p.terminate()
-        try: p.join(1)
-        except Exception: pass
-        return "", "Execution timed out.", timeout_s
-
-    try:
-        result = q.get_nowait()
-    except Exception:
-        return "", "No output captured (possibly killed).", 0.0
+        return "", "Error: Timeout (execution took too long)"
+    return parent_conn.recv()
 
-    return result.get("stdout", ""), result.get("stderr", ""), float(result.get("elapsed", 0.0))
-
-
-# -----------------------------
+# ----------------------
 # Gradio UI
-# -----------------------------
-EXAMPLE_CODE = """
-import requests
-r = requests.get("https://httpbin.org/get")
-print("Status:", r.status_code)
-print("JSON:", r.json())
-""".strip()
-
-
-def interface_run(code, stdin_text, timeout_s, mem_limit):
-    stdout, stderr, elapsed = run_code(code or "", stdin_text or "", timeout_s or 5.0, mem_limit or 512)
-    status = f"Finished in {elapsed:.3f}s"
-    return stdout, stderr, status
-
-
-with gr.Blocks(title="Python Code Runner (Free Sandbox)") as demo:
-    gr.Markdown("""
-    # 🐍 Python Code Runner (Free Sandbox)
-    اجرای کد پایتون با نصب خودکار هر کتابخانه‌ای که import کنید. ⚠️
-    
-    > توجه: این نسخه ناامن است و می‌تواند ریسک امنیتی داشته باشد.
-    """)
-
-    with gr.Row():
-        code = gr.Code(language="python", value=EXAMPLE_CODE, lines=18, label="Python code")
-    with gr.Row():
-        stdin_tb = gr.Textbox(lines=3, label="stdin (optional)")
-    with gr.Row():
-        timeout = gr.Slider(1, 30, value=5, step=1, label="Time limit (seconds)")
-        mem_limit = gr.Slider(64, 4096, value=512, step=64, label="Memory limit (MB)")
-    run_btn = gr.Button("Run", variant="primary")
+# ----------------------
+with gr.Blocks(title="Python Runner (Auto Install)") as demo:
+    gr.Markdown("### 🐍 Python Sandbox with Auto-Package Install (USE WITH CAUTION)")
 
-    with gr.Row():
-        stdout = gr.Textbox(lines=12, label="stdout", interactive=False)
-    with gr.Row():
-        stderr = gr.Textbox(lines=8, label="stderr", interactive=False)
-    status = gr.Label(value="Idle", label="Status")
+    code = gr.Code(label="Python code", language="python")
+    stdin = gr.Textbox(label="stdin", placeholder="Optional input for `input()` calls")
+    run_btn = gr.Button("▶ Run")
+    stdout = gr.Textbox(label="stdout")
+    stderr = gr.Textbox(label="stderr", elem_classes="stderr")
 
-    run_btn.click(interface_run, inputs=[code, stdin_tb, timeout, mem_limit], outputs=[stdout, stderr, status])
+    def execute(c, s):
+        return run_code(c, s)
 
-    gr.Examples(
-        examples=[
-            ["print('Hello from Space!')"],
-            ["import numpy as np\nprint(np.arange(10))"],
-            ["import requests\nprint(requests.get('https://httpbin.org/get').status_code)"],
-        ],
-        inputs=[code],
-        label="Quick examples"
-    )
+    run_btn.click(execute, [code, stdin], [stdout, stderr])
 
 if __name__ == "__main__":
     demo.launch()