app.py

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)