Adding browsing engines

scrapling/engines/__init__.py

@@ -0,0 +1,6 @@
+from .camo import CamoufoxEngine
+from .static import StaticEngine
+from .pw import PlaywrightEngine, DEFAULT_DISABLED_RESOURCES, DEFAULT_STEALTH_FLAGS
+from .tools import check_if_engine_usable
+
+__all__ = ['CamoufoxEngine', 'PlaywrightEngine']

scrapling/engines/camo.py

@@ -0,0 +1,63 @@
+import logging
+from scrapling._types import Union, Callable, Optional
+
+from .tools import check_type_validity, get_os_name, generate_convincing_referer
+
+from camoufox.sync_api import Camoufox
+
+
+def _do_nothing(page):
+    # Anything
+    return page
+
+
+class CamoufoxEngine:
+    def __init__(
+            self, headless: Union[bool, str] = True,
+            block_images: Optional[bool] = True,
+            block_webrtc: Optional[bool] = False,
+            network_idle: Optional[bool] = False,
+            timeout: Optional[float] = 30000,
+            page_action: Callable = _do_nothing,
+            wait_selector: Optional[str] = None,
+            wait_selector_state: str = 'attached',
+    ):
+        self.headless = headless
+        self.block_images = bool(block_images)
+        self.block_webrtc = bool(block_webrtc)
+        self.network_idle = bool(network_idle)
+        self.timeout = check_type_validity(timeout, [int, float], 30000)
+        if callable(page_action):
+            self.page_action = page_action
+        else:
+            self.page_action = _do_nothing
+            logging.error('[Ignored] Argument "page_action" must be callable')
+
+        self.wait_selector = wait_selector
+        self.wait_selector_state = wait_selector_state
+
+    def fetch(self, url: str):
+        with Camoufox(
+                headless=self.headless,
+                block_images=self.block_images,
+                os=get_os_name(),
+                block_webrtc=self.block_webrtc,
+        ) as browser:
+            page = browser.new_page()
+            page.set_default_navigation_timeout(self.timeout)
+            page.set_default_timeout(self.timeout)
+            page.goto(url, referer=generate_convincing_referer(url))
+            page.wait_for_load_state(state="load")
+            page.wait_for_load_state(state="domcontentloaded")
+            if self.network_idle:
+                page.wait_for_load_state('networkidle')
+
+            page = self.page_action(page)
+
+            if self.wait_selector and type(self.wait_selector) is str:
+                waiter = page.locator(self.wait_selector)
+                waiter.wait_for(state=self.wait_selector_state)
+
+            html = page.content()
+            page.close()
+        return html

scrapling/engines/pw.py

@@ -0,0 +1,209 @@
+import json
+import logging
+from scrapling._types import Union, Callable, Optional, List, Dict
+
+from .tools import check_type_validity, generate_headers, js_bypass_path, construct_websocket_url, generate_convincing_referer
+
+# Disable loading these resources for speed
+DEFAULT_DISABLED_RESOURCES = ['beacon', 'csp_report', 'font', 'image', 'imageset', 'media', 'object', 'texttrack', 'stylesheet', 'websocket']
+DEFAULT_STEALTH_FLAGS = [
+    # Explanation: https://peter.sh/experiments/chromium-command-line-switches/
+    # Generally this will make the browser faster and less detectable
+    '--incognito', '--accept-lang=en-US', '--lang=en-US', '--no-pings', '--mute-audio', '--no-first-run', '--no-default-browser-check', '--disable-cloud-import',
+    '--disable-gesture-typing', '--disable-offer-store-unmasked-wallet-cards', '--disable-offer-upload-credit-cards', '--disable-print-preview', '--disable-voice-input',
+    '--disable-wake-on-wifi', '--disable-cookie-encryption', '--ignore-gpu-blocklist', '--enable-async-dns', '--enable-simple-cache-backend', '--enable-tcp-fast-open',
+    '--prerender-from-omnibox=disabled', '--enable-web-bluetooth', '--disable-features=AudioServiceOutOfProcess,IsolateOrigins,site-per-process,TranslateUI,BlinkGenPropertyTrees',
+    '--aggressive-cache-discard', '--disable-ipc-flooding-protection', '--disable-blink-features=AutomationControlled', '--test-type',
+    '--enable-features=NetworkService,NetworkServiceInProcess,TrustTokens,TrustTokensAlwaysAllowIssuance',
+    '--disable-breakpad', '--disable-component-update', '--disable-domain-reliability', '--disable-sync', '--disable-client-side-phishing-detection',
+    '--disable-hang-monitor', '--disable-popup-blocking', '--disable-prompt-on-repost', '--metrics-recording-only', '--safebrowsing-disable-auto-update', '--password-store=basic',
+    '--autoplay-policy=no-user-gesture-required', '--use-mock-keychain', '--force-webrtc-ip-handling-policy=disable_non_proxied_udp',
+    '--webrtc-ip-handling-policy=disable_non_proxied_udp', '--disable-session-crashed-bubble', '--disable-crash-reporter', '--disable-dev-shm-usage', '--force-color-profile=srgb',
+    '--disable-translate', '--disable-background-networking', '--disable-background-timer-throttling', '--disable-backgrounding-occluded-windows', '--disable-infobars',
+    '--hide-scrollbars', '--disable-renderer-backgrounding', '--font-render-hinting=none', '--disable-logging', '--enable-surface-synchronization',
+    '--run-all-compositor-stages-before-draw', '--disable-threaded-animation', '--disable-threaded-scrolling', '--disable-checker-imaging',
+    '--disable-new-content-rendering-timeout', '--disable-image-animation-resync', '--disable-partial-raster',
+    '--blink-settings=primaryHoverType=2,availableHoverTypes=2,primaryPointerType=4,availablePointerTypes=4',
+    '--disable-layer-tree-host-memory-pressure',
+    '--window-position=0,0',
+    '--disable-features=site-per-process',
+    '--disable-default-apps',
+    '--disable-component-extensions-with-background-pages',
+    '--disable-extensions',
+    # "--disable-reading-from-canvas",  # For Firefox
+    '--start-maximized'  # For headless check bypass
+]
+
+
+def _do_nothing(page):
+    # Anything
+    return page
+
+
+class PlaywrightEngine:
+    def __init__(
+            self, headless: Union[bool, str] = True,
+            disable_resources: Optional[List] = None,
+            useragent: Optional[str] = None,
+            network_idle: Optional[bool] = False,
+            timeout: Optional[float] = 30000,
+            page_action: Callable = _do_nothing,
+            wait_selector: Optional[str] = None,
+            wait_selector_state: Optional[str] = 'attached',
+            stealth: bool = False,
+            hide_canvas: bool = True,
+            disable_webgl: bool = False,
+            cdp_url: Optional[str] = None,
+            nstbrowser_mode: bool = False,
+            nstbrowser_config: Optional[Dict] = None,
+    ):
+        self.headless = headless
+        self.disable_resources = disable_resources
+        self.network_idle = bool(network_idle)
+        self.stealth = bool(stealth)
+        self.hide_canvas = bool(hide_canvas)
+        self.disable_webgl = bool(disable_webgl)
+        self.cdp_url = cdp_url
+        self.useragent = useragent
+        self.timeout = check_type_validity(timeout, [int, float], 30000)
+        if callable(page_action):
+            self.page_action = page_action
+        else:
+            self.page_action = _do_nothing
+            logging.error('[Ignored] Argument "page_action" must be callable')
+
+        self.wait_selector = wait_selector
+        self.wait_selector_state = wait_selector_state
+        self.nstbrowser_mode = bool(nstbrowser_mode)
+        self.nstbrowser_config = nstbrowser_config
+
+    def _cdp_url_logic(self, flags: Optional[dict] = None):
+        cdp_url = self.cdp_url
+        if self.nstbrowser_mode:
+            if self.nstbrowser_config and type(self.nstbrowser_config) is Dict:
+                config = self.nstbrowser_config
+            else:
+                # Defaulting to the docker mode, token doesn't matter in it as it's passed for the container
+                query = {
+                    "once": True,
+                    "headless": True,
+                    "autoClose": True,
+                    "fingerprint": {
+                        "flags": {
+                            "timezone": "BasedOnIp",
+                            "screen": "Custom"
+                        },
+                        "platform": 'linux',  # support: windows, mac, linux
+                        "kernel": 'chromium',  # only support: chromium
+                        "kernelMilestone": '128',
+                        "hardwareConcurrency": 8,
+                        "deviceMemory": 8,
+                    },
+                }
+                if flags:
+                    query.update({
+                        "args": dict(zip(flags, [''] * len(flags))),  # browser args should be a dictionary
+                    })
+
+                config = {
+                    'config': json.dumps(query),
+                    # 'token': ''
+                }
+            cdp_url = construct_websocket_url(cdp_url, config)
+
+        return cdp_url
+
+    def fetch(self, url):
+        if not self.stealth:
+            from playwright.sync_api import sync_playwright
+        else:
+            from rebrowser_playwright.sync_api import sync_playwright
+
+        with sync_playwright() as p:
+            # Handle the UserAgent early
+            if self.useragent:
+                extra_headers = {}
+                useragent = self.useragent
+            else:
+                extra_headers = generate_headers(browser_mode=True)
+                useragent = extra_headers.get('User-Agent')
+
+            # Prepare the flags before diving
+            flags = DEFAULT_STEALTH_FLAGS
+            if self.hide_canvas:
+                flags += ['--fingerprinting-canvas-image-data-noise']
+            if self.disable_webgl:
+                flags += ['--disable-webgl', '--disable-webgl-image-chromium', '--disable-webgl2']
+
+            # Creating the browser
+            if self.cdp_url:
+                cdp_url = self._cdp_url_logic(flags if self.stealth else None)
+                browser = p.chromium.connect_over_cdp(endpoint_url=cdp_url)
+            else:
+                if self.stealth:
+                    browser = p.chromium.launch(headless=self.headless, args=flags, ignore_default_args=['--enable-automation'], chromium_sandbox=True)
+                else:
+                    browser = p.chromium.launch(headless=self.headless, ignore_default_args=['--enable-automation'])
+
+            # Creating the context
+            if self.stealth:
+                context = browser.new_context(
+                    locale='en-US',
+                    is_mobile=False,
+                    has_touch=False,
+                    color_scheme='dark',  # Bypasses the 'prefersLightColor' check in creepjs
+                    user_agent=useragent,
+                    device_scale_factor=2,
+                    # 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,
+                    extra_http_headers=extra_headers,
+                    screen={"width": 1920, "height": 1080},
+                    viewport={"width": 1920, "height": 1080},
+                    permissions=["geolocation", 'notifications'],
+                )
+            else:
+                context = browser.new_context(
+                    color_scheme='dark',
+                    user_agent=useragent,
+                    device_scale_factor=2,
+                    extra_http_headers=extra_headers
+                )
+
+            # Finally we are in business
+            page = context.new_page()
+            page.set_default_navigation_timeout(self.timeout)
+            page.set_default_timeout(self.timeout)
+            if self.stealth:
+                # Basic bypasses nothing fancy as I'm still working on it
+                # But with adding these bypasses to the above config, it bypasses many online tests like
+                # https://bot.sannysoft.com/
+                # https://kaliiiiiiiiii.github.io/brotector/
+                # https://pixelscan.net/
+                # https://iphey.com/
+                # https://www.browserscan.net/bot-detection <== this one also checks for the CDP runtime fingerprint
+                # https://arh.antoinevastel.com/bots/areyouheadless/
+                # https://prescience-data.github.io/execution-monitor.html
+                page.add_init_script(path=js_bypass_path('webdriver_fully.js'))
+                page.add_init_script(path=js_bypass_path('window_chrome.js'))
+                page.add_init_script(path=js_bypass_path('navigator_plugins.js'))
+                page.add_init_script(path=js_bypass_path('pdf_viewer.js'))
+                page.add_init_script(path=js_bypass_path('notification_permission.js'))
+                page.add_init_script(path=js_bypass_path('screen_props.js'))
+                page.add_init_script(path=js_bypass_path('playwright_fingerprint.js'))
+
+            page.goto(url, referer=generate_convincing_referer(url) if self.stealth else None)
+            page.wait_for_load_state(state="load")
+            page.wait_for_load_state(state="domcontentloaded")
+            if self.network_idle:
+                page.wait_for_load_state('networkidle')
+
+            page = self.page_action(page)
+
+            if self.wait_selector and type(self.wait_selector) is str:
+                waiter = page.locator(self.wait_selector)
+                waiter.wait_for(state=self.wait_selector_state)
+
+            html = page.content()
+            page.close()
+        return html

scrapling/engines/static.py

@@ -0,0 +1,53 @@
+import logging
+from scrapling._types import Union, Optional, Dict
+
+from .tools import generate_convincing_referer, generate_headers
+
+import httpx
+
+
+class StaticEngine:
+    def __init__(
+            self,
+            follow_redirects: bool = True,
+            timeout: Optional[Union[int, float]] = None,
+    ):
+        self.timeout = timeout
+        self.follow_redirects = bool(follow_redirects)
+        self._extra_headers = generate_headers(browser_mode=False)
+
+    @staticmethod
+    def _headers_job(headers, url, stealth):
+        headers = headers or {}
+
+        # Validate headers
+        if not headers.get('user-agent') and not headers.get('User-Agent'):
+            headers['User-Agent'] = generate_headers(browser_mode=False).get('User-Agent')
+            logging.info(f"Can't find useragent in headers so '{headers['User-Agent']}' was used.")
+
+        if stealth:
+            extra_headers = generate_headers(browser_mode=False)
+            headers.update(extra_headers)
+            headers.update({'referer': generate_convincing_referer(url)})
+
+        return headers
+
+    def get(self, url: str, stealthy_headers: Optional[bool] = True, **kwargs: Dict):
+        headers = self._headers_job(kwargs.get('headers'), url, stealthy_headers)
+        request = httpx.get(url=url, headers=headers, follow_redirects=self.follow_redirects, timeout=self.timeout, **kwargs)
+        return request.text
+
+    def post(self, url: str, stealthy_headers: Optional[bool] = True, **kwargs: Dict):
+        headers = self._headers_job(kwargs.get('headers'), url, stealthy_headers)
+        request = httpx.post(url=url, headers=headers, follow_redirects=self.follow_redirects, timeout=self.timeout, **kwargs)
+        return request.text
+
+    def delete(self, url: str, stealthy_headers: Optional[bool] = True, **kwargs: Dict):
+        headers = self._headers_job(kwargs.get('headers'), url, stealthy_headers)
+        request = httpx.delete(url=url, headers=headers, follow_redirects=self.follow_redirects, timeout=self.timeout, **kwargs)
+        return request.text
+
+    def put(self, url: str, stealthy_headers: Optional[bool] = True, **kwargs: Dict):
+        headers = self._headers_job(kwargs.get('headers'), url, stealthy_headers)
+        request = httpx.put(url=url, headers=headers, follow_redirects=self.follow_redirects, timeout=self.timeout, **kwargs)
+        return request.text

scrapling/engines/tools.py

@@ -0,0 +1,174 @@
+import os
+import logging
+import inspect
+import platform
+from scrapling._types import Any, List, Type, Union, Optional
+from urllib.parse import urlparse, urlencode
+
+from tldextract import extract
+from browserforge.fingerprints import FingerprintGenerator
+from browserforge.headers import HeaderGenerator, Browser
+
+
+def generate_convincing_referer(url):
+    """
+    Takes the domain from the URL without the subdomain/suffix and make it look like you were searching google for this website
+
+    >>> generate_convincing_referer('https://www.somewebsite.com/blah')
+    'https://www.google.com/search?q=somewebsite'
+
+    :param url: The URL you are about to fetch.
+    :return:
+    """
+    website_name = extract(url).domain
+    return f'https://www.google.com/search?q={website_name}'
+
+
+def check_if_engine_usable(engine):
+    if isinstance(engine, type):
+        raise TypeError("Expected an engine instance, not a class definition of the engine")
+
+    if hasattr(engine, 'fetch'):
+        fetch_function = getattr(engine, "fetch")
+        if callable(fetch_function):
+            if len(inspect.signature(fetch_function).parameters) > 0:
+                return engine
+            else:
+                raise TypeError("Engine class instance must have a callable method 'fetch' with the first argument used for the url.")
+        else:
+            raise TypeError("Invalid engine instance! Engine class must have a callable method 'fetch'")
+    else:
+        raise TypeError("Invalid engine instance! Engine class must have the method 'fetch'")
+
+
+def construct_websocket_url(base_url, query_params):
+    # Validate the base URL structure
+    try:
+        parsed = urlparse(base_url)
+
+        # Check scheme
+        if parsed.scheme not in ('ws', 'wss'):
+            raise ValueError("URL must use 'ws://' or 'wss://' scheme")
+
+        # Validate hostname and port
+        if not parsed.netloc:
+            raise ValueError("Invalid hostname")
+
+        # Ensure path starts with /
+        path = parsed.path
+        if not path.startswith('/'):
+            path = '/' + path
+
+        # Reconstruct the base URL with validated parts
+        validated_base = f"{parsed.scheme}://{parsed.netloc}{path}"
+
+        # Add query parameters
+        if query_params:
+            query_string = urlencode(query_params)
+            return f"{validated_base}?{query_string}"
+
+        return validated_base
+
+    except Exception as e:
+        raise ValueError(f"Invalid WebSocket URL: {str(e)}")
+
+
+def js_bypass_path(filename):
+    current_directory = os.path.dirname(__file__)
+    return os.path.join(current_directory, 'bypasses', filename)
+
+
+def get_os_name():
+    # Get the OS name in the same format needed for browserforge
+    os_name = platform.system()
+    return {
+        'Linux': 'linux',
+        'Darwin': 'macos',
+        'Windows': 'windows',
+        # For the future? because why not
+        'iOS': 'ios',
+    }.get(os_name)
+
+
+def generate_suitable_fingerprint():
+    # This would be for Browserforge playwright injector
+    os_name = get_os_name()
+    return FingerprintGenerator(
+        browser=[Browser(name='chrome', min_version=128)],
+        os=os_name,  # None is ignored
+        device='desktop'
+    ).generate()
+
+
+def generate_headers(browser_mode=False):
+    if browser_mode:
+        # In this mode we don't care about anything other than matching the OS and the browser type with the browser we are using
+        # So we don't raise any inconsistency red flags while websites fingerprinting us
+        os_name = get_os_name()
+        return HeaderGenerator(
+            browser=[Browser(name='chrome', min_version=128)],
+            os=os_name,  # None is ignored
+            device='desktop'
+        ).generate()
+    else:
+        # Here it's used for normal requests that aren't done through browsers so we can take it lightly
+        browsers = [
+            Browser(name='chrome', min_version=120),
+            Browser(name='firefox', min_version=120),
+            Browser(name='edge', min_version=120),
+        ]
+        return HeaderGenerator(browser=browsers, device='desktop').generate()
+
+
+def get_variable_name(var: Any) -> Optional[str]:
+    """Get the name of a variable using global and local scopes.
+    :param var: The variable to find the name for
+    :return: The name of the variable if found, None otherwise
+    """
+    for scope in [globals(), locals()]:
+        for name, value in scope.items():
+            if value is var:
+                return name
+    return None
+
+
+def check_type_validity(variable: Any, valid_types: Union[List[Type], None], default_value: Any = None, critical: bool = False, param_name: Optional[str] = None) -> Any:
+    """Check if a variable matches the specified type constraints.
+    :param variable: The variable to check
+    :param valid_types: List of valid types for the variable
+    :param default_value: Value to return if type check fails
+    :param critical: If True, raises TypeError instead of logging error
+    :param param_name: Optional parameter name for error messages
+    :return: The original variable if valid, default_value if invalid
+    :raise TypeError: If critical=True and type check fails
+    """
+    # Use provided param_name or try to get it automatically
+    var_name = param_name or get_variable_name(variable) or "Unknown"
+
+    # Convert valid_types to a list if None
+    valid_types = valid_types or []
+
+    # Handle None value
+    if variable is None:
+        if type(None) in valid_types:
+            return variable
+        error_msg = f'Argument "{var_name}" cannot be None'
+        if critical:
+            raise TypeError(error_msg)
+        logging.error(f'[Ignored] {error_msg}')
+        return default_value
+
+    # If no valid_types specified and variable has a value, return it
+    if not valid_types:
+        return variable
+
+    # Check if variable type matches any of the valid types
+    if not any(isinstance(variable, t) for t in valid_types):
+        type_names = [t.__name__ for t in valid_types]
+        error_msg = f'Argument "{var_name}" must be of type {" or ".join(type_names)}'
+        if critical:
+            raise TypeError(error_msg)
+        logging.error(f'[Ignored] {error_msg}')
+        return default_value
+
+    return variable