diff --git a/scrapling/cli.py b/scrapling/cli.py index a914ffe..e4b3c7f 100644 --- a/scrapling/cli.py +++ b/scrapling/cli.py @@ -72,14 +72,10 @@ def __ParseExtractArguments( return parsed_headers, parsed_cookies, parsed_params, parsed_json -def __BuildRequest( - headers: List[str], cookies: str, params: str, json: Optional[str] = None, **kwargs -) -> Dict: +def __BuildRequest(headers: List[str], cookies: str, params: str, json: Optional[str] = None, **kwargs) -> Dict: """Build a request object using the specified arguments""" # Parse parameters - parsed_headers, parsed_cookies, parsed_params, parsed_json = ( - __ParseExtractArguments(headers, cookies, params, json) - ) + parsed_headers, parsed_cookies, parsed_params, parsed_json = __ParseExtractArguments(headers, cookies, params, json) # Build request arguments request_kwargs = { "headers": parsed_headers if parsed_headers else None, @@ -106,10 +102,7 @@ def __BuildRequest( help="Force Scrapling to reinstall all Fetchers dependencies", ) def install(force): # pragma: no cover - if ( - force - or not __PACKAGE_DIR__.joinpath(".scrapling_dependencies_installed").exists() - ): + if force or not __PACKAGE_DIR__.joinpath(".scrapling_dependencies_installed").exists(): __Execute( [python_executable, "-m", "playwright", "install", "chromium"], "Playwright browsers", @@ -158,9 +151,7 @@ def mcp(): "level", is_flag=False, default="debug", - type=Choice( - ["debug", "info", "warning", "error", "critical", "fatal"], case_sensitive=False - ), + type=Choice(["debug", "info", "warning", "error", "critical", "fatal"], case_sensitive=False), help="Log level (default: DEBUG)", ) def shell(code, level): @@ -178,9 +169,7 @@ def extract(): pass -@extract.command( - help=f"Perform a GET request and save the content to a file.\n\n{__OUTPUT_FILE_HELP__}" -) +@extract.command(help=f"Perform a GET request and save the content to a file.\n\n{__OUTPUT_FILE_HELP__}") @argument("url", required=True) @argument("output_file", required=True) @option( @@ -190,9 +179,7 @@ def extract(): help='HTTP headers in format "Key: Value" (can be used multiple times)', ) @option("--cookies", help='Cookies string in format "name1=value1; name2=value2"') -@option( - "--timeout", type=int, default=30, help="Request timeout in seconds (default: 30)" -) +@option("--timeout", type=int, default=30, help="Request timeout in seconds (default: 30)") @option("--proxy", help='Proxy URL in format "http://username:password@host:port"') @option( "--css-selector", @@ -267,9 +254,7 @@ def get( __Request_and_Save(Fetcher.get, url, output_file, css_selector, **kwargs) -@extract.command( - help=f"Perform a POST request and save the content to a file.\n\n{__OUTPUT_FILE_HELP__}" -) +@extract.command(help=f"Perform a POST request and save the content to a file.\n\n{__OUTPUT_FILE_HELP__}") @argument("url", required=True) @argument("output_file", required=True) @option( @@ -285,9 +270,7 @@ def get( help='HTTP headers in format "Key: Value" (can be used multiple times)', ) @option("--cookies", help='Cookies string in format "name1=value1; name2=value2"') -@option( - "--timeout", type=int, default=30, help="Request timeout in seconds (default: 30)" -) +@option("--timeout", type=int, default=30, help="Request timeout in seconds (default: 30)") @option("--proxy", help='Proxy URL in format "http://username:password@host:port"') @option( "--css-selector", @@ -367,9 +350,7 @@ def post( __Request_and_Save(Fetcher.post, url, output_file, css_selector, **kwargs) -@extract.command( - help=f"Perform a PUT request and save the content to a file.\n\n{__OUTPUT_FILE_HELP__}" -) +@extract.command(help=f"Perform a PUT request and save the content to a file.\n\n{__OUTPUT_FILE_HELP__}") @argument("url", required=True) @argument("output_file", required=True) @option("--data", "-d", help="Form data to include in the request body") @@ -381,9 +362,7 @@ def post( help='HTTP headers in format "Key: Value" (can be used multiple times)', ) @option("--cookies", help='Cookies string in format "name1=value1; name2=value2"') -@option( - "--timeout", type=int, default=30, help="Request timeout in seconds (default: 30)" -) +@option("--timeout", type=int, default=30, help="Request timeout in seconds (default: 30)") @option("--proxy", help='Proxy URL in format "http://username:password@host:port"') @option( "--css-selector", @@ -463,9 +442,7 @@ def put( __Request_and_Save(Fetcher.put, url, output_file, css_selector, **kwargs) -@extract.command( - help=f"Perform a DELETE request and save the content to a file.\n\n{__OUTPUT_FILE_HELP__}" -) +@extract.command(help=f"Perform a DELETE request and save the content to a file.\n\n{__OUTPUT_FILE_HELP__}") @argument("url", required=True) @argument("output_file", required=True) @option( @@ -475,9 +452,7 @@ def put( help='HTTP headers in format "Key: Value" (can be used multiple times)', ) @option("--cookies", help='Cookies string in format "name1=value1; name2=value2"') -@option( - "--timeout", type=int, default=30, help="Request timeout in seconds (default: 30)" -) +@option("--timeout", type=int, default=30, help="Request timeout in seconds (default: 30)") @option("--proxy", help='Proxy URL in format "http://username:password@host:port"') @option( "--css-selector", @@ -552,9 +527,7 @@ def delete( __Request_and_Save(Fetcher.delete, url, output_file, css_selector, **kwargs) -@extract.command( - help=f"Use DynamicFetcher to fetch content with browser automation.\n\n{__OUTPUT_FILE_HELP__}" -) +@extract.command(help=f"Use DynamicFetcher to fetch content with browser automation.\n\n{__OUTPUT_FILE_HELP__}") @argument("url", required=True) @argument("output_file", required=True) @option( @@ -591,9 +564,7 @@ def delete( ) @option("--wait-selector", help="CSS selector to wait for before proceeding") @option("--locale", default="en-US", help="Browser locale (default: en-US)") -@option( - "--stealth/--no-stealth", default=False, help="Enable stealth mode (default: False)" -) +@option("--stealth/--no-stealth", default=False, help="Enable stealth mode (default: False)") @option( "--hide-canvas/--show-canvas", default=False, @@ -675,9 +646,7 @@ def fetch( __Request_and_Save(DynamicFetcher.fetch, url, output_file, css_selector, **kwargs) -@extract.command( - help=f"Use StealthyFetcher to fetch content with advanced stealth features.\n\n{__OUTPUT_FILE_HELP__}" -) +@extract.command(help=f"Use StealthyFetcher to fetch content with advanced stealth features.\n\n{__OUTPUT_FILE_HELP__}") @argument("url", required=True) @argument("output_file", required=True) @option( diff --git a/scrapling/core/_html_utils.py b/scrapling/core/_html_utils.py index 99776af..6b09830 100644 --- a/scrapling/core/_html_utils.py +++ b/scrapling/core/_html_utils.py @@ -269,17 +269,13 @@ name2codepoint = { } -def to_unicode( - text: StrOrBytes, encoding: Optional[str] = None, errors: str = "strict" -) -> str: +def to_unicode(text: StrOrBytes, encoding: Optional[str] = None, errors: str = "strict") -> str: """Return the Unicode representation of a bytes object `text`. If `text` is already a Unicode object, return it as-is.""" if isinstance(text, str): return text if not isinstance(text, (bytes, str)): - raise TypeError( - f"to_unicode must receive bytes or str, got {type(text).__name__}" - ) + raise TypeError(f"to_unicode must receive bytes or str, got {type(text).__name__}") if encoding is None: encoding = "utf-8" return text.decode(encoding, errors) @@ -328,9 +324,7 @@ def _replace_entities( entity_name = groups["named"] if entity_name.lower() in keep: return m.group(0) - number = name2codepoint.get(entity_name) or name2codepoint.get( - entity_name.lower() - ) + number = name2codepoint.get(entity_name) or name2codepoint.get(entity_name.lower()) if number is not None: # Browsers typically # interpret numeric character references in the 80-9F range as representing the characters mapped diff --git a/scrapling/core/ai.py b/scrapling/core/ai.py index aa2d52b..7777c21 100644 --- a/scrapling/core/ai.py +++ b/scrapling/core/ai.py @@ -32,21 +32,13 @@ class ResponseModel(BaseModel): """Request's response information structure.""" status: int = Field(description="The status code returned by the website.") - content: list[str] = Field( - description="The content as Markdown/HTML or the text content of the page." - ) - url: str = Field( - description="The URL given by the user that resulted in this response." - ) + content: list[str] = Field(description="The content as Markdown/HTML or the text content of the page.") + url: str = Field(description="The URL given by the user that resulted in this response.") -def _ContentTranslator( - content: Generator[str, None, None], page: _ScraplingResponse -) -> ResponseModel: +def _ContentTranslator(content: Generator[str, None, None], page: _ScraplingResponse) -> ResponseModel: """Convert a content generator to a list of ResponseModel objects.""" - return ResponseModel( - status=page.status, content=[result for result in content], url=page.url - ) + return ResponseModel(status=page.status, content=[result for result in content], url=page.url) class ScraplingMCPServer: diff --git a/scrapling/core/custom_types.py b/scrapling/core/custom_types.py index ef76880..eb7fa34 100644 --- a/scrapling/core/custom_types.py +++ b/scrapling/core/custom_types.py @@ -31,15 +31,11 @@ class TextHandler(str): __slots__ = () - def __getitem__( - self, key: SupportsIndex | slice - ) -> "TextHandler": # pragma: no cover + 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": # pragma: no cover + def split(self, sep: str = None, maxsplit: SupportsIndex = -1) -> "TextHandlers": # pragma: no cover return TextHandlers( cast( List[_TextHandlerType], @@ -50,14 +46,10 @@ class TextHandler(str): 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"]: # pragma: no cover + 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"]: # pragma: no cover + def rstrip(self, chars: str = None) -> Union[str, "TextHandler"]: # pragma: no cover return TextHandler(super().rstrip(chars)) def capitalize(self) -> Union[str, "TextHandler"]: # pragma: no cover @@ -66,37 +58,25 @@ class TextHandler(str): def casefold(self) -> Union[str, "TextHandler"]: # pragma: no cover return TextHandler(super().casefold()) - def center( - self, width: SupportsIndex, fillchar: str = " " - ) -> Union[str, "TextHandler"]: # pragma: no cover + def center(self, width: SupportsIndex, fillchar: str = " ") -> Union[str, "TextHandler"]: # pragma: no cover return TextHandler(super().center(width, fillchar)) - def expandtabs( - self, tabsize: SupportsIndex = 8 - ) -> Union[str, "TextHandler"]: # pragma: no cover + 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: str, **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 return TextHandler(super().format_map(mapping)) - def join( - self, iterable: Iterable[str] - ) -> Union[str, "TextHandler"]: # pragma: no cover + 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"]: # pragma: no cover + def ljust(self, width: SupportsIndex, fillchar: str = " ") -> Union[str, "TextHandler"]: # pragma: no cover return TextHandler(super().ljust(width, fillchar)) - def rjust( - self, width: SupportsIndex, fillchar: str = " " - ) -> Union[str, "TextHandler"]: # pragma: no cover + def rjust(self, width: SupportsIndex, fillchar: str = " ") -> Union[str, "TextHandler"]: # pragma: no cover return TextHandler(super().rjust(width, fillchar)) def swapcase(self) -> Union[str, "TextHandler"]: # pragma: no cover @@ -108,14 +88,10 @@ class TextHandler(str): def translate(self, table) -> Union[str, "TextHandler"]: # pragma: no cover return TextHandler(super().translate(table)) - def zfill( - self, width: SupportsIndex - ) -> Union[str, "TextHandler"]: # pragma: no cover + def zfill(self, width: SupportsIndex) -> Union[str, "TextHandler"]: # pragma: no cover return TextHandler(super().zfill(width)) - def replace( - self, old: str, new: str, count: SupportsIndex = -1 - ) -> Union[str, "TextHandler"]: + def replace(self, old: str, new: str, count: SupportsIndex = -1) -> Union[str, "TextHandler"]: return TextHandler(super().replace(old, new, count)) def upper(self) -> Union[str, "TextHandler"]: @@ -203,11 +179,7 @@ class TextHandler(str): results = flatten(results) if not replace_entities: - return TextHandlers( - cast( - List[_TextHandlerType], [TextHandler(string) for string in results] - ) - ) + return TextHandlers(cast(List[_TextHandlerType], [TextHandler(string) for string in results])) return TextHandlers( cast( @@ -257,9 +229,7 @@ class TextHandlers(List[TextHandler]): def __getitem__(self, pos: slice) -> "TextHandlers": # pragma: no cover pass - def __getitem__( - self, pos: SupportsIndex | slice - ) -> Union[TextHandler, "TextHandlers"]: + def __getitem__(self, pos: SupportsIndex | slice) -> Union[TextHandler, "TextHandlers"]: lst = super().__getitem__(pos) if isinstance(pos, slice): return TextHandlers(cast(List[_TextHandlerType], lst)) @@ -280,9 +250,7 @@ class TextHandlers(List[TextHandler]): :param clean_match: if enabled, this will ignore all whitespaces and consecutive spaces while matching :param case_sensitive: if disabled, the function will set the regex to ignore the letters-case while compiling it """ - results = [ - n.re(regex, replace_entities, clean_match, case_sensitive) for n in self - ] + results = [n.re(regex, replace_entities, clean_match, case_sensitive) for n in self] return TextHandlers(flatten(results)) def re_first( @@ -330,34 +298,24 @@ class AttributesHandler(Mapping[str, _TextHandlerType]): def __init__(self, mapping=None, **kwargs): mapping = ( - { - key: TextHandler(value) if isinstance(value, str) else value - for key, value in mapping.items() - } + {key: TextHandler(value) if isinstance(value, str) else value for key, value in mapping.items()} if mapping is not None else {} ) if kwargs: mapping.update( - { - key: TextHandler(value) if isinstance(value, str) else value - for key, value in kwargs.items() - } + {key: TextHandler(value) if isinstance(value, str) else value for key, value in kwargs.items()} ) # Fastest read-only mapping type self._data = MappingProxyType(mapping) - def get( - self, key: str, default: Optional[str] = None - ) -> Optional[_TextHandlerType]: + def get(self, key: str, default: Optional[str] = None) -> Optional[_TextHandlerType]: """Acts like the standard dictionary `.get()` method""" return self._data.get(key, default) - def search_values( - self, keyword: str, partial: bool = False - ) -> Generator["AttributesHandler", None, None]: + def search_values(self, keyword: str, partial: bool = False) -> Generator["AttributesHandler", None, None]: """Search current attributes by values and return a dictionary of each matching item :param keyword: The keyword to search for in the attribute values :param partial: If True, the function will search if keyword in each value instead of perfect match diff --git a/scrapling/core/mixins.py b/scrapling/core/mixins.py index afad094..4087020 100644 --- a/scrapling/core/mixins.py +++ b/scrapling/core/mixins.py @@ -5,9 +5,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, selection: str = "css", full_path: bool = False) -> str: """Generate a selector for the current element. :return: A string of the generated selector. """ @@ -18,18 +16,10 @@ class SelectorsGeneration: if target.parent: if target.attrib.get("id"): # id is enough - part = ( - f"#{target.attrib['id']}" - if css - else f"[@id='{target.attrib['id']}']" - ) + part = f"#{target.attrib['id']}" if css else f"[@id='{target.attrib['id']}']" selectorPath.append(part) if not full_path: - return ( - " > ".join(reversed(selectorPath)) - if css - else "//*" + "/".join(reversed(selectorPath)) - ) + return " > ".join(reversed(selectorPath)) if css else "//*" + "/".join(reversed(selectorPath)) else: part = f"{target.tag}" # We won't use classes anymore because I some websites share exact classes between elements @@ -45,28 +35,16 @@ class SelectorsGeneration: break if counter[target.tag] > 1: - part += ( - f":nth-of-type({counter[target.tag]})" - if css - else f"[{counter[target.tag]}]" - ) + part += f":nth-of-type({counter[target.tag]})" if css else f"[{counter[target.tag]}]" selectorPath.append(part) target = target.parent if target is None or target.tag == "html": - return ( - " > ".join(reversed(selectorPath)) - if css - else "//" + "/".join(reversed(selectorPath)) - ) + return " > ".join(reversed(selectorPath)) if css else "//" + "/".join(reversed(selectorPath)) else: break - return ( - " > ".join(reversed(selectorPath)) - if css - else "//" + "/".join(reversed(selectorPath)) - ) + return " > ".join(reversed(selectorPath)) if css else "//" + "/".join(reversed(selectorPath)) @property def generate_css_selector(self) -> str: diff --git a/scrapling/core/shell.py b/scrapling/core/shell.py index 24e504c..8459a60 100644 --- a/scrapling/core/shell.py +++ b/scrapling/core/shell.py @@ -79,9 +79,7 @@ def _CookieParser(cookie_string): yield key, morsel.value -def _ParseHeaders( - header_lines: List[str], parse_cookies: bool = True -) -> Tuple[Dict[str, str], Dict[str, str]]: +def _ParseHeaders(header_lines: List[str], parse_cookies: bool = True) -> Tuple[Dict[str, str], Dict[str, str]]: """Parses headers into separate header and cookie dictionaries.""" header_dict = dict() cookie_dict = dict() @@ -93,9 +91,7 @@ def _ParseHeaders( header_value = "" header_dict[header_key] = header_value else: - raise ValueError( - f"Could not parse header without colon: '{header_line}'." - ) + raise ValueError(f"Could not parse header without colon: '{header_line}'.") else: header_key, header_value = header_line.split(":", 1) header_key = header_key.strip() @@ -104,13 +100,9 @@ def _ParseHeaders( if parse_cookies: if header_key.lower() == "cookie": try: - cookie_dict = { - key: value for key, value in _CookieParser(header_value) - } + cookie_dict = {key: value for key, value in _CookieParser(header_value)} except Exception as e: # pragma: no cover - raise ValueError( - f"Could not parse cookie string from header '{header_value}': {e}" - ) + raise ValueError(f"Could not parse cookie string from header '{header_value}': {e}") else: header_dict[header_key] = header_value else: @@ -129,9 +121,7 @@ class NoExitArgumentParser(ArgumentParser): # pragma: no cover if message: log.error(f"Scrapling shell exited with status {status}: {message}") self._print_message(message, stderr) - raise ValueError( - f"Scrapling shell exited with status {status}: {message or 'Unknown reason'}" - ) + raise ValueError(f"Scrapling shell exited with status {status}: {message or 'Unknown reason'}") class CurlParser: @@ -152,15 +142,11 @@ class CurlParser: # Data arguments (prioritizing types common from DevTools) _parser.add_argument("-d", "--data", default=None) - _parser.add_argument( - "--data-raw", default=None - ) # Often used by browsers for JSON body + _parser.add_argument("--data-raw", default=None) # Often used by browsers for JSON body _parser.add_argument("--data-binary", default=None) # Keep urlencode for completeness, though less common from browser copy/paste _parser.add_argument("--data-urlencode", action="append", default=[]) - _parser.add_argument( - "-G", "--get", action="store_true" - ) # Use GET and put data in URL + _parser.add_argument("-G", "--get", action="store_true") # Use GET and put data in URL _parser.add_argument( "-b", @@ -175,9 +161,7 @@ class CurlParser: # Connection/Security _parser.add_argument("-k", "--insecure", action="store_true") - _parser.add_argument( - "--compressed", action="store_true" - ) # Very common from browsers + _parser.add_argument("--compressed", action="store_true") # Very common from browsers # Other flags often included but may not map directly to request args _parser.add_argument("-i", "--include", action="store_true") @@ -194,9 +178,7 @@ class CurlParser: clean_command = curl_command.strip().lstrip("curl").strip().replace("\\\n", " ") try: - tokens = shlex_split( - clean_command - ) # Split the string using shell-like syntax + tokens = shlex_split(clean_command) # Split the string using shell-like syntax except ValueError as e: # pragma: no cover log.error(f"Could not split command line: {e}") return None @@ -213,9 +195,7 @@ class CurlParser: raise except Exception as e: # pragma: no cover - log.error( - f"An unexpected error occurred during curl arguments parsing: {e}" - ) + log.error(f"An unexpected error occurred during curl arguments parsing: {e}") return None # --- Determine Method --- @@ -247,9 +227,7 @@ class CurlParser: cookies[key] = value log.debug(f"Parsed cookies from -b argument: {list(cookies.keys())}") except Exception as e: # pragma: no cover - log.error( - f"Could not parse cookie string from -b '{parsed_args.cookie}': {e}" - ) + log.error(f"Could not parse cookie string from -b '{parsed_args.cookie}': {e}") # --- Process Data Payload --- params = dict() @@ -280,9 +258,7 @@ class CurlParser: try: data_payload = dict(parse_qsl(combined_data, keep_blank_values=True)) except Exception as e: - log.warning( - f"Could not parse urlencoded data '{combined_data}': {e}. Treating as raw string." - ) + log.warning(f"Could not parse urlencoded data '{combined_data}': {e}. Treating as raw string.") data_payload = combined_data # Check if raw data looks like JSON, prefer 'json' param if so @@ -303,9 +279,7 @@ class CurlParser: try: params.update(dict(parse_qsl(data_payload, keep_blank_values=True))) except ValueError: - log.warning( - f"Could not parse data '{data_payload}' into GET parameters for -G." - ) + log.warning(f"Could not parse data '{data_payload}' into GET parameters for -G.") if params: data_payload = None # Clear data as it's moved to params @@ -314,21 +288,13 @@ class CurlParser: # --- Process Proxy --- proxies: Optional[Dict[str, str]] = None if parsed_args.proxy: - proxy_url = ( - f"http://{parsed_args.proxy}" - if "://" not in parsed_args.proxy - else parsed_args.proxy - ) + proxy_url = f"http://{parsed_args.proxy}" if "://" not in parsed_args.proxy else parsed_args.proxy if parsed_args.proxy_user: user_pass = parsed_args.proxy_user parts = urlparse(proxy_url) netloc_parts = parts.netloc.split("@") - netloc = ( - f"{user_pass}@{netloc_parts[-1]}" - if len(netloc_parts) > 1 - else f"{user_pass}@{parts.netloc}" - ) + netloc = f"{user_pass}@{netloc_parts[-1]}" if len(netloc_parts) > 1 else f"{user_pass}@{parts.netloc}" proxy_url = urlunparse( ( parts.scheme, @@ -359,11 +325,7 @@ class CurlParser: def convert2fetcher(self, curl_command: Request | str) -> Optional[Response]: if isinstance(curl_command, (Request, str)): - request = ( - self.parse(curl_command) - if isinstance(curl_command, str) - else curl_command - ) + request = self.parse(curl_command) if isinstance(curl_command, str) else curl_command # Ensure request parsing was successful before proceeding if request is None: # pragma: no cover @@ -386,9 +348,7 @@ class CurlParser: log.error(f"Error calling Fetcher.{method}: {e}") return None else: # pragma: no cover - log.error( - f'Request method "{method}" isn\'t supported by Scrapling yet' - ) + log.error(f'Request method "{method}" isn\'t supported by Scrapling yet') return None else: # pragma: no cover @@ -621,18 +581,14 @@ class Convertor: yield "" @classmethod - def write_content_to_file( - cls, page: Selector, filename: str, css_selector: Optional[str] = None - ) -> None: + def write_content_to_file(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): # 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") elif not filename.endswith((".md", ".html", ".txt")): - raise ValueError( - "Unknown file type: filename must end with '.md', '.html', or '.txt'" - ) + raise ValueError("Unknown file type: filename must end with '.md', '.html', or '.txt'") else: with open(filename, "w", encoding="utf-8") as f: extension = filename.split(".")[-1] diff --git a/scrapling/core/storage.py b/scrapling/core/storage.py index 089a9ec..03c05a2 100644 --- a/scrapling/core/storage.py +++ b/scrapling/core/storage.py @@ -27,11 +27,7 @@ class StorageSystemMixin(ABC): # pragma: no cover try: extracted = tld(self.url) - return ( - extracted.top_domain_under_public_suffix - or extracted.domain - or default_value - ) + return extracted.top_domain_under_public_suffix or extracted.domain or default_value except AttributeError: return default_value @@ -90,9 +86,7 @@ class SQLiteStorageSystem(StorageSystemMixin): self.connection.execute("PRAGMA journal_mode=WAL") self.cursor = self.connection.cursor() self._setup_database() - log.debug( - f'Storage system loaded with arguments (storage_file="{storage_file}", url="{url}")' - ) + log.debug(f'Storage system loaded with arguments (storage_file="{storage_file}", url="{url}")') def _setup_database(self) -> None: self.cursor.execute(""" diff --git a/scrapling/core/translator.py b/scrapling/core/translator.py index e0a91bc..88dfbff 100644 --- a/scrapling/core/translator.py +++ b/scrapling/core/translator.py @@ -89,9 +89,7 @@ class TranslatorMixin: xpath = super().xpath_element(selector) # type: ignore[safe-super] return XPathExpr.from_xpath(xpath) - def xpath_pseudo_element( - self, xpath: OriginalXPathExpr, pseudo_element: PseudoElement - ) -> OriginalXPathExpr: + def xpath_pseudo_element(self, xpath: OriginalXPathExpr, pseudo_element: PseudoElement) -> OriginalXPathExpr: """ Dispatch method that transforms XPath to support the pseudo-element. """ @@ -99,31 +97,21 @@ class TranslatorMixin: method_name = f"xpath_{pseudo_element.name.replace('-', '_')}_functional_pseudo_element" method = getattr(self, method_name, None) if not method: # pragma: no cover - raise ExpressionError( - f"The functional pseudo-element ::{pseudo_element.name}() is unknown" - ) + raise ExpressionError(f"The functional pseudo-element ::{pseudo_element.name}() is unknown") xpath = method(xpath, pseudo_element) else: - method_name = ( - f"xpath_{pseudo_element.replace('-', '_')}_simple_pseudo_element" - ) + method_name = f"xpath_{pseudo_element.replace('-', '_')}_simple_pseudo_element" method = getattr(self, method_name, None) if not method: # pragma: no cover - raise ExpressionError( - f"The pseudo-element ::{pseudo_element} is unknown" - ) + raise ExpressionError(f"The pseudo-element ::{pseudo_element} is unknown") xpath = method(xpath) return xpath @staticmethod - def xpath_attr_functional_pseudo_element( - xpath: OriginalXPathExpr, function: FunctionalPseudoElement - ) -> XPathExpr: + def xpath_attr_functional_pseudo_element(xpath: OriginalXPathExpr, function: FunctionalPseudoElement) -> XPathExpr: """Support selecting attribute values using ::attr() pseudo-element""" 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}" - ) + raise ExpressionError(f"Expected a single string or ident for ::attr(), got {function.arguments!r}") return XPathExpr.from_xpath(xpath, attribute=function.arguments[0].value) @staticmethod diff --git a/scrapling/core/utils.py b/scrapling/core/utils.py index 5607f8d..2f57cfa 100644 --- a/scrapling/core/utils.py +++ b/scrapling/core/utils.py @@ -24,9 +24,7 @@ def setup_logger(): logger = logging.getLogger("scrapling") logger.setLevel(logging.INFO) - formatter = logging.Formatter( - fmt="[%(asctime)s] %(levelname)s: %(message)s", datefmt="%Y-%m-%d %H:%M:%S" - ) + formatter = logging.Formatter(fmt="[%(asctime)s] %(levelname)s: %(message)s", datefmt="%Y-%m-%d %H:%M:%S") console_handler = logging.StreamHandler() console_handler.setFormatter(formatter) @@ -61,11 +59,7 @@ class _StorageTools: def __clean_attributes(element: html.HtmlElement, forbidden: tuple = ()) -> Dict: if not element.attrib: return {} - return { - k: v.strip() - for k, v in element.attrib.items() - if v and v.strip() and k not in forbidden - } + return {k: v.strip() for k, v in element.attrib.items() if v and v.strip() and k not in forbidden} @classmethod def element_to_dict(cls, element: html.HtmlElement) -> Dict: @@ -85,17 +79,11 @@ class _StorageTools: } ) - siblings = [ - child.tag for child in parent.iterchildren() if child != element - ] + siblings = [child.tag for child in parent.iterchildren() if child != element] if siblings: result.update({"siblings": tuple(siblings)}) - children = [ - child.tag - for child in element.iterchildren() - if not isinstance(child, html_forbidden) - ] + children = [child.tag for child in element.iterchildren() if not isinstance(child, html_forbidden)] if children: result.update({"children": tuple(children)}) @@ -104,11 +92,7 @@ class _StorageTools: @classmethod def _get_element_path(cls, element: html.HtmlElement): parent = element.getparent() - return tuple( - (element.tag,) - if parent is None - else (cls._get_element_path(parent) + (element.tag,)) - ) + return tuple((element.tag,) if parent is None else (cls._get_element_path(parent) + (element.tag,))) @lru_cache(128, typed=True) diff --git a/scrapling/engines/_browsers/_base.py b/scrapling/engines/_browsers/_base.py index 23a77d6..6412698 100644 --- a/scrapling/engines/_browsers/_base.py +++ b/scrapling/engines/_browsers/_base.py @@ -80,9 +80,7 @@ class SyncSession: return self.page_pool.add_page(page) @staticmethod - def _get_with_precedence( - request_value: Any, session_value: Any, sentinel_value: object - ) -> Any: + def _get_with_precedence(request_value: Any, session_value: Any, sentinel_value: object) -> Any: """Get value with request-level priority over session-level""" return request_value if request_value is not sentinel_value else session_value @@ -169,11 +167,7 @@ class DynamicSessionMixin: self.wait_selector_state = config.wait_selector_state self.selector_config = config.selector_config self.page_action = config.page_action - self._headers_keys = ( - set(map(str.lower, self.extra_headers.keys())) - if self.extra_headers - else set() - ) + self._headers_keys = set(map(str.lower, self.extra_headers.keys())) if self.extra_headers else set() self.__initiate_browser_options__() def __initiate_browser_options__(self): @@ -184,9 +178,7 @@ class DynamicSessionMixin: self.headless, self.proxy, self.locale, - tuple(self.extra_headers.items()) - if self.extra_headers - else tuple(), + tuple(self.extra_headers.items()) if self.extra_headers else tuple(), self.useragent, self.real_chrome, self.stealth, @@ -194,9 +186,7 @@ class DynamicSessionMixin: self.disable_webgl, ) ) - self.launch_options["extra_http_headers"] = dict( - self.launch_options["extra_http_headers"] - ) + self.launch_options["extra_http_headers"] = dict(self.launch_options["extra_http_headers"]) self.launch_options["proxy"] = dict(self.launch_options["proxy"]) or None self.context_options = dict() else: @@ -206,16 +196,12 @@ class DynamicSessionMixin: _context_kwargs( self.proxy, self.locale, - tuple(self.extra_headers.items()) - if self.extra_headers - else tuple(), + tuple(self.extra_headers.items()) if self.extra_headers else tuple(), self.useragent, self.stealth, ) ) - self.context_options["extra_http_headers"] = dict( - self.context_options["extra_http_headers"] - ) + self.context_options["extra_http_headers"] = dict(self.context_options["extra_http_headers"]) self.context_options["proxy"] = dict(self.context_options["proxy"]) or None @@ -249,11 +235,7 @@ class StealthySessionMixin: self.selector_config = config.selector_config self.additional_args = config.additional_args self.page_action = config.page_action - self._headers_keys = ( - set(map(str.lower, self.extra_headers.keys())) - if self.extra_headers - else set() - ) + self._headers_keys = set(map(str.lower, self.extra_headers.keys())) if self.extra_headers else set() self.__initiate_browser_options__() def __initiate_browser_options__(self): diff --git a/scrapling/engines/_browsers/_page.py b/scrapling/engines/_browsers/_page.py index 61dd51b..ffa4b22 100644 --- a/scrapling/engines/_browsers/_page.py +++ b/scrapling/engines/_browsers/_page.py @@ -6,9 +6,7 @@ from playwright.async_api import Page as AsyncPage from scrapling.core._types import Optional, List, Literal -PageState = Literal[ - "finished", "ready", "busy", "error" -] # States that a page can be in +PageState = Literal["finished", "ready", "busy", "error"] # States that a page can be in @dataclass diff --git a/scrapling/engines/_browsers/_validators.py b/scrapling/engines/_browsers/_validators.py index 6363df7..c34d6fe 100644 --- a/scrapling/engines/_browsers/_validators.py +++ b/scrapling/engines/_browsers/_validators.py @@ -25,9 +25,7 @@ class PlaywrightConfig(Struct, kw_only=True, frozen=False): stealth: bool = False wait: int | float = 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]] = 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 @@ -46,10 +44,8 @@ class PlaywrightConfig(Struct, kw_only=True, frozen=False): raise ValueError("max_pages must be between 1 and 50") if self.timeout < 0: raise ValueError("timeout must be >= 0") - if self.page_action is not None and not callable(self.page_action): - raise TypeError( - f"page_action must be callable, got {type(self.page_action).__name__}" - ) + if self.page_action and not callable(self.page_action): + raise TypeError(f"page_action must be callable, got {type(self.page_action).__name__}") if self.proxy: self.proxy = construct_proxy_dict(self.proxy, as_tuple=True) if self.cdp_url: @@ -108,9 +104,7 @@ class CamoufoxConfig(Struct, kw_only=True, frozen=False): cookies: Optional[List[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]] = None # The default value for proxy in Playwright's source is `None` os_randomize: bool = False disable_ads: bool = False geoip: bool = False @@ -123,10 +117,8 @@ class CamoufoxConfig(Struct, kw_only=True, frozen=False): raise ValueError("max_pages must be between 1 and 50") if self.timeout < 0: raise ValueError("timeout must be >= 0") - if self.page_action is not None and not callable(self.page_action): - raise TypeError( - f"page_action must be callable, got {type(self.page_action).__name__}" - ) + if self.page_action and not callable(self.page_action): + raise TypeError(f"page_action must be callable, got {type(self.page_action).__name__}") if self.proxy: self.proxy = construct_proxy_dict(self.proxy, as_tuple=True) diff --git a/scrapling/engines/static.py b/scrapling/engines/static.py index 59f68fa..e028feb 100644 --- a/scrapling/engines/static.py +++ b/scrapling/engines/static.py @@ -108,13 +108,9 @@ class FetcherSession: headers = self.get_with_precedence(kwargs, "headers", self.default_headers) stealth = self.get_with_precedence(kwargs, "stealth", self.stealth) - impersonate = self.get_with_precedence( - kwargs, "impersonate", self.default_impersonate - ) + impersonate = self.get_with_precedence(kwargs, "impersonate", self.default_impersonate) - if self.get_with_precedence( - kwargs, "http3", self.default_http3 - ): # pragma: no cover + if self.get_with_precedence(kwargs, "http3", self.default_http3): # pragma: no cover request_args["http_version"] = CurlHttpVersion.V3ONLY if impersonate: log.warning( @@ -126,25 +122,13 @@ class FetcherSession: "url": url, # Curl automatically generates the suitable browser headers when you use `impersonate` "headers": self._headers_job(url, headers, stealth, bool(impersonate)), - "proxies": self.get_with_precedence( - kwargs, "proxies", self.default_proxies - ), + "proxies": self.get_with_precedence(kwargs, "proxies", self.default_proxies), "proxy": self.get_with_precedence(kwargs, "proxy", self.default_proxy), - "proxy_auth": self.get_with_precedence( - kwargs, "proxy_auth", self.default_proxy_auth - ), - "timeout": self.get_with_precedence( - kwargs, "timeout", self.default_timeout - ), - "allow_redirects": self.get_with_precedence( - kwargs, "allow_redirects", self.default_follow_redirects - ), - "max_redirects": self.get_with_precedence( - kwargs, "max_redirects", self.default_max_redirects - ), - "verify": self.get_with_precedence( - kwargs, "verify", self.default_verify - ), + "proxy_auth": self.get_with_precedence(kwargs, "proxy_auth", self.default_proxy_auth), + "timeout": self.get_with_precedence(kwargs, "timeout", self.default_timeout), + "allow_redirects": self.get_with_precedence(kwargs, "allow_redirects", self.default_follow_redirects), + "max_redirects": self.get_with_precedence(kwargs, "max_redirects", self.default_max_redirects), + "verify": self.get_with_precedence(kwargs, "verify", self.default_verify), "cert": self.get_with_precedence(kwargs, "cert", self.default_cert), "impersonate": impersonate, **{ @@ -192,18 +176,12 @@ class FetcherSession: extra_headers = generate_headers(browser_mode=False) # Don't overwrite user-supplied headers - extra_headers = { - key: value - for key, value in extra_headers.items() - if key.lower() not in headers_keys - } + extra_headers = {key: value for key, value in extra_headers.items() if key.lower() not in headers_keys} headers.update(extra_headers) elif "user-agent" not in headers_keys and not impersonate_enabled: headers["User-Agent"] = __default_useragent__ - log.debug( - f"Can't find useragent in headers so '{headers['User-Agent']}' was used." - ) + log.debug(f"Can't find useragent in headers so '{headers['User-Agent']}' was used.") return headers @@ -215,9 +193,7 @@ class FetcherSession: "Create a new FetcherSession instance for a new independent session, " "or use the current instance sequentially after the previous context has exited." ) - if ( - self._async_curl_session - ): # Prevent mixing if async is active from this instance + if self._async_curl_session: # Prevent mixing if async is active from this instance raise RuntimeError( "This FetcherSession instance has an active asynchronous session. " "Cannot enter a synchronous context simultaneously with the same manager instance." @@ -275,9 +251,7 @@ class FetcherSession: :return: A `Response` object for synchronous requests or an awaitable for asynchronous. """ session = self._curl_session - if session is True and not any( - (self.__enter__, self.__exit__, self.__aenter__, self.__aexit__) - ): + if session is True and not any((self.__enter__, self.__exit__, self.__aenter__, self.__aexit__)): # For usage inside FetcherClient # It turns out `curl_cffi` caches impersonation state, so if you turned it off, then on then off, it won't be off on the last time. session = CurlSession() @@ -290,9 +264,7 @@ class FetcherSession: return ResponseFactory.from_http_request(response, selector_config) except CurlError as e: # pragma: no cover if attempt < max_retries - 1: - log.error( - f"Attempt {attempt + 1} failed: {e}. Retrying in {retry_delay} seconds..." - ) + log.error(f"Attempt {attempt + 1} failed: {e}. Retrying in {retry_delay} seconds...") time_sleep(retry_delay) else: log.error(f"Failed after {max_retries} attempts: {e}") @@ -320,9 +292,7 @@ class FetcherSession: :return: A `Response` object for synchronous requests or an awaitable for asynchronous. """ session = self._async_curl_session - if session is True and not any( - (self.__enter__, self.__exit__, self.__aenter__, self.__aexit__) - ): + if session is True and not any((self.__enter__, self.__exit__, self.__aenter__, self.__aexit__)): # For usage inside the ` AsyncFetcherClient ` class, and that's for several reasons # 1. It turns out `curl_cffi` caches impersonation state, so if you turned it off, then on then off, it won't be off on the last time. # 2. `curl_cffi` doesn't support making async requests without sessions @@ -337,9 +307,7 @@ class FetcherSession: return ResponseFactory.from_http_request(response, selector_config) except CurlError as e: # pragma: no cover if attempt < max_retries - 1: - log.error( - f"Attempt {attempt + 1} failed: {e}. Retrying in {retry_delay} seconds..." - ) + log.error(f"Attempt {attempt + 1} failed: {e}. Retrying in {retry_delay} seconds...") await asyncio_sleep(retry_delay) else: log.error(f"Failed after {max_retries} attempts: {e}") @@ -372,19 +340,13 @@ class FetcherSession: selector_config = kwargs.pop("selector_config", {}) or self.selector_config max_retries = self.get_with_precedence(kwargs, "retries", self.default_retries) - retry_delay = self.get_with_precedence( - kwargs, "retry_delay", self.default_retry_delay - ) + retry_delay = self.get_with_precedence(kwargs, "retry_delay", self.default_retry_delay) request_args = self._merge_request_args(stealth=stealth, **kwargs) if self._curl_session: - return self.__make_request( - method, request_args, max_retries, retry_delay, selector_config - ) + return self.__make_request(method, request_args, max_retries, retry_delay, selector_config) elif self._async_curl_session: # The returned value is a Coroutine - return self.__make_async_request( - method, request_args, max_retries, retry_delay, selector_config - ) + return self.__make_async_request(method, request_args, max_retries, retry_delay, selector_config) raise RuntimeError("No active session available.") @@ -455,9 +417,7 @@ class FetcherSession: "http3": http3, **kwargs, } - return self.__prepare_and_dispatch( - "GET", stealth=stealthy_headers, **request_args - ) + return self.__prepare_and_dispatch("GET", stealth=stealthy_headers, **request_args) def post( self, @@ -532,9 +492,7 @@ class FetcherSession: "http3": http3, **kwargs, } - return self.__prepare_and_dispatch( - "POST", stealth=stealthy_headers, **request_args - ) + return self.__prepare_and_dispatch("POST", stealth=stealthy_headers, **request_args) def put( self, @@ -609,9 +567,7 @@ class FetcherSession: "http3": http3, **kwargs, } - return self.__prepare_and_dispatch( - "PUT", stealth=stealthy_headers, **request_args - ) + return self.__prepare_and_dispatch("PUT", stealth=stealthy_headers, **request_args) def delete( self, @@ -688,9 +644,7 @@ class FetcherSession: "http3": http3, **kwargs, } - return self.__prepare_and_dispatch( - "DELETE", stealth=stealthy_headers, **request_args - ) + return self.__prepare_and_dispatch("DELETE", stealth=stealthy_headers, **request_args) class FetcherClient(FetcherSession): diff --git a/scrapling/engines/toolbelt/convertor.py b/scrapling/engines/toolbelt/convertor.py index 12bfd55..6b0ca41 100644 --- a/scrapling/engines/toolbelt/convertor.py +++ b/scrapling/engines/toolbelt/convertor.py @@ -18,9 +18,7 @@ class ResponseFactory: """ @classmethod - def _process_response_history( - cls, first_response: SyncResponse, parser_arguments: Dict - ) -> list[Response]: + def _process_response_history(cls, first_response: SyncResponse, parser_arguments: Dict) -> list[Response]: """Process response history to build a list of `Response` objects""" history = [] current_request = first_response.request.redirected_from @@ -36,18 +34,12 @@ class ResponseFactory: # using current_response.text() will trigger "Error: Response.text: Response body is unavailable for redirect responses" content="", status=current_response.status if current_response else 301, - reason=( - current_response.status_text - or StatusText.get(current_response.status) - ) + reason=(current_response.status_text or StatusText.get(current_response.status)) if current_response else StatusText.get(301), - encoding=current_response.headers.get("content-type", "") - or "utf-8", + encoding=current_response.headers.get("content-type", "") or "utf-8", cookies=tuple(), - headers=current_response.all_headers() - if current_response - else {}, + headers=current_response.all_headers() if current_response else {}, request_headers=current_request.all_headers(), **parser_arguments, ), @@ -94,13 +86,9 @@ class ResponseFactory: raise ValueError("Failed to get a response from the page") # This will be parsed inside `Response` - encoding = ( - final_response.headers.get("content-type", "") or "utf-8" - ) # default encoding + encoding = final_response.headers.get("content-type", "") or "utf-8" # default encoding # PlayWright API sometimes give empty status text for some reason! - status_text = final_response.status_text or StatusText.get( - final_response.status - ) + status_text = final_response.status_text or StatusText.get(final_response.status) history = cls._process_response_history(first_response, parser_arguments) try: @@ -141,18 +129,12 @@ class ResponseFactory: # using current_response.text() will trigger "Error: Response.text: Response body is unavailable for redirect responses" content="", status=current_response.status if current_response else 301, - reason=( - current_response.status_text - or StatusText.get(current_response.status) - ) + reason=(current_response.status_text or StatusText.get(current_response.status)) if current_response else StatusText.get(301), - encoding=current_response.headers.get("content-type", "") - or "utf-8", + encoding=current_response.headers.get("content-type", "") or "utf-8", cookies=tuple(), - headers=await current_response.all_headers() - if current_response - else {}, + headers=await current_response.all_headers() if current_response else {}, request_headers=await current_request.all_headers(), **parser_arguments, ), @@ -199,17 +181,11 @@ class ResponseFactory: raise ValueError("Failed to get a response from the page") # This will be parsed inside `Response` - encoding = ( - final_response.headers.get("content-type", "") or "utf-8" - ) # default encoding + encoding = final_response.headers.get("content-type", "") or "utf-8" # default encoding # PlayWright API sometimes give empty status text for some reason! - status_text = final_response.status_text or StatusText.get( - final_response.status - ) + status_text = final_response.status_text or StatusText.get(final_response.status) - history = await cls._async_process_response_history( - first_response, parser_arguments - ) + history = await cls._async_process_response_history(first_response, parser_arguments) try: page_content = await page.content() except Exception as e: # pragma: no cover @@ -239,9 +215,7 @@ class ResponseFactory: """ return Response( url=response.url, - content=response.content - if isinstance(response.content, bytes) - else response.content.encode(), + content=response.content if isinstance(response.content, bytes) else response.content.encode(), status=response.status_code, reason=response.reason, encoding=response.encoding or "utf-8", diff --git a/scrapling/engines/toolbelt/custom.py b/scrapling/engines/toolbelt/custom.py index 79b52d1..72c39e9 100644 --- a/scrapling/engines/toolbelt/custom.py +++ b/scrapling/engines/toolbelt/custom.py @@ -49,9 +49,7 @@ class ResponseEncoding: @classmethod @lru_cache(maxsize=128) - def get_value( - cls, content_type: Optional[str], text: Optional[str] = "test" - ) -> str: + def get_value(cls, content_type: Optional[str], text: Optional[str] = "test") -> str: """Determine the appropriate character encoding from a content-type header. The encoding is determined by these rules in order: @@ -84,9 +82,7 @@ class ResponseEncoding: encoding = cls.__DEFAULT_ENCODING if encoding: - _ = text.encode( - encoding - ) # Validate encoding and validate it can encode the given text + _ = text.encode(encoding) # Validate encoding and validate it can encode the given text return encoding return cls.__DEFAULT_ENCODING @@ -129,9 +125,7 @@ class Response(Selector): **selector_config, ) # For easier debugging while working from a Python shell - log.info( - f"Fetched ({status}) <{method} {url}> (referer: {request_headers.get('referer')})" - ) + log.info(f"Fetched ({status}) <{method} {url}> (referer: {request_headers.get('referer')})") class BaseFetcher: @@ -190,18 +184,12 @@ class BaseFetcher: setattr(cls, key, value) else: # Yup, no fun allowed LOL - raise AttributeError( - f'Unknown parser argument: "{key}"; maybe you meant {cls.parser_keywords}?' - ) + raise AttributeError(f'Unknown parser argument: "{key}"; maybe you meant {cls.parser_keywords}?') else: - raise ValueError( - f'Unknown parser argument: "{key}"; maybe you meant {cls.parser_keywords}?' - ) + raise ValueError(f'Unknown parser argument: "{key}"; maybe you meant {cls.parser_keywords}?') if not kwargs: - raise AttributeError( - f"You must pass a keyword to configure, current keywords: {cls.parser_keywords}?" - ) + raise AttributeError(f"You must pass a keyword to configure, current keywords: {cls.parser_keywords}?") @classmethod def _generate_parser_arguments(cls) -> Dict: @@ -217,9 +205,7 @@ class BaseFetcher: ) if cls.adaptive_domain: if not isinstance(cls.adaptive_domain, str): - log.warning( - '[Ignored] The argument "adaptive_domain" must be of string type' - ) + log.warning('[Ignored] The argument "adaptive_domain" must be of string type') else: parser_arguments.update({"adaptive_domain": cls.adaptive_domain}) diff --git a/scrapling/engines/toolbelt/navigation.py b/scrapling/engines/toolbelt/navigation.py index ab91174..6d666a8 100644 --- a/scrapling/engines/toolbelt/navigation.py +++ b/scrapling/engines/toolbelt/navigation.py @@ -30,9 +30,7 @@ def intercept_route(route: Route): :return: PlayWright `Route` object """ if route.request.resource_type in DEFAULT_DISABLED_RESOURCES: - log.debug( - f'Blocking background resource "{route.request.url}" of type "{route.request.resource_type}"' - ) + log.debug(f'Blocking background resource "{route.request.url}" of type "{route.request.resource_type}"') route.abort() else: route.continue_() @@ -45,17 +43,13 @@ async def async_intercept_route(route: async_Route): :return: PlayWright `Route` object """ if route.request.resource_type in DEFAULT_DISABLED_RESOURCES: - log.debug( - f'Blocking background resource "{route.request.url}" of type "{route.request.resource_type}"' - ) + log.debug(f'Blocking background resource "{route.request.url}" of type "{route.request.resource_type}"') await route.abort() else: await route.continue_() -def construct_proxy_dict( - proxy_string: str | Dict[str, str], as_tuple=False -) -> Optional[Dict | Tuple]: +def construct_proxy_dict(proxy_string: str | Dict[str, str], as_tuple=False) -> Optional[Dict | Tuple]: """Validate a proxy and return it in the acceptable format for Playwright Reference: https://playwright.dev/python/docs/network#http-proxy @@ -65,10 +59,7 @@ def construct_proxy_dict( """ if isinstance(proxy_string, str): proxy = urlparse(proxy_string) - if ( - proxy.scheme not in ("http", "https", "socks4", "socks5") - or not proxy.hostname - ): + if proxy.scheme not in ("http", "https", "socks4", "socks5") or not proxy.hostname: raise ValueError("Invalid proxy string!") try: diff --git a/scrapling/fetchers.py b/scrapling/fetchers.py index ea7f749..fc72de4 100644 --- a/scrapling/fetchers.py +++ b/scrapling/fetchers.py @@ -112,9 +112,7 @@ 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__}" - ) + ValueError(f"The custom parser config must be of type dictionary, got {cls.__class__}") with StealthySession( wait=wait, @@ -210,9 +208,7 @@ 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__}" - ) + ValueError(f"The custom parser config must be of type dictionary, got {cls.__class__}") async with AsyncStealthySession( wait=wait, @@ -318,9 +314,7 @@ class DynamicFetcher(BaseFetcher): if not custom_config: custom_config = {} elif not isinstance(custom_config, dict): - raise ValueError( - f"The custom parser config must be of type dictionary, got {cls.__class__}" - ) + raise ValueError(f"The custom parser config must be of type dictionary, got {cls.__class__}") with DynamicSession( wait=wait, @@ -404,9 +398,7 @@ class DynamicFetcher(BaseFetcher): if not custom_config: custom_config = {} elif not isinstance(custom_config, dict): - raise ValueError( - f"The custom parser config must be of type dictionary, got {cls.__class__}" - ) + raise ValueError(f"The custom parser config must be of type dictionary, got {cls.__class__}") async with AsyncDynamicSession( wait=wait, diff --git a/scrapling/parser.py b/scrapling/parser.py index 730f490..f943db8 100644 --- a/scrapling/parser.py +++ b/scrapling/parser.py @@ -110,22 +110,16 @@ class Selector(SelectorsGeneration): If empty, default values will be used. """ if root is None and content is None: - raise ValueError( - "Selector class needs HTML content, or root arguments to work" - ) + raise ValueError("Selector class needs HTML content, or root arguments to work") self.__text = None if root is None: if isinstance(content, str): - body = ( - content.strip().replace("\x00", "").encode(encoding) or b"" - ) + body = content.strip().replace("\x00", "").encode(encoding) or b"" elif isinstance(content, bytes): body = content.replace(b"\x00", b"").strip() else: - raise TypeError( - f"content argument must be str or bytes, got {type(content)}" - ) + raise TypeError(f"content argument must be str or bytes, got {type(content)}") # https://lxml.de/api/lxml.etree.HTMLParser-class.html parser = HTMLParser( @@ -165,16 +159,10 @@ class Selector(SelectorsGeneration): } if not hasattr(storage, "__wrapped__"): - raise ValueError( - "Storage class must be wrapped with lru_cache decorator, see docs for info" - ) + raise ValueError("Storage class must be wrapped with lru_cache decorator, see docs for info") - if not issubclass( - storage.__wrapped__, StorageSystemMixin - ): # pragma: no cover - raise ValueError( - "Storage system must be inherited from class `StorageSystemMixin`" - ) + if not issubclass(storage.__wrapped__, StorageSystemMixin): # pragma: no cover + raise ValueError("Storage system must be inherited from class `StorageSystemMixin`") self._storage = storage(**storage_args) @@ -239,9 +227,7 @@ class Selector(SelectorsGeneration): def __element_convertor(self, element: HtmlElement) -> "Selector": """Used internally to convert a single HtmlElement to Selector directly without checks""" - db_instance = ( - self._storage if (hasattr(self, "_storage") and self._storage) else None - ) + db_instance = self._storage if (hasattr(self, "_storage") and self._storage) else None return Selector( root=element, url=self.url, @@ -355,9 +341,7 @@ class Selector(SelectorsGeneration): @property def html_content(self) -> TextHandler: """Return the inner HTML code of the element""" - return TextHandler( - tostring(self._root, encoding="unicode", method="html", with_tail=False) - ) + return TextHandler(tostring(self._root, encoding="unicode", method="html", with_tail=False)) body = html_content @@ -404,9 +388,7 @@ class Selector(SelectorsGeneration): def siblings(self) -> "Selectors": """Return other children of the current element's parent or empty list otherwise""" if self.parent: - return Selectors( - child for child in self.parent.children if child._root != self._root - ) + return Selectors(child for child in self.parent.children if child._root != self._root) return Selectors() def iterancestors(self) -> Generator["Selector", None, None]: @@ -519,9 +501,7 @@ 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.__handle_elements(score_table[percent])}") if not selector_type: return score_table[highest_probability] @@ -658,9 +638,7 @@ class Selector(SelectorsGeneration): SelectorError, SelectorSyntaxError, ) as e: - raise SelectorSyntaxError( - f"Invalid CSS selector '{selector}': {str(e)}" - ) from e + raise SelectorSyntaxError(f"Invalid CSS selector '{selector}': {str(e)}") from e def xpath( self, @@ -702,9 +680,7 @@ class Selector(SelectorsGeneration): elif self.__adaptive_enabled and auto_save: self.save(elements[0], identifier or selector) - return self.__handle_elements( - elements[0:1] if (_first_match and elements) else elements - ) + return self.__handle_elements(elements[0:1] if (_first_match and elements) else elements) elif self.__adaptive_enabled: if adaptive: element_data = self.retrieve(identifier or selector) @@ -713,9 +689,7 @@ class Selector(SelectorsGeneration): if elements is not None and auto_save: self.save(elements[0], identifier or selector) - return self.__handle_elements( - elements[0:1] if (_first_match and elements) else elements - ) + return self.__handle_elements(elements[0:1] if (_first_match and elements) else elements) else: if adaptive: log.warning( @@ -726,9 +700,7 @@ class Selector(SelectorsGeneration): "Argument `auto_save` will be ignored because `adaptive` wasn't enabled on initialization. Check docs for more info." ) - return self.__handle_elements( - elements[0:1] if (_first_match and elements) else elements - ) + return self.__handle_elements(elements[0:1] if (_first_match and elements) else elements) except ( SelectorError, @@ -751,9 +723,7 @@ class Selector(SelectorsGeneration): """ if not args and not kwargs: - raise TypeError( - "You have to pass something to search with, like tag name(s), tag attributes, or both." - ) + raise TypeError("You have to pass something to search with, like tag name(s), tag attributes, or both.") attributes = dict() tags, patterns = set(), set() @@ -766,18 +736,11 @@ class Selector(SelectorsGeneration): elif type(arg) in (list, tuple, set): if not all(map(lambda x: isinstance(x, str), arg)): - raise TypeError( - "Nested Iterables are not accepted, only iterables of tag names are accepted" - ) + raise TypeError("Nested Iterables are not accepted, only iterables of tag names are accepted") tags.update(set(arg)) elif isinstance(arg, dict): - if not all( - [ - (isinstance(k, str) and isinstance(v, str)) - for k, v in arg.items() - ] - ): + if not all([(isinstance(k, str) and isinstance(v, str)) for k, v in arg.items()]): raise TypeError( "Nested dictionaries are not accepted, only string keys and string values are accepted" ) @@ -795,13 +758,9 @@ class Selector(SelectorsGeneration): ) else: - raise TypeError( - f'Argument with type "{type(arg)}" is not accepted, please read the docs.' - ) + raise TypeError(f'Argument with type "{type(arg)}" is not accepted, please read the docs.') - if not all( - [(isinstance(k, str) and isinstance(v, str)) for k, v in kwargs.items()] - ): + if not all([(isinstance(k, str) and isinstance(v, str)) for k, v in kwargs.items()]): raise TypeError("Only string values are accepted for arguments") for attribute_name, value in kwargs.items(): @@ -825,9 +784,7 @@ class Selector(SelectorsGeneration): if results: # From the results, get the ones that fulfill passed regex patterns for pattern in patterns: - results = results.filter( - lambda e: e.text.re(pattern, check_match=True) - ) + results = results.filter(lambda e: e.text.re(pattern, check_match=True)) # From the results, get the ones that fulfill passed functions for function in functions: @@ -858,9 +815,7 @@ class Selector(SelectorsGeneration): return element return None - def __calculate_similarity_score( - self, original: Dict, candidate: HtmlElement - ) -> float: + def __calculate_similarity_score(self, original: Dict, candidate: HtmlElement) -> float: """Used internally to calculate a score that shows how a candidate element similar to the original one :param original: The original element in the form of the dictionary generated from `element_to_dict` function @@ -877,15 +832,11 @@ class Selector(SelectorsGeneration): checks += 1 if original["text"]: - score += SequenceMatcher( - None, original["text"], candidate.get("text") or "" - ).ratio() # * 0.3 # 30% + score += SequenceMatcher(None, original["text"], candidate.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"], candidate["attributes"]) # * 0.3 # 30% checks += 1 # Separate similarity test for class, id, href,... this will help in full structural changes @@ -903,9 +854,7 @@ class Selector(SelectorsGeneration): ).ratio() # * 0.3 # 30% checks += 1 - score += SequenceMatcher( - None, original["path"], candidate["path"] - ).ratio() # * 0.1 # 10% + score += SequenceMatcher(None, original["path"], candidate["path"]).ratio() # * 0.1 # 10% checks += 1 if original.get("parent_name"): @@ -944,14 +893,8 @@ class Selector(SelectorsGeneration): @staticmethod def __calculate_dict_diff(dict1: Dict, dict2: Dict) -> float: """Used internally to calculate similarity between two dictionaries as SequenceMatcher doesn't accept dictionaries""" - score = ( - SequenceMatcher(None, tuple(dict1.keys()), tuple(dict2.keys())).ratio() - * 0.5 - ) - score += ( - SequenceMatcher(None, tuple(dict1.values()), tuple(dict2.values())).ratio() - * 0.5 - ) + score = SequenceMatcher(None, tuple(dict1.keys()), tuple(dict2.keys())).ratio() * 0.5 + score += SequenceMatcher(None, tuple(dict1.values()), tuple(dict2.values())).ratio() * 0.5 return score def save(self, element: Union["Selector", HtmlElement], identifier: str) -> None: @@ -1031,9 +974,7 @@ class Selector(SelectorsGeneration): :param clean_match: if enabled, this will ignore all whitespaces and consecutive spaces while matching :param case_sensitive: if disabled, the function will set the regex to ignore the letters case while compiling it """ - return self.text.re_first( - regex, default, replace_entities, clean_match, case_sensitive - ) + return self.text.re_first(regex, default, replace_entities, clean_match, case_sensitive) @staticmethod def __get_attributes(element: HtmlElement, ignore_attributes: List | Tuple) -> Dict: @@ -1052,9 +993,7 @@ class Selector(SelectorsGeneration): """Calculate a score of how much these elements are alike and return True if the score is higher or equals the threshold""" candidate_attributes = ( - self.__get_attributes(candidate, ignore_attributes) - if ignore_attributes - else candidate.attrib + self.__get_attributes(candidate, ignore_attributes) if ignore_attributes else candidate.attrib ) score, checks = 0, 0 @@ -1116,11 +1055,7 @@ class Selector(SelectorsGeneration): similar_elements = list() current_depth = len(list(root.iterancestors())) - target_attrs = ( - self.__get_attributes(root, ignore_attributes) - if ignore_attributes - else root.attrib - ) + target_attrs = self.__get_attributes(root, ignore_attributes) if ignore_attributes else root.attrib path_parts = [self.tag] if (parent := root.getparent()) is not None: @@ -1129,9 +1064,7 @@ class Selector(SelectorsGeneration): path_parts.insert(0, grandparent.tag) xpath_path = "//{}".format("/".join(path_parts)) - potential_matches = root.xpath( - f"{xpath_path}[count(ancestor::*) = {current_depth}]" - ) + potential_matches = root.xpath(f"{xpath_path}[count(ancestor::*) = {current_depth}]") for potential_match in potential_matches: if potential_match != root and self.__are_alike( @@ -1275,12 +1208,7 @@ class Selectors(List[Selector]): :return: `Selectors` class. """ - results = [ - n.xpath( - selector, identifier or selector, False, auto_save, percentage, **kwargs - ) - for n in self - ] + results = [n.xpath(selector, identifier or selector, False, auto_save, percentage, **kwargs) for n in self] return self.__class__(flatten(results)) def css( @@ -1308,10 +1236,7 @@ class Selectors(List[Selector]): :return: `Selectors` class. """ - results = [ - n.css(selector, identifier or selector, False, auto_save, percentage) - for n in self - ] + results = [n.css(selector, identifier or selector, False, auto_save, percentage) for n in self] return self.__class__(flatten(results)) def re( @@ -1329,10 +1254,7 @@ class Selectors(List[Selector]): :param clean_match: if enabled, this will ignore all whitespaces and consecutive spaces while matching :param case_sensitive: if disabled, the function will set the regex to ignore the letters case while compiling it """ - results = [ - n.text.re(regex, replace_entities, clean_match, case_sensitive) - for n in self - ] + results = [n.text.re(regex, replace_entities, clean_match, case_sensitive) for n in self] return TextHandlers(flatten(results)) def re_first(