From 59ec6be20108350fcc03ad0696545a49175d1c2c Mon Sep 17 00:00:00 2001 From: Karim shoair Date: Sun, 26 Oct 2025 05:04:19 +0300 Subject: [PATCH] refactor(fetchers): Less duplicated code and better handling for unstable websites - Websites that never finish loading their requests won't crash the code now if you used `network_idle` with them - Fixed a typo in `load_dom` in DynamicSession's async_fetch - Removed dead code --- scrapling/engines/_browsers/_base.py | 146 +++++++++++++++++-- scrapling/engines/_browsers/_camoufox.py | 150 ++++---------------- scrapling/engines/_browsers/_controllers.py | 111 +++------------ scrapling/engines/toolbelt/custom.py | 12 -- scrapling/engines/toolbelt/fingerprints.py | 14 +- 5 files changed, 185 insertions(+), 248 deletions(-) diff --git a/scrapling/engines/_browsers/_base.py b/scrapling/engines/_browsers/_base.py index e01a820..4157494 100644 --- a/scrapling/engines/_browsers/_base.py +++ b/scrapling/engines/_browsers/_base.py @@ -2,17 +2,27 @@ from time import time from asyncio import sleep as asyncio_sleep, Lock from camoufox import DefaultAddons -from playwright.sync_api import BrowserContext, Playwright -from playwright.async_api import ( - BrowserContext as AsyncBrowserContext, - Playwright as AsyncPlaywright, +from playwright.sync_api import ( + Page, + Frame, + BrowserContext, + Playwright, + Response as SyncPlaywrightResponse, ) +from playwright.async_api import ( + Page as AsyncPage, + Frame as AsyncFrame, + Playwright as AsyncPlaywright, + Response as AsyncPlaywrightResponse, + BrowserContext as AsyncBrowserContext, +) +from playwright._impl._errors import Error as PlaywrightError from camoufox.pkgman import installed_verstr as camoufox_version from camoufox.utils import launch_options as generate_launch_options from ._page import PageInfo, PagePool from scrapling.parser import Selector -from scrapling.core._types import Any, cast, Dict, Optional, TYPE_CHECKING +from scrapling.core._types import Any, cast, Dict, List, Optional, Callable, TYPE_CHECKING from scrapling.engines.toolbelt.fingerprints import get_os_name from ._validators import validate, PlaywrightConfig, CamoufoxConfig from ._config_tools import _compiled_stealth_scripts, _launch_kwargs, _context_kwargs @@ -26,10 +36,35 @@ class SyncSession: self.max_pages = max_pages self.page_pool = PagePool(max_pages) self._max_wait_for_page = 60 - self.playwright: Optional[Playwright] = None - self.context: Optional[BrowserContext] = None + self.playwright: Playwright | Any = None + self.context: BrowserContext | Any = None self._closed = False + def __create__(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.__create__() + return self + + def __exit__(self, exc_type, exc_val, exc_tb): + self.close() + def _get_page( self, timeout: int | float, @@ -65,17 +100,76 @@ class SyncSession: "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: + 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, 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: Optional[AsyncPlaywright] = None - self.context: Optional[AsyncBrowserContext] = None + self.playwright: AsyncPlaywright | Any = None + self.context: AsyncBrowserContext | Any = None self._closed = False self._lock = Lock() + async def __create__(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.__create__() + return self + + async def __aexit__(self, exc_type, exc_val, exc_tb): + await self.close() + async def _get_page( self, timeout: int | float, @@ -122,6 +216,40 @@ class AsyncSession: "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: + 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, 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 DynamicSessionMixin: def __validate__(self, **params): diff --git a/scrapling/engines/_browsers/_camoufox.py b/scrapling/engines/_browsers/_camoufox.py index e66a826..ac66870 100644 --- a/scrapling/engines/_browsers/_camoufox.py +++ b/scrapling/engines/_browsers/_camoufox.py @@ -2,18 +2,16 @@ from random import randint from re import compile as re_compile from playwright.sync_api import ( - Response as SyncPlaywrightResponse, - sync_playwright, - Locator, Page, + Locator, + sync_playwright, ) from playwright.async_api import ( async_playwright, - Response as AsyncPlaywrightResponse, - BrowserContext as AsyncBrowserContext, - Playwright as AsyncPlaywright, - Locator as AsyncLocator, Page as async_Page, + Locator as AsyncLocator, + Playwright as AsyncPlaywright, + BrowserContext as AsyncBrowserContext, ) from playwright._impl._errors import Error as PlaywrightError @@ -184,38 +182,13 @@ class StealthySession(StealthySessionMixin, SyncSession): if self.cookies: # pragma: no cover self.context.add_cookies(self.cookies) - def __enter__(self): # pragma: no cover - self.__create__() - return self - - def __exit__(self, exc_type, exc_val, exc_tb): - self.close() - - def close(self): # pragma: no cover - """Close all resources""" - if self._closed: # pragma: no cover - return - - if self.context: - self.context.close() - self.context = None - - if self.playwright: - self.playwright.stop() - self.playwright = None - - self._closed = True - def _solve_cloudflare(self, page: Page) -> None: # pragma: no cover """Solve the cloudflare challenge displayed on the playwright page passed :param page: The targeted page :return: """ - try: - page.wait_for_load_state("networkidle", timeout=5000) - except PlaywrightError: - pass + self._wait_for_networkidle(page, timeout=5000) challenge_type = self._detect_cloudflare(ResponseFactory._get_page_content(page)) if not challenge_type: log.error("No Cloudflare challenge found.") @@ -241,8 +214,7 @@ class StealthySession(StealthySessionMixin, SyncSession): outer_box = {} iframe = page.frame(url=__CF_PATTERN__) if iframe is not None: - iframe.wait_for_load_state(state="domcontentloaded") - iframe.wait_for_load_state("networkidle") + self._wait_for_page_stability(iframe, True, True) if challenge_type != "embedded": while not iframe.frame_element().is_visible(): @@ -258,7 +230,7 @@ class StealthySession(StealthySessionMixin, SyncSession): # Move the mouse to the center of the window, then press and hold the left mouse button page.mouse.click(captcha_x, captcha_y, delay=60, button="left") - page.wait_for_load_state("networkidle") + self._wait_for_networkidle(page) if iframe is not None: # Wait for the frame to be removed from the page while iframe in page.frames: @@ -266,8 +238,7 @@ class StealthySession(StealthySessionMixin, SyncSession): if challenge_type != "embedded": page.locator(box_selector).last.wait_for(state="detached") page.locator(".zone-name-title").wait_for(state="hidden") - page.wait_for_load_state(state="load") - page.wait_for_load_state(state="domcontentloaded") + self._wait_for_page_stability(page, True, False) log.info("Cloudflare captcha is solved") return @@ -328,32 +299,19 @@ class StealthySession(StealthySessionMixin, SyncSession): if self._closed: # pragma: no cover raise RuntimeError("Context manager has been closed") - final_response = None referer = ( generate_convincing_referer(url) if (params.google_search and "referer" not in self._headers_keys) else None ) - def handle_response(finished_response: SyncPlaywrightResponse): - nonlocal final_response - if ( - finished_response.request.resource_type == "document" - and finished_response.request.is_navigation_request() - and finished_response.request.frame == page_info.page.main_frame - ): - final_response = finished_response - page_info = self._get_page(params.timeout, params.extra_headers, params.disable_resources) - page_info.mark_busy(url=url) + final_response = [None] + handle_response = self._create_response_handler(page_info, final_response) try: # pragma: no cover # Navigate to URL and wait for a specified state page_info.page.on("response", handle_response) first_response = page_info.page.goto(url, referer=referer) - if params.load_dom: - page_info.page.wait_for_load_state(state="domcontentloaded") - - if params.network_idle: - page_info.page.wait_for_load_state("networkidle") + self._wait_for_page_stability(page_info.page, params.load_dom, params.network_idle) if not first_response: raise RuntimeError(f"Failed to get response for {url}") @@ -361,11 +319,7 @@ class StealthySession(StealthySessionMixin, SyncSession): if params.solve_cloudflare: self._solve_cloudflare(page_info.page) # Make sure the page is fully loaded after the captcha - page_info.page.wait_for_load_state(state="load") - if params.load_dom: - page_info.page.wait_for_load_state(state="domcontentloaded") - if params.network_idle: - page_info.page.wait_for_load_state("networkidle") + self._wait_for_page_stability(page_info.page, params.load_dom, params.network_idle) if params.page_action: try: @@ -378,17 +332,13 @@ class StealthySession(StealthySessionMixin, SyncSession): waiter: Locator = page_info.page.locator(params.wait_selector) waiter.first.wait_for(state=params.wait_selector_state) # Wait again after waiting for the selector, helpful with protections like Cloudflare - page_info.page.wait_for_load_state(state="load") - if params.load_dom: - page_info.page.wait_for_load_state(state="domcontentloaded") - if params.network_idle: - page_info.page.wait_for_load_state("networkidle") + self._wait_for_page_stability(page_info.page, params.load_dom, params.network_idle) except Exception as e: log.error(f"Error waiting for selector {params.wait_selector}: {e}") page_info.page.wait_for_timeout(params.wait) response = ResponseFactory.from_playwright_response( - page_info.page, first_response, final_response, params.selector_config, bool(params.page_action) + page_info.page, first_response, final_response[0], params.selector_config, bool(params.page_action) ) # Close the page to free up resources @@ -513,38 +463,13 @@ class AsyncStealthySession(StealthySessionMixin, AsyncSession): if self.cookies: await self.context.add_cookies(self.cookies) # pyright: ignore [reportArgumentType] - async def __aenter__(self): - await self.__create__() - return self - - async def __aexit__(self, exc_type, exc_val, exc_tb): - await self.close() - - 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 _solve_cloudflare(self, page: async_Page): + async def _solve_cloudflare(self, page: async_Page): # pragma: no cover """Solve the cloudflare challenge displayed on the playwright page passed. The async version :param page: The async targeted page :return: """ - try: - await page.wait_for_load_state("networkidle", timeout=5000) - except PlaywrightError: - pass + await self._wait_for_networkidle(page, timeout=5000) challenge_type = self._detect_cloudflare(await ResponseFactory._get_async_page_content(page)) if not challenge_type: log.error("No Cloudflare challenge found.") @@ -570,8 +495,7 @@ class AsyncStealthySession(StealthySessionMixin, AsyncSession): outer_box = {} iframe = page.frame(url=__CF_PATTERN__) if iframe is not None: - await iframe.wait_for_load_state(state="domcontentloaded") - await iframe.wait_for_load_state("networkidle") + await self._wait_for_page_stability(iframe, True, True) if challenge_type != "embedded": while not await (await iframe.frame_element()).is_visible(): @@ -587,7 +511,7 @@ class AsyncStealthySession(StealthySessionMixin, AsyncSession): # Move the mouse to the center of the window, then press and hold the left mouse button await page.mouse.click(captcha_x, captcha_y, delay=60, button="left") - await page.wait_for_load_state("networkidle") + await self._wait_for_networkidle(page) if iframe is not None: # Wait for the frame to be removed from the page while iframe in page.frames: @@ -595,8 +519,7 @@ class AsyncStealthySession(StealthySessionMixin, AsyncSession): if challenge_type != "embedded": await page.locator(box_selector).wait_for(state="detached") await page.locator(".zone-name-title").wait_for(state="hidden") - await page.wait_for_load_state(state="load") - await page.wait_for_load_state(state="domcontentloaded") + await self._wait_for_page_stability(page, True, False) log.info("Cloudflare captcha is solved") return @@ -657,22 +580,13 @@ class AsyncStealthySession(StealthySessionMixin, AsyncSession): if self._closed: # pragma: no cover raise RuntimeError("Context manager has been closed") - final_response = None referer = ( generate_convincing_referer(url) if (params.google_search and "referer" not in self._headers_keys) else None ) - async def handle_response(finished_response: AsyncPlaywrightResponse): - nonlocal final_response - if ( - finished_response.request.resource_type == "document" - and finished_response.request.is_navigation_request() - and finished_response.request.frame == page_info.page.main_frame - ): - final_response = finished_response - page_info = await self._get_page(params.timeout, params.extra_headers, params.disable_resources) - page_info.mark_busy(url=url) + final_response = [None] + handle_response = self._create_response_handler(page_info, final_response) if TYPE_CHECKING: if not isinstance(page_info.page, async_Page): @@ -682,11 +596,7 @@ class AsyncStealthySession(StealthySessionMixin, AsyncSession): # Navigate to URL and wait for a specified state page_info.page.on("response", handle_response) first_response = await page_info.page.goto(url, referer=referer) - if params.load_dom: - await page_info.page.wait_for_load_state(state="domcontentloaded") - - if params.network_idle: - await page_info.page.wait_for_load_state("networkidle") + await self._wait_for_page_stability(page_info.page, params.load_dom, params.network_idle) if not first_response: raise RuntimeError(f"Failed to get response for {url}") @@ -694,11 +604,7 @@ class AsyncStealthySession(StealthySessionMixin, AsyncSession): if params.solve_cloudflare: await self._solve_cloudflare(page_info.page) # Make sure the page is fully loaded after the captcha - await page_info.page.wait_for_load_state(state="load") - if params.load_dom: - await page_info.page.wait_for_load_state(state="domcontentloaded") - if params.network_idle: - await page_info.page.wait_for_load_state("networkidle") + await self._wait_for_page_stability(page_info.page, params.load_dom, params.network_idle) if params.page_action: try: @@ -711,11 +617,7 @@ class AsyncStealthySession(StealthySessionMixin, AsyncSession): waiter: AsyncLocator = page_info.page.locator(params.wait_selector) await waiter.first.wait_for(state=params.wait_selector_state) # Wait again after waiting for the selector, helpful with protections like Cloudflare - await page_info.page.wait_for_load_state(state="load") - if params.load_dom: - await page_info.page.wait_for_load_state(state="domcontentloaded") - if params.network_idle: - await page_info.page.wait_for_load_state("networkidle") + await self._wait_for_page_stability(page_info.page, params.load_dom, params.network_idle) except Exception as e: log.error(f"Error waiting for selector {params.wait_selector}: {e}") @@ -723,7 +625,7 @@ class AsyncStealthySession(StealthySessionMixin, AsyncSession): # Create response object response = await ResponseFactory.from_async_playwright_response( - page_info.page, first_response, final_response, params.selector_config, bool(params.page_action) + page_info.page, first_response, final_response[0], params.selector_config, bool(params.page_action) ) # Close the page to free up resources diff --git a/scrapling/engines/_browsers/_controllers.py b/scrapling/engines/_browsers/_controllers.py index 45a6938..155789a 100644 --- a/scrapling/engines/_browsers/_controllers.py +++ b/scrapling/engines/_browsers/_controllers.py @@ -1,16 +1,13 @@ from playwright.sync_api import ( - Response as SyncPlaywrightResponse, - sync_playwright, - Playwright, Locator, + Playwright, + sync_playwright, ) from playwright.async_api import ( async_playwright, - Response as AsyncPlaywrightResponse, - BrowserContext as AsyncBrowserContext, - Playwright as AsyncPlaywright, Locator as AsyncLocator, - Page as async_Page, + Playwright as AsyncPlaywright, + BrowserContext as AsyncBrowserContext, ) from patchright.sync_api import sync_playwright as sync_patchright from patchright.async_api import async_playwright as async_patchright @@ -178,28 +175,6 @@ class DynamicSession(DynamicSessionMixin, SyncSession): if self.cookies: # pragma: no cover self.context.add_cookies(self.cookies) - def __enter__(self): - self.__create__() - return self - - def __exit__(self, exc_type, exc_val, exc_tb): - self.close() - - 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 fetch( self, url: str, @@ -253,32 +228,19 @@ class DynamicSession(DynamicSessionMixin, SyncSession): if self._closed: # pragma: no cover raise RuntimeError("Context manager has been closed") - final_response = None referer = ( generate_convincing_referer(url) if (params.google_search and "referer" not in self._headers_keys) else None ) - def handle_response(finished_response: SyncPlaywrightResponse): - nonlocal final_response - if ( - finished_response.request.resource_type == "document" - and finished_response.request.is_navigation_request() - and finished_response.request.frame == page_info.page.main_frame - ): - final_response = finished_response - page_info = self._get_page(params.timeout, params.extra_headers, params.disable_resources) - page_info.mark_busy(url=url) + final_response = [None] + handle_response = self._create_response_handler(page_info, final_response) try: # pragma: no cover # Navigate to URL and wait for a specified state page_info.page.on("response", handle_response) first_response = page_info.page.goto(url, referer=referer) - if params.load_dom: - page_info.page.wait_for_load_state(state="domcontentloaded") - - if params.network_idle: - page_info.page.wait_for_load_state("networkidle") + self._wait_for_page_stability(page_info.page, params.load_dom, params.network_idle) if not first_response: raise RuntimeError(f"Failed to get response for {url}") @@ -294,11 +256,7 @@ class DynamicSession(DynamicSessionMixin, SyncSession): waiter: Locator = page_info.page.locator(params.wait_selector) waiter.first.wait_for(state=params.wait_selector_state) # Wait again after waiting for the selector, helpful with protections like Cloudflare - page_info.page.wait_for_load_state(state="load") - if params.load_dom: - page_info.page.wait_for_load_state(state="domcontentloaded") - if params.network_idle: - page_info.page.wait_for_load_state("networkidle") + self._wait_for_page_stability(page_info.page, params.load_dom, params.network_idle) except Exception as e: # pragma: no cover log.error(f"Error waiting for selector {params.wait_selector}: {e}") @@ -306,7 +264,7 @@ class DynamicSession(DynamicSessionMixin, SyncSession): # Create response object response = ResponseFactory.from_playwright_response( - page_info.page, first_response, final_response, params.selector_config, bool(params.page_action) + page_info.page, first_response, final_response[0], params.selector_config, bool(params.page_action) ) # Close the page to free up resources @@ -431,28 +389,6 @@ class AsyncDynamicSession(DynamicSessionMixin, AsyncSession): if self.cookies: await self.context.add_cookies(self.cookies) # pyright: ignore - async def __aenter__(self): - await self.__create__() - return self - - async def __aexit__(self, exc_type, exc_val, exc_tb): - await self.close() - - 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 fetch( self, url: str, @@ -506,24 +442,17 @@ class AsyncDynamicSession(DynamicSessionMixin, AsyncSession): if self._closed: # pragma: no cover raise RuntimeError("Context manager has been closed") - final_response = None referer = ( generate_convincing_referer(url) if (params.google_search and "referer" not in self._headers_keys) else None ) - async def handle_response(finished_response: AsyncPlaywrightResponse): - nonlocal final_response - if ( - finished_response.request.resource_type == "document" - and finished_response.request.is_navigation_request() - and finished_response.request.frame == page_info.page.main_frame - ): - final_response = finished_response - page_info = await self._get_page(params.timeout, params.extra_headers, params.disable_resources) - page_info.mark_busy(url=url) + final_response = [None] + handle_response = self._create_response_handler(page_info, final_response) if TYPE_CHECKING: + from playwright.async_api import Page as async_Page + if not isinstance(page_info.page, async_Page): raise TypeError @@ -531,11 +460,7 @@ class AsyncDynamicSession(DynamicSessionMixin, AsyncSession): # Navigate to URL and wait for a specified state page_info.page.on("response", handle_response) first_response = await page_info.page.goto(url, referer=referer) - if self.load_dom: - await page_info.page.wait_for_load_state(state="domcontentloaded") - - if params.network_idle: - await page_info.page.wait_for_load_state("networkidle") + await self._wait_for_page_stability(page_info.page, params.load_dom, params.network_idle) if not first_response: raise RuntimeError(f"Failed to get response for {url}") @@ -551,11 +476,7 @@ class AsyncDynamicSession(DynamicSessionMixin, AsyncSession): waiter: AsyncLocator = page_info.page.locator(params.wait_selector) await waiter.first.wait_for(state=params.wait_selector_state) # Wait again after waiting for the selector, helpful with protections like Cloudflare - await page_info.page.wait_for_load_state(state="load") - if self.load_dom: - await page_info.page.wait_for_load_state(state="domcontentloaded") - if params.network_idle: - await page_info.page.wait_for_load_state("networkidle") + await self._wait_for_page_stability(page_info.page, params.load_dom, params.network_idle) except Exception as e: log.error(f"Error waiting for selector {params.wait_selector}: {e}") @@ -563,7 +484,7 @@ class AsyncDynamicSession(DynamicSessionMixin, AsyncSession): # Create response object response = await ResponseFactory.from_async_playwright_response( - page_info.page, first_response, final_response, params.selector_config, bool(params.page_action) + page_info.page, first_response, final_response[0], params.selector_config, bool(params.page_action) ) # Close the page to free up resources diff --git a/scrapling/engines/toolbelt/custom.py b/scrapling/engines/toolbelt/custom.py index 43ef61c..1f20038 100644 --- a/scrapling/engines/toolbelt/custom.py +++ b/scrapling/engines/toolbelt/custom.py @@ -209,15 +209,3 @@ class StatusText: def get(cls, status_code: int) -> str: """Get the phrase for a given HTTP status code.""" return cls._phrases.get(status_code, "Unknown Status Code") - - -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 diff --git a/scrapling/engines/toolbelt/fingerprints.py b/scrapling/engines/toolbelt/fingerprints.py index 7bcad8a..c006ec2 100644 --- a/scrapling/engines/toolbelt/fingerprints.py +++ b/scrapling/engines/toolbelt/fingerprints.py @@ -7,8 +7,9 @@ from platform import system as platform_system from tldextract import extract from browserforge.headers import Browser, HeaderGenerator +from browserforge.headers.generator import SUPPORTED_OPERATING_SYSTEMS -from scrapling.core._types import Dict, Literal +from scrapling.core._types import Dict, Literal, Tuple __OS_NAME__ = platform_system() OSName = Literal["linux", "macos", "windows"] @@ -29,12 +30,12 @@ def generate_convincing_referer(url: str) -> str: @lru_cache(1, typed=True) -def get_os_name() -> OSName | None: +def get_os_name() -> OSName | Tuple: """Get the current OS name in the same format needed for browserforge, if the OS is Unknown, return None so browserforge uses all. :return: Current OS name or `None` otherwise """ - match __OS_NAME__: + match __OS_NAME__: # pragma: no cover case "Linux": return "linux" case "Darwin": @@ -42,7 +43,7 @@ def get_os_name() -> OSName | None: case "Windows": return "windows" case _: - return None + return SUPPORTED_OPERATING_SYSTEMS def generate_headers(browser_mode: bool = False) -> Dict: @@ -63,10 +64,7 @@ def generate_headers(browser_mode: bool = False) -> Dict: Browser(name="edge", min_version=130), ] ) - if os_name: - return HeaderGenerator(browser=browsers, os=os_name, device="desktop").generate() - else: - return HeaderGenerator(browser=browsers, device="desktop").generate() + return HeaderGenerator(browser=browsers, os=os_name, device="desktop").generate() __default_useragent__ = generate_headers(browser_mode=False).get("User-Agent")