File size: 3,984 Bytes
c52954a | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 | 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)}
|