Blablablab/audio-classification
0
1"""2Playwright Fallback for Web Agent Recording3 4Server-side headless browser for sites that block iframe embedding.5Uses Playwright to render pages and stream screenshots back to the client.6 7This is an optional dependency - only imported when needed.8Install with: pip install playwright && playwright install chromium9 10Usage:11 session = PlaywrightSession()12 await session.start("https://example.com")13 screenshot_bytes = await session.screenshot()14 await session.click(100, 200)15 await session.stop()16"""17 18import asyncio19import logging20import os21from typing import Optional, Dict, Any, Tuple22 23logger = logging.getLogger(__name__)24 25 26class PlaywrightSession:27 """28 Manages a headless browser session for web agent recording.29 30 Each session runs a Chromium instance that navigates pages,31 captures screenshots, and executes user interactions.32 """33 34 def __init__(self, width: int = 1280, height: int = 720):35 self.width = width36 self.height = height37 self.browser = None38 self.context = None39 self.page = None40 self._playwright = None41 42 async def start(self, url: str) -> bool:43 """Launch browser and navigate to URL."""44 try:45 from playwright.async_api import async_playwright46 except ImportError:47 logger.error(48 "Playwright is not installed. Install with: "49 "pip install playwright && playwright install chromium"50 )51 return False52 53 try:54 self._playwright = await async_playwright().start()55 56 # Launch with flags that reduce bot detection signals.57 # Sites use these JS-side checks to trigger captchas:58 # navigator.webdriver, missing plugins, headless UA, etc.59 self.browser = await self._playwright.chromium.launch(60 headless=True,61 args=[62 "--disable-blink-features=AutomationControlled",63 "--disable-infobars",64 "--no-first-run",65 "--no-default-browser-check",66 "--disable-background-timer-throttling",67 "--disable-backgrounding-occluded-windows",68 "--disable-renderer-backgrounding",69 ],70 )71 72 # Current, realistic user-agent and locale/timezone73 self.context = await self.browser.new_context(74 viewport={"width": self.width, "height": self.height},75 user_agent=(76 "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) "77 "AppleWebKit/537.36 (KHTML, like Gecko) "78 "Chrome/131.0.0.0 Safari/537.36"79 ),80 locale="en-US",81 timezone_id="America/New_York",82 color_scheme="light",83 )84 85 self.page = await self.context.new_page()86 87 # Remove the navigator.webdriver flag that headless Chromium sets.88 # This is the single most common bot-detection signal.89 await self.page.add_init_script("""90 Object.defineProperty(navigator, 'webdriver', {91 get: () => undefined,92 });93 // Fake plugins array (headless has 0 plugins)94 Object.defineProperty(navigator, 'plugins', {95 get: () => [1, 2, 3, 4, 5],96 });97 // Fake languages98 Object.defineProperty(navigator, 'languages', {99 get: () => ['en-US', 'en'],100 });101 // Remove chrome.runtime that automation sets102 if (window.chrome) {103 window.chrome.runtime = undefined;104 }105 """)106 107 await self.page.goto(url, wait_until="domcontentloaded", timeout=30000)108 logger.info(f"Playwright session started at {url}")109 return True110 except Exception as e:111 logger.error(f"Failed to start Playwright session: {e}")112 await self.stop()113 return False114 115 async def screenshot(self) -> Optional[bytes]:116 """Capture current page screenshot as PNG bytes."""117 if not self.page:118 return None119 try:120 return await self.page.screenshot(type="png")121 except Exception as e:122 logger.error(f"Screenshot failed: {e}")123 return None124 125 async def click(self, x: int, y: int) -> bool:126 """Execute click at coordinates."""127 if not self.page:128 return False129 try:130 await self.page.mouse.click(x, y)131 await self.page.wait_for_load_state("domcontentloaded", timeout=5000)132 return True133 except Exception as e:134 logger.warning(f"Click failed at ({x}, {y}): {e}")135 return False136 137 async def type_text(self, text: str) -> bool:138 """Type text into the currently focused element."""139 if not self.page:140 return False141 try:142 await self.page.keyboard.type(text)143 return True144 except Exception as e:145 logger.warning(f"Type failed: {e}")146 return False147 148 async def scroll(self, dx: int = 0, dy: int = 0) -> bool:149 """Scroll the page."""150 if not self.page:151 return False152 try:153 await self.page.mouse.wheel(dx, dy)154 return True155 except Exception as e:156 logger.warning(f"Scroll failed: {e}")157 return False158 159 async def navigate(self, url: str) -> bool:160 """Navigate to a new URL."""161 if not self.page:162 return False163 try:164 await self.page.goto(url, wait_until="domcontentloaded", timeout=30000)165 return True166 except Exception as e:167 logger.warning(f"Navigation failed to {url}: {e}")168 return False169 170 async def get_state(self) -> Dict[str, Any]:171 """Get current page state."""172 if not self.page:173 return {}174 try:175 return {176 "url": self.page.url,177 "title": await self.page.title(),178 "viewport": {"width": self.width, "height": self.height},179 }180 except Exception:181 return {}182 183 async def stop(self):184 """Close browser and clean up."""185 try:186 if self.browser:187 await self.browser.close()188 if self._playwright:189 await self._playwright.stop()190 except Exception as e:191 logger.warning(f"Cleanup error: {e}")192 finally:193 self.browser = None194 self.context = None195 self.page = None196 self._playwright = None197 198 199def check_playwright_available() -> bool:200 """Check if Playwright is installed and has browsers."""201 try:202 import playwright203 return True204 except ImportError:205 return False206 