import gradio as gr import spaces import json import os import subprocess import asyncio import threading import time # ─── Install browser deps + Xvfb on startup ─────────────────── def setup(): """Install chromium deps and start Xvfb""" # Install system deps for chromium subprocess.run( "apt-get update && apt-get install -y libnss3 libnspr4 libatk1.0-0 " "libatk-bridge2.0-0 libcups2 libdrm2 libxkbcommon0 libxcomposite1 " "libxdamage1 libxfixes3 libxrandr2 libgbm1 libpango-1.0-0 libcairo2 " "libasound2 libx11-xcb1 libxcb1 libxext6 libx11-6 libxshmfence1 " "xvfb fonts-liberation 2>/dev/null", shell=True, capture_output=True ) # Install playwright browsers subprocess.run(["python", "-m", "playwright", "install", "chromium"], capture_output=True) # Start Xvfb subprocess.Popen( ["Xvfb", ":99", "-screen", "0", "1280x720x24", "-nolisten", "tcp"], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL ) os.environ["DISPLAY"] = ":99" time.sleep(1) print("[✓] Setup done: deps + Xvfb") setup() # ─── Playwright Async Browser ───────────────────────────────── from playwright.async_api import async_playwright pw = None browser = None sessions = {} # session_id -> page ready = False lock = asyncio.Lock() async def init_browser(): global pw, browser, ready try: pw = await async_playwright().start() browser = await pw.chromium.launch( headless=True, args=[ "--no-sandbox", "--disable-gpu", "--disable-dev-shm-usage", "--disable-setuid-sandbox", "--no-first-run", "--single-process", ] ) ready = True print("[✓] Browser ready (headless)") except Exception as e: print(f"[✗] Browser init failed: {e}") async def get_page(session_id="default"): if session_id not in sessions: ctx = await browser.new_context( viewport={"width": 1280, "height": 720}, user_agent="Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36" ) page = await ctx.new_page() await page.set_extra_http_headers({"Accept-Language": "en-US,en;q=0.9"}) sessions[session_id] = page return sessions[session_id] # ─── API Functions (async) ───────────────────────────────────── async def run_code(url: str, script: str, wait: float, session: str) -> str: """Navigate to URL and execute JavaScript""" global ready if not ready: await init_browser() if not ready: return json.dumps({"success": False, "error": "Browser failed to start"}) async with lock: try: page = await get_page(session or "default") if url and url.strip(): await page.goto(url.strip(), wait_until="domcontentloaded", timeout=60000) if wait and wait > 0: await page.wait_for_timeout(int(wait * 1000)) if not script or not script.strip(): return json.dumps({"success": True, "result": None, "url": page.url}) result = await page.evaluate(script.strip()) return json.dumps({"success": True, "result": result}, ensure_ascii=False, default=str) except Exception as e: return json.dumps({"success": False, "error": str(e)}) async def eval_code(script: str, session: str) -> str: """Execute JS on existing page""" if not ready: return json.dumps({"success": False, "error": "Browser not ready"}) async with lock: try: sid = session or "default" if sid not in sessions: return json.dumps({"success": False, "error": f"Session '{sid}' not found. Use Run tab first."}) page = sessions[sid] result = await page.evaluate(script.strip()) return json.dumps({"success": True, "result": result}, ensure_ascii=False, default=str) except Exception as e: return json.dumps({"success": False, "error": str(e)}) async def list_sessions() -> str: info = {} for sid, page in sessions.items(): try: info[sid] = page.url except: info[sid] = "?" return json.dumps({"ready": ready, "sessions": info}) async def close_session(session: str) -> str: async with lock: if session in sessions: try: await sessions[session].context.close() except: pass del sessions[session] return json.dumps({"success": True, "closed": session}) return json.dumps({"success": False, "error": "not found"}) async def health_check() -> str: return json.dumps({"ready": ready, "sessions": len(sessions)}) @spaces.GPU(duration=0) def _gpu_dummy(): """Dummy function to satisfy ZeroGPU detection""" pass # ─── Init browser at startup ────────────────────────────────── def startup_sync(): """Run async init in a new event loop (avoids deprecation)""" loop = asyncio.new_event_loop() loop.run_until_complete(init_browser()) loop.close() threading.Thread(target=startup_sync, daemon=True).start() time.sleep(5) # ─── Gradio UI ──────────────────────────────────────────────── with gr.Blocks(title="Playwright Runner") as demo: gr.Markdown("# 🎭 Playwright Remote Runner\nSend JS to execute in Chromium. Use API for programmatic access.") with gr.Tab("Run"): with gr.Row(): url_input = gr.Textbox(label="URL", placeholder="https://example.com") session_input = gr.Textbox(label="Session", value="default") wait_input = gr.Number(label="Wait (sec)", value=3) script_input = gr.Code(label="JavaScript", language="javascript", value='() => document.title') run_btn = gr.Button("Execute", variant="primary") run_output = gr.Code(label="Result", language="json") run_btn.click(run_code, [url_input, script_input, wait_input, session_input], run_output, queue=False) with gr.Tab("Eval"): eval_session = gr.Textbox(label="Session", value="default") eval_script = gr.Code(label="JavaScript", language="javascript", value='() => ({ title: document.title, url: location.href })') eval_btn = gr.Button("Eval", variant="primary") eval_output = gr.Code(label="Result", language="json") eval_btn.click(eval_code, [eval_script, eval_session], eval_output, queue=False) with gr.Tab("Sessions"): list_btn = gr.Button("List") list_output = gr.Code(language="json") list_btn.click(list_sessions, [], list_output, queue=False) close_input = gr.Textbox(label="Close session") close_btn = gr.Button("Close", variant="stop") close_output = gr.Code(language="json") close_btn.click(close_session, [close_input], close_output, queue=False) with gr.Tab("Health"): h_btn = gr.Button("Check") h_out = gr.Code(language="json") h_btn.click(health_check, [], h_out, queue=False) demo.launch(server_name="0.0.0.0", server_port=7860, ssr_mode=False, show_api=True)