diff --git a/scrapling/core/_types.py b/scrapling/core/_types.py index f19e706..f2fc097 100644 --- a/scrapling/core/_types.py +++ b/scrapling/core/_types.py @@ -32,6 +32,7 @@ from typing import ( Coroutine, SupportsIndex, ) +from typing_extensions import Self, Unpack # Proxy can be a string URL or a dict (Playwright format: {"server": "...", "username": "...", "password": "..."}) ProxyType = Union[str, Dict[str, str]] @@ -41,27 +42,6 @@ PageLoadStates = Literal["commit", "domcontentloaded", "load", "networkidle"] extraction_types = Literal["text", "html", "markdown"] StrOrBytes = Union[str, bytes] -if TYPE_CHECKING: # pragma: no cover - from typing_extensions import Unpack -else: # pragma: no cover - - class _Unpack: - @staticmethod - def __getitem__(*args, **kwargs): - pass - - Unpack = _Unpack() - - -try: - # Python 3.11+ - from typing import Self # novermin -except ImportError: # pragma: no cover - try: - from typing_extensions import Self # Backport - except ImportError: - Self = object - # Copied from `playwright._impl._api_structures.SetCookieParam` class SetCookieParam(TypedDict, total=False): diff --git a/scrapling/core/custom_types.py b/scrapling/core/custom_types.py index 45ab4d5..d06ac29 100644 --- a/scrapling/core/custom_types.py +++ b/scrapling/core/custom_types.py @@ -35,9 +35,7 @@ class TextHandler(str): lst = super().__getitem__(key) return TextHandler(lst) - def split( - self, sep: str | None = None, maxsplit: SupportsIndex = -1 - ) -> Union[List, "TextHandlers"]: # pragma: no cover + def split(self, sep: str | None = None, maxsplit: SupportsIndex = -1) -> list[Any]: # pragma: no cover return TextHandlers([TextHandler(s) for s in super().split(sep, maxsplit)]) def strip(self, chars: str | None = None) -> Union[str, "TextHandler"]: # pragma: no cover @@ -61,7 +59,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: object, **kwargs: str) -> Union[str, "TextHandler"]: # pragma: no cover + def format(self, *args: object, **kwargs: object) -> Union[str, "TextHandler"]: # pragma: no cover return TextHandler(super().format(*args, **kwargs)) def format_map(self, mapping) -> Union[str, "TextHandler"]: # pragma: no cover @@ -291,7 +289,7 @@ class AttributesHandler(Mapping[str, _TextHandlerType]): __slots__ = ("_data",) - def __init__(self, mapping=None, **kwargs): + def __init__(self, mapping: Any = None, **kwargs: Any) -> None: mapping = ( {key: TextHandler(value) if isinstance(value, str) else value for key, value in mapping.items()} if mapping is not None @@ -324,8 +322,8 @@ class AttributesHandler(Mapping[str, _TextHandlerType]): yield AttributesHandler({key: value}) @property - def json_string(self): - """Convert current attributes to JSON string if the attributes are JSON serializable otherwise throws error""" + def json_string(self) -> bytes: + """Convert current attributes to JSON bytes if the attributes are JSON serializable otherwise throws error""" return dumps(dict(self._data)) def __getitem__(self, key: str) -> _TextHandlerType: diff --git a/scrapling/core/mixins.py b/scrapling/core/mixins.py index 0b66c07..c2e7420 100644 --- a/scrapling/core/mixins.py +++ b/scrapling/core/mixins.py @@ -1,7 +1,4 @@ -from scrapling.core._types import TYPE_CHECKING - -if TYPE_CHECKING: - from scrapling.parser import Selector +from scrapling.core._types import Any, Dict class SelectorsGeneration: @@ -11,7 +8,11 @@ class SelectorsGeneration: Inspiration: https://searchfox.org/mozilla-central/source/devtools/shared/inspector/css-logic.js#591 """ - def _general_selection(self: "Selector", selection: str = "css", full_path: bool = False) -> str: # type: ignore[name-defined] + # Note: This is a mixin class meant to be used with Selector. + # The methods access Selector attributes (._root, .parent, .attrib, .tag, etc.) + # through self, which will be a Selector instance at runtime. + + def _general_selection(self: Any, selection: str = "css", full_path: bool = False) -> str: """Generate a selector for the current element. :return: A string of the generated selector. """ @@ -36,7 +37,7 @@ class SelectorsGeneration: # if classes and css: # part += f".{'.'.join(classes)}" # else: - counter = {} + counter: Dict[str, int] = {} for child in target.parent.children: counter.setdefault(child.tag, 0) counter[child.tag] += 1 @@ -56,28 +57,28 @@ class SelectorsGeneration: return " > ".join(reversed(selectorPath)) if css else "//" + "/".join(reversed(selectorPath)) @property - def generate_css_selector(self: "Selector") -> str: # type: ignore[name-defined] + def generate_css_selector(self: Any) -> str: """Generate a CSS selector for the current element :return: A string of the generated selector. """ return self._general_selection() @property - def generate_full_css_selector(self: "Selector") -> str: # type: ignore[name-defined] + def generate_full_css_selector(self: Any) -> str: """Generate a complete CSS selector for the current element :return: A string of the generated selector. """ return self._general_selection(full_path=True) @property - def generate_xpath_selector(self: "Selector") -> str: # type: ignore[name-defined] + def generate_xpath_selector(self: Any) -> str: """Generate an XPath selector for the current element :return: A string of the generated selector. """ return self._general_selection("xpath") @property - def generate_full_xpath_selector(self: "Selector") -> str: # type: ignore[name-defined] + def generate_full_xpath_selector(self: Any) -> str: """Generate a complete XPath selector for the current element :return: A string of the generated selector. """ diff --git a/scrapling/core/shell.py b/scrapling/core/shell.py index 3a1511e..0bd6efc 100644 --- a/scrapling/core/shell.py +++ b/scrapling/core/shell.py @@ -30,6 +30,7 @@ from scrapling.core.custom_types import TextHandler from scrapling.engines.toolbelt.custom import Response from scrapling.core.utils._shell import _ParseHeaders, _CookieParser from scrapling.core._types import ( + Callable, Dict, Any, cast, @@ -82,7 +83,7 @@ class NoExitArgumentParser(ArgumentParser): # pragma: no cover class CurlParser: """Builds the argument parser for relevant curl flags from DevTools.""" - def __init__(self): + def __init__(self) -> None: from scrapling.fetchers import Fetcher as __Fetcher self.__fetcher = __Fetcher @@ -467,19 +468,21 @@ Type 'exit' or press Ctrl+D to exit. return result - def create_wrapper(self, func, get_signature=True, signature_name=None): + def create_wrapper( + self, func: Callable, get_signature: bool = True, signature_name: Optional[str] = None + ) -> Callable: """Create a wrapper that preserves function signature but updates page""" @wraps(func) - def wrapper(*args, **kwargs): + def wrapper(*args: Any, **kwargs: Any) -> Any: result = func(*args, **kwargs) return self.update_page(result) if get_signature: # Explicitly preserve and unpack signature for IPython introspection and autocompletion - wrapper.__signature__ = _unpack_signature(func, signature_name) # pyright: ignore + setattr(wrapper, "__signature__", _unpack_signature(func, signature_name)) else: - wrapper.__signature__ = signature(func) # pyright: ignore + setattr(wrapper, "__signature__", signature(func)) return wrapper @@ -601,7 +604,7 @@ class Convertor: " ", ): # Remove consecutive white-spaces - txt_content = re_sub(f"[{s}]+", s, txt_content) + txt_content = TextHandler(re_sub(f"[{s}]+", s, txt_content)) yield txt_content yield "" diff --git a/scrapling/core/storage.py b/scrapling/core/storage.py index 8e04599..d93a320 100644 --- a/scrapling/core/storage.py +++ b/scrapling/core/storage.py @@ -63,12 +63,11 @@ class StorageSystemMixin(ABC): # pragma: no cover 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): - # Hash functions have to take bytes - _identifier = _identifier.encode("utf-8") + # Hash functions have to take bytes + _identifier_bytes = _identifier.encode("utf-8") - hash_value = sha256(_identifier).hexdigest() - return f"{hash_value}_{len(_identifier)}" # Length to reduce collision chance + hash_value = sha256(_identifier_bytes).hexdigest() + return f"{hash_value}_{len(_identifier_bytes)}" # Length to reduce collision chance @lru_cache(1, typed=True) diff --git a/scrapling/engines/_browsers/_base.py b/scrapling/engines/_browsers/_base.py index 8bd442d..e379882 100644 --- a/scrapling/engines/_browsers/_base.py +++ b/scrapling/engines/_browsers/_base.py @@ -5,16 +5,12 @@ from contextlib import contextmanager, asynccontextmanager from playwright.sync_api._generated import Page from playwright.sync_api import ( Frame, - Browser, BrowserContext, - Playwright, Response as SyncPlaywrightResponse, ) from playwright.async_api._generated import Page as AsyncPage from playwright.async_api import ( Frame as AsyncFrame, - Browser as AsyncBrowser, - Playwright as AsyncPlaywright, Response as AsyncPlaywrightResponse, BrowserContext as AsyncBrowserContext, ) @@ -37,6 +33,7 @@ from scrapling.core._types import ( Optional, Callable, TYPE_CHECKING, + cast, overload, Tuple, ProxyType, @@ -61,12 +58,12 @@ class SyncSession: self.max_pages = max_pages self.page_pool = PagePool(max_pages) self._max_wait_for_page = 60 - self.playwright: Playwright | Any = None - self.context: BrowserContext | Any = None - self.browser: Optional[Browser] = None + self.playwright: Any = None + self.context: Any = None + self.browser: Any = None self._is_alive = False - def start(self): + def start(self) -> None: pass def close(self): # pragma: no cover @@ -215,13 +212,13 @@ class AsyncSession: self.max_pages = max_pages self.page_pool = PagePool(max_pages) self._max_wait_for_page = 60 - self.playwright: AsyncPlaywright | Any = None - self.context: AsyncBrowserContext | Any = None - self.browser: Optional[AsyncBrowser] = None + self.playwright: Any = None + self.context: Any = None + self.browser: Any = None self._is_alive = False self._lock = Lock() - async def start(self): + async def start(self) -> None: pass async def close(self): @@ -378,6 +375,8 @@ class AsyncSession: class BaseSessionMixin: + _config: "PlaywrightConfig | StealthConfig" + @overload def __validate_routine__(self, params: Dict, model: type[StealthConfig]) -> StealthConfig: ... @@ -404,7 +403,7 @@ class BaseSessionMixin: return config def __generate_options__(self, extra_flags: Tuple | None = None) -> None: - config: PlaywrightConfig | StealthConfig = self._config # type: ignore[has-type] + config: PlaywrightConfig | StealthConfig = self._config self._context_options.update( { "proxy": config.proxy, @@ -466,7 +465,7 @@ class DynamicSessionMixin(BaseSessionMixin): class StealthySessionMixin(BaseSessionMixin): def __validate__(self, **params): - self._config: StealthConfig = self.__validate_routine__(params, model=StealthConfig) + self._config = self.__validate_routine__(params, model=StealthConfig) self._context_options.update( { "is_mobile": False, @@ -482,22 +481,23 @@ class StealthySessionMixin(BaseSessionMixin): self.__generate_stealth_options() def __generate_stealth_options(self) -> None: - flags = tuple() - if not self._config.cdp_url: + config = cast(StealthConfig, self._config) + flags: Tuple[str, ...] = tuple() + if not config.cdp_url: flags = DEFAULT_FLAGS + DEFAULT_STEALTH_FLAGS - if self._config.block_webrtc: + if config.block_webrtc: flags += ( "--webrtc-ip-handling-policy=disable_non_proxied_udp", "--force-webrtc-ip-handling-policy", # Ensures the policy is enforced ) - if not self._config.allow_webgl: + if not config.allow_webgl: flags += ( "--disable-webgl", "--disable-webgl-image-chromium", "--disable-webgl2", ) - if self._config.hide_canvas: + if config.hide_canvas: flags += ("--fingerprinting-canvas-image-data-noise",) super(StealthySessionMixin, self).__generate_options__(flags) diff --git a/scrapling/engines/_browsers/_controllers.py b/scrapling/engines/_browsers/_controllers.py index 5f0b247..5d9e801 100644 --- a/scrapling/engines/_browsers/_controllers.py +++ b/scrapling/engines/_browsers/_controllers.py @@ -8,11 +8,10 @@ from playwright.sync_api import ( from playwright.async_api import ( async_playwright, Locator as AsyncLocator, - BrowserContext as AsyncBrowserContext, ) from scrapling.core.utils import log -from scrapling.core._types import Unpack +from scrapling.core._types import Optional, ProxyType, Unpack from scrapling.engines.toolbelt.proxy_rotation import is_proxy_error from scrapling.engines.toolbelt.convertor import Response, ResponseFactory from scrapling.engines.toolbelt.fingerprints import generate_convincing_referer @@ -134,6 +133,7 @@ class DynamicSession(SyncSession, DynamicSessionMixin): ) for attempt in range(self._config.retries): + proxy: Optional[ProxyType] = None if self._config.proxy_rotator and static_proxy is None: proxy = self._config.proxy_rotator.get_proxy() else: @@ -238,7 +238,7 @@ class AsyncDynamicSession(AsyncSession, DynamicSessionMixin): self.__validate__(**kwargs) super().__init__(max_pages=self._config.max_pages) - async def start(self): + async def start(self) -> None: """Create a browser for this instance and context.""" if not self.playwright: self.playwright = await async_playwright().start() @@ -246,16 +246,14 @@ class AsyncDynamicSession(AsyncSession, DynamicSessionMixin): if self._config.cdp_url: self.browser = await self.playwright.chromium.connect_over_cdp(endpoint_url=self._config.cdp_url) if not self._config.proxy_rotator and self.browser: - self.context: AsyncBrowserContext = await self.browser.new_context(**self._context_options) + self.context = await self.browser.new_context(**self._context_options) elif self._config.proxy_rotator: self.browser = await self.playwright.chromium.launch(**self._browser_options) else: persistent_options = ( self._browser_options | self._context_options | {"user_data_dir": self._user_data_dir} ) - self.context: AsyncBrowserContext = await self.playwright.chromium.launch_persistent_context( - **persistent_options - ) + self.context = await self.playwright.chromium.launch_persistent_context(**persistent_options) if self.context: self.context = await self._initialize_context(self._config, self.context) @@ -304,6 +302,7 @@ class AsyncDynamicSession(AsyncSession, DynamicSessionMixin): ) for attempt in range(self._config.retries): + proxy: Optional[ProxyType] = None if self._config.proxy_rotator and static_proxy is None: proxy = self._config.proxy_rotator.get_proxy() else: diff --git a/scrapling/engines/_browsers/_page.py b/scrapling/engines/_browsers/_page.py index 655d3d1..481016e 100644 --- a/scrapling/engines/_browsers/_page.py +++ b/scrapling/engines/_browsers/_page.py @@ -61,7 +61,9 @@ class PagePool: raise RuntimeError(f"Maximum page limit ({self.max_pages}) reached") if isinstance(page, AsyncPage): - page_info = cast(PageInfo[AsyncPage], PageInfo(page, "ready", "")) + page_info: PageInfo[SyncPage] | PageInfo[AsyncPage] = cast( + PageInfo[AsyncPage], PageInfo(page, "ready", "") + ) else: page_info = cast(PageInfo[SyncPage], PageInfo(page, "ready", "")) diff --git a/scrapling/engines/_browsers/_stealth.py b/scrapling/engines/_browsers/_stealth.py index 730cd7e..3c9ea58 100644 --- a/scrapling/engines/_browsers/_stealth.py +++ b/scrapling/engines/_browsers/_stealth.py @@ -13,7 +13,7 @@ from patchright.sync_api import sync_playwright from patchright.async_api import async_playwright from scrapling.core.utils import log -from scrapling.core._types import Any, Unpack +from scrapling.core._types import Any, Optional, ProxyType, Unpack from scrapling.engines.toolbelt.proxy_rotation import is_proxy_error from scrapling.engines.toolbelt.convertor import Response, ResponseFactory from scrapling.engines.toolbelt.fingerprints import generate_convincing_referer @@ -78,7 +78,7 @@ class StealthySession(SyncSession, StealthySessionMixin): self.__validate__(**kwargs) super().__init__() - def start(self): + def start(self) -> None: """Create a browser for this instance and context.""" if not self.playwright: self.playwright = sync_playwright().start() @@ -146,7 +146,7 @@ class StealthySession(SyncSession, StealthySessionMixin): # Waiting for the verify spinner to disappear, checking every 1s if it disappeared page.wait_for_timeout(500) - outer_box = {} + outer_box: Any = {} iframe = page.frame(url=__CF_PATTERN__) if iframe is not None: self._wait_for_page_stability(iframe, True, False) @@ -156,14 +156,14 @@ class StealthySession(SyncSession, StealthySessionMixin): # Double-checking that the iframe is loaded page.wait_for_timeout(500) - outer_box: Any = iframe.frame_element().bounding_box() + outer_box = iframe.frame_element().bounding_box() if not iframe or not outer_box: if "