from time import time from asyncio import sleep as asyncio_sleep, Lock from playwright.sync_api._generated import Page from playwright.sync_api import ( Frame, BrowserContext, Playwright, Response as SyncPlaywrightResponse, ) from playwright.async_api._generated import Page as AsyncPage from playwright.async_api import ( Frame as AsyncFrame, Playwright as AsyncPlaywright, Response as AsyncPlaywrightResponse, BrowserContext as AsyncBrowserContext, ) from playwright._impl._errors import Error as PlaywrightError from ._page import PageInfo, PagePool from scrapling.parser import Selector from ._validators import validate, PlaywrightConfig, StealthConfig from ._config_tools import __default_chrome_useragent__, __default_useragent__ from scrapling.engines.toolbelt.navigation import intercept_route, async_intercept_route from scrapling.core._types import Any, cast, Dict, List, Optional, Callable, TYPE_CHECKING, overload, Tuple from scrapling.engines.constants import ( DEFAULT_STEALTH_FLAGS, HARMFUL_DEFAULT_ARGS, DEFAULT_FLAGS, ) class SyncSession: def __init__(self, max_pages: int = 1): self.max_pages = max_pages self.page_pool = PagePool(max_pages) self._max_wait_for_page = 60 self.playwright: Playwright | Any = None self.context: BrowserContext | Any = None self._closed = False def start(self): pass def close(self): # pragma: no cover """Close all resources""" if self._closed: return if self.context: self.context.close() self.context = None if self.playwright: self.playwright.stop() self.playwright = None # pyright: ignore self._closed = True def __enter__(self): self.start() return self def __exit__(self, exc_type, exc_val, exc_tb): self.close() def _get_page( self, timeout: int | float, extra_headers: Optional[Dict[str, str]], disable_resources: bool, ) -> PageInfo[Page]: # pragma: no cover """Get a new page to use""" # No need to check if a page is available or not in sync code because the code blocked before reaching here till the page closed, ofc. assert self.context is not None, "Browser context not initialized" page = self.context.new_page() page.set_default_navigation_timeout(timeout) page.set_default_timeout(timeout) if extra_headers: page.set_extra_http_headers(extra_headers) if disable_resources: page.route("**/*", intercept_route) page_info = self.page_pool.add_page(page) page_info.mark_busy() return page_info def get_pool_stats(self) -> Dict[str, int]: """Get statistics about the current page pool""" return { "total_pages": self.page_pool.pages_count, "busy_pages": self.page_pool.busy_count, "max_pages": self.max_pages, } @staticmethod def _wait_for_networkidle(page: Page | Frame, timeout: Optional[int] = None): """Wait for the page to become idle (no network activity) even if there are never-ending requests.""" try: page.wait_for_load_state("networkidle", timeout=timeout) except (PlaywrightError, Exception): pass def _wait_for_page_stability(self, page: Page | Frame, load_dom: bool, network_idle: bool): page.wait_for_load_state(state="load") if load_dom: page.wait_for_load_state(state="domcontentloaded") if network_idle: self._wait_for_networkidle(page) @staticmethod def _create_response_handler(page_info: PageInfo[Page], response_container: List) -> Callable: """Create a response handler that captures the final navigation response. :param page_info: The PageInfo object containing the page :param response_container: A list to store the final response (mutable container) :return: A callback function for page.on("response", ...) """ def handle_response(finished_response: SyncPlaywrightResponse): if ( finished_response.request.resource_type == "document" and finished_response.request.is_navigation_request() and finished_response.request.frame == page_info.page.main_frame ): response_container[0] = finished_response return handle_response class AsyncSession: def __init__(self, max_pages: int = 1): self.max_pages = max_pages self.page_pool = PagePool(max_pages) self._max_wait_for_page = 60 self.playwright: AsyncPlaywright | Any = None self.context: AsyncBrowserContext | Any = None self._closed = False self._lock = Lock() async def start(self): pass async def close(self): """Close all resources""" if self._closed: # pragma: no cover return if self.context: await self.context.close() self.context = None # pyright: ignore if self.playwright: await self.playwright.stop() self.playwright = None # pyright: ignore self._closed = True async def __aenter__(self): await self.start() return self async def __aexit__(self, exc_type, exc_val, exc_tb): await self.close() async def _get_page( self, timeout: int | float, extra_headers: Optional[Dict[str, str]], disable_resources: bool, ) -> PageInfo[AsyncPage]: # pragma: no cover """Get a new page to use""" if TYPE_CHECKING: assert self.context is not None, "Browser context not initialized" async with self._lock: # If we're at max capacity after cleanup, wait for busy pages to finish if self.page_pool.pages_count >= self.max_pages: start_time = time() while time() - start_time < self._max_wait_for_page: await asyncio_sleep(0.05) if self.page_pool.pages_count < self.max_pages: break else: raise TimeoutError( f"No pages finished to clear place in the pool within the {self._max_wait_for_page}s timeout period" ) page = await self.context.new_page() page.set_default_navigation_timeout(timeout) page.set_default_timeout(timeout) if extra_headers: await page.set_extra_http_headers(extra_headers) if disable_resources: await page.route("**/*", async_intercept_route) return self.page_pool.add_page(page) def get_pool_stats(self) -> Dict[str, int]: """Get statistics about the current page pool""" return { "total_pages": self.page_pool.pages_count, "busy_pages": self.page_pool.busy_count, "max_pages": self.max_pages, } @staticmethod async def _wait_for_networkidle(page: AsyncPage | AsyncFrame, timeout: Optional[int] = None): """Wait for the page to become idle (no network activity) even if there are never-ending requests.""" try: await page.wait_for_load_state("networkidle", timeout=timeout) except (PlaywrightError, Exception): pass async def _wait_for_page_stability(self, page: AsyncPage | AsyncFrame, load_dom: bool, network_idle: bool): await page.wait_for_load_state(state="load") if load_dom: await page.wait_for_load_state(state="domcontentloaded") if network_idle: await self._wait_for_networkidle(page) @staticmethod def _create_response_handler(page_info: PageInfo[AsyncPage], response_container: List) -> Callable: """Create an async response handler that captures the final navigation response. :param page_info: The PageInfo object containing the page :param response_container: A list to store the final response (mutable container) :return: A callback function for page.on("response", ...) """ async def handle_response(finished_response: AsyncPlaywrightResponse): if ( finished_response.request.resource_type == "document" and finished_response.request.is_navigation_request() and finished_response.request.frame == page_info.page.main_frame ): response_container[0] = finished_response return handle_response class BaseSessionMixin: @overload def __validate_routine__(self, params: Dict, model: type[StealthConfig]) -> StealthConfig: ... @overload def __validate_routine__(self, params: Dict, model: type[PlaywrightConfig]) -> PlaywrightConfig: ... def __validate_routine__( self, params: Dict, model: type[PlaywrightConfig] | type[StealthConfig] ) -> PlaywrightConfig | StealthConfig: # Dark color scheme bypasses the 'prefersLightColor' check in creepjs self._context_options: Dict[str, Any] = {"color_scheme": "dark", "device_scale_factor": 2} self._launch_options: Dict[str, Any] = self._context_options | { "args": DEFAULT_FLAGS, "ignore_default_args": HARMFUL_DEFAULT_ARGS, } if "__max_pages" in params: params["max_pages"] = params.pop("__max_pages") config = validate(params, model=model) self._headers_keys = ( {header.lower() for header in config.extra_headers.keys()} if config.extra_headers else set() ) return config def __generate_options__(self, extra_flags: Tuple | None = None) -> None: config = cast(PlaywrightConfig, getattr(self, "_config", None)) self._context_options.update( { "proxy": config.proxy, "locale": config.locale, "timezone_id": config.timezone_id, "extra_http_headers": config.extra_headers, } ) # The default useragent in the headful is always correct now in the current versions of Playwright if config.useragent: self._context_options["user_agent"] = config.useragent elif not config.useragent and config.headless: self._context_options["user_agent"] = ( __default_chrome_useragent__ if config.real_chrome else __default_useragent__ ) if not config.cdp_url: self._launch_options |= self._context_options self._context_options = {} flags = self._launch_options["args"] if config.extra_flags or extra_flags: flags = list(set(flags + (config.extra_flags or extra_flags))) self._launch_options.update( { "args": flags, "headless": config.headless, "user_data_dir": config.user_data_dir, "channel": "chrome" if config.real_chrome else "chromium", } ) if config.additional_args: self._launch_options.update(config.additional_args) else: # while `context_options` is left to be used when cdp mode is enabled self._launch_options = dict() if config.additional_args: self._context_options.update(config.additional_args) class DynamicSessionMixin(BaseSessionMixin): def __validate__(self, **params): self._config = self.__validate_routine__(params, model=PlaywrightConfig) self.__generate_options__() class StealthySessionMixin(BaseSessionMixin): def __validate__(self, **params): self._config: StealthConfig = self.__validate_routine__(params, model=StealthConfig) self._context_options.update( { "is_mobile": False, "has_touch": False, # I'm thinking about disabling it to rest from all Service Workers' headache, but let's keep it as it is for now "service_workers": "allow", "ignore_https_errors": True, "screen": {"width": 1920, "height": 1080}, "viewport": {"width": 1920, "height": 1080}, "permissions": ["geolocation", "notifications"], } ) self.__generate_stealth_options() def __generate_stealth_options(self) -> None: flags = tuple() if not self._config.cdp_url: flags = DEFAULT_FLAGS + DEFAULT_STEALTH_FLAGS if self._config.block_webrtc: flags += ( "--webrtc-ip-handling-policy=disable_non_proxied_udp", "--force-webrtc-ip-handling-policy", # Ensures the policy is enforced ) if not self._config.allow_webgl: flags += ( "--disable-webgl", "--disable-webgl-image-chromium", "--disable-webgl2", ) if self._config.hide_canvas: flags += ("--fingerprinting-canvas-image-data-noise",) super(StealthySessionMixin, self).__generate_options__(flags) @staticmethod def _detect_cloudflare(page_content: str) -> str | None: """ Detect the type of Cloudflare challenge present in the provided page content. This function analyzes the given page content to identify whether a specific type of Cloudflare challenge is present. It checks for three predefined challenge types: non-interactive, managed, and interactive. If a challenge type is detected, it returns the corresponding type as a string. If no challenge type is detected, it returns None. Args: page_content (str): The content of the page to analyze for Cloudflare challenge types. Returns: str: A string representing the detected Cloudflare challenge type, if found. Returns None if no challenge matches. """ challenge_types = ( "non-interactive", "managed", "interactive", ) for ctype in challenge_types: if f"cType: '{ctype}'" in page_content: return ctype # Check if turnstile captcha is embedded inside the page (Usually inside a closed Shadow iframe) selector = Selector(content=page_content) if selector.css('script[src*="challenges.cloudflare.com/turnstile/v"]'): return "embedded" return None