diff --git a/scrapling/spiders/links.py b/scrapling/spiders/links.py new file mode 100644 index 0000000..7b9a4ff --- /dev/null +++ b/scrapling/spiders/links.py @@ -0,0 +1,295 @@ +"""Pure URL discovery primitive""" + +import re +from urllib.parse import urlsplit + +from w3lib.html import strip_html5_whitespace +from w3lib.url import canonicalize_url, safe_url_string + +from scrapling.core._types import ( + TYPE_CHECKING, + Iterable, + Callable, + List, + Optional, + Pattern, + Set, + Tuple, + Union, + Any, +) +from scrapling.core.utils import log + +if TYPE_CHECKING: + from scrapling.engines.toolbelt.custom import Response + + +__all__ = ["LinkExtractor"] +valid_schemas = {"http", "https", "file"} + + +IGNORED_EXTENSIONS = { + # archives + "7z", + "7zip", + "bz2", + "rar", + "tar", + "tar.gz", + "xz", + "zip", + # images + "mng", + "pct", + "bmp", + "gif", + "jpg", + "jpeg", + "png", + "pst", + "psp", + "tif", + "tiff", + "ai", + "drw", + "dxf", + "eps", + "ps", + "svg", + "cdr", + "ico", + "webp", + # audio + "mp3", + "wma", + "ogg", + "wav", + "ra", + "aac", + "mid", + "au", + "aiff", + # video + "3gp", + "asf", + "asx", + "avi", + "mov", + "mp4", + "mpg", + "qt", + "rm", + "swf", + "wmv", + "m4a", + "m4v", + "flv", + "webm", + # office suites + "xls", + "xlsm", + "xlsx", + "xltm", + "xltx", + "potm", + "potx", + "ppt", + "pptm", + "pptx", + "pps", + "doc", + "docb", + "docm", + "docx", + "dotm", + "dotx", + "odt", + "ods", + "odg", + "odp", + # other + "css", + "pdf", + "exe", + "bin", + "rss", + "dmg", + "iso", + "apk", + "jar", + "sh", + "rb", + "js", + "hta", + "bat", + "cpl", + "msi", + "msp", + "py", +} + + +PatternInput = Iterable[Union[str, Pattern[str]]] +StrOrIterable = Union[str, Iterable[str]] + + +def _to_str_tuple(value: StrOrIterable) -> Tuple[str, ...]: + if not value: + return () + if isinstance(value, str): + return (value,) + return tuple(value) + + +def _compile_patterns(patterns: Union[str, Pattern[str], PatternInput, None]) -> Tuple[Pattern[str], ...]: + if not patterns: + return () + if isinstance(patterns, (str, re.Pattern)): + patterns = (patterns,) + return tuple(p if isinstance(p, re.Pattern) else re.compile(p) for p in patterns) + + +def _url_extension(url: str) -> str: + path = urlsplit(url).path + _, _, last = path.rpartition("/") + if "." not in last: + return "" + return last.rsplit(".", 1)[1].lower() + + +def _filler(x): + return x + + +class LinkExtractor: + """Extracts and filters URLs from a `Response` (or a single URL via `matches`). + + All matching is regex-based; allow/deny patterns can be plain strings (compiled + with `re.compile`) or pre-compiled `re.Pattern` objects, individually or as an + iterable. + + :param allow: Regex pattern(s) URLs must match to be kept. String, compiled `re.Pattern`, + or an iterable of either. Empty means match all. + :param deny: Regex pattern(s) URLs must NOT match. Takes precedence over `allow`. + :param allow_domains: Domain(s) to keep. Matches the exact host or any subdomain + (e.g. `"example.com"` matches `"api.example.com"`). String or iterable. + :param deny_domains: Domain(s) to exclude. Same matching rules as `allow_domains`. + :param restrict_css: CSS selectors to scope DOM extraction to. Empty means whole page. + :param restrict_xpath: XPath selectors to scope DOM extraction to. Empty means whole page. + :param tags: Element tags to look for links in. Default ("a", "area"). + :param attrs: Attributes on those tags to read URLs from. Default ("href",). + :param canonicalize: Canonicalize URLs (sort query params, normalize path). Default True. + :param strip: Strip whitespace from extracted URLs. Default True. + :param keep_fragment: Preserve the URL fragment when canonicalizing. Default False. + :param deny_extensions: File extensions to drop. Default `IGNORED_EXTENSIONS`. + :param process: A function to do a process on the values extracted before using them. Return None to drop any value. + """ + + def __init__( + self, + allow: Union[str, Pattern[str], PatternInput] = (), + deny: Union[str, Pattern[str], PatternInput] = (), + allow_domains: StrOrIterable = (), + deny_domains: StrOrIterable = (), + restrict_css: StrOrIterable = (), + restrict_xpath: StrOrIterable = (), + tags: Iterable[str] = ("a", "area"), + attrs: Iterable[str] = ("href",), + canonicalize: bool = True, + strip: bool = True, + keep_fragment: bool = False, + deny_extensions: Optional[Iterable[str]] = None, + process: Callable[[Any], Any] | None = None, + ) -> None: + self.allow: Tuple[Pattern[str], ...] = _compile_patterns(allow) + self.deny: Tuple[Pattern[str], ...] = _compile_patterns(deny) + self.allow_domains: Tuple[str, ...] = tuple(d.lower() for d in _to_str_tuple(allow_domains)) + self.deny_domains: Tuple[str, ...] = tuple(d.lower() for d in _to_str_tuple(deny_domains)) + self.restrict_css: Tuple[str, ...] = _to_str_tuple(restrict_css) + self.restrict_xpath: Tuple[str, ...] = _to_str_tuple(restrict_xpath) + self.tags: Tuple[str, ...] = tuple(tags) + self.attrs: Tuple[str, ...] = tuple(attrs) + self.canonicalize = canonicalize + self.strip = strip + self.keep_fragment = keep_fragment + self.deny_extensions: Set[str] = set( + (ext.lower().lstrip(".") for ext in deny_extensions) if deny_extensions is not None else IGNORED_EXTENSIONS + ) + self.process: Callable[[Any], Any] = process if callable(process) else _filler + + def extract(self, response: "Response") -> List[str]: + """Return absolute, filtered, deduped URLs from `response`.""" + scopes: List[Any] = [] + if self.restrict_xpath: + for xp in self.restrict_xpath: + scopes.extend(response.xpath(xp)) + if self.restrict_css: + for cs in self.restrict_css: + scopes.extend(response.css(cs)) + if not scopes: + scopes = [response] + + out: List[str] = [] + search_selector = "| ".join([f".//{tag}/@{attr}" for tag in self.tags for attr in self.attrs]) + for scope in scopes: + for url in scope._root.xpath(search_selector): + if not url: + continue + url = str(url) + if self.strip: + url = strip_html5_whitespace(url) + if not url: + continue + url = str(response.urljoin(url)) + url = self.process(url) + if not url: + continue + + if self.canonicalize: + url = canonicalize_url(url, keep_fragments=self.keep_fragment) + + try: + url = safe_url_string(url, encoding=response.encoding) + except ValueError: + log.debug(f"Skipping the extraction of bad URL {url!r}") + continue + + if not self._url_passes(url): + continue + + out.append(url) + + # Switching to dict for deduplication instead of Set will keep the insertion order of the links. + return list(dict.fromkeys(out)) + + def matches(self, url: str) -> bool: + """URL-only filter (no response extraction). + + Applies allow/deny/allow_domains/deny_domains/deny_extensions to a single URL. + Used by `SitemapSpider` to dispatch sitemap URLs through `CrawlRule`s without + needing a `Response`. + """ + if self.canonicalize: + url = canonicalize_url(url, keep_fragments=self.keep_fragment) + return self._url_passes(url) + + def _url_passes(self, url: str) -> bool: + if url.split("://", 1)[0] not in valid_schemas: + return False + + ext = _url_extension(url) + if ext and ext in self.deny_extensions: + return False + + if self.allow and not any(p.search(url) for p in self.allow): + return False + if self.deny and any(p.search(url) for p in self.deny): + return False + + if self.allow_domains or self.deny_domains: + host = (urlsplit(url).hostname or "").lower() + if self.allow_domains and not any(host == d or host.endswith("." + d) for d in self.allow_domains): + return False + if self.deny_domains and any(host == d or host.endswith("." + d) for d in self.deny_domains): + return False + return True