diff --git a/scrapling/engines/__init__.py b/scrapling/engines/__init__.py new file mode 100644 index 0000000..89b5d62 --- /dev/null +++ b/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'] diff --git a/scrapling/engines/camo.py b/scrapling/engines/camo.py new file mode 100644 index 0000000..26464ab --- /dev/null +++ b/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 diff --git a/scrapling/engines/pw.py b/scrapling/engines/pw.py new file mode 100644 index 0000000..db353fa --- /dev/null +++ b/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 diff --git a/scrapling/engines/static.py b/scrapling/engines/static.py new file mode 100644 index 0000000..d4110e7 --- /dev/null +++ b/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 diff --git a/scrapling/engines/tools.py b/scrapling/engines/tools.py new file mode 100644 index 0000000..3d4564e --- /dev/null +++ b/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