feat(spiders): Add pure URL discovery primitive
This commit is contained in:
Karim shoair
@@ -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
|
||||
Reference in New Issue
Block a user