pinchtab_browser.py

import httpx
import logging
from typing import Dict, Any, Optional

logger = logging.getLogger(__name__)

class PinchTabBrowser:
    """Asynchronous wrapper for the PinchTab HTTP API."""

    def __init__(self, base_url: str = "http://localhost:9867"):
        self.base_url = base_url.rstrip("/")
        self.instance_id = None
        self.tab_id = None
        self._async_client = httpx.AsyncClient(timeout=60.0)

    async def close(self):
        await self._async_client.aclose()

    async def _get_instance(self) -> str:
        if self.instance_id:
            return self.instance_id

        try:
            resp = await self._async_client.get(f"{self.base_url}/instances")
            instances = resp.json()
            if instances and isinstance(instances, list):
                self.instance_id = instances[0]["id"]
                return self.instance_id
        except Exception as e:
            logger.error(f"Failed to get instances: {e}")

        try:
            resp = await self._async_client.post(f"{self.base_url}/instances/start", json={"mode": "headless"})
            data = resp.json()
            self.instance_id = data.get("id")
            return self.instance_id
        except Exception as e:
            logger.error(f"Failed to start instance: {e}")
            raise

    async def _instance_url(self) -> str:
        iid = await self._get_instance()
        try:
            resp = await self._async_client.get(f"{self.base_url}/instances")
            instances = resp.json()
            for inst in instances:
                if inst["id"] == iid:
                    # Use the same hostname as base_url but with the instance's port
                    from urllib.parse import urlparse
                    parsed = urlparse(self.base_url)
                    return f"{parsed.scheme}://{parsed.hostname}:{inst['port']}"
        except:
            pass
        return f"{self.base_url}/instances/{iid}"

    async def navigate(self, url: str) -> Dict[str, Any]:
        try:
            target_url = await self._instance_url()
            resp = await self._async_client.post(f"{target_url}/navigate", json={"url": url})
            try:
                data = resp.json()
            except:
                data = {"status": "ok", "text": resp.text}
            if "tabId" in data:
                self.tab_id = data["tabId"]
            return data
        except Exception as e:
            return {"status": "error", "message": str(e)}

    async def snapshot(self, filter_mode: str = "interactive") -> Dict[str, Any]:
        try:
            target_url = await self._instance_url()
            params = {"filter": filter_mode}
            if self.tab_id:
                params["tabId"] = self.tab_id
            resp = await self._async_client.get(f"{target_url}/snapshot", params=params)
            return resp.json()
        except Exception as e:
            return {"status": "error", "message": str(e)}

    async def action(self, kind: str, ref: Optional[str] = None, text: Optional[str] = None, key: Optional[str] = None) -> Dict[str, Any]:
        try:
            target_url = await self._instance_url()
            body = {"kind": kind}
            if ref: body["ref"] = ref
            if text: body["text"] = text
            if key: body["key"] = key
            if self.tab_id: body["tabId"] = self.tab_id

            resp = await self._async_client.post(f"{target_url}/action", json=body)
            return resp.json()
        except Exception as e:
            return {"status": "error", "message": str(e)}

    async def get_text(self) -> Dict[str, Any]:
        try:
            target_url = await self._instance_url()
            params = {}
            if self.tab_id: params["tabId"] = self.tab_id
            resp = await self._async_client.get(f"{target_url}/text", params=params)
            return resp.json()
        except Exception as e:
            return {"status": "error", "message": str(e)}