From debe03256bbb250543b7d78bda9e7988795028f5 Mon Sep 17 00:00:00 2001 From: Karim shoair Date: Sun, 5 Oct 2025 04:03:39 +0300 Subject: [PATCH] refactor: Making all the codebase acceptable by PyRight Also fixes #97 --- scrapling/core/_types.py | 3 + scrapling/core/ai.py | 3 +- scrapling/core/custom_types.py | 47 ++--- scrapling/core/mixins.py | 24 ++- scrapling/core/shell.py | 7 +- scrapling/core/storage.py | 10 +- scrapling/core/translator.py | 21 ++- scrapling/engines/_browsers/_base.py | 39 ++-- scrapling/engines/_browsers/_camoufox.py | 17 +- scrapling/engines/_browsers/_config_tools.py | 2 +- scrapling/engines/_browsers/_controllers.py | 21 ++- scrapling/engines/_browsers/_validators.py | 39 ++-- scrapling/engines/static.py | 8 +- scrapling/engines/toolbelt/convertor.py | 10 +- scrapling/engines/toolbelt/custom.py | 15 +- scrapling/engines/toolbelt/fingerprints.py | 27 ++- scrapling/engines/toolbelt/navigation.py | 14 +- scrapling/fetchers/__init__.py | 12 +- scrapling/fetchers/chrome.py | 7 +- scrapling/fetchers/firefox.py | 4 - scrapling/parser.py | 181 +++++++++++-------- 21 files changed, 306 insertions(+), 205 deletions(-) diff --git a/scrapling/core/_types.py b/scrapling/core/_types.py index 59422f2..51016d4 100644 --- a/scrapling/core/_types.py +++ b/scrapling/core/_types.py @@ -12,9 +12,11 @@ from typing import ( Generator, Iterable, List, + Set, Literal, Optional, Pattern, + Sequence, Tuple, TypeVar, Union, @@ -22,6 +24,7 @@ from typing import ( Mapping, Awaitable, Protocol, + Coroutine, SupportsIndex, ) diff --git a/scrapling/core/ai.py b/scrapling/core/ai.py index 283f183..d497e57 100644 --- a/scrapling/core/ai.py +++ b/scrapling/core/ai.py @@ -20,6 +20,7 @@ from scrapling.core._types import ( Mapping, Dict, List, + Any, SelectorWaitStates, Generator, ) @@ -171,7 +172,7 @@ class ScraplingMCPServer: :param stealthy_headers: If enabled (default), it creates and adds real browser headers. It also sets the referer header as if this request came from a Google search of URL's domain. """ async with FetcherSession() as session: - tasks = [ + tasks: List[Any] = [ session.get( url, auth=auth, diff --git a/scrapling/core/custom_types.py b/scrapling/core/custom_types.py index e9ece7d..3675372 100644 --- a/scrapling/core/custom_types.py +++ b/scrapling/core/custom_types.py @@ -5,6 +5,7 @@ from re import compile as re_compile, UNICODE, IGNORECASE from orjson import dumps, loads from scrapling.core._types import ( + Any, cast, Dict, List, @@ -14,7 +15,6 @@ from scrapling.core._types import ( Literal, Pattern, Iterable, - Optional, Generator, SupportsIndex, ) @@ -33,23 +33,20 @@ class TextHandler(str): def __getitem__(self, key: SupportsIndex | slice) -> "TextHandler": # pragma: no cover lst = super().__getitem__(key) - return cast(_TextHandlerType, TextHandler(lst)) + return TextHandler(lst) - def split(self, sep: str = None, maxsplit: SupportsIndex = -1) -> "TextHandlers": # pragma: no cover - return TextHandlers( - cast( - List[_TextHandlerType], - [TextHandler(s) for s in super().split(sep, maxsplit)], - ) - ) + def split( + self, sep: str | None = None, maxsplit: SupportsIndex = -1 + ) -> Union[List, "TextHandlers"]: # pragma: no cover + return TextHandlers([TextHandler(s) for s in super().split(sep, maxsplit)]) - def strip(self, chars: str = None) -> Union[str, "TextHandler"]: # pragma: no cover + def strip(self, chars: str | None = None) -> Union[str, "TextHandler"]: # pragma: no cover return TextHandler(super().strip(chars)) - def lstrip(self, chars: str = None) -> Union[str, "TextHandler"]: # pragma: no cover + def lstrip(self, chars: str | None = None) -> Union[str, "TextHandler"]: # pragma: no cover return TextHandler(super().lstrip(chars)) - def rstrip(self, chars: str = None) -> Union[str, "TextHandler"]: # pragma: no cover + def rstrip(self, chars: str | None = None) -> Union[str, "TextHandler"]: # pragma: no cover return TextHandler(super().rstrip(chars)) def capitalize(self) -> Union[str, "TextHandler"]: # pragma: no cover @@ -64,7 +61,7 @@ class TextHandler(str): def expandtabs(self, tabsize: SupportsIndex = 8) -> Union[str, "TextHandler"]: # pragma: no cover return TextHandler(super().expandtabs(tabsize)) - def format(self, *args: str, **kwargs: str) -> Union[str, "TextHandler"]: # pragma: no cover + def format(self, *args: object, **kwargs: str) -> Union[str, "TextHandler"]: # pragma: no cover return TextHandler(super().format(*args, **kwargs)) def format_map(self, mapping) -> Union[str, "TextHandler"]: # pragma: no cover @@ -131,10 +128,11 @@ class TextHandler(str): def re( self, regex: str | Pattern, - check_match: Literal[True], replace_entities: bool = True, clean_match: bool = False, case_sensitive: bool = True, + *, + check_match: Literal[True], ) -> bool: ... @overload @@ -179,19 +177,14 @@ class TextHandler(str): results = flatten(results) if not replace_entities: - return TextHandlers(cast(List[_TextHandlerType], [TextHandler(string) for string in results])) + return TextHandlers([TextHandler(string) for string in results]) - return TextHandlers( - cast( - List[_TextHandlerType], - [TextHandler(_replace_entities(s)) for s in results], - ) - ) + return TextHandlers([TextHandler(_replace_entities(s)) for s in results]) def re_first( self, regex: str | Pattern, - default=None, + default: Any = None, replace_entities: bool = True, clean_match: bool = False, case_sensitive: bool = True, @@ -232,8 +225,8 @@ class TextHandlers(List[TextHandler]): def __getitem__(self, pos: SupportsIndex | slice) -> Union[TextHandler, "TextHandlers"]: lst = super().__getitem__(pos) if isinstance(pos, slice): - return TextHandlers(cast(List[_TextHandlerType], lst)) - return cast(_TextHandlerType, TextHandler(lst)) + return TextHandlers(cast(List[TextHandler], lst)) + return TextHandler(cast(TextHandler, lst)) def re( self, @@ -256,7 +249,7 @@ class TextHandlers(List[TextHandler]): def re_first( self, regex: str | Pattern, - default=None, + default: Any = None, replace_entities: bool = True, clean_match: bool = False, case_sensitive: bool = True, @@ -309,9 +302,9 @@ class AttributesHandler(Mapping[str, _TextHandlerType]): ) # Fastest read-only mapping type - self._data = MappingProxyType(mapping) + self._data: Mapping[str, Any] = MappingProxyType(mapping) - def get(self, key: str, default: Optional[str] = None) -> Optional[_TextHandlerType]: + def get(self, key: str, default: Any = None) -> _TextHandlerType: """Acts like the standard dictionary `.get()` method""" return self._data.get(key, default) diff --git a/scrapling/core/mixins.py b/scrapling/core/mixins.py index 4087020..3a96bda 100644 --- a/scrapling/core/mixins.py +++ b/scrapling/core/mixins.py @@ -1,3 +1,9 @@ +from scrapling.core._types import TYPE_CHECKING + +if TYPE_CHECKING: + from scrapling.parser import Selector + + class SelectorsGeneration: """ Functions for generating selectors @@ -5,7 +11,7 @@ class SelectorsGeneration: Inspiration: https://searchfox.org/mozilla-central/source/devtools/shared/inspector/css-logic.js#591 """ - def __general_selection(self, selection: str = "css", full_path: bool = False) -> str: + def _general_selection(self: "Selector", selection: str = "css", full_path: bool = False) -> str: # type: ignore[name-defined] """Generate a selector for the current element. :return: A string of the generated selector. """ @@ -47,29 +53,29 @@ class SelectorsGeneration: return " > ".join(reversed(selectorPath)) if css else "//" + "/".join(reversed(selectorPath)) @property - def generate_css_selector(self) -> str: + def generate_css_selector(self: "Selector") -> str: # type: ignore[name-defined] """Generate a CSS selector for the current element :return: A string of the generated selector. """ - return self.__general_selection() + return self._general_selection() @property - def generate_full_css_selector(self) -> str: + def generate_full_css_selector(self: "Selector") -> str: # type: ignore[name-defined] """Generate a complete CSS selector for the current element :return: A string of the generated selector. """ - return self.__general_selection(full_path=True) + return self._general_selection(full_path=True) @property - def generate_xpath_selector(self) -> str: + def generate_xpath_selector(self: "Selector") -> str: # type: ignore[name-defined] """Generate an XPath selector for the current element :return: A string of the generated selector. """ - return self.__general_selection("xpath") + return self._general_selection("xpath") @property - def generate_full_xpath_selector(self) -> str: + def generate_full_xpath_selector(self: "Selector") -> str: # type: ignore[name-defined] """Generate a complete XPath selector for the current element :return: A string of the generated selector. """ - return self.__general_selection("xpath", full_path=True) + return self._general_selection("xpath", full_path=True) diff --git a/scrapling/core/shell.py b/scrapling/core/shell.py index 5ef4458..0799496 100644 --- a/scrapling/core/shell.py +++ b/scrapling/core/shell.py @@ -31,6 +31,7 @@ from scrapling.core._types import ( Optional, Dict, Any, + cast, extraction_types, Generator, ) @@ -540,15 +541,15 @@ class Convertor: raise ValueError(f"Unknown extraction type: {extraction_type}") else: if main_content_only: - page = page.css_first("body") or page + page = cast(Selector, page.css_first("body")) or page - pages = [page] if not css_selector else page.css(css_selector) + pages = [page] if not css_selector else cast(Selectors, page.css(css_selector)) for page in pages: match extraction_type: case "markdown": yield cls._convert_to_markdown(page.html_content) case "html": - yield page.body + yield page.html_content case "text": txt_content = page.get_all_text(strip=True) for s in ( diff --git a/scrapling/core/storage.py b/scrapling/core/storage.py index e832cf0..50258c5 100644 --- a/scrapling/core/storage.py +++ b/scrapling/core/storage.py @@ -56,13 +56,13 @@ class StorageSystemMixin(ABC): # pragma: no cover @lru_cache(128, typed=True) def _get_hash(identifier: str) -> str: """If you want to hash identifier in your storage system, use this safer""" - identifier = identifier.lower().strip() - if isinstance(identifier, str): + _identifier = identifier.lower().strip() + if isinstance(_identifier, str): # Hash functions have to take bytes - identifier = identifier.encode("utf-8") + _identifier = _identifier.encode("utf-8") - hash_value = sha256(identifier).hexdigest() - return f"{hash_value}_{len(identifier)}" # Length to reduce collision chance + hash_value = sha256(_identifier).hexdigest() + return f"{hash_value}_{len(_identifier)}" # Length to reduce collision chance @lru_cache(1, typed=True) diff --git a/scrapling/core/translator.py b/scrapling/core/translator.py index d98092e..bb6d405 100644 --- a/scrapling/core/translator.py +++ b/scrapling/core/translator.py @@ -10,24 +10,23 @@ So you don't have to learn a new selectors/api method like what bs4 done with so from functools import lru_cache -from cssselect.xpath import ExpressionError -from cssselect.xpath import XPathExpr as OriginalXPathExpr from cssselect import HTMLTranslator as OriginalHTMLTranslator +from cssselect.xpath import ExpressionError, XPathExpr as OriginalXPathExpr from cssselect.parser import Element, FunctionalPseudoElement, PseudoElement -from scrapling.core._types import Any, Optional, Protocol, Self +from scrapling.core._types import Any, Protocol, Self class XPathExpr(OriginalXPathExpr): textnode: bool = False - attribute: Optional[str] = None + attribute: str | None = None @classmethod def from_xpath( cls, xpath: OriginalXPathExpr, textnode: bool = False, - attribute: Optional[str] = None, + attribute: str | None = None, ) -> Self: x = cls(path=xpath.path, element=xpath.element, condition=xpath.condition) x.textnode = textnode @@ -71,10 +70,10 @@ class XPathExpr(OriginalXPathExpr): # e.g. cssselect.GenericTranslator, cssselect.HTMLTranslator class TranslatorProtocol(Protocol): - def xpath_element(self, selector: Element) -> OriginalXPathExpr: # pragma: no cover + def xpath_element(self, selector: Element) -> OriginalXPathExpr: # pyright: ignore # pragma: no cover pass - def css_to_xpath(self, css: str, prefix: str = ...) -> str: # pragma: no cover + def css_to_xpath(self, css: str, prefix: str = ...) -> str: # pyright: ignore # pragma: no cover pass @@ -121,9 +120,15 @@ class TranslatorMixin: class HTMLTranslator(TranslatorMixin, OriginalHTMLTranslator): - @lru_cache(maxsize=256) def css_to_xpath(self, css: str, prefix: str = "descendant-or-self::") -> str: return super().css_to_xpath(css, prefix) translator = HTMLTranslator() +# Using a function instead of the translator directly to avoid Pyright override error + + +@lru_cache(maxsize=256) +def css_to_xpath(query: str) -> str: + """Return translated XPath version of a given CSS query""" + return translator.css_to_xpath(query) diff --git a/scrapling/engines/_browsers/_base.py b/scrapling/engines/_browsers/_base.py index d1a872d..f16ef14 100644 --- a/scrapling/engines/_browsers/_base.py +++ b/scrapling/engines/_browsers/_base.py @@ -7,14 +7,12 @@ from playwright.async_api import ( BrowserContext as AsyncBrowserContext, Playwright as AsyncPlaywright, ) -from camoufox.utils import ( - launch_options as generate_launch_options, - installed_verstr as camoufox_version, -) +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 Dict, Optional +from scrapling.core._types import Any, cast, Dict, Optional, 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 @@ -41,6 +39,7 @@ class SyncSession: """Get a new page to use""" # No need to check if a page is available or not in sync code because the code blocked before reaching here till the page closed, ofc. + assert self.context is not None, "Browser context not initialized" page = self.context.new_page() page.set_default_navigation_timeout(timeout) page.set_default_timeout(timeout) @@ -65,11 +64,14 @@ class SyncSession: } -class AsyncSession(SyncSession): +class AsyncSession: def __init__(self, max_pages: int = 1): - super().__init__(max_pages) + 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._closed = False self._lock = Lock() async def _get_page( @@ -79,6 +81,9 @@ class AsyncSession(SyncSession): disable_resources: bool, ) -> PageInfo: # pragma: no cover """Get a new page to use""" + if TYPE_CHECKING: + assert self.context is not None, "Browser context not initialized" + async with self._lock: # If we're at max capacity after cleanup, wait for busy pages to finish if self.page_pool.pages_count >= self.max_pages: @@ -92,6 +97,7 @@ class AsyncSession(SyncSession): f"No pages finished to clear place in the pool within the {self._max_wait_for_page}s timeout period" ) + assert self.context is not None, "Browser context not initialized" page = await self.context.new_page() page.set_default_navigation_timeout(timeout) page.set_default_timeout(timeout) @@ -107,6 +113,14 @@ class AsyncSession(SyncSession): return self.page_pool.add_page(page) + def get_pool_stats(self) -> Dict[str, int]: + """Get statistics about the current page pool""" + return { + "total_pages": self.page_pool.pages_count, + "busy_pages": self.page_pool.busy_count, + "max_pages": self.max_pages, + } + class DynamicSessionMixin: def __validate__(self, **params): @@ -139,6 +153,9 @@ class DynamicSessionMixin: self.__initiate_browser_options__() def __initiate_browser_options__(self): + if TYPE_CHECKING: + assert isinstance(self.proxy, tuple) + if not self.cdp_url: # `launch_options` is used with persistent context self.launch_options = dict( @@ -175,7 +192,7 @@ class DynamicSessionMixin: class StealthySessionMixin: def __validate__(self, **params): - config = validate(params, model=CamoufoxConfig) + config: CamoufoxConfig = validate(params, model=CamoufoxConfig) self.max_pages = config.max_pages self.headless = config.headless @@ -209,10 +226,10 @@ class StealthySessionMixin: def __initiate_browser_options__(self): """Initiate browser options.""" - self.launch_options = generate_launch_options( + self.launch_options: Dict[str, Any] = generate_launch_options( **{ "geoip": self.geoip, - "proxy": dict(self.proxy) if self.proxy else self.proxy, + "proxy": dict(self.proxy) if self.proxy and isinstance(self.proxy, tuple) else self.proxy, "addons": self.addons, "exclude_addons": [] if self.disable_ads else [DefaultAddons.UBO], "headless": self.headless, @@ -232,7 +249,7 @@ class StealthySessionMixin: "browser.cache.disk_cache_ssl": True, "browser.cache.disk.smart_size.enabled": True, }, - **self.additional_args, + **cast(Dict, self.additional_args), } ) diff --git a/scrapling/engines/_browsers/_camoufox.py b/scrapling/engines/_browsers/_camoufox.py index 6347a6c..7f42e65 100644 --- a/scrapling/engines/_browsers/_camoufox.py +++ b/scrapling/engines/_browsers/_camoufox.py @@ -26,6 +26,7 @@ from scrapling.core._types import ( List, Optional, Callable, + TYPE_CHECKING, SelectorWaitStates, ) from scrapling.engines.toolbelt.convertor import ( @@ -205,7 +206,7 @@ class StealthySession(StealthySessionMixin, SyncSession): self._closed = True @staticmethod - def _get_page_content(page: Page) -> str | None: + def _get_page_content(page: Page) -> str: """ A workaround for Playwright issue with `page.content()` on Windows. Ref.: https://github.com/microsoft/playwright/issues/16108 :param page: The page to extract content from. @@ -217,6 +218,7 @@ class StealthySession(StealthySessionMixin, SyncSession): except PlaywrightError: page.wait_for_timeout(1000) continue + return "" # pyright: ignore def _solve_cloudflare(self, page: Page) -> None: # pragma: no cover """Solve the cloudflare challenge displayed on the playwright page passed @@ -502,8 +504,8 @@ class AsyncStealthySession(StealthySessionMixin, AsyncSession): async def __create__(self): """Create a browser for this instance and context.""" - self.playwright: AsyncPlaywright = await async_playwright().start() - self.context: AsyncBrowserContext = await self.playwright.firefox.launch_persistent_context( + self.playwright: AsyncPlaywright | None = await async_playwright().start() + self.context: AsyncBrowserContext | None = await self.playwright.firefox.launch_persistent_context( **self.launch_options ) @@ -511,7 +513,7 @@ class AsyncStealthySession(StealthySessionMixin, AsyncSession): await self.context.add_init_script(path=self.init_script) if self.cookies: - await self.context.add_cookies(self.cookies) + await self.context.add_cookies(self.cookies) # pyright: ignore [reportArgumentType] async def __aenter__(self): await self.__create__() @@ -536,7 +538,7 @@ class AsyncStealthySession(StealthySessionMixin, AsyncSession): self._closed = True @staticmethod - async def _get_page_content(page: async_Page) -> str | None: + async def _get_page_content(page: async_Page) -> str: """ A workaround for Playwright issue with `page.content()` on Windows. Ref.: https://github.com/microsoft/playwright/issues/16108 :param page: The page to extract content from. @@ -548,6 +550,7 @@ class AsyncStealthySession(StealthySessionMixin, AsyncSession): except PlaywrightError: await page.wait_for_timeout(1000) continue + return "" # pyright: ignore async def _solve_cloudflare(self, page: async_Page): """Solve the cloudflare challenge displayed on the playwright page passed. The async version @@ -679,6 +682,10 @@ class AsyncStealthySession(StealthySessionMixin, AsyncSession): page_info = await self._get_page(params.timeout, params.extra_headers, params.disable_resources) page_info.mark_busy(url=url) + if TYPE_CHECKING: + if not isinstance(page_info.page, async_Page): + raise TypeError + try: # Navigate to URL and wait for a specified state page_info.page.on("response", handle_response) diff --git a/scrapling/engines/_browsers/_config_tools.py b/scrapling/engines/_browsers/_config_tools.py index 4b405e6..322d7ad 100644 --- a/scrapling/engines/_browsers/_config_tools.py +++ b/scrapling/engines/_browsers/_config_tools.py @@ -62,7 +62,7 @@ def _set_flags(hide_canvas, disable_webgl): # pragma: no cover @lru_cache(2, typed=True) def _launch_kwargs( headless, - proxy, + proxy: Tuple, locale, extra_headers, useragent, diff --git a/scrapling/engines/_browsers/_controllers.py b/scrapling/engines/_browsers/_controllers.py index b895f9a..576808b 100644 --- a/scrapling/engines/_browsers/_controllers.py +++ b/scrapling/engines/_browsers/_controllers.py @@ -10,6 +10,7 @@ from playwright.async_api import ( BrowserContext as AsyncBrowserContext, Playwright as AsyncPlaywright, Locator as AsyncLocator, + Page as async_Page, ) from patchright.sync_api import sync_playwright as sync_patchright from patchright.async_api import async_playwright as async_patchright @@ -18,10 +19,12 @@ from scrapling.core.utils import log from ._base import SyncSession, AsyncSession, DynamicSessionMixin from ._validators import validate_fetch as _validate from scrapling.core._types import ( + Any, Dict, List, Optional, Callable, + TYPE_CHECKING, SelectorWaitStates, ) from scrapling.engines.toolbelt.convertor import ( @@ -30,7 +33,7 @@ from scrapling.engines.toolbelt.convertor import ( ) from scrapling.engines.toolbelt.fingerprints import generate_convincing_referer -_UNSET = object() +_UNSET: Any = object() class DynamicSession(DynamicSessionMixin, SyncSession): @@ -154,7 +157,7 @@ class DynamicSession(DynamicSessionMixin, SyncSession): """Create a browser for this instance and context.""" sync_context = sync_patchright if self.stealth else sync_playwright - self.playwright: Playwright = sync_context().start() + self.playwright: Playwright = sync_context().start() # pyright: ignore [reportAttributeAccessIssue] if self.cdp_url: # pragma: no cover self.context = self.playwright.chromium.connect_over_cdp(endpoint_url=self.cdp_url).new_context( @@ -187,7 +190,7 @@ class DynamicSession(DynamicSessionMixin, SyncSession): if self.playwright: self.playwright.stop() - self.playwright = None + self.playwright = None # pyright: ignore self._closed = True @@ -399,7 +402,7 @@ class AsyncDynamicSession(DynamicSessionMixin, AsyncSession): """Create a browser for this instance and context.""" async_context = async_patchright if self.stealth else async_playwright - self.playwright: AsyncPlaywright = await async_context().start() + self.playwright: AsyncPlaywright = await async_context().start() # pyright: ignore [reportAttributeAccessIssue] if self.cdp_url: browser = await self.playwright.chromium.connect_over_cdp(endpoint_url=self.cdp_url) @@ -413,7 +416,7 @@ class AsyncDynamicSession(DynamicSessionMixin, AsyncSession): await self.context.add_init_script(path=self.init_script) if self.cookies: - await self.context.add_cookies(self.cookies) + await self.context.add_cookies(self.cookies) # pyright: ignore async def __aenter__(self): await self.__create__() @@ -429,11 +432,11 @@ class AsyncDynamicSession(DynamicSessionMixin, AsyncSession): if self.context: await self.context.close() - self.context = None + self.context = None # pyright: ignore if self.playwright: await self.playwright.stop() - self.playwright = None + self.playwright = None # pyright: ignore self._closed = True @@ -506,6 +509,10 @@ class AsyncDynamicSession(DynamicSessionMixin, AsyncSession): page_info = await self._get_page(params.timeout, params.extra_headers, params.disable_resources) page_info.mark_busy(url=url) + if TYPE_CHECKING: + if not isinstance(page_info.page, async_Page): + raise TypeError + try: # Navigate to URL and wait for a specified state page_info.page.on("response", handle_response) diff --git a/scrapling/engines/_browsers/_validators.py b/scrapling/engines/_browsers/_validators.py index a2d7c60..1c2dbbc 100644 --- a/scrapling/engines/_browsers/_validators.py +++ b/scrapling/engines/_browsers/_validators.py @@ -11,7 +11,10 @@ from scrapling.core._types import ( Tuple, Optional, Callable, + Iterable, SelectorWaitStates, + cast, + overload, ) from scrapling.engines.toolbelt.navigation import construct_proxy_dict @@ -73,7 +76,7 @@ class PlaywrightConfig(Struct, kw_only=True, frozen=False): stealth: bool = False wait: Seconds = 0 page_action: Optional[Callable] = None - proxy: Optional[str | Dict[str, str]] = None # The default value for proxy in Playwright's source is `None` + proxy: Optional[str | Dict[str, str] | Tuple] = None # The default value for proxy in Playwright's source is `None` locale: str = "en-US" extra_headers: Optional[Dict[str, str]] = None useragent: Optional[str] = None @@ -81,11 +84,11 @@ class PlaywrightConfig(Struct, kw_only=True, frozen=False): init_script: Optional[str] = None disable_resources: bool = False wait_selector: Optional[str] = None - cookies: Optional[List[Dict]] = None + cookies: Optional[Iterable[Dict]] = None network_idle: bool = False load_dom: bool = True wait_selector_state: SelectorWaitStates = "attached" - selector_config: Optional[Dict] = None + selector_config: Optional[Dict] = {} def __post_init__(self): """Custom validation after msgspec validation""" @@ -125,15 +128,15 @@ class CamoufoxConfig(Struct, kw_only=True, frozen=False): wait_selector: Optional[str] = None addons: Optional[List[str]] = None wait_selector_state: SelectorWaitStates = "attached" - cookies: Optional[List[Dict]] = None + cookies: Optional[Iterable[Dict]] = None google_search: bool = True extra_headers: Optional[Dict[str, str]] = None - proxy: Optional[str | Dict[str, str]] = None # The default value for proxy in Playwright's source is `None` + proxy: Optional[str | Dict[str, str] | Tuple] = None # The default value for proxy in Playwright's source is `None` os_randomize: bool = False disable_ads: bool = False geoip: bool = False - selector_config: Optional[Dict] = None - additional_args: Optional[Dict] = None + selector_config: Optional[Dict] = {} + additional_args: Optional[Dict] = {} def __post_init__(self): """Custom validation after msgspec validation""" @@ -177,7 +180,7 @@ class FetchConfig(Struct, kw_only=True): network_idle: bool = False load_dom: bool = True solve_cloudflare: bool = False - selector_config: Optional[Dict] = {} + selector_config: Dict = {} def to_dict(self): return {f: getattr(self, f) for f in self.__struct_fields__} @@ -198,7 +201,7 @@ class _fetch_params: network_idle: bool load_dom: bool solve_cloudflare: bool - selector_config: Optional[Dict] + selector_config: Dict def validate_fetch(params: List[Tuple], sentinel=None) -> _fetch_params: @@ -212,7 +215,7 @@ def validate_fetch(params: List[Tuple], sentinel=None) -> _fetch_params: result[arg] = session_value if overrides: - overrides = validate(overrides, FetchConfig).to_dict() + overrides = cast(FetchConfig, validate(overrides, FetchConfig)).to_dict() overrides.update(result) return _fetch_params(**overrides) @@ -222,7 +225,21 @@ def validate_fetch(params: List[Tuple], sentinel=None) -> _fetch_params: return _fetch_params(**result) -def validate(params: Dict, model) -> PlaywrightConfig | CamoufoxConfig | FetchConfig: +@overload +def validate(params: Dict, model: type[PlaywrightConfig]) -> PlaywrightConfig: ... + + +@overload +def validate(params: Dict, model: type[CamoufoxConfig]) -> CamoufoxConfig: ... + + +@overload +def validate(params: Dict, model: type[FetchConfig]) -> FetchConfig: ... + + +def validate( + params: Dict, model: type[PlaywrightConfig] | type[CamoufoxConfig] | type[FetchConfig] +) -> PlaywrightConfig | CamoufoxConfig | FetchConfig: try: return convert(params, model) except ValidationError as e: diff --git a/scrapling/engines/static.py b/scrapling/engines/static.py index 3c8bef9..d49b5da 100644 --- a/scrapling/engines/static.py +++ b/scrapling/engines/static.py @@ -182,7 +182,7 @@ class FetcherSession: return headers - def __enter__(self): + def __enter__(self) -> "FetcherClient": """Creates and returns a new synchronous Fetcher Session""" if self._curl_session: raise RuntimeError( @@ -197,7 +197,7 @@ class FetcherSession: ) self._curl_session = CurlSession() - return self + return cast("FetcherClient", self) def __exit__(self, exc_type, exc_val, exc_tb): """Closes the active synchronous session managed by this instance, if any.""" @@ -205,7 +205,7 @@ class FetcherSession: self._curl_session.close() self._curl_session = None - async def __aenter__(self): + async def __aenter__(self) -> "AsyncFetcherClient": """Creates and returns a new asynchronous Session.""" if self._async_curl_session: raise RuntimeError( @@ -220,7 +220,7 @@ class FetcherSession: ) self._async_curl_session = AsyncCurlSession() - return self + return cast("AsyncFetcherClient", self) async def __aexit__(self, exc_type, exc_val, exc_tb): """Closes the active asynchronous session managed by this instance, if any.""" diff --git a/scrapling/engines/toolbelt/convertor.py b/scrapling/engines/toolbelt/convertor.py index 05dce87..cc810db 100644 --- a/scrapling/engines/toolbelt/convertor.py +++ b/scrapling/engines/toolbelt/convertor.py @@ -58,7 +58,8 @@ class ResponseFactory: "encoding": cls.__extract_browser_encoding( current_response.headers.get("content-type", "") ) - or "utf-8", + if current_response + else "utf-8", "cookies": tuple(), "headers": current_response.all_headers() if current_response else {}, "request_headers": current_request.all_headers(), @@ -161,7 +162,8 @@ class ResponseFactory: "encoding": cls.__extract_browser_encoding( current_response.headers.get("content-type", "") ) - or "utf-8", + if current_response + else "utf-8", "cookies": tuple(), "headers": await current_response.all_headers() if current_response else {}, "request_headers": await current_request.all_headers(), @@ -255,8 +257,8 @@ class ResponseFactory: "encoding": response.encoding or "utf-8", "cookies": dict(response.cookies), "headers": dict(response.headers), - "request_headers": dict(response.request.headers), - "method": response.request.method, + "request_headers": dict(response.request.headers) if response.request else {}, + "method": response.request.method if response.request else "GET", "history": response.history, # https://github.com/lexiforest/curl_cffi/issues/82 **parser_arguments, } diff --git a/scrapling/engines/toolbelt/custom.py b/scrapling/engines/toolbelt/custom.py index 774eec7..43ef61c 100644 --- a/scrapling/engines/toolbelt/custom.py +++ b/scrapling/engines/toolbelt/custom.py @@ -8,6 +8,7 @@ from scrapling.core.utils import log from scrapling.core._types import ( Any, Dict, + cast, List, Optional, Tuple, @@ -30,10 +31,10 @@ class Response(Selector): request_headers: Dict, encoding: str = "utf-8", method: str = "GET", - history: List = None, - **selector_config: Dict, + history: List | None = None, + **selector_config: Any, ): - adaptive_domain = selector_config.pop("adaptive_domain", None) + adaptive_domain: str = cast(str, selector_config.pop("adaptive_domain", "")) self.status = status self.reason = reason self.cookies = cookies @@ -58,7 +59,7 @@ class BaseFetcher: keep_cdata: Optional[bool] = False storage_args: Optional[Dict] = None keep_comments: Optional[bool] = False - adaptive_domain: Optional[str] = None + adaptive_domain: str = "" parser_keywords: Tuple = ( "huge_tree", "adaptive", @@ -124,12 +125,8 @@ class BaseFetcher: adaptive=cls.adaptive, storage=cls.storage, storage_args=cls.storage_args, + adaptive_domain=cls.adaptive_domain, ) - if cls.adaptive_domain: - if not isinstance(cls.adaptive_domain, str): - log.warning('[Ignored] The argument "adaptive_domain" must be of string type') - else: - parser_arguments.update({"adaptive_domain": cls.adaptive_domain}) return parser_arguments diff --git a/scrapling/engines/toolbelt/fingerprints.py b/scrapling/engines/toolbelt/fingerprints.py index bd836e7..7bcad8a 100644 --- a/scrapling/engines/toolbelt/fingerprints.py +++ b/scrapling/engines/toolbelt/fingerprints.py @@ -8,9 +8,10 @@ from platform import system as platform_system from tldextract import extract from browserforge.headers import Browser, HeaderGenerator -from scrapling.core._types import Dict, Optional +from scrapling.core._types import Dict, Literal __OS_NAME__ = platform_system() +OSName = Literal["linux", "macos", "windows"] @lru_cache(10, typed=True) @@ -28,16 +29,20 @@ def generate_convincing_referer(url: str) -> str: @lru_cache(1, typed=True) -def get_os_name() -> Optional[str]: - """Get the current OS name in the same format needed for browserforge +def get_os_name() -> OSName | None: + """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 """ - return { - "Linux": "linux", - "Darwin": "macos", - "Windows": "windows", - }.get(__OS_NAME__) + match __OS_NAME__: + case "Linux": + return "linux" + case "Darwin": + return "macos" + case "Windows": + return "windows" + case _: + return None def generate_headers(browser_mode: bool = False) -> Dict: @@ -58,8 +63,10 @@ def generate_headers(browser_mode: bool = False) -> Dict: Browser(name="edge", min_version=130), ] ) - - return HeaderGenerator(browser=browsers, os=os_name, device="desktop").generate() + if os_name: + return HeaderGenerator(browser=browsers, os=os_name, device="desktop").generate() + else: + return HeaderGenerator(browser=browsers, device="desktop").generate() __default_useragent__ = generate_headers(browser_mode=False).get("User-Agent") diff --git a/scrapling/engines/toolbelt/navigation.py b/scrapling/engines/toolbelt/navigation.py index ea5991b..959aa5f 100644 --- a/scrapling/engines/toolbelt/navigation.py +++ b/scrapling/engines/toolbelt/navigation.py @@ -11,7 +11,7 @@ from msgspec import Struct, structs, convert, ValidationError from playwright.sync_api import Route from scrapling.core.utils import log -from scrapling.core._types import Dict, Optional, Tuple +from scrapling.core._types import Dict, Tuple, overload, Literal from scrapling.engines.constants import DEFAULT_DISABLED_RESOURCES __BYPASSES_DIR__ = Path(__file__).parent / "bypasses" @@ -49,7 +49,15 @@ async def async_intercept_route(route: async_Route): await route.continue_() -def construct_proxy_dict(proxy_string: str | Dict[str, str], as_tuple=False) -> Optional[Dict | Tuple]: +@overload +def construct_proxy_dict(proxy_string: str | Dict[str, str] | Tuple, as_tuple: Literal[True]) -> Tuple: ... + + +@overload +def construct_proxy_dict(proxy_string: str | Dict[str, str] | Tuple, as_tuple: Literal[False] = False) -> Dict: ... + + +def construct_proxy_dict(proxy_string: str | Dict[str, str] | Tuple, as_tuple: bool = False) -> Dict | Tuple: """Validate a proxy and return it in the acceptable format for Playwright Reference: https://playwright.dev/python/docs/network#http-proxy @@ -83,7 +91,7 @@ def construct_proxy_dict(proxy_string: str | Dict[str, str], as_tuple=False) -> except ValidationError as e: raise TypeError(f"Invalid proxy dictionary: {e}") - return None + raise TypeError(f"Invalid proxy string: {proxy_string}") @lru_cache(10, typed=True) diff --git a/scrapling/fetchers/__init__.py b/scrapling/fetchers/__init__.py index 9c64659..c135273 100644 --- a/scrapling/fetchers/__init__.py +++ b/scrapling/fetchers/__init__.py @@ -19,7 +19,17 @@ _LAZY_IMPORTS = { "AsyncStealthySession": ("scrapling.fetchers.firefox", "AsyncStealthySession"), } -__all__ = ["Fetcher", "AsyncFetcher", "StealthyFetcher", "DynamicFetcher"] +__all__ = [ + "Fetcher", + "AsyncFetcher", + "FetcherSession", + "DynamicFetcher", + "DynamicSession", + "AsyncDynamicSession", + "StealthyFetcher", + "StealthySession", + "AsyncStealthySession", +] def __getattr__(name: str) -> Any: diff --git a/scrapling/fetchers/chrome.py b/scrapling/fetchers/chrome.py index 49fe009..3c33c47 100644 --- a/scrapling/fetchers/chrome.py +++ b/scrapling/fetchers/chrome.py @@ -1,10 +1,9 @@ from scrapling.core._types import ( Callable, - Dict, List, + Dict, Optional, SelectorWaitStates, - Iterable, ) from scrapling.engines.toolbelt.custom import BaseFetcher, Response from scrapling.engines._browsers._controllers import DynamicSession, AsyncDynamicSession @@ -47,7 +46,7 @@ class DynamicFetcher(BaseFetcher): disable_resources: bool = False, wait_selector: Optional[str] = None, init_script: Optional[str] = None, - cookies: Optional[Iterable[Dict]] = None, + cookies: Optional[List[Dict]] = None, network_idle: bool = False, load_dom: bool = True, wait_selector_state: SelectorWaitStates = "attached", @@ -134,7 +133,7 @@ class DynamicFetcher(BaseFetcher): disable_resources: bool = False, wait_selector: Optional[str] = None, init_script: Optional[str] = None, - cookies: Optional[Iterable[Dict]] = None, + cookies: Optional[List[Dict]] = None, network_idle: bool = False, load_dom: bool = True, wait_selector_state: SelectorWaitStates = "attached", diff --git a/scrapling/fetchers/firefox.py b/scrapling/fetchers/firefox.py index ca6db40..5986096 100644 --- a/scrapling/fetchers/firefox.py +++ b/scrapling/fetchers/firefox.py @@ -83,8 +83,6 @@ class StealthyFetcher(BaseFetcher): """ if not custom_config: custom_config = {} - elif not isinstance(custom_config, dict): - ValueError(f"The custom parser config must be of type dictionary, got {cls.__class__}") with StealthySession( wait=wait, @@ -182,8 +180,6 @@ class StealthyFetcher(BaseFetcher): """ if not custom_config: custom_config = {} - elif not isinstance(custom_config, dict): - ValueError(f"The custom parser config must be of type dictionary, got {cls.__class__}") async with AsyncStealthySession( wait=wait, diff --git a/scrapling/parser.py b/scrapling/parser.py index 6d0a575..6a4934c 100644 --- a/scrapling/parser.py +++ b/scrapling/parser.py @@ -17,17 +17,21 @@ from lxml.etree import ( from scrapling.core._types import ( Any, + Set, Dict, + cast, List, Tuple, Union, Pattern, Callable, + Literal, Optional, Iterable, overload, Generator, SupportsIndex, + TYPE_CHECKING, ) from scrapling.core.custom_types import AttributesHandler, TextHandler, TextHandlers from scrapling.core.mixins import SelectorsGeneration @@ -36,7 +40,7 @@ from scrapling.core.storage import ( StorageSystemMixin, _StorageTools, ) -from scrapling.core.translator import translator as _translator +from scrapling.core.translator import css_to_xpath as _css_to_xpath from scrapling.core.utils import clean_spaces, flatten, html_forbidden, log __DEFAULT_DB_FILE__ = str(Path(__file__).parent / "elements_storage.db") @@ -70,20 +74,23 @@ class Selector(SelectorsGeneration): "_raw_body", ) + if TYPE_CHECKING: + _storage: StorageSystemMixin + def __init__( self, content: Optional[str | bytes] = None, - url: Optional[str] = None, + url: str = "", encoding: str = "utf-8", huge_tree: bool = True, root: Optional[HtmlElement] = None, keep_comments: Optional[bool] = False, keep_cdata: Optional[bool] = False, adaptive: Optional[bool] = False, - _storage: object = None, + _storage: Optional[StorageSystemMixin] = None, storage: Any = SQLiteStorageSystem, storage_args: Optional[Dict] = None, - **kwargs, + **_, ): """The main class that works as a wrapper for the HTML input data. Using this class, you can search for elements with expressions in CSS, XPath, or with simply text. Check the docs for more info. @@ -131,7 +138,7 @@ class Selector(SelectorsGeneration): default_doctype=True, strip_cdata=(not keep_cdata), ) - self._root = fromstring(body, parser=parser, base_url=url) + self._root = cast(HtmlElement, fromstring(body, parser=parser, base_url=url or None)) self._raw_body = content else: @@ -141,7 +148,7 @@ class Selector(SelectorsGeneration): f"Root have to be a valid element of `html` module types to work, not of type {type(root)}" ) - self._root = root + self._root = cast(HtmlElement, root) self._raw_body = "" self.__adaptive_enabled = adaptive @@ -238,6 +245,9 @@ class Selector(SelectorsGeneration): **self.__response_data, ) + def __elements_convertor(self, elements: List[HtmlElement]) -> "Selectors": + return Selectors(map(self.__element_convertor, elements)) + def __handle_element( self, element: Optional[HtmlElement | _ElementUnicodeResult] ) -> Optional[Union[TextHandler, "Selector"]]: @@ -262,7 +272,7 @@ class Selector(SelectorsGeneration): if self._is_text_node(result[0]): return TextHandlers(map(TextHandler, result)) - return Selectors(map(self.__element_convertor, result)) + return self.__elements_convertor(result) def __getstate__(self) -> Any: # lxml don't like it :) @@ -323,7 +333,7 @@ class Selector(SelectorsGeneration): if not valid_values or processed_text.strip(): _all_strings.append(processed_text) - return TextHandler(separator).join(_all_strings) + return cast(TextHandler, TextHandler(separator).join(_all_strings)) def urljoin(self, relative_url: str) -> str: """Join this Selector's url with a relative url to form an absolute full URL.""" @@ -372,13 +382,14 @@ class Selector(SelectorsGeneration): @property def parent(self) -> Optional["Selector"]: """Return the direct parent of the element or ``None`` otherwise""" - return self.__handle_element(self._root.getparent()) + _parent = self._root.getparent() + return self.__element_convertor(_parent) if _parent is not None else None @property def below_elements(self) -> "Selectors": """Return all elements under the current element in the DOM tree""" below = _find_all_elements(self._root) - return self.__handle_elements(below) + return self.__elements_convertor(below) if below is not None else Selectors() @property def children(self) -> "Selectors": @@ -425,7 +436,7 @@ class Selector(SelectorsGeneration): # Ignore HTML comments and unwanted types next_element = next_element.getnext() - return self.__handle_element(next_element) + return self.__element_convertor(next_element) if next_element is not None else None @property def previous(self) -> Optional["Selector"]: @@ -435,10 +446,10 @@ class Selector(SelectorsGeneration): # Ignore HTML comments and unwanted types prev_element = prev_element.getprevious() - return self.__handle_element(prev_element) + return self.__element_convertor(prev_element) if prev_element is not None else None # For easy copy-paste from Scrapy/parsel code when needed :) - def get(self, default=None): + def get(self, default=None): # pyright: ignore return self def get_all(self): @@ -468,6 +479,16 @@ class Selector(SelectorsGeneration): return data + ">" # From here we start with the selecting functions + @overload + def relocate( + self, element: Union[Dict, HtmlElement, "Selector"], percentage: int, selector_type: Literal[True] + ) -> "Selectors": ... + + @overload + def relocate( + self, element: Union[Dict, HtmlElement, "Selector"], percentage: int, selector_type: Literal[False] = False + ) -> List[HtmlElement]: ... + def relocate( self, element: Union[Dict, HtmlElement, "Selector"], @@ -506,11 +527,11 @@ class Selector(SelectorsGeneration): log.debug(f"Highest probability was {highest_probability}%") log.debug("Top 5 best matching elements are: ") for percent in tuple(sorted(score_table.keys(), reverse=True))[:5]: - log.debug(f"{percent} -> {self.__handle_elements(score_table[percent])}") + log.debug(f"{percent} -> {self.__elements_convertor(score_table[percent])}") if not selector_type: return score_table[highest_probability] - return self.__handle_elements(score_table[highest_probability]) + return self.__elements_convertor(score_table[highest_probability]) return [] def css_first( @@ -593,7 +614,7 @@ class Selector(SelectorsGeneration): auto_save: bool = False, percentage: int = 0, **kwargs: Any, - ) -> Union["Selectors", List, "TextHandlers"]: + ) -> Union["Selectors", List[Any], "TextHandlers"]: """Search the current tree with CSS3 selectors **Important: @@ -614,7 +635,7 @@ class Selector(SelectorsGeneration): try: if not self.__adaptive_enabled or "," not in selector: # No need to split selectors in this case, let's save some CPU cycles :) - xpath_selector = _translator.css_to_xpath(selector) + xpath_selector = _css_to_xpath(selector) return self.xpath( xpath_selector, identifier or selector, @@ -628,7 +649,7 @@ class Selector(SelectorsGeneration): for single_selector in split_selectors(selector): # I'm doing this only so the `save` function saves data correctly for combined selectors # Like using the ',' to combine two different selectors that point to different elements. - xpath_selector = _translator.css_to_xpath(single_selector.canonical()) + xpath_selector = _css_to_xpath(single_selector.canonical()) results += self.xpath( xpath_selector, identifier or single_selector.canonical(), @@ -731,7 +752,8 @@ class Selector(SelectorsGeneration): raise TypeError("You have to pass something to search with, like tag name(s), tag attributes, or both.") attributes = dict() - tags, patterns = set(), set() + tags: Set[str] = set() + patterns: Set[Pattern] = set() results, functions, selectors = Selectors(), [], [] # Brace yourself for a wonderful journey! @@ -740,6 +762,7 @@ class Selector(SelectorsGeneration): tags.add(arg) elif type(arg) in (list, tuple, set): + arg = cast(Iterable, arg) # Type narrowing for type checkers like pyright if not all(map(lambda x: isinstance(x, str), arg)): raise TypeError("Nested Iterables are not accepted, only iterables of tag names are accepted") tags.update(set(arg)) @@ -774,7 +797,7 @@ class Selector(SelectorsGeneration): attributes[attribute_name] = value # It's easier and faster to build a selector than traversing the tree - tags = tags or ["*"] + tags = tags or set("*") for tag in tags: selector = tag for key, value in attributes.items(): @@ -785,7 +808,7 @@ class Selector(SelectorsGeneration): selectors.append(selector) if selectors: - results = self.css(", ".join(selectors)) + results = cast(Selectors, self.css(", ".join(selectors))) if results: # From the results, get the ones that fulfill passed regex patterns for pattern in patterns: @@ -828,20 +851,20 @@ class Selector(SelectorsGeneration): :return: A percentage score of how similar is the candidate to the original element """ score, checks = 0, 0 - candidate = _StorageTools.element_to_dict(candidate) + data = _StorageTools.element_to_dict(candidate) # Possible TODO: # Study the idea of giving weight to each test below so some are more important than others # Current results: With weights some websites had better score while it was worse for others - score += 1 if original["tag"] == candidate["tag"] else 0 # * 0.3 # 30% + score += 1 if original["tag"] == data["tag"] else 0 # * 0.3 # 30% checks += 1 if original["text"]: - score += SequenceMatcher(None, original["text"], candidate.get("text") or "").ratio() # * 0.3 # 30% + score += SequenceMatcher(None, original["text"], data.get("text") or "").ratio() # * 0.3 # 30% checks += 1 # if both don't have attributes, it still counts for something! - score += self.__calculate_dict_diff(original["attributes"], candidate["attributes"]) # * 0.3 # 30% + score += self.__calculate_dict_diff(original["attributes"], data["attributes"]) # * 0.3 # 30% checks += 1 # Separate similarity test for class, id, href,... this will help in full structural changes @@ -855,23 +878,23 @@ class Selector(SelectorsGeneration): score += SequenceMatcher( None, original["attributes"][attrib], - candidate["attributes"].get(attrib) or "", + data["attributes"].get(attrib) or "", ).ratio() # * 0.3 # 30% checks += 1 - score += SequenceMatcher(None, original["path"], candidate["path"]).ratio() # * 0.1 # 10% + score += SequenceMatcher(None, original["path"], data["path"]).ratio() # * 0.1 # 10% checks += 1 if original.get("parent_name"): # Then we start comparing parents' data - if candidate.get("parent_name"): + if data.get("parent_name"): score += SequenceMatcher( - None, original["parent_name"], candidate.get("parent_name") or "" + None, original["parent_name"], data.get("parent_name") or "" ).ratio() # * 0.2 # 20% checks += 1 score += self.__calculate_dict_diff( - original["parent_attribs"], candidate.get("parent_attribs") or {} + original["parent_attribs"], data.get("parent_attribs") or {} ) # * 0.2 # 20% checks += 1 @@ -879,7 +902,7 @@ class Selector(SelectorsGeneration): score += SequenceMatcher( None, original["parent_text"], - candidate.get("parent_text") or "", + data.get("parent_text") or "", ).ratio() # * 0.1 # 10% checks += 1 # else: @@ -887,9 +910,7 @@ class Selector(SelectorsGeneration): # score -= 0.1 if original.get("siblings"): - score += SequenceMatcher( - None, original["siblings"], candidate.get("siblings") or [] - ).ratio() # * 0.1 # 10% + score += SequenceMatcher(None, original["siblings"], data.get("siblings") or []).ratio() # * 0.1 # 10% checks += 1 # How % sure? let's see @@ -902,7 +923,7 @@ class Selector(SelectorsGeneration): score += SequenceMatcher(None, tuple(dict1.values()), tuple(dict2.values())).ratio() * 0.5 return score - def save(self, element: Union["Selector", HtmlElement], identifier: str) -> None: + def save(self, element: HtmlElement, identifier: str) -> None: """Saves the element's unique properties to the storage for retrieval and relocation later :param element: The element itself that we want to save to storage, it can be a ` Selector ` or pure ` HtmlElement ` @@ -910,15 +931,16 @@ class Selector(SelectorsGeneration): the docs for more info. """ if self.__adaptive_enabled: - if isinstance(element, self.__class__): - element = element._root + target = element + if isinstance(target, self.__class__): + target: HtmlElement = target._root - if self._is_text_node(element): - element = element.getparent() + if self._is_text_node(target): + target: HtmlElement = target.getparent() - self._storage.save(element, identifier) + self._storage.save(target, identifier) else: - log.critical( + raise RuntimeError( "Can't use `adaptive` features while it's disabled globally, you have to start a new class instance." ) @@ -932,10 +954,9 @@ class Selector(SelectorsGeneration): if self.__adaptive_enabled: return self._storage.retrieve(identifier) - log.critical( + raise RuntimeError( "Can't use `adaptive` features while it's disabled globally, you have to start a new class instance." ) - return None # Operations on text functions def json(self) -> Dict: @@ -1104,28 +1125,30 @@ class Selector(SelectorsGeneration): if not case_sensitive: text = text.lower() - for node in self.__handle_elements(_find_all_elements_with_spaces(self._root)): - """Check if element matches given text otherwise, traverse the children tree and iterate""" - node_text = node.text - if clean_match: - node_text = node_text.clean() + possible_targets = _find_all_elements_with_spaces(self._root) + if possible_targets: + for node in self.__elements_convertor(possible_targets): + """Check if element matches given text otherwise, traverse the children tree and iterate""" + node_text = node.text + if clean_match: + node_text = node_text.clean() - if not case_sensitive: - node_text = node_text.lower() + if not case_sensitive: + node_text = node_text.lower() - if partial: - if text in node_text: + if partial: + if text in node_text: + results.append(node) + elif text == node_text: results.append(node) - elif text == node_text: - results.append(node) - if first_match and results: - # we got an element so we should stop - break + if first_match and results: + # we got an element so we should stop + break - if first_match: - if results: - return results[0] + if first_match: + if results: + return results[0] return results def find_by_regex( @@ -1143,23 +1166,25 @@ class Selector(SelectorsGeneration): """ results = Selectors() - for node in self.__handle_elements(_find_all_elements_with_spaces(self._root)): - """Check if element matches given regex otherwise, traverse the children tree and iterate""" - node_text = node.text - if node_text.re( - query, - check_match=True, - clean_match=clean_match, - case_sensitive=case_sensitive, - ): - results.append(node) + possible_targets = _find_all_elements_with_spaces(self._root) + if possible_targets: + for node in self.__elements_convertor(possible_targets): + """Check if element matches given regex otherwise, traverse the children tree and iterate""" + node_text = node.text + if node_text.re( + query, + check_match=True, + clean_match=clean_match, + case_sensitive=case_sensitive, + ): + results.append(node) - if first_match and results: - # we got an element so we should stop - break + if first_match and results: + # we got an element so we should stop + break - if results and first_match: - return results[0] + if results and first_match: + return results[0] return results @@ -1181,9 +1206,9 @@ class Selectors(List[Selector]): def __getitem__(self, pos: SupportsIndex | slice) -> Union[Selector, "Selectors"]: lst = super().__getitem__(pos) if isinstance(pos, slice): - return self.__class__(lst) + return self.__class__(cast(List[Selector], lst)) else: - return lst + return cast(Selector, lst) def xpath( self, @@ -1265,7 +1290,7 @@ class Selectors(List[Selector]): def re_first( self, regex: str | Pattern, - default=None, + default: Any = None, replace_entities: bool = True, clean_match: bool = False, case_sensitive: bool = True,