Fetcher-Klassen¶

@classmethod
def display_config(cls):
    return dict(
        huge_tree=cls.huge_tree,
        keep_comments=cls.keep_comments,
        adaptive=cls.adaptive,
        storage=cls.storage,
        storage_args=cls.storage_args,
        adaptive_domain=cls.adaptive_domain,
    )

@classmethod
def configure(cls, **kwargs):
    """Set multiple arguments for the parser at once globally

    :param kwargs: The keywords can be any arguments of the following: huge_tree, keep_comments, adaptive, storage, storage_args, adaptive_domain
    """
    for key, value in kwargs.items():
        key = key.strip().lower()
        if hasattr(cls, key):
            if key in cls.parser_keywords:
                setattr(cls, key, value)
            else:
                # Yup, no fun allowed LOL
                raise AttributeError(f'Unknown parser argument: "{key}"; maybe you meant {cls.parser_keywords}?')
        else:
            raise ValueError(f'Unknown parser argument: "{key}"; maybe you meant {cls.parser_keywords}?')

    if not kwargs:
        raise AttributeError(f"You must pass a keyword to configure, current keywords: {cls.parser_keywords}?")

@classmethod
def get(cls, url: str, **kwargs: Unpack[GetRequestParams]) -> Response:
    return __FetcherClientInstance__.get(url, **_merge_selector_config(cls, kwargs))

@classmethod
def post(cls, url: str, **kwargs: Unpack[DataRequestParams]) -> Response:
    return __FetcherClientInstance__.post(url, **_merge_selector_config(cls, kwargs))

@classmethod
def put(cls, url: str, **kwargs: Unpack[DataRequestParams]) -> Response:
    return __FetcherClientInstance__.put(url, **_merge_selector_config(cls, kwargs))

@classmethod
def delete(cls, url: str, **kwargs: Unpack[DataRequestParams]) -> Response:
    return __FetcherClientInstance__.delete(url, **_merge_selector_config(cls, kwargs))

@classmethod
def display_config(cls):
    return dict(
        huge_tree=cls.huge_tree,
        keep_comments=cls.keep_comments,
        adaptive=cls.adaptive,
        storage=cls.storage,
        storage_args=cls.storage_args,
        adaptive_domain=cls.adaptive_domain,
    )

@classmethod
def configure(cls, **kwargs):
    """Set multiple arguments for the parser at once globally

    :param kwargs: The keywords can be any arguments of the following: huge_tree, keep_comments, adaptive, storage, storage_args, adaptive_domain
    """
    for key, value in kwargs.items():
        key = key.strip().lower()
        if hasattr(cls, key):
            if key in cls.parser_keywords:
                setattr(cls, key, value)
            else:
                # Yup, no fun allowed LOL
                raise AttributeError(f'Unknown parser argument: "{key}"; maybe you meant {cls.parser_keywords}?')
        else:
            raise ValueError(f'Unknown parser argument: "{key}"; maybe you meant {cls.parser_keywords}?')

    if not kwargs:
        raise AttributeError(f"You must pass a keyword to configure, current keywords: {cls.parser_keywords}?")

@classmethod
def get(cls, url: str, **kwargs: Unpack[GetRequestParams]) -> Awaitable[Response]:
    return __AsyncFetcherClientInstance__.get(url, **_merge_selector_config(cls, kwargs))

@classmethod
def post(cls, url: str, **kwargs: Unpack[DataRequestParams]) -> Awaitable[Response]:
    return __AsyncFetcherClientInstance__.post(url, **_merge_selector_config(cls, kwargs))

@classmethod
def put(cls, url: str, **kwargs: Unpack[DataRequestParams]) -> Awaitable[Response]:
    return __AsyncFetcherClientInstance__.put(url, **_merge_selector_config(cls, kwargs))

@classmethod
def delete(cls, url: str, **kwargs: Unpack[DataRequestParams]) -> Awaitable[Response]:
    return __AsyncFetcherClientInstance__.delete(url, **_merge_selector_config(cls, kwargs))

@classmethod
def display_config(cls):
    return dict(
        huge_tree=cls.huge_tree,
        keep_comments=cls.keep_comments,
        adaptive=cls.adaptive,
        storage=cls.storage,
        storage_args=cls.storage_args,
        adaptive_domain=cls.adaptive_domain,
    )

@classmethod
def configure(cls, **kwargs):
    """Set multiple arguments for the parser at once globally

    :param kwargs: The keywords can be any arguments of the following: huge_tree, keep_comments, adaptive, storage, storage_args, adaptive_domain
    """
    for key, value in kwargs.items():
        key = key.strip().lower()
        if hasattr(cls, key):
            if key in cls.parser_keywords:
                setattr(cls, key, value)
            else:
                # Yup, no fun allowed LOL
                raise AttributeError(f'Unknown parser argument: "{key}"; maybe you meant {cls.parser_keywords}?')
        else:
            raise ValueError(f'Unknown parser argument: "{key}"; maybe you meant {cls.parser_keywords}?')

    if not kwargs:
        raise AttributeError(f"You must pass a keyword to configure, current keywords: {cls.parser_keywords}?")

@classmethod
def fetch(cls, url: str, **kwargs: Unpack[StealthSession]) -> Response:
    """
    Opens up a browser and do your request based on your chosen options below.

    :param url: Target url.
    :param headless: Run the browser in headless/hidden (default), or headful/visible mode.
    :param disable_resources: Drop requests for unnecessary resources for a speed boost.
        Requests dropped are of type `font`, `image`, `media`, `beacon`, `object`, `imageset`, `texttrack`, `websocket`, `csp_report`, and `stylesheet`.
    :param blocked_domains: A set of domain names to block requests to. Subdomains are also matched (e.g., ``"example.com"`` blocks ``"sub.example.com"`` too).
    :param block_ads: Block requests to ~3,500 known ad/tracking domains. Can be combined with ``blocked_domains``.
    :param dns_over_https: Route DNS queries through Cloudflare's DNS-over-HTTPS to prevent DNS leaks when using proxies.
    :param useragent: Pass a useragent string to be used. Otherwise the fetcher will generate a real Useragent of the same browser and use it.
    :param cookies: Set cookies for the next request.
    :param network_idle: Wait for the page until there are no network connections for at least 500 ms.
    :param timeout: The timeout in milliseconds that is used in all operations and waits through the page. The default is 30,000
    :param wait: The time (milliseconds) the fetcher will wait after everything finishes before closing the page and returning the ` Response ` object.
    :param page_action: Added for automation. A function that takes the `page` object, runs after navigation, and does the automation you need.
    :param page_setup: A function that takes the `page` object, runs before navigation. Use it to register event listeners or routes that must be set up before the page loads.
    :param wait_selector: Wait for a specific CSS selector to be in a specific state.
    :param init_script: An absolute path to a JavaScript file to be executed on page creation for all pages in this session.
    :param locale: Specify user locale, for example, `en-GB`, `de-DE`, etc. Locale will affect navigator.language value, Accept-Language request header value as well as number and date formatting
        rules. Defaults to the system default locale.
    :param timezone_id: Changes the timezone of the browser. Defaults to the system timezone.
    :param wait_selector_state: The state to wait for the selector given with `wait_selector`. The default state is `attached`.
    :param solve_cloudflare: Solves all types of the Cloudflare's Turnstile/Interstitial challenges before returning the response to you.
    :param real_chrome: If you have a Chrome browser installed on your device, enable this, and the Fetcher will launch an instance of your browser and use it.
    :param hide_canvas: Add random noise to canvas operations to prevent fingerprinting.
    :param block_webrtc: Forces WebRTC to respect proxy settings to prevent local IP address leak.
    :param allow_webgl: Enabled by default. Disabling it disables WebGL and WebGL 2.0 support entirely. Disabling WebGL is not recommended as many WAFs now check if WebGL is enabled.
    :param load_dom: Enabled by default, wait for all JavaScript on page(s) to fully load and execute.
    :param pierce_shadow: Include open Shadow DOM content in the response HTML. Defaults to False.
    :param google_search: Enabled by default, Scrapling will set a Google referer header.
    :param extra_headers: A dictionary of extra headers to add to the request. _The referer set by `google_search` takes priority over the referer set here if used together._
    :param proxy: The proxy to be used with requests, it can be a string or a dictionary with the keys 'server', 'username', and 'password' only.
    :param user_data_dir: Path to a User Data Directory, which stores browser session data like cookies and local storage. The default is to create a temporary directory.
    :param extra_flags: A list of additional browser flags to pass to the browser on launch.
    :param selector_config: The arguments that will be passed in the end while creating the final Selector's class.
    :param additional_args: Additional arguments to be passed to Playwright's context as additional settings, and it takes higher priority than Scrapling's settings.
    :return: A `Response` object.
    """
    selector_config = kwargs.get("selector_config", {}) or kwargs.get(
        "custom_config", {}
    )  # Checking `custom_config` for backward compatibility
    if not isinstance(selector_config, dict):
        raise TypeError("Argument `selector_config` must be a dictionary.")

    kwargs["selector_config"] = {**cls._generate_parser_arguments(), **selector_config}

    with StealthySession(**kwargs) as engine:
        return engine.fetch(url)

@classmethod
async def async_fetch(cls, url: str, **kwargs: Unpack[StealthSession]) -> Response:
    """
    Opens up a browser and do your request based on your chosen options below.

    :param url: Target url.
    :param headless: Run the browser in headless/hidden (default), or headful/visible mode.
    :param disable_resources: Drop requests for unnecessary resources for a speed boost.
        Requests dropped are of type `font`, `image`, `media`, `beacon`, `object`, `imageset`, `texttrack`, `websocket`, `csp_report`, and `stylesheet`.
    :param blocked_domains: A set of domain names to block requests to. Subdomains are also matched (e.g., ``"example.com"`` blocks ``"sub.example.com"`` too).
    :param block_ads: Block requests to ~3,500 known ad/tracking domains. Can be combined with ``blocked_domains``.
    :param dns_over_https: Route DNS queries through Cloudflare's DNS-over-HTTPS to prevent DNS leaks when using proxies.
    :param useragent: Pass a useragent string to be used. Otherwise the fetcher will generate a real Useragent of the same browser and use it.
    :param cookies: Set cookies for the next request.
    :param network_idle: Wait for the page until there are no network connections for at least 500 ms.
    :param timeout: The timeout in milliseconds that is used in all operations and waits through the page. The default is 30,000
    :param wait: The time (milliseconds) the fetcher will wait after everything finishes before closing the page and returning the ` Response ` object.
    :param page_action: Added for automation. A function that takes the `page` object, runs after navigation, and does the automation you need.
    :param page_setup: A function that takes the `page` object, runs before navigation. Use it to register event listeners or routes that must be set up before the page loads.
    :param wait_selector: Wait for a specific CSS selector to be in a specific state.
    :param init_script: An absolute path to a JavaScript file to be executed on page creation for all pages in this session.
    :param locale: Specify user locale, for example, `en-GB`, `de-DE`, etc. Locale will affect navigator.language value, Accept-Language request header value as well as number and date formatting
        rules. Defaults to the system default locale.
    :param timezone_id: Changes the timezone of the browser. Defaults to the system timezone.
    :param wait_selector_state: The state to wait for the selector given with `wait_selector`. The default state is `attached`.
    :param solve_cloudflare: Solves all types of the Cloudflare's Turnstile/Interstitial challenges before returning the response to you.
    :param real_chrome: If you have a Chrome browser installed on your device, enable this, and the Fetcher will launch an instance of your browser and use it.
    :param hide_canvas: Add random noise to canvas operations to prevent fingerprinting.
    :param block_webrtc: Forces WebRTC to respect proxy settings to prevent local IP address leak.
    :param allow_webgl: Enabled by default. Disabling it disables WebGL and WebGL 2.0 support entirely. Disabling WebGL is not recommended as many WAFs now check if WebGL is enabled.
    :param load_dom: Enabled by default, wait for all JavaScript on page(s) to fully load and execute.
    :param pierce_shadow: Include open Shadow DOM content in the response HTML. Defaults to False.
    :param google_search: Enabled by default, Scrapling will set a Google referer header.
    :param extra_headers: A dictionary of extra headers to add to the request. _The referer set by `google_search` takes priority over the referer set here if used together._
    :param proxy: The proxy to be used with requests, it can be a string or a dictionary with the keys 'server', 'username', and 'password' only.
    :param user_data_dir: Path to a User Data Directory, which stores browser session data like cookies and local storage. The default is to create a temporary directory.
    :param extra_flags: A list of additional browser flags to pass to the browser on launch.
    :param selector_config: The arguments that will be passed in the end while creating the final Selector's class.
    :param additional_args: Additional arguments to be passed to Playwright's context as additional settings, and it takes higher priority than Scrapling's settings.
    :return: A `Response` object.
    """
    selector_config = kwargs.get("selector_config", {}) or kwargs.get(
        "custom_config", {}
    )  # Checking `custom_config` for backward compatibility
    if not isinstance(selector_config, dict):
        raise TypeError("Argument `selector_config` must be a dictionary.")

    kwargs["selector_config"] = {**cls._generate_parser_arguments(), **selector_config}

    async with AsyncStealthySession(**kwargs) as engine:
        return await engine.fetch(url)

def __enter__(self) -> _SyncSessionLogic:
    """Creates and returns a new synchronous Fetcher Session"""
    if self._client is None:
        # Use **vars(self) to avoid repeating all parameters
        config = {k.replace("_default_", ""): getattr(self, k) for k in self.__slots__ if k.startswith("_default")}
        config["stealthy_headers"] = self._stealth
        config["selector_config"] = self.selector_config
        config["proxy_rotator"] = self._proxy_rotator
        self._client = _SyncSessionLogic(**config)
        try:
            result = self._client.__enter__()
        except Exception:
            self._client = None
            raise
        self._is_alive = True
        return result
    raise RuntimeError("This FetcherSession instance already has an active synchronous session.")

def __exit__(self, exc_type, exc_val, exc_tb):
    if self._client is not None and isinstance(self._client, _SyncSessionLogic):
        self._client.__exit__(exc_type, exc_val, exc_tb)
        self._client = None
        self._is_alive = False
        return
    raise RuntimeError("Cannot exit invalid session")

async def __aenter__(self) -> _ASyncSessionLogic:
    """Creates and returns a new asynchronous Session."""
    if self._client is None:
        # Use **vars(self) to avoid repeating all parameters
        config = {k.replace("_default_", ""): getattr(self, k) for k in self.__slots__ if k.startswith("_default")}
        config["stealthy_headers"] = self._stealth
        config["selector_config"] = self.selector_config
        config["proxy_rotator"] = self._proxy_rotator
        self._client = _ASyncSessionLogic(**config)
        try:
            result = await self._client.__aenter__()
        except Exception:
            self._client = None
            raise
        self._is_alive = True
        return result
    raise RuntimeError("This FetcherSession instance already has an active asynchronous session.")

async def __aexit__(self, exc_type, exc_val, exc_tb):
    if self._client is not None and isinstance(self._client, _ASyncSessionLogic):
        await self._client.__aexit__(exc_type, exc_val, exc_tb)
        self._client = None
        self._is_alive = False
        return
    raise RuntimeError("Cannot exit invalid session")

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._browser_options: Dict[str, Any] = {
        "args": DEFAULT_ARGS,
        "ignore_default_args": HARMFUL_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_context_options__(self) -> None:
    config: PlaywrightConfig | StealthConfig = self._config
    self._context_options.update(
        {
            "proxy": config.proxy,
            "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 config.additional_args:
        self._context_options.update(config.additional_args)

def __generate_options__(self, extra_flags: Tuple | None = None) -> None:
    self.__generate_context_options__()
    config = self._config
    flags = self._browser_options["args"]
    if config.extra_flags or extra_flags:
        flags = list(set(tuple(flags) + tuple(extra_flags or ()) + tuple(config.extra_flags or ())))
    if config.dns_over_https:
        flags = list(flags) + ["--dns-over-https-templates=https://cloudflare-dns.com/dns-query"]
    if config.locale:
        base_lang = config.locale.split("-")[0].lower()
        accept_lang = f"{config.locale},{base_lang}" if base_lang != config.locale.lower() else config.locale
        flags = list(flags) + [f"--lang={config.locale}", f"--accept-lang={accept_lang}"]
    self._browser_options.update(
        {"args": flags, "headless": config.headless, "channel": "chrome" if config.real_chrome else "chromium"}
    )
    if config.executable_path:
        self._browser_options["executable_path"] = config.executable_path
    self._user_data_dir = config.user_data_dir

def __validate__(self, **params):
    self._config = self.__validate_routine__(params, model=self._config_class)
    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 save_state(self, path: str | Path) -> None:
    """Save cookies, local storage and IndexedDB to a Playwright-compatible JSON file.

    Requires a started session without proxy rotation. Does not save sessionStorage.
    """
    if not self._is_alive or self.context is None:
        raise RuntimeError("Session state requires a started browser session without proxy rotation")
    self.context.storage_state(path=Path(path), indexed_db=True)

def load_state(self, path: str | Path) -> None:
    """Replace cookies, local storage and IndexedDB from a Playwright-compatible JSON file.

    Requires a started session without proxy rotation. Fetch or reload pages to use the restored login.
    """
    if not self._is_alive or self.context is None:
        raise RuntimeError("Session state requires a started browser session without proxy rotation")
    self.context.set_storage_state(path)

def close_pages(self) -> None:
    """Close every open tab in the session's pool. The next request opens a fresh tab."""
    for page_info in self.page_pool.clear():
        with suppress(Exception):
            page_info.page.close()

def close(self):  # pragma: no cover
    """Close all resources"""
    self.network._detach()
    if not self._is_alive:
        return

    self.close_pages()
    try:
        if self.context and self._owns_context:
            self.context.close()
    finally:
        self.context = None
        try:
            if self.browser:
                self.browser.close()
        finally:
            self.browser = None
            try:
                if self.playwright:
                    self.playwright.stop()
            finally:
                self.playwright = None
                self._is_alive = False