Going online (first public version)
This commit is contained in:
Karim shoair
@@ -0,0 +1,10 @@
|
||||
# Declare top-level shortcuts
|
||||
from scrapling.parser import Adaptor, Adaptors
|
||||
from scrapling.custom_types import TextHandler, AttributesHandler
|
||||
|
||||
__author__ = "Karim Shoair (karim.shoair@pm.me)"
|
||||
__version__ = "0.1"
|
||||
__copyright__ = "Copyright (c) 2024 Karim Shoair"
|
||||
|
||||
|
||||
__all__ = ['Adaptor', 'Adaptors', 'TextHandler', 'AttributesHandler']
|
||||
@@ -0,0 +1,146 @@
|
||||
import re
|
||||
from types import MappingProxyType
|
||||
from collections.abc import Mapping
|
||||
from typing import Dict, List, Union, Pattern
|
||||
|
||||
from scrapling.utils import _is_iterable, flatten
|
||||
|
||||
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,903 @@
|
||||
import os
|
||||
from difflib import SequenceMatcher
|
||||
from typing import Any, Dict, List, Tuple, Optional, Pattern, SupportsIndex, Union, Callable, Generator
|
||||
|
||||
from scrapling.translator import HTMLTranslator
|
||||
from scrapling.mixins import SelectorsGeneration
|
||||
from scrapling.custom_types import TextHandler, AttributesHandler
|
||||
from scrapling.storage_adaptors import SQLiteStorageSystem, StorageSystemMixin, _StorageTools
|
||||
from scrapling.utils import setup_basic_logging, logging, clean_spaces, flatten, html_forbidden
|
||||
|
||||
from lxml import etree, html
|
||||
from cssselect import SelectorError, SelectorSyntaxError, parse as split_selectors
|
||||
|
||||
|
||||
class Adaptor(SelectorsGeneration):
|
||||
__slots__ = (
|
||||
'url', 'encoding', '__auto_match_enabled', '_root', '_storage', '__debug',
|
||||
'__keep_comments', '__huge_tree_enabled', '__attributes', '__text', '__tag',
|
||||
)
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
text: Optional[str] = None,
|
||||
url: Optional[str] = None,
|
||||
body: bytes = b"",
|
||||
encoding: str = "utf8",
|
||||
huge_tree: bool = True,
|
||||
root: Optional[html.HtmlElement] = None,
|
||||
keep_comments: Optional[bool] = False,
|
||||
auto_match: Optional[bool] = False,
|
||||
storage: Any = SQLiteStorageSystem,
|
||||
storage_args: Optional[Dict] = None,
|
||||
debug: Optional[bool] = True,
|
||||
):
|
||||
"""The main class that works as a wrapper for the HTML input data. Using this class, you can search for elements
|
||||
with expressions in CSS, XPath, or with simply text. Check the docs for more info.
|
||||
|
||||
Here we try to extend module ``lxml.html.HtmlElement`` while maintaining a simpler interface, We are not
|
||||
inheriting from the ``lxml.html.HtmlElement`` because it's not pickleable which makes a lot of reference jobs
|
||||
not possible. You can test it here and see code explodes with `AssertionError: invalid Element proxy at...`.
|
||||
It's an old issue with lxml, see `this entry <https://bugs.launchpad.net/lxml/+bug/736708>`
|
||||
|
||||
:param text: HTML body passed as text.
|
||||
:param url: allows storing a URL with the html data for retrieving later.
|
||||
:param body: HTML body as ``bytes`` object. It can be used instead of the ``text`` argument.
|
||||
:param encoding: The encoding type that will be used in HTML parsing, default is `UTF-8`
|
||||
:param huge_tree: Enabled by default, should always be enabled when parsing large HTML documents. This controls
|
||||
libxml2 feature that forbids parsing certain large documents to protect from possible memory exhaustion.
|
||||
:param root: Used internally to pass etree objects instead of text/body arguments, it takes highest priority.
|
||||
Don't use it unless you know what you are doing!
|
||||
:param keep_comments: While parsing the HTML body, drop comments or not. Disabled by default for obvious reasons
|
||||
:param auto_match: Globally turn-off the auto-match feature in all functions, this argument takes higher
|
||||
priority over all auto-match related arguments/functions in the class.
|
||||
:param storage: The storage class to be passed for auto-matching functionalities, see ``Docs`` for more info.
|
||||
:param storage_args: A dictionary of ``argument->value`` pairs to be passed for the storage class.
|
||||
If empty, default values will be used.
|
||||
:param debug: Enable debug mode
|
||||
"""
|
||||
if root is None and not body and text is None:
|
||||
raise ValueError("Adaptor class needs text, body, or root arguments to work")
|
||||
|
||||
if root is None:
|
||||
if text is None:
|
||||
if not body or not isinstance(body, bytes):
|
||||
raise TypeError(f"body argument must be valid and of type bytes, got {body.__class__}")
|
||||
|
||||
body = body.replace(b"\x00", b"").strip()
|
||||
else:
|
||||
if not isinstance(text, str):
|
||||
raise TypeError(f"text argument must be of type str, got {text.__class__}")
|
||||
|
||||
body = text.strip().replace("\x00", "").encode(encoding) or b"<html/>"
|
||||
|
||||
parser = html.HTMLParser(
|
||||
# https://lxml.de/api/lxml.etree.HTMLParser-class.html
|
||||
recover=True, remove_blank_text=True, remove_comments=(keep_comments is True), encoding=encoding,
|
||||
compact=True, huge_tree=huge_tree, default_doctype=True
|
||||
)
|
||||
self._root = etree.fromstring(body, parser=parser, base_url=url)
|
||||
|
||||
else:
|
||||
# All html types inherits from HtmlMixin so this to check for all at once
|
||||
if not issubclass(type(root), html.HtmlMixin):
|
||||
raise TypeError(
|
||||
f"Root have to be a valid element of `html` module types to work, not of type {type(root)}"
|
||||
)
|
||||
|
||||
self._root = root
|
||||
|
||||
setup_basic_logging(level='debug' if debug else 'info')
|
||||
self.__auto_match_enabled = auto_match
|
||||
|
||||
if self.__auto_match_enabled:
|
||||
if not storage_args:
|
||||
storage_args = {
|
||||
'storage_file': os.path.join(os.path.dirname(__file__), 'elements_storage.db'),
|
||||
'url': url
|
||||
}
|
||||
|
||||
if not hasattr(storage, '__wrapped__'):
|
||||
raise ValueError("Storage class must be wrapped with cache decorator, see docs for info")
|
||||
|
||||
if not issubclass(storage.__wrapped__, StorageSystemMixin):
|
||||
raise ValueError("Storage system must be inherited from class `StorageSystemMixin`")
|
||||
|
||||
self._storage = storage(**storage_args)
|
||||
|
||||
self.__keep_comments = keep_comments
|
||||
self.__huge_tree_enabled = huge_tree
|
||||
self.encoding = encoding
|
||||
self.url = url
|
||||
# For selector stuff
|
||||
self.__attributes = None
|
||||
self.__text = None
|
||||
self.__tag = None
|
||||
self.__debug = debug
|
||||
|
||||
# Node functionalities, I wanted to move to separate Mixin class but it had slight impact on performance
|
||||
@staticmethod
|
||||
def _is_text_node(element: Union[html.HtmlElement, etree._ElementUnicodeResult]) -> bool:
|
||||
"""Return True if given element is a result of a string expression
|
||||
Examples:
|
||||
Xpath -> '/text()', '/@attribute' etc...
|
||||
CSS3 -> '::text', '::attr(attrib)'...
|
||||
"""
|
||||
# Faster than checking `element.is_attribute or element.is_text or element.is_tail`
|
||||
return issubclass(type(element), etree._ElementUnicodeResult)
|
||||
|
||||
def __get_correct_result(
|
||||
self, element: Union[html.HtmlElement, etree._ElementUnicodeResult]
|
||||
) -> Union[TextHandler, html.HtmlElement, 'Adaptor', str]:
|
||||
"""Used internally in all functions to convert results to type (Adaptor|Adaptors) when possible"""
|
||||
if self._is_text_node(element):
|
||||
# etree._ElementUnicodeResult basically inherit from `str` so it's fine
|
||||
return TextHandler(str(element))
|
||||
else:
|
||||
if issubclass(type(element), html.HtmlMixin):
|
||||
return self.__class__(
|
||||
root=element, url=self.url, encoding=self.encoding, auto_match=self.__auto_match_enabled,
|
||||
keep_comments=self.__keep_comments, huge_tree=self.__huge_tree_enabled, debug=self.__debug
|
||||
)
|
||||
return element
|
||||
|
||||
def __convert_results(
|
||||
self, result: Union[List[html.HtmlElement], html.HtmlElement]
|
||||
) -> Union['Adaptors[Adaptor]', 'Adaptor', List, None]:
|
||||
"""Used internally in all functions to convert results to type (Adaptor|Adaptors) in bulk when possible"""
|
||||
if result is None:
|
||||
return None
|
||||
elif result == []: # Lxml will give a warning if I used something like `not result`
|
||||
return []
|
||||
|
||||
if isinstance(result, Adaptors):
|
||||
return result
|
||||
|
||||
if type(result) is list:
|
||||
results = [self.__get_correct_result(n) for n in result]
|
||||
if all(isinstance(res, self.__class__) for res in results):
|
||||
return Adaptors(results)
|
||||
return results
|
||||
|
||||
return self.__get_correct_result(result)
|
||||
|
||||
def __getstate__(self) -> Any:
|
||||
# lxml don't like it :)
|
||||
raise TypeError("Can't pickle Adaptor objects")
|
||||
|
||||
# The following four properties I made them into functions instead of variables directly
|
||||
# So they don't slow down the process of initializing many instances of the class and gets executed only
|
||||
# when the user need them for the first time for that specific element and gets cached for next times
|
||||
# Doing that only made the library performance test sky rocked multiple times faster than before
|
||||
# because I was executing them on initialization before :))
|
||||
@property
|
||||
def tag(self) -> str:
|
||||
"""Get tag name of the element"""
|
||||
if not self.__tag:
|
||||
self.__tag = self._root.tag
|
||||
return self.__tag
|
||||
|
||||
@property
|
||||
def text(self) -> TextHandler:
|
||||
"""Get text content of the element"""
|
||||
if not self.__text:
|
||||
self.__text = TextHandler(self._root.text)
|
||||
return self.__text
|
||||
|
||||
def get_all_text(self, separator: str = "\n", strip: bool = False, ignore_tags: Tuple = ('script', 'style',), valid_values: bool = True) -> TextHandler:
|
||||
"""Get all child strings of this element, concatenated using the given separator.
|
||||
|
||||
:param separator: Strings will be concatenated using this separator.
|
||||
:param strip: If True, strings will be stripped before being concatenated.
|
||||
:param ignore_tags: A tuple of all tag names you want to ignore
|
||||
:param valid_values: If enabled, elements with text-content that is empty or only whitespaces will be ignored
|
||||
|
||||
:return: A TextHandler
|
||||
"""
|
||||
_all_strings = []
|
||||
|
||||
def _traverse(node: html.HtmlElement) -> None:
|
||||
"""Traverse element children and get text content of each
|
||||
|
||||
:param node: Current node in the tree structure
|
||||
:return:
|
||||
"""
|
||||
if node.tag not in ignore_tags:
|
||||
text = node.text
|
||||
if text and type(text) is str:
|
||||
if valid_values:
|
||||
if text.strip():
|
||||
_all_strings.append(text if not strip else text.strip())
|
||||
else:
|
||||
_all_strings.append(text if not strip else text.strip())
|
||||
|
||||
for branch in node.iterchildren():
|
||||
_traverse(branch)
|
||||
|
||||
# We will start using Lxml directly for the speed boost
|
||||
_traverse(self._root)
|
||||
|
||||
return TextHandler(separator.join([s for s in _all_strings]))
|
||||
|
||||
@property
|
||||
def attrib(self) -> AttributesHandler:
|
||||
"""Get attributes of the element"""
|
||||
if not self.__attributes:
|
||||
self.__attributes = AttributesHandler(self._root.attrib)
|
||||
return self.__attributes
|
||||
|
||||
@property
|
||||
def html_content(self) -> str:
|
||||
"""Return the inner html code of the element"""
|
||||
return etree.tostring(self._root, encoding='unicode', method='html', with_tail=False)
|
||||
|
||||
body = html_content
|
||||
|
||||
def prettify(self) -> str:
|
||||
"""Return a prettified version of the element's inner html-code"""
|
||||
return etree.tostring(self._root, encoding='unicode', pretty_print=True, method='html', with_tail=False)
|
||||
|
||||
def has_class(self, class_name: str) -> bool:
|
||||
"""Check if element has a specific class
|
||||
:param class_name: The class name to check for
|
||||
:return: True if element has class with that name otherwise False
|
||||
"""
|
||||
return class_name in self._root.classes
|
||||
|
||||
@property
|
||||
def parent(self) -> Union['Adaptor', None]:
|
||||
"""Return the direct parent of the element or ``None`` otherwise"""
|
||||
return self.__convert_results(self._root.getparent())
|
||||
|
||||
@property
|
||||
def children(self) -> Union['Adaptors[Adaptor]', List]:
|
||||
"""Return the children elements of the current element or empty list otherwise"""
|
||||
return self.__convert_results(list(
|
||||
child for child in self._root.iterchildren() if type(child) not in html_forbidden
|
||||
))
|
||||
|
||||
@property
|
||||
def siblings(self) -> Union['Adaptors[Adaptor]', List]:
|
||||
"""Return other children of the current element's parent or empty list otherwise"""
|
||||
if self.parent:
|
||||
return Adaptors([child for child in self.parent.children if child._root != self._root])
|
||||
return []
|
||||
|
||||
def iterancestors(self) -> Generator['Adaptor', None, None]:
|
||||
"""Return a generator that loops over all ancestors of the element, starting with element's parent."""
|
||||
for ancestor in self._root.iterancestors():
|
||||
yield self.__convert_results(ancestor)
|
||||
|
||||
def find_ancestor(self, func: Callable[['Adaptor'], bool]) -> Union['Adaptor', None]:
|
||||
"""Loop over all ancestors of the element till one match the passed function
|
||||
:param func: A function that takes each ancestor as an argument and returns True/False
|
||||
:return: The first ancestor that match the function or ``None`` otherwise.
|
||||
"""
|
||||
for ancestor in self.iterancestors():
|
||||
if func(ancestor):
|
||||
return ancestor
|
||||
return None
|
||||
|
||||
@property
|
||||
def path(self) -> 'Adaptors[Adaptor]':
|
||||
"""Returns list of type :class:`Adaptors` that contains the path leading to the current element from the root."""
|
||||
lst = list(self.iterancestors())
|
||||
return Adaptors(lst)
|
||||
|
||||
@property
|
||||
def next(self) -> Union['Adaptor', None]:
|
||||
"""Returns the next element of the current element in the children of the parent or ``None`` otherwise."""
|
||||
next_element = self._root.getnext()
|
||||
if next_element is not None:
|
||||
while type(next_element) in html_forbidden:
|
||||
# Ignore html comments and unwanted types
|
||||
next_element = next_element.getnext()
|
||||
|
||||
return self.__convert_results(next_element)
|
||||
|
||||
@property
|
||||
def previous(self) -> Union['Adaptor', None]:
|
||||
"""Returns the previous element of the current element in the children of the parent or ``None`` otherwise."""
|
||||
prev_element = self._root.getprevious()
|
||||
if prev_element is not None:
|
||||
while type(prev_element) in html_forbidden:
|
||||
# Ignore html comments and unwanted types
|
||||
prev_element = prev_element.getprevious()
|
||||
|
||||
return self.__convert_results(prev_element)
|
||||
|
||||
def __str__(self) -> str:
|
||||
return self.html_content
|
||||
|
||||
def __repr__(self) -> str:
|
||||
length_limit = 40
|
||||
data = "<"
|
||||
content = clean_spaces(self.html_content)
|
||||
if len(content) > length_limit:
|
||||
content = content[:length_limit].strip() + '...'
|
||||
data += f"data='{content}'"
|
||||
|
||||
if self.parent:
|
||||
parent_content = clean_spaces(self.parent.html_content)
|
||||
if len(parent_content) > length_limit:
|
||||
parent_content = parent_content[:length_limit].strip() + '...'
|
||||
|
||||
data += f" parent='{parent_content}'"
|
||||
|
||||
return data + ">"
|
||||
|
||||
# From here we start the selecting functions
|
||||
def relocate(
|
||||
self, element: Union[Dict, html.HtmlElement, 'Adaptor'], percentage: int = 0, adaptor_type: bool = False
|
||||
) -> Union[List[Union[html.HtmlElement, None]], 'Adaptors']:
|
||||
"""This function will search again for the element in the page tree, used automatically on page structure change
|
||||
|
||||
:param element: The element we want to relocate in the tree
|
||||
:param percentage: The minimum percentage to accept and not going lower than that. Be aware that the percentage
|
||||
calculation depends solely on the page structure so don't play with this number unless you must know
|
||||
what you are doing!
|
||||
:param adaptor_type: If True, the return result will be converted to `Adaptors` object
|
||||
:return: List of pure HTML elements that got the highest matching score or 'Adaptors' object
|
||||
"""
|
||||
score_table = {}
|
||||
# Note: `element` will be most likely always be a dictionary at this point.
|
||||
if isinstance(element, self.__class__):
|
||||
element = element._root
|
||||
|
||||
if issubclass(type(element), html.HtmlElement):
|
||||
element = _StorageTools.element_to_dict(element)
|
||||
|
||||
# TODO: Optimize the traverse logic a bit, maybe later
|
||||
def _traverse(node: html.HtmlElement, ele: Dict) -> None:
|
||||
"""Get the matching score of the given element against the node then traverse the children
|
||||
|
||||
:param node: Current node in the tree structure
|
||||
:param ele: The element we are searching for as dictionary
|
||||
:return:
|
||||
"""
|
||||
# Hence: the code doesn't stop even if the score was 100%
|
||||
# because there might be another element(s) left in page with the same score
|
||||
score = self.__calculate_similarity_score(ele, node)
|
||||
score_table.setdefault(score, []).append(node)
|
||||
for branch in node.iterchildren():
|
||||
_traverse(branch, ele)
|
||||
|
||||
# This will block until we traverse all children/branches
|
||||
_traverse(self._root, element)
|
||||
|
||||
if score_table:
|
||||
highest_probability = max(score_table.keys())
|
||||
if score_table[highest_probability] and highest_probability >= percentage:
|
||||
logging.debug(f'Highest probability was {highest_probability}%')
|
||||
logging.debug('Top 5 best matching elements are: ')
|
||||
for percent in tuple(sorted(score_table.keys(), reverse=True))[:5]:
|
||||
logging.debug(f'{percent} -> {self.__convert_results(score_table[percent])}')
|
||||
if not adaptor_type:
|
||||
return score_table[highest_probability]
|
||||
return self.__convert_results(score_table[highest_probability])
|
||||
return []
|
||||
|
||||
def css(self, selector: str, identifier: str = '',
|
||||
auto_match: bool = False, auto_save: bool = False, percentage: int = 0
|
||||
) -> Union['Adaptors[Adaptor]', List]:
|
||||
"""Search current tree with CSS3 selectors
|
||||
|
||||
**Important:
|
||||
It's recommended to use the identifier argument if you plan to use different selector later
|
||||
and want to relocate the same element(s)**
|
||||
|
||||
:param selector: The CSS3 selector to be used.
|
||||
:param auto_match: Enabled will make function try to relocate the element if it was 'saved' before
|
||||
:param identifier: A string that will be used to save/retrieve element's data in auto-matching
|
||||
otherwise the selector will be used.
|
||||
:param auto_save: Automatically save new elements for `auto_match` later
|
||||
:param percentage: The minimum percentage to accept while auto-matching and not going lower than that.
|
||||
Be aware that the percentage calculation depends solely on the page structure so don't play with this
|
||||
number unless you must know what you are doing!
|
||||
|
||||
:return: List as :class:`Adaptors`
|
||||
"""
|
||||
try:
|
||||
if not self.__auto_match_enabled:
|
||||
# No need to split selectors in this case, let's save some CPU cycles :)
|
||||
xpath_selector = HTMLTranslator().css_to_xpath(selector)
|
||||
return self.xpath(xpath_selector, identifier or selector, auto_match, auto_save, percentage)
|
||||
|
||||
results = []
|
||||
if ',' in selector:
|
||||
for single_selector in split_selectors(selector):
|
||||
# I'm doing this only so the `save` function save data correctly for combined selectors
|
||||
# Like using the ',' to combine two different selectors that point to different elements.
|
||||
xpath_selector = HTMLTranslator().css_to_xpath(single_selector.canonical())
|
||||
results += self.xpath(
|
||||
xpath_selector, identifier or single_selector.canonical(), auto_match, auto_save, percentage
|
||||
)
|
||||
else:
|
||||
xpath_selector = HTMLTranslator().css_to_xpath(selector)
|
||||
return self.xpath(xpath_selector, identifier or selector, auto_match, auto_save, percentage)
|
||||
|
||||
return self.__convert_results(results)
|
||||
except (SelectorError, SelectorSyntaxError,):
|
||||
raise SelectorSyntaxError(f"Invalid CSS selector: {selector}")
|
||||
|
||||
def xpath(self, selector: str, identifier: str = '',
|
||||
auto_match: bool = False, auto_save: bool = False, percentage: int = 0, **kwargs: Any
|
||||
) -> Union['Adaptors[Adaptor]', List]:
|
||||
"""Search current tree with XPath selectors
|
||||
|
||||
**Important:
|
||||
It's recommended to use the identifier argument if you plan to use different selector later
|
||||
and want to relocate the same element(s)**
|
||||
|
||||
Note: **Additional keyword arguments will be passed as XPath variables in the XPath expression!**
|
||||
|
||||
:param selector: The XPath selector to be used.
|
||||
:param auto_match: Enabled will make function try to relocate the element if it was 'saved' before
|
||||
:param identifier: A string that will be used to save/retrieve element's data in auto-matching
|
||||
otherwise the selector will be used.
|
||||
:param auto_save: Automatically save new elements for `auto_match` later
|
||||
:param percentage: The minimum percentage to accept while auto-matching and not going lower than that.
|
||||
Be aware that the percentage calculation depends solely on the page structure so don't play with this
|
||||
number unless you must know what you are doing!
|
||||
|
||||
:return: List as :class:`Adaptors`
|
||||
"""
|
||||
try:
|
||||
selected_elements = self._root.xpath(selector, **kwargs)
|
||||
|
||||
if selected_elements:
|
||||
if not self.__auto_match_enabled and auto_save:
|
||||
logging.warning("Argument `auto_save` will be ignored because `auto_match` wasn't enabled on initialization. Check docs for more info.")
|
||||
|
||||
elif self.__auto_match_enabled and auto_save:
|
||||
self.save(selected_elements[0], identifier or selector)
|
||||
|
||||
return self.__convert_results(selected_elements)
|
||||
else:
|
||||
if self.__auto_match_enabled and auto_match:
|
||||
element_data = self.retrieve(identifier or selector)
|
||||
if element_data:
|
||||
relocated = self.relocate(element_data, percentage)
|
||||
if relocated is not None and auto_save:
|
||||
self.save(relocated[0], identifier or selector)
|
||||
|
||||
return self.__convert_results(relocated)
|
||||
else:
|
||||
return self.__convert_results(selected_elements)
|
||||
|
||||
elif not self.__auto_match_enabled and auto_match:
|
||||
logging.warning("Argument `auto_match` will be ignored because `auto_match` wasn't enabled on initialization. Check docs for more info.")
|
||||
|
||||
return self.__convert_results(selected_elements)
|
||||
|
||||
except (SelectorError, SelectorSyntaxError, etree.XPathError, etree.XPathEvalError):
|
||||
raise SelectorSyntaxError(f"Invalid XPath selector: {selector}")
|
||||
|
||||
def __calculate_similarity_score(self, original: Dict, candidate: html.HtmlElement) -> float:
|
||||
"""Used internally to calculate a score that shows how candidate element similar to the original one
|
||||
|
||||
:param original: The original element in the form of the dictionary generated from `element_to_dict` function
|
||||
:param candidate: The element to compare with the original element.
|
||||
:return: A percentage score of how similar is the candidate to the original element
|
||||
"""
|
||||
score, checks = 0, 0
|
||||
candidate = _StorageTools.element_to_dict(candidate)
|
||||
|
||||
# Possible TODO:
|
||||
# Study the idea of giving weight to each test below so some are more important than others
|
||||
# Current results: With weights some websites had better score while it was worse for others
|
||||
score += 1 if original['tag'] == candidate['tag'] else 0 # * 0.3 # 30%
|
||||
checks += 1
|
||||
|
||||
if original['text']:
|
||||
score += SequenceMatcher(None, original['text'], candidate.get('text') or '').ratio() # * 0.3 # 30%
|
||||
checks += 1
|
||||
|
||||
# if both doesn't have attributes, it still count for something!
|
||||
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
|
||||
for attrib in ('class', 'id', 'href', 'src',):
|
||||
if original['attributes'].get(attrib):
|
||||
score += SequenceMatcher(
|
||||
None, original['attributes'][attrib], candidate['attributes'].get(attrib) or ''
|
||||
).ratio() # * 0.3 # 30%
|
||||
checks += 1
|
||||
|
||||
score += SequenceMatcher(None, original['path'], candidate['path']).ratio() # * 0.1 # 10%
|
||||
checks += 1
|
||||
|
||||
if original.get('parent_name'):
|
||||
# Then we start comparing parents' data
|
||||
if candidate.get('parent_name'):
|
||||
score += SequenceMatcher(
|
||||
None, original['parent_name'], candidate.get('parent_name') or ''
|
||||
).ratio() # * 0.2 # 20%
|
||||
checks += 1
|
||||
|
||||
score += self.__calculate_dict_diff(
|
||||
original['parent_attribs'], candidate.get('parent_attribs') or {}
|
||||
) # * 0.2 # 20%
|
||||
checks += 1
|
||||
|
||||
if original['parent_text']:
|
||||
score += SequenceMatcher(
|
||||
None, original['parent_text'], candidate.get('parent_text') or ''
|
||||
).ratio() # * 0.1 # 10%
|
||||
checks += 1
|
||||
# else:
|
||||
# # The original element have a parent and this one not, this is not a good sign
|
||||
# score -= 0.1
|
||||
|
||||
if original.get('siblings'):
|
||||
score += SequenceMatcher(
|
||||
None, original['siblings'], candidate.get('siblings') or []
|
||||
).ratio() # * 0.1 # 10%
|
||||
checks += 1
|
||||
|
||||
# How % sure? let's see
|
||||
return round((score / checks) * 100, 2)
|
||||
|
||||
@staticmethod
|
||||
def __calculate_dict_diff(dict1: dict, dict2: dict) -> float:
|
||||
"""Used internally 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
|
||||
return score
|
||||
|
||||
def save(self, element: Union['Adaptor', 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, it can be a `Adaptor` or pure `HtmlElement`
|
||||
:param identifier: This is the identifier that will be used to retrieve the element later from the storage. See
|
||||
the docs for more info.
|
||||
"""
|
||||
if self.__auto_match_enabled:
|
||||
if isinstance(element, self.__class__):
|
||||
element = element._root
|
||||
|
||||
if self._is_text_node(element):
|
||||
element = element.getparent()
|
||||
|
||||
self._storage.save(element, identifier)
|
||||
else:
|
||||
logging.critical(
|
||||
"Can't use Auto-match features with disabled globally, you have to start a new class instance."
|
||||
)
|
||||
|
||||
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
|
||||
"""
|
||||
if self.__auto_match_enabled:
|
||||
return self._storage.retrieve(identifier)
|
||||
|
||||
logging.critical(
|
||||
"Can't use Auto-match features with disabled globally, you have to start a new class instance."
|
||||
)
|
||||
|
||||
# Operations on text functions
|
||||
def json(self) -> Dict:
|
||||
"""Return json response if the response is jsonable otherwise throws error"""
|
||||
return self.text.json()
|
||||
|
||||
def re(self, regex: Union[str, Pattern[str]], replace_entities: bool = True) -> 'List[str]':
|
||||
"""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
|
||||
"""
|
||||
return self.text.re(regex, replace_entities)
|
||||
|
||||
def re_first(self, regex: Union[str, Pattern[str]], default=None, replace_entities: bool = True):
|
||||
"""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
|
||||
|
||||
"""
|
||||
return self.text.re_first(regex, default, replace_entities)
|
||||
|
||||
def find_similar(
|
||||
self,
|
||||
similarity_threshold: float = 0.2,
|
||||
ignore_attributes: Union[List, Tuple] = ('href', 'src',),
|
||||
match_text: bool = False
|
||||
) -> Union['Adaptors[Adaptor]', List]:
|
||||
"""Find elements that are in the same tree depth in the page with the same tag name and same parent tag etc...
|
||||
then return the ones that match the current element attributes with percentage higher than the input threshold.
|
||||
|
||||
This function is inspired by AutoScraper and made for cases where you, for example, found a product div inside
|
||||
a products-list container and want to find other products using that that element as a starting point EXCEPT
|
||||
this function works in any case without depending on the element type.
|
||||
|
||||
:param similarity_threshold: The percentage to use while comparing elements attributes.
|
||||
Note: Elements found before attributes matching/comparison will be sharing the same depth, same tag name,
|
||||
same parent tag name, and same grand parent tag name. So they are 99% likely to be correct unless your are
|
||||
extremely unlucky then attributes matching comes into play so basically don't play with this number unless
|
||||
you are getting the results you don't want.
|
||||
Also, if current element doesn't have attributes and the similar element as well, then it's a 100% match.
|
||||
:param ignore_attributes: Attribute names passed will be ignored while matching the attributes in last step.
|
||||
The default value is to ignore `href` and `src` as URLs can change a lot between elements so it's unreliable
|
||||
:param match_text: If True, elements text content will be taken into calculation while matching.
|
||||
Not recommended to use in normal cases but it depends.
|
||||
|
||||
:return: A ``Adaptors`` container of ``Adaptor`` objects or empty list
|
||||
"""
|
||||
def get_attributes(element: html.HtmlElement) -> Dict:
|
||||
"""Return attributes dictionary without the ignored list"""
|
||||
return {k: v for k, v in element.attrib.items() if k not in ignore_attributes}
|
||||
|
||||
def are_alike(original: html.HtmlElement, original_attributes: Dict, candidate: html.HtmlElement) -> bool:
|
||||
"""Calculate a score of how much these elements are alike and return True
|
||||
if score is higher or equal the threshold"""
|
||||
candidate_attributes = get_attributes(candidate) if ignore_attributes else candidate.attrib
|
||||
score, checks = 0, 0
|
||||
|
||||
if original_attributes:
|
||||
score += sum(
|
||||
SequenceMatcher(None, v, candidate_attributes.get(k, '')).ratio()
|
||||
for k, v in original_attributes.items()
|
||||
)
|
||||
checks += len(candidate_attributes)
|
||||
else:
|
||||
if not candidate_attributes:
|
||||
# Both doesn't have attributes, this must mean something
|
||||
score += 1
|
||||
checks += 1
|
||||
|
||||
if match_text:
|
||||
score += SequenceMatcher(
|
||||
None, clean_spaces(original.text or ''), clean_spaces(candidate.text or '')
|
||||
).ratio()
|
||||
checks += 1
|
||||
|
||||
if checks:
|
||||
return round(score / checks, 2) >= similarity_threshold
|
||||
return False
|
||||
|
||||
# We will use the elements root from now on to get the speed boost of using Lxml directly
|
||||
root = self._root
|
||||
current_depth = len(list(root.iterancestors()))
|
||||
target_attrs = get_attributes(root) if ignore_attributes else root.attrib
|
||||
similar_elements = list()
|
||||
# + root.xpath(f"//{self.tag}[count(ancestor::*) = {current_depth-1}]")
|
||||
parent = root.getparent()
|
||||
if parent is not None:
|
||||
grandparent = parent.getparent() # lol
|
||||
if grandparent is not None:
|
||||
potential_matches = root.xpath(
|
||||
f"//{grandparent.tag}/{parent.tag}/{self.tag}[count(ancestor::*) = {current_depth}]"
|
||||
)
|
||||
else:
|
||||
potential_matches = root.xpath(f"//{parent.tag}/{self.tag}[count(ancestor::*) = {current_depth}]")
|
||||
else:
|
||||
potential_matches = root.xpath(f"//{self.tag}[count(ancestor::*) = {current_depth}]")
|
||||
|
||||
for potential_match in potential_matches:
|
||||
if potential_match != root and are_alike(root, target_attrs, potential_match):
|
||||
similar_elements.append(potential_match)
|
||||
|
||||
return self.__convert_results(similar_elements)
|
||||
|
||||
def find_by_text(
|
||||
self, text: str, first_match: bool = True, partial: bool = False,
|
||||
case_sensitive: bool = False, clean_match: bool = True
|
||||
) -> Union['Adaptors[Adaptor]', 'Adaptor', List]:
|
||||
"""Find elements that its text content fully/partially matches input.
|
||||
:param text: Text query to match
|
||||
:param first_match: Return first element that matches conditions, enabled by default
|
||||
:param partial: If enabled, function return elements that contains the input text
|
||||
:param case_sensitive: if enabled, letters case will be taken into consideration
|
||||
:param clean_match: if enabled, this will ignore all whitespaces and consecutive spaces while matching
|
||||
"""
|
||||
|
||||
results = []
|
||||
if not case_sensitive:
|
||||
text = text.lower()
|
||||
|
||||
def _traverse(node: Adaptor) -> None:
|
||||
"""Check if element matches given text otherwise, traverse the children tree and iterate"""
|
||||
node_text = node.text
|
||||
# if there's already no text in this node, dodge it to save CPU cycles and time
|
||||
if node_text:
|
||||
if clean_match:
|
||||
node_text = node_text.clean()
|
||||
|
||||
if not case_sensitive:
|
||||
node_text = node_text.lower()
|
||||
|
||||
if partial:
|
||||
if text in node_text:
|
||||
results.append(node)
|
||||
elif text == node_text:
|
||||
results.append(node)
|
||||
|
||||
if results and first_match:
|
||||
# we got an element so we should stop
|
||||
return
|
||||
|
||||
for branch in node.children:
|
||||
_traverse(branch)
|
||||
|
||||
# This will block until we traverse all children/branches
|
||||
_traverse(self)
|
||||
|
||||
if first_match:
|
||||
if results:
|
||||
return results[0]
|
||||
return self.__convert_results(results)
|
||||
|
||||
def find_by_regex(
|
||||
self, query: str, first_match: bool = True, case_sensitive: bool = False, clean_match: bool = True
|
||||
) -> Union['Adaptors[Adaptor]', 'Adaptor', List]:
|
||||
"""Find elements that its text content matches the input regex pattern.
|
||||
:param query: Regex query to match
|
||||
:param first_match: Return first element that matches conditions, enabled by default
|
||||
:param case_sensitive: if enabled, letters case will be taken into consideration in the regex
|
||||
:param clean_match: if enabled, this will ignore all whitespaces and consecutive spaces while matching
|
||||
"""
|
||||
results = []
|
||||
|
||||
def _traverse(node: Adaptor) -> None:
|
||||
"""Check if element matches given regex otherwise, traverse the children tree and iterate"""
|
||||
node_text = node.text
|
||||
# if there's already no text in this node, dodge it to save CPU cycles and time
|
||||
if node_text:
|
||||
if node_text.re(query, check_match=True, clean_match=clean_match, case_sensitive=case_sensitive):
|
||||
results.append(node)
|
||||
|
||||
if results and first_match:
|
||||
# we got an element so we should stop
|
||||
return
|
||||
|
||||
for branch in node.children:
|
||||
_traverse(branch)
|
||||
|
||||
# This will block until we traverse all children/branches
|
||||
_traverse(self)
|
||||
|
||||
if results and first_match:
|
||||
return results[0]
|
||||
return self.__convert_results(results)
|
||||
|
||||
|
||||
class Adaptors(List[Adaptor]):
|
||||
"""
|
||||
The :class:`Adaptors` class is a subclass of the builtin ``List`` class, which provides a few additional methods.
|
||||
"""
|
||||
__slots__ = ()
|
||||
|
||||
def __getitem__(self, pos: Union[SupportsIndex, slice]) -> Union[Adaptor, "Adaptors[Adaptor]"]:
|
||||
lst = super().__getitem__(pos)
|
||||
if isinstance(pos, slice):
|
||||
return self.__class__(lst)
|
||||
else:
|
||||
return lst
|
||||
|
||||
def xpath(
|
||||
self, selector: str, identifier: str = '', auto_save: bool = False, percentage: int = 0, **kwargs: Any
|
||||
) -> Union["Adaptors[Adaptor]", List]:
|
||||
"""
|
||||
Call the ``.xpath()`` method for each element in this list and return
|
||||
their results as another :class:`Adaptors`.
|
||||
|
||||
**Important:
|
||||
It's recommended to use the identifier argument if you plan to use different selector later
|
||||
and want to relocate the same element(s)**
|
||||
|
||||
Note: **Additional keyword arguments will be passed as XPath variables in the XPath expression!**
|
||||
|
||||
:param selector: The XPath selector to be used.
|
||||
:param identifier: A string that will be used to retrieve element's data in auto-matching
|
||||
otherwise the selector will be used.
|
||||
:param auto_save: Automatically save new elements for `auto_match` later
|
||||
:param percentage: The minimum percentage to accept while auto-matching and not going lower than that.
|
||||
Be aware that the percentage calculation depends solely on the page structure so don't play with this
|
||||
number unless you must know what you are doing!
|
||||
|
||||
:return: List as :class:`Adaptors`
|
||||
"""
|
||||
results = [
|
||||
n.xpath(selector, identifier or selector, False, auto_save, percentage, **kwargs) for n in self
|
||||
]
|
||||
return self.__class__(flatten(results))
|
||||
|
||||
def css(self, selector: str, identifier: str = '', auto_save: bool = False, percentage: int = 0) -> Union["Adaptors[Adaptor]", List]:
|
||||
"""
|
||||
Call the ``.css()`` method for each element in this list and return
|
||||
their results flattened as another :class:`Adaptors`.
|
||||
|
||||
**Important:
|
||||
It's recommended to use the identifier argument if you plan to use different selector later
|
||||
and want to relocate the same element(s)**
|
||||
|
||||
:param selector: The CSS3 selector to be used.
|
||||
:param identifier: A string that will be used to retrieve element's data in auto-matching
|
||||
otherwise the selector will be used.
|
||||
:param auto_save: Automatically save new elements for `auto_match` later
|
||||
:param percentage: The minimum percentage to accept while auto-matching and not going lower than that.
|
||||
Be aware that the percentage calculation depends solely on the page structure so don't play with this
|
||||
number unless you must know what you are doing!
|
||||
|
||||
:return: List as :class:`Adaptors`
|
||||
"""
|
||||
results = [
|
||||
n.css(selector, identifier or selector, False, auto_save, percentage) for n in self
|
||||
]
|
||||
return self.__class__(flatten(results))
|
||||
|
||||
def re(self, regex: Union[str, Pattern[str]], replace_entities: bool = True) -> 'List[str]':
|
||||
"""Call the ``.re()`` method for each element in this list and return
|
||||
their results flattened as List of TextHandler.
|
||||
|
||||
: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
|
||||
"""
|
||||
results = [
|
||||
n.text.re(regex, replace_entities) for n in self
|
||||
]
|
||||
return flatten(results)
|
||||
|
||||
def re_first(self, regex: Union[str, Pattern[str]], default=None, replace_entities: bool = True):
|
||||
"""Call the ``.re_first()`` method for each element in this list and return
|
||||
their results flattened as List of TextHandler.
|
||||
|
||||
: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
|
||||
|
||||
"""
|
||||
results = [
|
||||
n.text.re_first(regex, default, replace_entities) for n in self
|
||||
]
|
||||
return flatten(results)
|
||||
|
||||
# def __getattr__(self, name):
|
||||
# if name in dir(self.__class__):
|
||||
# return super().__getattribute__(name)
|
||||
#
|
||||
# # Execute the method itself on each Adaptor
|
||||
# results = []
|
||||
# for item in self:
|
||||
# results.append(getattr(item, name))
|
||||
#
|
||||
# if all(callable(r) for r in results):
|
||||
# def call_all(*args, **kwargs):
|
||||
# final_results = [r(*args, **kwargs) for r in results]
|
||||
# if all([isinstance(r, (Adaptor, Adaptors,)) for r in results]):
|
||||
# return self.__class__(final_results)
|
||||
# return final_results
|
||||
#
|
||||
# return call_all
|
||||
# else:
|
||||
# # Flatten the result if it's a single-item list containing a list
|
||||
# if len(self) == 1 and isinstance(results[0], list):
|
||||
# return self.__class__(results[0])
|
||||
# return self.__class__(results)
|
||||
|
||||
def get(self, default=None):
|
||||
"""Returns the first item of the current list
|
||||
:param default: the default value to return if the current list is empty
|
||||
"""
|
||||
return self[0] if len(self) > 0 else default
|
||||
|
||||
@property
|
||||
def first(self):
|
||||
"""Returns the first item of the current list or `None` if the list is empty"""
|
||||
return self.get()
|
||||
|
||||
@property
|
||||
def last(self):
|
||||
"""Returns the last item of the current list or `None` if the list is empty"""
|
||||
return self[-1] if len(self) > 0 else None
|
||||
|
||||
def __getstate__(self) -> Any:
|
||||
# lxml don't like it :)
|
||||
raise TypeError("Can't pickle Adaptors object")
|
||||
@@ -0,0 +1,149 @@
|
||||
import orjson
|
||||
import sqlite3
|
||||
import logging
|
||||
import threading
|
||||
from hashlib import sha256
|
||||
from abc import ABC, abstractmethod
|
||||
from typing import Dict, Optional, Union
|
||||
|
||||
from scrapling.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
|
||||
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
|
||||
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,148 @@
|
||||
"""
|
||||
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 typing import TYPE_CHECKING, Any, Optional, Protocol
|
||||
|
||||
from scrapling.utils import cache
|
||||
|
||||
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
|
||||
|
||||
if TYPE_CHECKING:
|
||||
# typing.Self requires Python 3.11
|
||||
from typing_extensions import Self
|
||||
|
||||
|
||||
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,164 @@
|
||||
import re
|
||||
import os
|
||||
import logging
|
||||
from itertools import chain
|
||||
from logging import handlers
|
||||
# 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 typing 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,))
|
||||
|
||||
|
||||
@cache(None, typed=True)
|
||||
class _Logger(object):
|
||||
# I will leave this class here for now in case I decide I want to come back to use it :)
|
||||
__slots__ = ('console_logger', 'logger_file_path',)
|
||||
levels = {
|
||||
'debug': logging.DEBUG,
|
||||
'info': logging.INFO,
|
||||
'warning': logging.WARNING,
|
||||
'error': logging.ERROR,
|
||||
'critical': logging.CRITICAL
|
||||
}
|
||||
|
||||
def __init__(self, filename: str = 'debug.log', level: str = 'debug', when: str = 'midnight', backcount: int = 1):
|
||||
os.makedirs(os.path.join(os.path.dirname(__file__), 'logs'), exist_ok=True)
|
||||
format_str = logging.Formatter("[%(asctime)s] %(levelname)s: %(message)s", "%Y-%m-%d %H:%M:%S")
|
||||
|
||||
# on-screen output
|
||||
lvl = self.levels[level.lower()]
|
||||
self.console_logger = logging.getLogger('Scrapling')
|
||||
self.console_logger.setLevel(lvl)
|
||||
console_handler = logging.StreamHandler()
|
||||
console_handler.setLevel(lvl)
|
||||
console_handler.setFormatter(format_str)
|
||||
self.console_logger.addHandler(console_handler)
|
||||
|
||||
if lvl == logging.DEBUG:
|
||||
filename = os.path.join(os.path.dirname(__file__), 'logs', filename)
|
||||
self.logger_file_path = filename
|
||||
# Automatically generates the logging file at specified intervals
|
||||
file_handler = handlers.TimedRotatingFileHandler(
|
||||
# If more than (backcount+1) existed, oldest logs will be deleted
|
||||
filename=filename, when=when, backupCount=backcount, encoding='utf-8'
|
||||
)
|
||||
file_handler.setLevel(lvl)
|
||||
file_handler.setFormatter(format_str)
|
||||
# This for the logger when it appends the date to the new log
|
||||
file_handler.namer = lambda name: name.replace(".log", "") + ".log"
|
||||
self.console_logger.addHandler(file_handler)
|
||||
self.debug(f'Debug log path: {self.logger_file_path}')
|
||||
else:
|
||||
self.logger_file_path = None
|
||||
|
||||
def debug(self, message: str) -> None:
|
||||
self.console_logger.debug(message)
|
||||
|
||||
def info(self, message: str) -> None:
|
||||
self.console_logger.info(message)
|
||||
|
||||
def warning(self, message: str) -> None:
|
||||
self.console_logger.warning(message)
|
||||
|
||||
def error(self, message: str) -> None:
|
||||
self.console_logger.error(message)
|
||||
|
||||
def critical(self, message: str) -> None:
|
||||
self.console_logger.critical(message)
|
||||
|
||||
|
||||
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
|
||||
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