import asyncio from playwright.async_api import ( async_playwright, TimeoutError as PlaywrightTimeoutError, Error as PlaywrightError ) from utils.logger import logger from utils.retry import retry_async from urllib.parse import urlparse def is_binary_file(url: str): path = urlparse(url).path.lower() binary_ext = ( ".xls", ".xlsx", ".pdf", ".zip", ".tar", ".gz", ".doc", ".docx", ".ppt", ".pptx", ".xlsm", ".csv" ) return any(path.endswith(ext) for ext in binary_ext) class PlaywrightClient: """ Production-ready Playwright wrapper. """ def __init__( self, timeout=45000, headless=True, browser_type="chromium", content_selectors=None, selector_wait_ms=3000, blocked_resource_types=None, ): self.timeout = timeout self.headless = headless self.browser_type = browser_type # Selectors the caller considers "main content". If # provided, we wait for any of them to appear instead # of doing a blind fixed sleep — saves seconds per # page on docs that DOM-render quickly. self.content_selectors = content_selectors or [] self.selector_wait_ms = selector_wait_ms # Resource types we abort at the network layer to # avoid downloading bytes Chromium will never use for # extraction. Images and stylesheets are also fetched # separately by AssetDownloader from the raw HTML, so # the browser doesn't need them. self.blocked_resource_types = set( blocked_resource_types if blocked_resource_types is not None else ("image", "stylesheet", "media", "font") ) self.playwright = None self.browser = None self.context = None async def start(self): """ Start Playwright browser. """ logger.info( "Starting Playwright..." ) self.playwright = await async_playwright().start() browser_launcher = getattr( self.playwright, self.browser_type ) self.browser = await browser_launcher.launch( headless=self.headless, args=[ "--disable-blink-features=AutomationControlled", "--disable-dev-shm-usage", "--no-sandbox", "--disable-setuid-sandbox", "--disable-gpu" ] ) self.context = await self.browser.new_context( viewport={ "width": 1600, "height": 900 }, user_agent=( "Mozilla/5.0 " "(Windows NT 10.0; Win64; x64) " "AppleWebKit/537.36 " "(KHTML, like Gecko) " "Chrome/124.0.0.0 " "Safari/537.36" ), locale="en-US", java_script_enabled=True, ignore_https_errors=True ) logger.info( "Playwright started." ) async def stop(self): """ Shutdown browser safely. """ logger.info( "Stopping Playwright..." ) try: if self.context: await self.context.close() if self.browser: await self.browser.close() if self.playwright: await self.playwright.stop() except Exception as e: logger.warning( f"Playwright stop error: {e}" ) logger.info( "Playwright stopped." ) @retry_async( attempts=3, min_wait=2, max_wait=10 ) async def fetch_page(self, url: str): """ Render and fetch page safely (production version). """ logger.info(f"Fetching page: {url}") if is_binary_file(url): logger.info(f"Detected binary file, skipping as page: {url}") return { "url": url, "title": url.split("/")[-1], "html": "", "status_code": None, "headers": {}, "is_binary": True, "content_type": "" } page = None try: page = await self.context.new_page() # Apply routing (ONLY HTML requests should be intercepted) await page.route( "**/*", self.route_interceptor ) response = await page.goto( url, wait_until="domcontentloaded", timeout=self.timeout ) # Wait only as long as it takes for actual content # to appear, instead of the previous fixed 5s # load-state wait + 0.5s sleep. If the caller did # not configure content selectors, fall back to # the cheaper load-state wait. if self.content_selectors: joined = ",".join(self.content_selectors) try: await page.wait_for_selector( joined, timeout=self.selector_wait_ms, ) except PlaywrightTimeoutError: logger.debug( f"Content selector not seen within " f"{self.selector_wait_ms}ms, " f"continuing anyway: {url}" ) else: try: await page.wait_for_load_state( "load", timeout=self.selector_wait_ms, ) except PlaywrightTimeoutError: logger.debug( f"Load state timeout, continuing " f"anyway: {url}" ) title = await page.title() html = await page.content() final_url = page.url status = response.status if response else None headers = response.headers if response else {} logger.info(f"Fetched page successfully: {url}") return { "url": final_url, "title": title, "html": html, "status_code": status, "headers": headers, "is_binary": False } except PlaywrightTimeoutError: logger.error(f"Timeout while fetching: {url}") raise except PlaywrightError as e: if self.is_transient_error(e): logger.warning( f"Transient Playwright error for {url}: {e}" ) else: logger.exception( f"Non-transient Playwright error for {url}: {e}" ) raise except Exception as e: logger.exception( f"Unexpected fetch failed {url}: {e}" ) raise finally: if page is not None: try: await page.close() except Exception as close_err: logger.warning( f"Page close error for {url}: {close_err}" ) async def route_interceptor( self, route ): """ Block unnecessary resources to speed up crawling. Image and stylesheet bytes never affect HTML extraction; assets are downloaded separately by AssetDownloader from the raw HTML. Aborting them at the network layer drops the per-page wire weight by a large factor on doc sites. """ if route.request.resource_type in self.blocked_resource_types: await route.abort() return await route.continue_() async def screenshot( self, url, output_path ): """ Debug screenshot utility. """ page = None try: page = await self.context.new_page() await page.goto( url, wait_until="domcontentloaded", timeout=self.timeout ) await page.screenshot( path=output_path, full_page=True ) logger.info( f"Screenshot saved: " f"{output_path}" ) finally: if page is not None: await page.close() async def evaluate_js( self, url, script ): """ Execute JS on page. """ page = None try: page = await self.context.new_page() await page.goto( url, wait_until="domcontentloaded", timeout=self.timeout ) result = await page.evaluate( script ) return result finally: if page is not None: await page.close() async def fetch_navigation_links( self, url ): """ Extract rendered navigation links. """ page = None try: page = await self.context.new_page() await page.goto( url, wait_until="domcontentloaded", timeout=self.timeout ) links = await page.evaluate(""" () => { return Array.from( document.querySelectorAll('a') ).map(a => a.href) } """) return links finally: if page is not None: await page.close() def is_transient_error(self, error: Exception): message = str(error).lower() transient_keywords = [ "err_connection_refused", "err_connection_reset", "err_connection_closed", "err_timed_out", "timeout", "net::err", "target closed", "browser has been closed", "page crashed" ] return any( keyword in message for keyword in transient_keywords )