freedak/Scrapling
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor(fetchers): Less duplicated code and better handling for unstable websites

Browse Source 
- 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
This commit is contained in:
Karim shoair
2025-10-26 05:04:19 +03:00
parent 74fa45daea
commit 59ec6be201
5 changed files with 185 additions and 248 deletions
Download Patch File Download Diff File Expand all files Collapse all files
scrapling/engines/_browsers/_base.py
+137 -9
View File
@@ -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):
scrapling/engines/_browsers/_camoufox.py
+26 -124
View File
@@ -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
scrapling/engines/_browsers/_controllers.py
+16 -95
View File
@@ -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
scrapling/engines/toolbelt/custom.py
-12
View File
@@ -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
scrapling/engines/toolbelt/fingerprints.py
+6 -8
View File
@@ -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")