Delete ui.py

ui.py

@@ -1,722 +0,0 @@
-import threading
-import queue
-from collections import deque
-from dataclasses import dataclass, field
-from typing import Dict, Optional, Tuple, Any
-
-from playwright.sync_api import sync_playwright, Page, BrowserContext
-from playwright_stealth import Stealth
-from PIL import Image
-from io import BytesIO
-
-import gradio as gr
-
-# ─────────────────────────────
-# Global Queues (worker <-> UI)
-# ─────────────────────────────
-task_queue: "queue.Queue[dict]" = queue.Queue()
-result_queue: "queue.Queue[Tuple[str, Optional[Image.Image]]]" = queue.Queue()
-
-# ─────────────────────────────
-# Browser State + Lock
-# ─────────────────────────────
-
-@dataclass
-class BrowserState:
-    running: bool = False
-    thread: Optional[threading.Thread] = None
-    pages: Dict[str, Page] = field(default_factory=dict)  # tab_name -> Page
-    active_page: Optional[str] = None
-    network_logs: deque = field(default_factory=lambda: deque(maxlen=500))
-    console_logs: deque = field(default_factory=lambda: deque(maxlen=500))
-    recording: bool = False
-    macro: list = field(default_factory=list)
-    tab_counter: int = 0  # to generate stable names
-
-
-BROWSER_STATE = BrowserState()
-BROWSER_LOCK = threading.Lock()
-
-
-# ─────────────────────────────
-# Helper: Safe screenshot
-# ─────────────────────────────
-
-def take_screenshot(page: Optional[Page]) -> Optional[Image.Image]:
-    if page is None:
-        return None
-    try:
-        img_bytes = page.screenshot()
-        return Image.open(BytesIO(img_bytes))
-    except Exception:
-        return None
-
-
-# ─────────────────────────────
-# Worker Class (Playwright)
-# ─────────────────────────────
-
-class PlaywrightWorker:
-    def __init__(self, state: BrowserState):
-        self.state = state
-        self.context: Optional[BrowserContext] = None
-
-    # ---- Console & Network hooks ----
-    def _attach_console_listener(self, page: Page):
-        def on_console(msg):
-            with BROWSER_LOCK:
-                self.state.console_logs.append(f"[{msg.type}] {msg.text}")
-        page.on("console", on_console)
-
-    def _attach_network_listeners(self, context: BrowserContext):
-        def on_request(request):
-            with BROWSER_LOCK:
-                self.state.network_logs.append(f"[REQUEST] {request.method} {request.url}")
-
-        def on_response(response):
-            with BROWSER_LOCK:
-                self.state.network_logs.append(f"[RESPONSE] {response.status} {response.url}")
-
-        context.on("request", on_request)
-        context.on("response", on_response)
-
-    # ---- Tabs Management helpers ----
-    def _create_new_tab(self) -> Tuple[str, Page]:
-        page = self.context.new_page()
-        with BROWSER_LOCK:
-            self.state.tab_counter += 1
-            tab_name = f"Tab-{self.state.tab_counter}"
-            self.state.pages[tab_name] = page
-            self.state.active_page = tab_name
-        self._attach_console_listener(page)
-        return tab_name, page
-
-    def _get_active_page(self) -> Optional[Page]:
-        with BROWSER_LOCK:
-            name = self.state.active_page
-            page = self.state.pages.get(name) if name else None
-        return page
-
-    # ---- Main loop ----
-    def run(self):
-        with Stealth().use_sync(sync_playwright()) as p:
-            browser = p.chromium.launch(headless=True, args=["--no-sandbox"])
-            self.context = browser.new_context()
-            self._attach_network_listeners(self.context)
-
-            # default first tab
-            tab_name, first_page = self._create_new_tab()
-
-            # Command dispatcher
-            handlers = {
-                "eval": self.handle_eval,
-                "goto": self.handle_goto,
-                "click": self.handle_click,
-                "click_xy": self.handle_click_xy,
-                "type": self.handle_type,
-                "new_tab": self.handle_new_tab,
-                "close_tab": self.handle_close_tab,
-                "switch_tab": self.handle_switch_tab,
-                "inspect": self.handle_inspect,
-                "get_network_logs": self.handle_get_network_logs,
-                "get_console_logs": self.handle_get_console_logs,
-                "clear_logs": self.handle_clear_logs,
-                "start_record": self.handle_start_record,
-                "stop_record": self.handle_stop_record,
-                "play_macro": self.handle_play_macro,
-            }
-
-            while True:
-                task = task_queue.get()
-                cmd = task.get("cmd")
-
-                if cmd == "__EXIT__":
-                    break
-
-                # default outputs
-                result_text = ""
-                screenshot = None
-
-                try:
-                    # decide active page per command
-                    page = self._get_active_page()
-
-                    # macro recording (only for top-level calls, not from_macro)
-                    recordable_cmds = {
-                        "goto", "click", "type", "new_tab", "close_tab", "switch_tab", "click_xy"
-                    }
-                    if (
-                        not task.get("from_macro", False)
-                        and cmd in recordable_cmds
-                    ):
-                        with BROWSER_LOCK:
-                            if self.state.recording:
-                                # store a shallow copy without from_macro
-                                rec = {k: v for k, v in task.items() if k != "from_macro"}
-                                self.state.macro.append(rec)
-
-                    handler = handlers.get(cmd, None)
-                    if handler is None:
-                        result_text = f"Unknown command: {cmd}"
-                        screenshot = take_screenshot(page)
-                    else:
-                        result_text, screenshot = handler(task, page)
-
-                except Exception as e:
-                    result_text = f"Error: {type(e).__name__}: {e}"
-                    # screenshot remains None on error
-
-                result_queue.put((result_text, screenshot))
-
-            # graceful shutdown
-            try:
-                browser.close()
-            except Exception:
-                pass
-
-
-    # ─────────────────────────────
-    # Command Handlers
-    # ─────────────────────────────
-
-    def handle_eval(self, task: dict, page: Optional[Page]):
-        """
-        Evaluate python code in a restricted environment.
-        ⚠ Still not perfectly 'secure', only for trusted usage.
-        """
-        if page is None:
-            return "No active page.", None
-
-        code = task.get("code", "")
-        safe_globals = {"__builtins__": {}}  # no builtins
-        safe_locals = {}
-
-        try:
-            result = eval(code, safe_globals, safe_locals)
-        except Exception as e:
-            return f"Eval error: {type(e).__name__}: {e}", take_screenshot(page)
-
-        text = f"Eval result: {result!r}"
-        return text, take_screenshot(page)
-
-    def handle_goto(self, task: dict, page: Optional[Page]):
-        if page is None:
-            return "No active page.", None
-        url = task.get("url", "")
-        try:
-            page.goto(url)
-            return f"Navigated to {url}", take_screenshot(page)
-        except Exception as e:
-            return f"Goto error: {e}", take_screenshot(page)
-
-    def handle_click(event: gr.SelectData, click_type, last_screenshot):
-        x, y = event.index
-        img_w, img_h = last_screenshot.size
-
-        task_queue.put({
-            "cmd": "click_xy",
-            "x": x,
-            "y": y,
-            "img_w": img_w,
-            "img_h": img_h,
-            "click_type": click_type
-        })
-        return result_queue.get()
-
-    def handle_click_xy(self, task: dict, page: Optional[Page]):
-        if page is None:
-            return "No active page.", None
-
-        x = task.get("x")
-        y = task.get("y")
-        img_w = task.get("img_w")
-        img_h = task.get("img_h")
-        click_type = task.get("click_type", "left")
-
-        # viewport size
-        vp = page.viewport_size or {"width": img_w, "height": img_h}
-
-        real_x = x * (vp["width"] / img_w)
-        real_y = y * (vp["height"] / img_h)
-
-        if click_type == "left":
-            page.mouse.click(real_x, real_y)
-        elif click_type == "double":
-            page.mouse.dblclick(real_x, real_y)
-        elif click_type == "right":
-            page.mouse.click(real_x, real_y, button="right")
-        elif click_type == "hover":
-            page.mouse.move(real_x, real_y)
-
-        return f"{click_type} click at {real_x},{real_y}", take_screenshot(page)
-
-    def handle_type(self, task: dict, page: Optional[Page]):
-        if page is None:
-            return "No active page.", None
-        selector = task.get("selector", "")
-        text = task.get("text", "")
-        try:
-            page.fill(selector, text)
-            return f"Typed into {selector}: {text}", take_screenshot(page)
-        except Exception as e:
-            return f"Type error: {e}", take_screenshot(page)
-
-    def handle_new_tab(self, task: dict, page: Optional[Page]):
-        tab_name, new_page = self._create_new_tab()
-        return f"Opened new tab: {tab_name}", take_screenshot(new_page)
-
-    def handle_close_tab(self, task: dict, page: Optional[Page]):
-        name = task.get("tab", "")
-        with BROWSER_LOCK:
-            if name in self.state.pages:
-                try:
-                    self.state.pages[name].close()
-                except Exception:
-                    pass
-                del self.state.pages[name]
-                msg = f"Closed {name}"
-                # re-select active tab
-                if self.state.active_page == name:
-                    if self.state.pages:
-                        self.state.active_page = list(self.state.pages.keys())[0]
-                    else:
-                        self.state.active_page = None
-            else:
-                msg = f"Tab {name} not found."
-
-            active_name = self.state.active_page
-            active_page = self.state.pages.get(active_name) if active_name else None
-
-        return msg, take_screenshot(active_page)
-
-    def handle_switch_tab(self, task: dict, page: Optional[Page]):
-        name = task.get("tab", "")
-        with BROWSER_LOCK:
-            if name in self.state.pages:
-                self.state.active_page = name
-                active_page = self.state.pages[name]
-                msg = f"Switched to {name}"
-            else:
-                active_page = self._get_active_page()
-                msg = f"Tab {name} not found."
-
-        return msg, take_screenshot(active_page)
-
-    def handle_inspect(self, task: dict, page: Optional[Page]):
-        if page is None:
-            return "No active page.", None
-        selector = task.get("selector", "")
-        try:
-            el = page.query_selector(selector)
-            if not el:
-                return f"No element found for selector: {selector}", take_screenshot(page)
-
-            inner_text = el.inner_text()
-            inner_html = el.inner_html()
-            attrs = page.evaluate(
-                """(el) => {
-                    const out = {};
-                    for (const a of el.attributes) out[a.name] = a.value;
-                    return out;
-                }""",
-                el
-            )
-            xpath = page.evaluate(
-                """(el) => {
-                    function getXPath(node) {
-                        if (node.id)
-                            return 'id(\"' + node.id + '\")';
-                        if (node === document.body)
-                            return '/html/body';
-                        let ix = 0;
-                        const siblings = node.parentNode ? node.parentNode.childNodes : [];
-                        for (let i=0; i<siblings.length; i++) {
-                            const sibling = siblings[i];
-                            if (sibling === node)
-                                return getXPath(node.parentNode) + '/' + node.tagName.toLowerCase() + '[' + (ix+1) + ']';
-                            if (sibling.nodeType === 1 && sibling.tagName === node.tagName)
-                                ix++;
-                        }
-                    }
-                    return getXPath(el);
-                }""",
-                el
-            )
-
-            info = (
-                f"Selector: {selector}\n"
-                f"XPath: {xpath}\n\n"
-                f"Inner Text:\n{inner_text}\n\n"
-                f"Attributes:\n{attrs}\n\n"
-                f"Inner HTML (truncated):\n{inner_html[:1000]}"
-            )
-            return info, take_screenshot(page)
-        except Exception as e:
-            return f"Inspect error: {e}", take_screenshot(page)
-
-    def handle_get_network_logs(self, task: dict, page: Optional[Page]):
-        with BROWSER_LOCK:
-            logs = list(self.state.network_logs)[-100:]
-        text = "=== Network Logs (last 100) ===\n" + "\n".join(logs)
-        return text, take_screenshot(page)
-
-    def handle_get_console_logs(self, task: dict, page: Optional[Page]):
-        with BROWSER_LOCK:
-            logs = list(self.state.console_logs)[-100:]
-        text = "=== Console Logs (last 100) ===\n" + "\n".join(logs)
-        return text, take_screenshot(page)
-
-    def handle_clear_logs(self, task: dict, page: Optional[Page]):
-        with BROWSER_LOCK:
-            self.state.network_logs.clear()
-            self.state.console_logs.clear()
-        return "Network & console logs cleared.", take_screenshot(page)
-
-    def handle_start_record(self, task: dict, page: Optional[Page]):
-        with BROWSER_LOCK:
-            self.state.recording = True
-            self.state.macro = []
-        return "Macro recording started.", take_screenshot(page)
-
-    def handle_stop_record(self, task: dict, page: Optional[Page]):
-        with BROWSER_LOCK:
-            self.state.recording = False
-            steps = len(self.state.macro)
-        return f"Macro recording stopped. {steps} steps recorded.", take_screenshot(page)
-
-    def handle_play_macro(self, task: dict, page: Optional[Page]):
-        with BROWSER_LOCK:
-            macro_steps = list(self.state.macro)
-
-        if not macro_steps:
-            return "Macro is empty.", take_screenshot(self._get_active_page())
-
-        last_result = ""
-        current_page = self._get_active_page()
-
-        for step in macro_steps:
-            step_cmd = dict(step)
-            step_cmd["from_macro"] = True
-            cmd = step_cmd.get("cmd")
-
-            # use same handlers (no re-queue, direct call)
-            if cmd == "goto":
-                current_page = self._get_active_page()
-                last_result, _ = self.handle_goto(step_cmd, current_page)
-            elif cmd == "click":
-                current_page = self._get_active_page()
-                last_result, _ = self.handle_click(step_cmd, current_page)
-            elif cmd == "type":
-                current_page = self._get_active_page()
-                last_result, _ = self.handle_type(step_cmd, current_page)
-            elif cmd == "new_tab":
-                last_result, _ = self.handle_new_tab(step_cmd, current_page)
-                current_page = self._get_active_page()
-            elif cmd == "close_tab":
-                last_result, _ = self.handle_close_tab(step_cmd, current_page)
-                current_page = self._get_active_page()
-            elif cmd == "switch_tab":
-                last_result, _ = self.handle_switch_tab(step_cmd, current_page)
-                current_page = self._get_active_page()
-            elif cmd == "click_xy":
-                current_page = self._get_active_page()
-                last_result, _ = self.handle_click_xy(step_cmd, current_page)
-
-        final_page = self._get_active_page()
-        return f"Macro executed. {len(macro_steps)} steps.\nLast step: {last_result}", take_screenshot(final_page)
-
-
-# ─────────────────────────────
-# Worker spawn / shutdown helpers
-# ─────────────────────────────
-
-def start_worker_thread():
-    worker = PlaywrightWorker(BROWSER_STATE)
-    t = threading.Thread(target=worker.run, daemon=True)
-    t.start()
-    with BROWSER_LOCK:
-        BROWSER_STATE.thread = t
-        BROWSER_STATE.running = True
-        # reset state
-        BROWSER_STATE.pages.clear()
-        BROWSER_STATE.active_page = None
-        BROWSER_STATE.network_logs.clear()
-        BROWSER_STATE.console_logs.clear()
-        BROWSER_STATE.macro.clear()
-        BROWSER_STATE.recording = False
-        BROWSER_STATE.tab_counter = 0
-
-
-def stop_worker_thread():
-    with BROWSER_LOCK:
-        if not BROWSER_STATE.running:
-            return
-        BROWSER_STATE.running = False
-        t = BROWSER_STATE.thread
-
-    task_queue.put({"cmd": "__EXIT__"})
-    if t is not None:
-        t.join(timeout=5)
-
-
-# ─────────────────────────────
-# Frontend functions (Gradio)
-# ─────────────────────────────
-
-def start_browser():
-    with BROWSER_LOCK:
-        running = BROWSER_STATE.running
-
-    if running:
-        with BROWSER_LOCK:
-            tabs = list(BROWSER_STATE.pages.keys())
-            active = BROWSER_STATE.active_page
-        return "Browser is already running.", None, gr.update(choices=tabs, value=active)
-
-    start_worker_thread()
-    # first tab name predictable is Tab-1; but worker will create it on start
-    return "Browser Started!", None, gr.update(choices=["Tab-1"], value="Tab-1")
-
-
-def stop_browser():
-    with BROWSER_LOCK:
-        running = BROWSER_STATE.running
-
-    if not running:
-        return "Browser is not running.", None, gr.update(choices=[], value=None)
-
-    stop_worker_thread()
-    with BROWSER_LOCK:
-        BROWSER_STATE.pages.clear()
-        BROWSER_STATE.active_page = None
-
-    return "Browser Closed!", None, gr.update(choices=[], value=None)
-
-
-def execute_code(code):
-    with BROWSER_LOCK:
-        if not BROWSER_STATE.running:
-            return "Start browser first.", None
-    task_queue.put({"cmd": "eval", "code": code})
-    return result_queue.get()
-
-
-def navigate(url):
-    with BROWSER_LOCK:
-        if not BROWSER_STATE.running:
-            return "Start browser first.", None
-    task_queue.put({"cmd": "goto", "url": url})
-    return result_queue.get()
-
-
-def click(selector):
-    with BROWSER_LOCK:
-        if not BROWSER_STATE.running:
-            return "Start browser first.", None
-    task_queue.put({"cmd": "click", "selector": selector})
-    return result_queue.get()
-
-
-def type_text(selector, text):
-    with BROWSER_LOCK:
-        if not BROWSER_STATE.running:
-            return "Start browser first.", None
-    task_queue.put({"cmd": "type", "selector": selector, "text": text})
-    return result_queue.get()
-
-
-def new_tab():
-    with BROWSER_LOCK:
-        if not BROWSER_STATE.running:
-            return "Start browser first.", None
-    task_queue.put({"cmd": "new_tab"})
-    r, screenshot = result_queue.get()
-    with BROWSER_LOCK:
-        tabs = list(BROWSER_STATE.pages.keys())
-        active = BROWSER_STATE.active_page
-    return r, screenshot, gr.update(choices=tabs, value=active)
-
-
-def close_tab(tab):
-    with BROWSER_LOCK:
-        if not BROWSER_STATE.running:
-            return "Start browser first.", None
-    task_queue.put({"cmd": "close_tab", "tab": tab})
-    r, screenshot = result_queue.get()
-    with BROWSER_LOCK:
-        tabs = list(BROWSER_STATE.pages.keys())
-        active = BROWSER_STATE.active_page
-    return r, screenshot, gr.update(choices=tabs, value=active)
-
-
-def switch_tab(tab):
-    with BROWSER_LOCK:
-        if not BROWSER_STATE.running:
-            return "Start browser first.", None
-    task_queue.put({"cmd": "switch_tab", "tab": tab})
-    r, screenshot = result_queue.get()
-    with BROWSER_LOCK:
-        tabs = list(BROWSER_STATE.pages.keys())
-        active = BROWSER_STATE.active_page
-    return r, screenshot, gr.update(choices=tabs, value=active)
-
-
-def inspect_element(selector):
-    with BROWSER_LOCK:
-        if not BROWSER_STATE.running:
-            return "Start browser first.", None
-    task_queue.put({"cmd": "inspect", "selector": selector})
-    return result_queue.get()
-
-
-def handle_click(event: gr.SelectData, click_type):
-    x, y = event.index
-    with BROWSER_LOCK:
-        if not BROWSER_STATE.running:
-            return "Start browser first.", None
-
-    task_queue.put({
-        "cmd": "click_xy",
-        "x": x,
-        "y": y,
-        "click_type": click_type
-    })
-    return result_queue.get()
-
-
-def show_network_logs():
-    with BROWSER_LOCK:
-        if not BROWSER_STATE.running:
-            return "Start browser first.", None
-    task_queue.put({"cmd": "get_network_logs"})
-    return result_queue.get()
-
-
-def show_console_logs():
-    with BROWSER_LOCK:
-        if not BROWSER_STATE.running:
-            return "Start browser first.", None
-    task_queue.put({"cmd": "get_console_logs"})
-    return result_queue.get()
-
-
-def clear_logs():
-    with BROWSER_LOCK:
-        if not BROWSER_STATE.running:
-            return "Start browser first.", None
-    task_queue.put({"cmd": "clear_logs"})
-    return result_queue.get()
-
-
-def start_recording():
-    with BROWSER_LOCK:
-        if not BROWSER_STATE.running:
-            return "Start browser first.", None
-    task_queue.put({"cmd": "start_record"})
-    return result_queue.get()
-
-
-def stop_recording():
-    with BROWSER_LOCK:
-        if not BROWSER_STATE.running:
-            return "Start browser first.", None
-    task_queue.put({"cmd": "stop_record"})
-    return result_queue.get()
-
-
-def play_macro():
-    with BROWSER_LOCK:
-        if not BROWSER_STATE.running:
-            return "Start browser first.", None
-    task_queue.put({"cmd": "play_macro"})
-    return result_queue.get()
-
-
-# ─────────────────────────────
-# Gradio UI (slightly cleaner layout)
-# ─────────────────────────────
-
-with gr.Blocks() as app:
-    gr.Markdown("## 🔥 Advanced Playwright Control Panel (DOM + Logs + Macro RPA)")
-
-    with gr.Row():
-        start_btn = gr.Button("Open Browser")
-        stop_btn = gr.Button("Close Browser")
-
-    tabs_dropdown = gr.Dropdown(label="Tabs", choices=[], value=None)
-
-    with gr.Tab("Browse"):
-        with gr.Row():
-            url_box = gr.Textbox(label="URL", scale=4)
-            nav_btn = gr.Button("Go", scale=1)
-
-        with gr.Row():
-            sel_box = gr.Textbox(label="Selector (CSS)", scale=3)
-            click_btn = gr.Button("Click", scale=1)
-            type_box = gr.Textbox(label="Type Text", scale=3)
-            type_btn = gr.Button("Type", scale=1)
-
-    with gr.Tab("Inspect / Code"):
-        with gr.Row():
-            inspect_sel = gr.Textbox(label="Inspect Selector (CSS)")
-            inspect_btn = gr.Button("Inspect + Generate XPath")
-        code_input = gr.TextArea(label="Python Code (eval in Playwright worker - restricted)")
-        run_btn = gr.Button("Run Code")
-
-    with gr.Tab("Logs"):
-        with gr.Row():
-            net_btn = gr.Button("Show Network Logs")
-            cons_btn = gr.Button("Show Console Logs")
-            clear_logs_btn = gr.Button("Clear Logs")
-
-    with gr.Tab("Tabs & Macros"):
-        with gr.Row():
-            new_tab_btn = gr.Button("New Tab")
-            close_tab_btn = gr.Button("Close Selected Tab")
-            switch_tab_btn = gr.Button("Switch Tab")
-
-        with gr.Row():
-            start_rec_btn = gr.Button("Start Macro Recording")
-            stop_rec_btn = gr.Button("Stop Recording")
-            play_macro_btn = gr.Button("Play Recorded Macro")
-
-    output_text = gr.TextArea(label="Output")
-    output_image = gr.Image(label="Screenshot")
-    click_type = gr.Radio(
-        ["left", "double", "right", "hover"],
-        value="left",
-        label="Click Type"
-    )
-
-    output_image.select(
-        handle_click,
-        inputs=[click_type, output_image],
-        outputs=[output_text, output_image]
-    )
-
-    # Bindings
-    start_btn.click(start_browser, outputs=[output_text, output_image, tabs_dropdown])
-    stop_btn.click(stop_browser, outputs=[output_text, output_image, tabs_dropdown])
-
-    new_tab_btn.click(new_tab, outputs=[output_text, output_image, tabs_dropdown])
-    close_tab_btn.click(close_tab, inputs=tabs_dropdown, outputs=[output_text, output_image, tabs_dropdown])
-    switch_tab_btn.click(switch_tab, inputs=tabs_dropdown, outputs=[output_text, output_image, tabs_dropdown])
-
-    nav_btn.click(navigate, inputs=url_box, outputs=[output_text, output_image])
-    click_btn.click(click, inputs=sel_box, outputs=[output_text, output_image])
-    type_btn.click(type_text, inputs=[sel_box, type_box], outputs=[output_text, output_image])
-
-    run_btn.click(execute_code, inputs=code_input, outputs=[output_text, output_image])
-    inspect_btn.click(inspect_element, inputs=inspect_sel, outputs=[output_text, output_image])
-
-    net_btn.click(show_network_logs, outputs=[output_text, output_image])
-    cons_btn.click(show_console_logs, outputs=[output_text, output_image])
-    clear_logs_btn.click(clear_logs, outputs=[output_text, output_image])
-
-    start_rec_btn.click(start_recording, outputs=[output_text, output_image])
-    stop_rec_btn.click(stop_recording, outputs=[output_text, output_image])
-    play_macro_btn.click(play_macro, outputs=[output_text, output_image])
-
-
-if __name__ == "__main__":
-    app.launch()