Big structure changes (check commit description)
- Moved most of the parser functions/files to the core package. - Converted tools file to a package and made separate files for similar functions. - Now all fetcher engines return a Response object - Instead of selecting an engine to use and passing config to it, we have separate fetcher classes so the user can choose what to use while importing. - I added a new custom fetcher so the user can create and use an engine. - More...
This commit is contained in:
Karim shoair
@@ -0,0 +1,25 @@
|
||||
"""
|
||||
Type definitions for type checking purposes.
|
||||
"""
|
||||
|
||||
from typing import (
|
||||
Dict, Optional, Union, Callable, Any, List, Tuple, Pattern, Generator, Iterable, Type, TYPE_CHECKING
|
||||
)
|
||||
|
||||
try:
|
||||
from typing import Protocol
|
||||
except ImportError:
|
||||
# Added in Python 3.8
|
||||
Protocol = object
|
||||
|
||||
try:
|
||||
from typing import SupportsIndex
|
||||
except ImportError:
|
||||
# 'SupportsIndex' got added in Python 3.8
|
||||
SupportsIndex = None
|
||||
|
||||
if TYPE_CHECKING:
|
||||
# typing.Self requires Python 3.11
|
||||
from typing_extensions import Self
|
||||
else:
|
||||
Self = object
|
||||
@@ -0,0 +1,146 @@
|
||||
import re
|
||||
from types import MappingProxyType
|
||||
from collections.abc import Mapping
|
||||
|
||||
from scrapling.core.utils import _is_iterable, flatten
|
||||
from scrapling.core._types import Dict, List, Union, Pattern
|
||||
|
||||
from orjson import loads, dumps
|
||||
from w3lib.html import replace_entities as _replace_entities
|
||||
|
||||
|
||||
class TextHandler(str):
|
||||
"""Extends standard Python string by adding more functionality"""
|
||||
__slots__ = ()
|
||||
|
||||
def __new__(cls, string):
|
||||
# Because str is immutable and we can't override __init__
|
||||
if type(string) is str:
|
||||
return super().__new__(cls, string)
|
||||
else:
|
||||
return super().__new__(cls, '')
|
||||
|
||||
def sort(self, reverse: bool = False) -> str:
|
||||
"""Return a sorted version of the string"""
|
||||
return self.__class__("".join(sorted(self, reverse=reverse)))
|
||||
|
||||
def clean(self) -> str:
|
||||
"""Return a new version of the string after removing all white spaces and consecutive spaces"""
|
||||
data = re.sub(r'[\t|\r|\n]', '', self)
|
||||
data = re.sub(' +', ' ', data)
|
||||
return self.__class__(data.strip())
|
||||
|
||||
def json(self) -> Dict:
|
||||
"""Return json response if the response is jsonable otherwise throw error"""
|
||||
# Using __str__ function as a workaround for orjson issue with subclasses of str
|
||||
# Check this out: https://github.com/ijl/orjson/issues/445
|
||||
return loads(self.__str__())
|
||||
|
||||
def re(
|
||||
self, regex: Union[str, Pattern[str]], replace_entities: bool = True, clean_match: bool = False,
|
||||
case_sensitive: bool = False, check_match: bool = False
|
||||
) -> Union[List[str], bool]:
|
||||
"""Apply the given regex to the current text and return a list of strings with the matches.
|
||||
|
||||
:param regex: Can be either a compiled regular expression or a string.
|
||||
:param replace_entities: if enabled character entity references are replaced by their corresponding character
|
||||
:param clean_match: if enabled, this will ignore all whitespaces and consecutive spaces while matching
|
||||
:param case_sensitive: if enabled, function will set the regex to ignore letters case while compiling it
|
||||
:param check_match: used to quickly check if this regex matches or not without any operations on the results
|
||||
|
||||
"""
|
||||
if isinstance(regex, str):
|
||||
if not case_sensitive:
|
||||
regex = re.compile(regex, re.UNICODE)
|
||||
else:
|
||||
regex = re.compile(regex, flags=re.UNICODE | re.IGNORECASE)
|
||||
|
||||
input_text = self.clean() if clean_match else self
|
||||
results = regex.findall(input_text)
|
||||
if check_match:
|
||||
return bool(results)
|
||||
|
||||
if all(_is_iterable(res) for res in results):
|
||||
results = flatten(results)
|
||||
|
||||
if not replace_entities:
|
||||
return [TextHandler(string) for string in results]
|
||||
|
||||
return [TextHandler(_replace_entities(s)) for s in results]
|
||||
|
||||
def re_first(self, regex: Union[str, Pattern[str]], default=None, replace_entities: bool = True,
|
||||
clean_match: bool = False, case_sensitive: bool = False,):
|
||||
"""Apply the given regex to text and return the first match if found, otherwise return the default value.
|
||||
|
||||
:param regex: Can be either a compiled regular expression or a string.
|
||||
:param default: The default value to be returned if there is no match
|
||||
:param replace_entities: if enabled character entity references are replaced by their corresponding character
|
||||
:param clean_match: if enabled, this will ignore all whitespaces and consecutive spaces while matching
|
||||
:param case_sensitive: if enabled, function will set the regex to ignore letters case while compiling it
|
||||
|
||||
"""
|
||||
result = self.re(regex, replace_entities, clean_match=clean_match, case_sensitive=case_sensitive)
|
||||
return result[0] if result else default
|
||||
|
||||
|
||||
class AttributesHandler(Mapping):
|
||||
"""A read-only mapping to use instead of the standard dictionary for the speed boost but
|
||||
at the same time I use it to add more functionalities.
|
||||
If standard dictionary is needed, just convert this class to dictionary with `dict` function
|
||||
"""
|
||||
__slots__ = ('_data',)
|
||||
|
||||
def __init__(self, mapping=None, **kwargs):
|
||||
mapping = {
|
||||
key: TextHandler(value) if type(value) is str else value
|
||||
for key, value in mapping.items()
|
||||
} if mapping is not None else {}
|
||||
|
||||
if kwargs:
|
||||
mapping.update({
|
||||
key: TextHandler(value) if type(value) is str else value
|
||||
for key, value in kwargs.items()
|
||||
})
|
||||
|
||||
# Fastest read-only mapping type
|
||||
self._data = MappingProxyType(mapping)
|
||||
|
||||
def get(self, key, default=None):
|
||||
"""Acts like standard dictionary `.get()` method"""
|
||||
return self._data.get(key, default)
|
||||
|
||||
def search_values(self, keyword, partial=False):
|
||||
"""Search current attributes by values and return dictionary of each matching item
|
||||
:param keyword: The keyword to search for in the attributes values
|
||||
:param partial: If True, the function will search if keyword in each value instead of perfect match
|
||||
"""
|
||||
for key, value in self._data.items():
|
||||
if partial:
|
||||
if keyword in value:
|
||||
yield AttributesHandler({key: value})
|
||||
else:
|
||||
if keyword == value:
|
||||
yield AttributesHandler({key: value})
|
||||
|
||||
@property
|
||||
def json_string(self):
|
||||
"""Convert current attributes to JSON string if the attributes are JSON serializable otherwise throws error"""
|
||||
return dumps(dict(self._data))
|
||||
|
||||
def __getitem__(self, key):
|
||||
return self._data[key]
|
||||
|
||||
def __iter__(self):
|
||||
return iter(self._data)
|
||||
|
||||
def __len__(self):
|
||||
return len(self._data)
|
||||
|
||||
def __repr__(self):
|
||||
return f"{self.__class__.__name__}({self._data})"
|
||||
|
||||
def __str__(self):
|
||||
return str(self._data)
|
||||
|
||||
def __contains__(self, key):
|
||||
return key in self._data
|
||||
@@ -0,0 +1,74 @@
|
||||
|
||||
class SelectorsGeneration:
|
||||
"""Selectors generation functions
|
||||
Trying to generate selectors like Firefox or maybe cleaner ones!? Ehm
|
||||
Inspiration: https://searchfox.org/mozilla-central/source/devtools/shared/inspector/css-logic.js#591"""
|
||||
|
||||
def __general_selection(self, selection: str = 'css') -> str:
|
||||
"""Generate a selector for the current element.
|
||||
:return: A string of the generated selector.
|
||||
"""
|
||||
selectorPath = []
|
||||
target = self
|
||||
css = selection.lower() == 'css'
|
||||
while target is not None:
|
||||
if target.parent:
|
||||
if target.attrib.get('id'):
|
||||
# id is enough
|
||||
part = (
|
||||
f'#{target.attrib["id"]}' if css
|
||||
else f"[@id='{target.attrib['id']}']"
|
||||
)
|
||||
selectorPath.append(part)
|
||||
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
|
||||
# classes = target.attrib.get('class', '').split()
|
||||
# if classes and css:
|
||||
# part += f".{'.'.join(classes)}"
|
||||
# else:
|
||||
counter = {}
|
||||
for child in target.parent.children:
|
||||
counter.setdefault(child.tag, 0)
|
||||
counter[child.tag] += 1
|
||||
if child._root == target._root:
|
||||
break
|
||||
|
||||
if counter[target.tag] > 1:
|
||||
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))
|
||||
)
|
||||
else:
|
||||
break
|
||||
|
||||
return (
|
||||
" > ".join(reversed(selectorPath)) if css
|
||||
else '//' + "/".join(reversed(selectorPath))
|
||||
)
|
||||
|
||||
@property
|
||||
def css_selector(self) -> str:
|
||||
"""Generate a CSS selector for the current element
|
||||
:return: A string of the generated selector.
|
||||
"""
|
||||
return self.__general_selection()
|
||||
|
||||
@property
|
||||
def xpath_selector(self) -> str:
|
||||
"""Generate a XPath selector for the current element
|
||||
:return: A string of the generated selector.
|
||||
"""
|
||||
return self.__general_selection('xpath')
|
||||
@@ -0,0 +1,149 @@
|
||||
import orjson
|
||||
import sqlite3
|
||||
import logging
|
||||
import threading
|
||||
from hashlib import sha256
|
||||
from abc import ABC, abstractmethod
|
||||
|
||||
from scrapling.core._types import Dict, Optional, Union
|
||||
from scrapling.core.utils import _StorageTools, cache
|
||||
|
||||
from lxml import html
|
||||
from tldextract import extract as tld
|
||||
|
||||
|
||||
class StorageSystemMixin(ABC):
|
||||
# If you want to make your own storage system, you have to inherit from this
|
||||
def __init__(self, url: Union[str, None] = None):
|
||||
"""
|
||||
:param url: URL of the website we are working on to separate it from other websites data
|
||||
"""
|
||||
self.url = url
|
||||
|
||||
@cache(None, typed=True)
|
||||
def _get_base_url(self, default_value: str = 'default') -> str:
|
||||
if not self.url or type(self.url) is not str:
|
||||
return default_value
|
||||
|
||||
try:
|
||||
extracted = tld(self.url)
|
||||
return extracted.registered_domain or extracted.domain or default_value
|
||||
except AttributeError:
|
||||
return default_value
|
||||
|
||||
@abstractmethod
|
||||
def save(self, element: html.HtmlElement, identifier: str) -> None:
|
||||
"""Saves the element's unique properties to the storage for retrieval and relocation later
|
||||
|
||||
:param element: The element itself that we want to save to storage.
|
||||
:param identifier: This is the identifier that will be used to retrieve the element later from the storage. See
|
||||
the docs for more info.
|
||||
"""
|
||||
raise NotImplementedError('Storage system must implement `save` method')
|
||||
|
||||
@abstractmethod
|
||||
def retrieve(self, identifier: str) -> Optional[Dict]:
|
||||
"""Using the identifier, we search the storage and return the unique properties of the element
|
||||
|
||||
:param identifier: This is the identifier that will be used to retrieve the element from the storage. See
|
||||
the docs for more info.
|
||||
:return: A dictionary of the unique properties
|
||||
"""
|
||||
raise NotImplementedError('Storage system must implement `save` method')
|
||||
|
||||
@staticmethod
|
||||
@cache(None, typed=True)
|
||||
def _get_hash(identifier: str) -> str:
|
||||
"""If you want to hash identifier in your storage system, use this safer"""
|
||||
identifier = identifier.lower().strip()
|
||||
if isinstance(identifier, str):
|
||||
# Hash functions have to take bytes
|
||||
identifier = identifier.encode('utf-8')
|
||||
|
||||
hash_value = sha256(identifier).hexdigest()
|
||||
return f"{hash_value}_{len(identifier)}" # Length to reduce collision chance
|
||||
|
||||
|
||||
@cache(None, typed=True)
|
||||
class SQLiteStorageSystem(StorageSystemMixin):
|
||||
"""The recommended system to use, it's race condition safe and thread safe.
|
||||
Mainly built so the library can run in threaded frameworks like scrapy or threaded tools
|
||||
> It's optimized for threaded applications but running it without threads shouldn't make it slow."""
|
||||
def __init__(self, storage_file: str, url: Union[str, None] = None):
|
||||
"""
|
||||
:param storage_file: File to be used to store elements
|
||||
:param url: URL of the website we are working on to separate it from other websites data
|
||||
|
||||
"""
|
||||
super().__init__(url)
|
||||
self.storage_file = storage_file
|
||||
# We use a threading.Lock to ensure thread-safety instead of relying on thread-local storage.
|
||||
self.lock = threading.Lock()
|
||||
# >SQLite default mode in earlier version is 1 not 2 (1=thread-safe 2=serialized)
|
||||
# `check_same_thread=False` to allow it to be used across different threads.
|
||||
self.connection = sqlite3.connect(self.storage_file, check_same_thread=False)
|
||||
# WAL (Write-Ahead Logging) allows for better concurrency.
|
||||
self.connection.execute("PRAGMA journal_mode=WAL")
|
||||
self.cursor = self.connection.cursor()
|
||||
self._setup_database()
|
||||
logging.debug(
|
||||
f'Storage system loaded with arguments (storage_file="{storage_file}", url="{url}")'
|
||||
)
|
||||
|
||||
def _setup_database(self) -> None:
|
||||
self.cursor.execute("""
|
||||
CREATE TABLE IF NOT EXISTS storage (
|
||||
id INTEGER PRIMARY KEY,
|
||||
url TEXT,
|
||||
identifier TEXT,
|
||||
element_data TEXT,
|
||||
UNIQUE (url, identifier)
|
||||
)
|
||||
""")
|
||||
self.connection.commit()
|
||||
|
||||
def save(self, element: html.HtmlElement, identifier: str):
|
||||
"""Saves the elements unique properties to the storage for retrieval and relocation later
|
||||
|
||||
:param element: The element itself that we want to save to storage.
|
||||
:param identifier: This is the identifier that will be used to retrieve the element later from the storage. See
|
||||
the docs for more info.
|
||||
"""
|
||||
url = self._get_base_url()
|
||||
element_data = _StorageTools.element_to_dict(element)
|
||||
with self.lock:
|
||||
self.cursor.execute("""
|
||||
INSERT OR REPLACE INTO storage (url, identifier, element_data)
|
||||
VALUES (?, ?, ?)
|
||||
""", (url, identifier, orjson.dumps(element_data)))
|
||||
self.cursor.fetchall()
|
||||
self.connection.commit()
|
||||
|
||||
def retrieve(self, identifier: str) -> Optional[Dict]:
|
||||
"""Using the identifier, we search the storage and return the unique properties of the element
|
||||
|
||||
:param identifier: This is the identifier that will be used to retrieve the element from the storage. See
|
||||
the docs for more info.
|
||||
:return: A dictionary of the unique properties
|
||||
"""
|
||||
url = self._get_base_url()
|
||||
with self.lock:
|
||||
self.cursor.execute(
|
||||
"SELECT element_data FROM storage WHERE url = ? AND identifier = ?",
|
||||
(url, identifier)
|
||||
)
|
||||
result = self.cursor.fetchone()
|
||||
if result:
|
||||
return orjson.loads(result[0])
|
||||
return None
|
||||
|
||||
def close(self):
|
||||
"""Close all connections, will be useful when with some things like scrapy Spider.closed() function/signal"""
|
||||
with self.lock:
|
||||
self.connection.commit()
|
||||
self.cursor.close()
|
||||
self.connection.close()
|
||||
|
||||
def __del__(self):
|
||||
"""To ensure all connections are closed when the object is destroyed."""
|
||||
self.close()
|
||||
@@ -0,0 +1,143 @@
|
||||
"""
|
||||
Most of this file is adapted version of the translator of parsel library with some modifications simply for 1 important reason...
|
||||
To add pseudo-elements ``::text`` and ``::attr(ATTR_NAME)`` so we match Parsel/Scrapy selectors format
|
||||
which will be important in future releases but most importantly...
|
||||
so you don't have to learn a new selectors/api method like what bs4 done with soupsieve :)
|
||||
> if you want to learn about this, head to https://cssselect.readthedocs.io/en/latest/#cssselect.FunctionalPseudoElement
|
||||
"""
|
||||
|
||||
import re
|
||||
|
||||
from w3lib.html import HTML5_WHITESPACE
|
||||
from scrapling.core.utils import cache
|
||||
from scrapling.core._types import Any, Optional, Protocol, Self
|
||||
|
||||
from cssselect.xpath import ExpressionError
|
||||
from cssselect.xpath import XPathExpr as OriginalXPathExpr
|
||||
from cssselect import HTMLTranslator as OriginalHTMLTranslator
|
||||
from cssselect.parser import Element, FunctionalPseudoElement, PseudoElement
|
||||
|
||||
|
||||
regex = f"[{HTML5_WHITESPACE}]+"
|
||||
replace_html5_whitespaces = re.compile(regex).sub
|
||||
|
||||
|
||||
class XPathExpr(OriginalXPathExpr):
|
||||
|
||||
textnode: bool = False
|
||||
attribute: Optional[str] = None
|
||||
|
||||
@classmethod
|
||||
def from_xpath(
|
||||
cls,
|
||||
xpath: OriginalXPathExpr,
|
||||
textnode: bool = False,
|
||||
attribute: Optional[str] = None,
|
||||
) -> "Self":
|
||||
x = cls(path=xpath.path, element=xpath.element, condition=xpath.condition)
|
||||
x.textnode = textnode
|
||||
x.attribute = attribute
|
||||
return x
|
||||
|
||||
def __str__(self) -> str:
|
||||
path = super().__str__()
|
||||
if self.textnode:
|
||||
if path == "*":
|
||||
path = "text()"
|
||||
elif path.endswith("::*/*"):
|
||||
path = path[:-3] + "text()"
|
||||
else:
|
||||
path += "/text()"
|
||||
|
||||
if self.attribute is not None:
|
||||
if path.endswith("::*/*"):
|
||||
path = path[:-2]
|
||||
path += f"/@{self.attribute}"
|
||||
|
||||
return path
|
||||
|
||||
def join(
|
||||
self: "Self",
|
||||
combiner: str,
|
||||
other: OriginalXPathExpr,
|
||||
*args: Any,
|
||||
**kwargs: Any,
|
||||
) -> "Self":
|
||||
if not isinstance(other, XPathExpr):
|
||||
raise ValueError(
|
||||
f"Expressions of type {__name__}.XPathExpr can ony join expressions"
|
||||
f" of the same type (or its descendants), got {type(other)}"
|
||||
)
|
||||
super().join(combiner, other, *args, **kwargs)
|
||||
self.textnode = other.textnode
|
||||
self.attribute = other.attribute
|
||||
return self
|
||||
|
||||
|
||||
# e.g. cssselect.GenericTranslator, cssselect.HTMLTranslator
|
||||
class TranslatorProtocol(Protocol):
|
||||
def xpath_element(self, selector: Element) -> OriginalXPathExpr:
|
||||
pass
|
||||
|
||||
def css_to_xpath(self, css: str, prefix: str = ...) -> str:
|
||||
pass
|
||||
|
||||
|
||||
class TranslatorMixin:
|
||||
"""This mixin adds support to CSS pseudo elements via dynamic dispatch.
|
||||
|
||||
Currently supported pseudo-elements are ``::text`` and ``::attr(ATTR_NAME)``.
|
||||
"""
|
||||
|
||||
def xpath_element(self: TranslatorProtocol, selector: Element) -> XPathExpr:
|
||||
# https://github.com/python/mypy/issues/12344
|
||||
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:
|
||||
"""
|
||||
Dispatch method that transforms XPath to support pseudo-elements.
|
||||
"""
|
||||
if isinstance(pseudo_element, FunctionalPseudoElement):
|
||||
method_name = f"xpath_{pseudo_element.name.replace('-', '_')}_functional_pseudo_element"
|
||||
method = getattr(self, method_name, None)
|
||||
if not method:
|
||||
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 = getattr(self, method_name, None)
|
||||
if not method:
|
||||
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:
|
||||
"""Support selecting attribute values using ::attr() pseudo-element"""
|
||||
if function.argument_types() not in (["STRING"], ["IDENT"]):
|
||||
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
|
||||
def xpath_text_simple_pseudo_element(xpath: OriginalXPathExpr) -> XPathExpr:
|
||||
"""Support selecting text nodes using ::text pseudo-element"""
|
||||
return XPathExpr.from_xpath(xpath, textnode=True)
|
||||
|
||||
|
||||
class HTMLTranslator(TranslatorMixin, OriginalHTMLTranslator):
|
||||
@cache(maxsize=256)
|
||||
def css_to_xpath(self, css: str, prefix: str = "descendant-or-self::") -> str:
|
||||
return super().css_to_xpath(css, prefix)
|
||||
@@ -0,0 +1,105 @@
|
||||
import re
|
||||
import logging
|
||||
from itertools import chain
|
||||
# Using cache on top of a class is brilliant way to achieve Singleton design pattern without much code
|
||||
from functools import lru_cache as cache # functools.cache is available on Python 3.9+ only so let's keep lru_cache
|
||||
|
||||
from scrapling.core._types import Dict, Iterable, Any
|
||||
|
||||
from lxml import html
|
||||
|
||||
html_forbidden = {html.HtmlComment, }
|
||||
logging.basicConfig(
|
||||
level=logging.ERROR,
|
||||
format='%(asctime)s - %(levelname)s - %(message)s',
|
||||
handlers=[
|
||||
logging.StreamHandler()
|
||||
]
|
||||
)
|
||||
|
||||
|
||||
@cache(None, typed=True)
|
||||
def setup_basic_logging(level: str = 'debug'):
|
||||
levels = {
|
||||
'debug': logging.DEBUG,
|
||||
'info': logging.INFO,
|
||||
'warning': logging.WARNING,
|
||||
'error': logging.ERROR,
|
||||
'critical': logging.CRITICAL
|
||||
}
|
||||
formatter = logging.Formatter("[%(asctime)s] %(levelname)s: %(message)s", "%Y-%m-%d %H:%M:%S")
|
||||
lvl = levels[level.lower()]
|
||||
handler = logging.StreamHandler()
|
||||
handler.setFormatter(formatter)
|
||||
# Configure the root logger
|
||||
logging.basicConfig(level=lvl, handlers=[handler])
|
||||
|
||||
|
||||
def flatten(lst: Iterable):
|
||||
return list(chain.from_iterable(lst))
|
||||
|
||||
|
||||
def _is_iterable(s: Any):
|
||||
# This will be used only in regex functions to make sure it's iterable but not string/bytes
|
||||
return isinstance(s, (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 type(child) not in 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,)
|
||||
)
|
||||
)
|
||||
|
||||
|
||||
# def _root_type_verifier(method):
|
||||
# # Just to make sure we are safe
|
||||
# @wraps(method)
|
||||
# def _impl(self, *args, **kw):
|
||||
# # All html types inherits from HtmlMixin so this to check for all at once
|
||||
# if not issubclass(type(self._root), html.HtmlMixin):
|
||||
# raise ValueError(f"Cannot use function on a Node of type {type(self._root)!r}")
|
||||
# return method(self, *args, **kw)
|
||||
# return _impl
|
||||
|
||||
|
||||
@cache(None, typed=True)
|
||||
def clean_spaces(string):
|
||||
string = string.replace('\t', ' ')
|
||||
string = re.sub('[\n|\r]', '', string)
|
||||
return re.sub(' +', ' ', string)
|
||||
Reference in New Issue
Block a user