diff --git a/scrapling/cli.py b/scrapling/cli.py index 7b5d703..dd012cd 100644 --- a/scrapling/cli.py +++ b/scrapling/cli.py @@ -15,7 +15,7 @@ __OUTPUT_FILE_HELP__ = "Output file path can be HTML content, Markdown of the HT __PACKAGE_DIR__ = Path(__file__).parent -def __Execute(cmd: List[str], help_line: str) -> None: +def __Execute(cmd: List[str], help_line: str) -> None: # pragma: no cover print(f"Installing {help_line}...") _ = check_output(cmd, shell=False) # nosec B603 # I meant to not use try except here @@ -28,7 +28,7 @@ def __ParseJSONData(json_string: Optional[str] = None) -> Optional[Dict[str, Any try: return json_loads(json_string) - except JSONDecodeError as e: + except JSONDecodeError as e: # pragma: no cover raise ValueError(f"Invalid JSON data '{json_string}': {e}") @@ -105,7 +105,7 @@ def __BuildRequest( type=bool, help="Force Scrapling to reinstall all Fetchers dependencies", ) -def install(force): +def install(force): # pragma: no cover if ( force or not __PACKAGE_DIR__.joinpath(".scrapling_dependencies_installed").exists() diff --git a/scrapling/core/_html_utils.py b/scrapling/core/_html_utils.py index c0cd45c..99776af 100644 --- a/scrapling/core/_html_utils.py +++ b/scrapling/core/_html_utils.py @@ -340,7 +340,7 @@ def _replace_entities( if 0x80 <= number <= 0x9F: return bytes((number,)).decode("cp1252") return chr(number) - except (ValueError, OverflowError): + except (ValueError, OverflowError): # pragma: no cover pass return "" if remove_illegal and groups.get("semicolon") else m.group(0) diff --git a/scrapling/core/_types.py b/scrapling/core/_types.py index a114e37..cd8c9c0 100644 --- a/scrapling/core/_types.py +++ b/scrapling/core/_types.py @@ -35,7 +35,7 @@ StrOrBytes = Union[str, bytes] try: # Python 3.11+ from typing import Self # novermin -except ImportError: +except ImportError: # pragma: no cover try: from typing_extensions import Self # Backport except ImportError: diff --git a/scrapling/core/custom_types.py b/scrapling/core/custom_types.py index d46fd0c..ef76880 100644 --- a/scrapling/core/custom_types.py +++ b/scrapling/core/custom_types.py @@ -31,11 +31,15 @@ class TextHandler(str): __slots__ = () - def __getitem__(self, key: SupportsIndex | slice) -> "TextHandler": + def __getitem__( + self, key: SupportsIndex | slice + ) -> "TextHandler": # pragma: no cover lst = super().__getitem__(key) return cast(_TextHandlerType, TextHandler(lst)) - def split(self, sep: str = None, maxsplit: SupportsIndex = -1) -> "TextHandlers": + def split( + self, sep: str = None, maxsplit: SupportsIndex = -1 + ) -> "TextHandlers": # pragma: no cover return TextHandlers( cast( List[_TextHandlerType], @@ -43,58 +47,70 @@ class TextHandler(str): ) ) - def strip(self, chars: str = None) -> Union[str, "TextHandler"]: + def strip(self, chars: str = None) -> Union[str, "TextHandler"]: # pragma: no cover return TextHandler(super().strip(chars)) - def lstrip(self, chars: str = None) -> Union[str, "TextHandler"]: + def lstrip( + self, chars: str = None + ) -> Union[str, "TextHandler"]: # pragma: no cover return TextHandler(super().lstrip(chars)) - def rstrip(self, chars: str = None) -> Union[str, "TextHandler"]: + def rstrip( + self, chars: str = None + ) -> Union[str, "TextHandler"]: # pragma: no cover return TextHandler(super().rstrip(chars)) - def capitalize(self) -> Union[str, "TextHandler"]: + def capitalize(self) -> Union[str, "TextHandler"]: # pragma: no cover return TextHandler(super().capitalize()) - def casefold(self) -> Union[str, "TextHandler"]: + def casefold(self) -> Union[str, "TextHandler"]: # pragma: no cover return TextHandler(super().casefold()) def center( self, width: SupportsIndex, fillchar: str = " " - ) -> Union[str, "TextHandler"]: + ) -> Union[str, "TextHandler"]: # pragma: no cover return TextHandler(super().center(width, fillchar)) - def expandtabs(self, tabsize: SupportsIndex = 8) -> Union[str, "TextHandler"]: + 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"]: + def format( + self, *args: str, **kwargs: str + ) -> Union[str, "TextHandler"]: # pragma: no cover return TextHandler(super().format(*args, **kwargs)) - def format_map(self, mapping) -> Union[str, "TextHandler"]: + def format_map(self, mapping) -> Union[str, "TextHandler"]: # pragma: no cover return TextHandler(super().format_map(mapping)) - def join(self, iterable: Iterable[str]) -> Union[str, "TextHandler"]: + def join( + self, iterable: Iterable[str] + ) -> Union[str, "TextHandler"]: # pragma: no cover return TextHandler(super().join(iterable)) def ljust( self, width: SupportsIndex, fillchar: str = " " - ) -> Union[str, "TextHandler"]: + ) -> Union[str, "TextHandler"]: # pragma: no cover return TextHandler(super().ljust(width, fillchar)) def rjust( self, width: SupportsIndex, fillchar: str = " " - ) -> Union[str, "TextHandler"]: + ) -> Union[str, "TextHandler"]: # pragma: no cover return TextHandler(super().rjust(width, fillchar)) - def swapcase(self) -> Union[str, "TextHandler"]: + def swapcase(self) -> Union[str, "TextHandler"]: # pragma: no cover return TextHandler(super().swapcase()) - def title(self) -> Union[str, "TextHandler"]: + def title(self) -> Union[str, "TextHandler"]: # pragma: no cover return TextHandler(super().title()) - def translate(self, table) -> Union[str, "TextHandler"]: + def translate(self, table) -> Union[str, "TextHandler"]: # pragma: no cover return TextHandler(super().translate(table)) - def zfill(self, width: SupportsIndex) -> Union[str, "TextHandler"]: + def zfill( + self, width: SupportsIndex + ) -> Union[str, "TextHandler"]: # pragma: no cover return TextHandler(super().zfill(width)) def replace( @@ -120,10 +136,10 @@ class TextHandler(str): return self.__class__(__CONSECUTIVE_SPACES_REGEX__.sub(" ", data).strip()) # For easy copy-paste from Scrapy/parsel code when needed :) - def get(self, default=None): + def get(self, default=None): # pragma: no cover return self - def get_all(self): + def get_all(self): # pragma: no cover return self extract = get_all @@ -234,11 +250,11 @@ class TextHandlers(List[TextHandler]): __slots__ = () @overload - def __getitem__(self, pos: SupportsIndex) -> TextHandler: + def __getitem__(self, pos: SupportsIndex) -> TextHandler: # pragma: no cover pass @overload - def __getitem__(self, pos: slice) -> "TextHandlers": + def __getitem__(self, pos: slice) -> "TextHandlers": # pragma: no cover pass def __getitem__( @@ -276,7 +292,7 @@ class TextHandlers(List[TextHandler]): replace_entities: bool = True, clean_match: bool = False, case_sensitive: bool = True, - ) -> TextHandler: + ) -> TextHandler: # pragma: no cover """Call the ``.re_first()`` method for each element in this list and return the first result or the default value otherwise. diff --git a/scrapling/core/shell.py b/scrapling/core/shell.py index 5020194..0125ee0 100644 --- a/scrapling/core/shell.py +++ b/scrapling/core/shell.py @@ -108,7 +108,7 @@ def _ParseHeaders( cookie_dict = { key: value for key, value in _CookieParser(header_value) } - except Exception as e: + except Exception as e: # pragma: no cover raise ValueError( f"Could not parse cookie string from header '{header_value}': {e}" ) @@ -121,7 +121,7 @@ def _ParseHeaders( # Suppress exit on error to handle parsing errors gracefully -class NoExitArgumentParser(ArgumentParser): +class NoExitArgumentParser(ArgumentParser): # pragma: no cover def error(self, message): log.error(f"Curl arguments parsing error: {message}") raise ValueError(f"Curl arguments parsing error: {message}") @@ -188,8 +188,6 @@ class CurlParser: self.parser: NoExitArgumentParser = _parser self._supported_methods = ("get", "post", "put", "delete") - # --- Helper Functions --- - # --- Main Parsing Logic --- def parse(self, curl_command: str) -> Optional[Request]: """Parses the curl command string into a structured context for Fetcher.""" @@ -200,7 +198,7 @@ class CurlParser: tokens = shlex_split( clean_command ) # Split the string using shell-like syntax - except ValueError as e: + except ValueError as e: # pragma: no cover log.error(f"Could not split command line: {e}") return None @@ -209,13 +207,13 @@ class CurlParser: if unknown: raise AttributeError(f"Unknown/Unsupported curl arguments: {unknown}") - except ValueError: + except ValueError: # pragma: no cover return None except AttributeError: raise - except Exception as e: + except Exception as e: # pragma: no cover log.error( f"An unexpected error occurred during curl arguments parsing: {e}" ) @@ -249,7 +247,7 @@ class CurlParser: # Update the cookie dict, potentially overwriting cookies with the same name from -H 'cookie:' cookies[key] = value log.debug(f"Parsed cookies from -b argument: {list(cookies.keys())}") - except Exception as e: + except Exception as e: # pragma: no cover log.error( f"Could not parse cookie string from -b '{parsed_args.cookie}': {e}" ) @@ -261,7 +259,7 @@ class CurlParser: # DevTools often uses --data-raw for JSON bodies # Precedence: --data-binary > --data-raw / -d > --data-urlencode - if parsed_args.data_binary is not None: + if parsed_args.data_binary is not None: # pragma: no cover try: data_payload = parsed_args.data_binary.encode("utf-8") log.debug("Using data from --data-binary as bytes.") @@ -277,7 +275,7 @@ class CurlParser: elif parsed_args.data is not None: data_payload = parsed_args.data - elif parsed_args.data_urlencode: + elif parsed_args.data_urlencode: # pragma: no cover # Combine and parse urlencoded data combined_data = "&".join(parsed_args.data_urlencode) try: @@ -299,7 +297,7 @@ class CurlParser: pass # Not JSON, keep it in data_payload # Handle `-G`: Move data to params if the method is GET - if method == "get" and data_payload: + if method == "get" and data_payload: # pragma: no cover if isinstance(data_payload, dict): # From --data-urlencode likely params.update(data_payload) elif isinstance(data_payload, str): @@ -369,7 +367,7 @@ class CurlParser: ) # Ensure request parsing was successful before proceeding - if request is None: + if request is None: # pragma: no cover log.error("Failed to parse curl command, cannot convert to fetcher.") return None @@ -385,22 +383,22 @@ class CurlParser: try: return getattr(Fetcher, method)(**request_args) - except Exception as e: + except Exception as e: # pragma: no cover log.error(f"Error calling Fetcher.{method}: {e}") return None - else: + else: # pragma: no cover log.error( f'Request method "{method}" isn\'t supported by Scrapling yet' ) return None - else: + else: # pragma: no cover log.error("Input must be a valid curl command string or a Request object.") return None -def show_page_in_browser(page: Selector): +def show_page_in_browser(page: Selector): # pragma: no cover if not page or not isinstance(page, Selector): log.error("Input must be of type `Selector`") return @@ -429,7 +427,7 @@ class CustomShell: if _known_logging_levels.get(log_level): self.log_level = _known_logging_levels[log_level] - else: + else: # pragma: no cover log.warning(f'Unknown log level "{log_level}", defaulting to "DEBUG"') self.log_level = DEBUG @@ -480,7 +478,7 @@ class CustomShell: Type 'exit' or press Ctrl+D to exit. """ - def update_page(self, result): + def update_page(self, result): # pragma: no cover """Update the current page and add to pages history""" self.page = result if isinstance(result, (Response, Selector)): @@ -540,11 +538,11 @@ Type 'exit' or press Ctrl+D to exit. "help": self.show_help, } - def show_help(self): + def show_help(self): # pragma: no cover """Show help information""" print(self.banner()) - def start(self): + def start(self): # pragma: no cover """Start the interactive shell""" # Get our namespace with application objects namespace = self.get_namespace() @@ -594,7 +592,7 @@ class Convertor: main_content_only: bool = False, ) -> Generator[str, None, None]: """Extract the content of a Selector""" - if not page or not isinstance(page, Selector): + if not page or not isinstance(page, Selector): # pragma: no cover raise TypeError("Input must be of type `Selector`") elif not extraction_type or extraction_type not in cls._extension_map.values(): raise ValueError(f"Unknown extraction type: {extraction_type}") @@ -627,7 +625,7 @@ class Convertor: cls, page: Selector, filename: str, css_selector: Optional[str] = None ) -> None: """Write a Selector's content to a file""" - if not page or not isinstance(page, Selector): + if not page or not isinstance(page, Selector): # pragma: no cover raise TypeError("Input must be of type `Selector`") elif not filename or not isinstance(filename, str) or not filename.strip(): raise ValueError("Filename must be provided") diff --git a/scrapling/core/storage.py b/scrapling/core/storage.py index 096b688..089a9ec 100644 --- a/scrapling/core/storage.py +++ b/scrapling/core/storage.py @@ -12,7 +12,7 @@ from scrapling.core.utils import _StorageTools, log from scrapling.core._types import Dict, Optional, Any -class StorageSystemMixin(ABC): +class StorageSystemMixin(ABC): # pragma: no cover # If you want to make your own storage system, you have to inherit from this def __init__(self, url: Optional[str] = None): """ diff --git a/scrapling/core/translator.py b/scrapling/core/translator.py index bdab7a5..e0a91bc 100644 --- a/scrapling/core/translator.py +++ b/scrapling/core/translator.py @@ -37,15 +37,15 @@ class XPathExpr(OriginalXPathExpr): def __str__(self) -> str: path = super().__str__() if self.textnode: - if path == "*": + if path == "*": # pragma: no cover path = "text()" - elif path.endswith("::*/*"): + elif path.endswith("::*/*"): # pragma: no cover path = path[:-3] + "text()" else: path += "/text()" if self.attribute is not None: - if path.endswith("::*/*"): + if path.endswith("::*/*"): # pragma: no cover path = path[:-2] path += f"/@{self.attribute}" @@ -59,7 +59,7 @@ class XPathExpr(OriginalXPathExpr): **kwargs: Any, ) -> Self: if not isinstance(other, XPathExpr): - raise ValueError( + raise ValueError( # pragma: no cover f"Expressions of type {__name__}.XPathExpr can ony join expressions" f" of the same type (or its descendants), got {type(other)}" ) @@ -71,10 +71,10 @@ class XPathExpr(OriginalXPathExpr): # e.g. cssselect.GenericTranslator, cssselect.HTMLTranslator class TranslatorProtocol(Protocol): - def xpath_element(self, selector: Element) -> OriginalXPathExpr: + def xpath_element(self, selector: Element) -> OriginalXPathExpr: # pragma: no cover pass - def css_to_xpath(self, css: str, prefix: str = ...) -> str: + def css_to_xpath(self, css: str, prefix: str = ...) -> str: # pragma: no cover pass @@ -98,7 +98,7 @@ class TranslatorMixin: if isinstance(pseudo_element, FunctionalPseudoElement): method_name = f"xpath_{pseudo_element.name.replace('-', '_')}_functional_pseudo_element" method = getattr(self, method_name, None) - if not method: + if not method: # pragma: no cover raise ExpressionError( f"The functional pseudo-element ::{pseudo_element.name}() is unknown" ) @@ -108,7 +108,7 @@ class TranslatorMixin: f"xpath_{pseudo_element.replace('-', '_')}_simple_pseudo_element" ) method = getattr(self, method_name, None) - if not method: + if not method: # pragma: no cover raise ExpressionError( f"The pseudo-element ::{pseudo_element} is unknown" ) @@ -120,7 +120,7 @@ class TranslatorMixin: xpath: OriginalXPathExpr, function: FunctionalPseudoElement ) -> XPathExpr: """Support selecting attribute values using ::attr() pseudo-element""" - if function.argument_types() not in (["STRING"], ["IDENT"]): + if function.argument_types() not in (["STRING"], ["IDENT"]): # pragma: no cover raise ExpressionError( f"Expected a single string or ident for ::attr(), got {function.arguments!r}" ) diff --git a/scrapling/engines/_browsers/_camoufox.py b/scrapling/engines/_browsers/_camoufox.py index b8e33bd..55f185d 100644 --- a/scrapling/engines/_browsers/_camoufox.py +++ b/scrapling/engines/_browsers/_camoufox.py @@ -229,22 +229,24 @@ class StealthySession: def __create__(self): """Create a browser for this instance and context.""" self.playwright = sync_playwright().start() - self.context = self.playwright.firefox.launch_persistent_context( - **self.launch_options + self.context = ( + self.playwright.firefox.launch_persistent_context( # pragma: no cover + **self.launch_options + ) ) - if self.cookies: + if self.cookies: # pragma: no cover self.context.add_cookies(self.cookies) - def __enter__(self): + def __enter__(self): # pragma: no cover self.__create__() return self def __exit__(self, exc_type, exc_val, exc_tb): self.close() - def close(self): + def close(self): # pragma: no cover """Close all resources""" - if self._closed: + if self._closed: # pragma: no cover return if self.context: @@ -257,7 +259,7 @@ class StealthySession: self._closed = True - def _get_or_create_page(self) -> PageInfo: + def _get_or_create_page(self) -> PageInfo: # pragma: no cover """Get an available page or create a new one""" # Try to get a ready page first page_info = self.page_pool.get_ready_page() @@ -319,7 +321,7 @@ class StealthySession: return None - def _solve_cloudflare(self, page: Page) -> None: + 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 @@ -371,7 +373,7 @@ class StealthySession: :param url: The Target url. :return: A `Response` object. """ - if self._closed: + if self._closed: # pragma: no cover raise RuntimeError("Context manager has been closed") final_response = None @@ -392,7 +394,7 @@ class StealthySession: page_info = self._get_or_create_page() page_info.mark_busy(url=url) - try: + 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) @@ -440,7 +442,7 @@ class StealthySession: return response - except Exception as e: + except Exception as e: # pragma: no cover page_info.mark_error() raise e @@ -567,7 +569,7 @@ class AsyncStealthySession(StealthySession): async def close(self): """Close all resources""" - if self._closed: + if self._closed: # pragma: no cover return if self.context: @@ -605,7 +607,7 @@ class AsyncStealthySession(StealthySession): max_wait = 30 start_time = time() - while time() - start_time < max_wait: + while time() - start_time < max_wait: # pragma: no cover page_info = self.page_pool.get_ready_page() if page_info: return page_info @@ -625,7 +627,7 @@ class AsyncStealthySession(StealthySession): return else: log.info(f'The turnstile version discovered is "{challenge_type}"') - if challenge_type == "non-interactive": + if challenge_type == "non-interactive": # pragma: no cover while "