refactor: Make all fetchers as an optional dependency group
+ Removing some dead code
This commit is contained in:
@@ -0,0 +1,10 @@
|
||||
from ._utils import (
|
||||
log,
|
||||
__CONSECUTIVE_SPACES_REGEX__,
|
||||
flatten,
|
||||
_is_iterable,
|
||||
_StorageTools,
|
||||
clean_spaces,
|
||||
html_forbidden,
|
||||
)
|
||||
from ._shell import _CookieParser, _ParseHeaders
|
||||
@@ -0,0 +1,48 @@
|
||||
from http import cookies as Cookie
|
||||
|
||||
|
||||
from scrapling.core._types import (
|
||||
List,
|
||||
Dict,
|
||||
Tuple,
|
||||
)
|
||||
|
||||
|
||||
def _CookieParser(cookie_string):
|
||||
# Errors will be handled on call so the log can be specified
|
||||
cookie_parser = Cookie.SimpleCookie()
|
||||
cookie_parser.load(cookie_string)
|
||||
for key, morsel in cookie_parser.items():
|
||||
yield key, morsel.value
|
||||
|
||||
|
||||
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()
|
||||
|
||||
for header_line in header_lines:
|
||||
if ":" not in header_line:
|
||||
if header_line.endswith(";"):
|
||||
header_key = header_line[:-1].strip()
|
||||
header_value = ""
|
||||
header_dict[header_key] = header_value
|
||||
else:
|
||||
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()
|
||||
header_value = header_value.strip()
|
||||
|
||||
if parse_cookies:
|
||||
if header_key.lower() == "cookie":
|
||||
try:
|
||||
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}")
|
||||
else:
|
||||
header_dict[header_key] = header_value
|
||||
else:
|
||||
header_dict[header_key] = header_value
|
||||
|
||||
return header_dict, cookie_dict
|
||||
@@ -0,0 +1,101 @@
|
||||
import logging
|
||||
from itertools import chain
|
||||
from re import compile as re_compile
|
||||
|
||||
from lxml import html
|
||||
|
||||
from scrapling.core._types import Any, Dict, Iterable, List
|
||||
|
||||
# Using cache on top of a class is a brilliant way to achieve a Singleton design pattern without much code
|
||||
from functools import lru_cache # isort:skip
|
||||
|
||||
html_forbidden = (html.HtmlComment,)
|
||||
|
||||
__CLEANING_TABLE__ = str.maketrans({"\t": " ", "\n": None, "\r": None})
|
||||
__CONSECUTIVE_SPACES_REGEX__ = re_compile(r" +")
|
||||
|
||||
|
||||
@lru_cache(1, typed=True)
|
||||
def setup_logger():
|
||||
"""Create and configure a logger with a standard format.
|
||||
|
||||
:returns: logging.Logger: Configured logger instance
|
||||
"""
|
||||
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")
|
||||
|
||||
console_handler = logging.StreamHandler()
|
||||
console_handler.setFormatter(formatter)
|
||||
|
||||
# Add handler to logger (if not already added)
|
||||
if not logger.handlers:
|
||||
logger.addHandler(console_handler)
|
||||
|
||||
return logger
|
||||
|
||||
|
||||
log = setup_logger()
|
||||
|
||||
|
||||
def flatten(lst: Iterable[Any]) -> List[Any]:
|
||||
return list(chain.from_iterable(lst))
|
||||
|
||||
|
||||
def _is_iterable(obj: Any) -> bool:
|
||||
# This will be used only in regex functions to make sure it's iterable but not string/bytes
|
||||
return isinstance(
|
||||
obj,
|
||||
(
|
||||
list,
|
||||
tuple,
|
||||
),
|
||||
)
|
||||
|
||||
|
||||
class _StorageTools:
|
||||
@staticmethod
|
||||
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}
|
||||
|
||||
@classmethod
|
||||
def element_to_dict(cls, element: html.HtmlElement) -> Dict:
|
||||
parent = element.getparent()
|
||||
result = {
|
||||
"tag": str(element.tag),
|
||||
"attributes": cls.__clean_attributes(element),
|
||||
"text": element.text.strip() if element.text else None,
|
||||
"path": cls._get_element_path(element),
|
||||
}
|
||||
if parent is not None:
|
||||
result.update(
|
||||
{
|
||||
"parent_name": parent.tag,
|
||||
"parent_attribs": dict(parent.attrib),
|
||||
"parent_text": parent.text.strip() if parent.text else None,
|
||||
}
|
||||
)
|
||||
|
||||
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)]
|
||||
if children:
|
||||
result.update({"children": tuple(children)})
|
||||
|
||||
return result
|
||||
|
||||
@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,)))
|
||||
|
||||
|
||||
@lru_cache(128, typed=True)
|
||||
def clean_spaces(string):
|
||||
string = string.translate(__CLEANING_TABLE__)
|
||||
return __CONSECUTIVE_SPACES_REGEX__.sub(" ", string)
|
||||
Reference in New Issue
Block a user