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

style: applying the new ruff rules to all files

Browse Source
This commit is contained in:
Karim shoair
2025-09-13 03:22:53 +03:00
parent 60be9dc816
commit 330d03559c
18 changed files with 176 additions and 572 deletions
Download Patch File Download Diff File Expand all files Collapse all files
scrapling/cli.py
+15 -46
View File
@@ -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(
scrapling/core/_html_utils.py
+3 -9
View File
@@ -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
scrapling/core/ai.py
+4 -12
View File
@@ -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:
scrapling/core/custom_types.py
+19 -61
View File
@@ -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
scrapling/core/mixins.py
+6 -28
View File
@@ -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:
scrapling/core/shell.py
+19 -63
View File
@@ -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]
scrapling/core/storage.py
+2 -8
View File
@@ -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("""
scrapling/core/translator.py
+6 -18
View File
@@ -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
scrapling/core/utils.py
+5 -21
View File
@@ -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)
scrapling/engines/_browsers/_base.py
+7 -25
View File
@@ -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):
scrapling/engines/_browsers/_page.py
+1 -3
View File
@@ -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
scrapling/engines/_browsers/_validators.py
+6 -14
View File
@@ -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)
scrapling/engines/static.py
+22 -68
View File
@@ -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):
scrapling/engines/toolbelt/convertor.py
+13 -39
View File
@@ -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",
scrapling/engines/toolbelt/custom.py
+7 -21
View File
@@ -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})
scrapling/engines/toolbelt/navigation.py
+4 -13
View File
@@ -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:
scrapling/fetchers.py
+4 -12
View File
@@ -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,
scrapling/parser.py
+33 -111
View File
@@ -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"<html/>"
)
body = content.strip().replace("\x00", "").encode(encoding) or b"<html/>"
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(