From c76dfc383f34514a5a2841d2e32ac4e0c8c751a3 Mon Sep 17 00:00:00 2001 From: Andrey Rakhmatullin Date: Sun, 5 May 2024 22:52:15 +0500 Subject: [PATCH 1/2] Full typing for scrapy/linkextractors. --- scrapy/commands/bench.py | 3 +- scrapy/linkextractors/__init__.py | 8 +- scrapy/linkextractors/lxmlhtml.py | 149 +++++++++++++++++------------- 3 files changed, 91 insertions(+), 69 deletions(-) diff --git a/scrapy/commands/bench.py b/scrapy/commands/bench.py index 2e6bb5d86..7523f3cfe 100644 --- a/scrapy/commands/bench.py +++ b/scrapy/commands/bench.py @@ -8,7 +8,7 @@ from urllib.parse import urlencode import scrapy from scrapy import Request from scrapy.commands import ScrapyCommand -from scrapy.http import Response +from scrapy.http import Response, TextResponse from scrapy.linkextractors import LinkExtractor @@ -61,5 +61,6 @@ class _BenchSpider(scrapy.Spider): return [scrapy.Request(url, dont_filter=True)] def parse(self, response: Response) -> Any: # type: ignore[override] + assert isinstance(Response, TextResponse) for link in self.link_extractor.extract_links(response): yield scrapy.Request(link.url, callback=self.parse) diff --git a/scrapy/linkextractors/__init__.py b/scrapy/linkextractors/__init__.py index 73a63651c..38dbe8135 100644 --- a/scrapy/linkextractors/__init__.py +++ b/scrapy/linkextractors/__init__.py @@ -7,6 +7,7 @@ For more info see docs/topics/link-extractors.rst """ import re +from typing import Iterable # common file extensions that are not followed if they occur in links IGNORED_EXTENSIONS = [ @@ -110,14 +111,11 @@ IGNORED_EXTENSIONS = [ ] -_re_type = type(re.compile("", 0)) - - -def _matches(url, regexs): +def _matches(url: str, regexs: Iterable[re.Pattern[str]]) -> bool: return any(r.search(url) for r in regexs) -def _is_valid_url(url): +def _is_valid_url(url: str) -> bool: return url.split("://", 1)[0] in {"http", "https", "file", "ftp"} diff --git a/scrapy/linkextractors/lxmlhtml.py b/scrapy/linkextractors/lxmlhtml.py index 71c6d08fc..3fa7d1e3a 100644 --- a/scrapy/linkextractors/lxmlhtml.py +++ b/scrapy/linkextractors/lxmlhtml.py @@ -5,21 +5,19 @@ Link extractor based on lxml.html import logging import operator from functools import partial +from typing import Any, Callable, Iterable, List, Optional, Set, Tuple, Union, cast from urllib.parse import urljoin, urlparse from lxml import etree # nosec +from lxml.html import HtmlElement # nosec from parsel.csstranslator import HTMLTranslator from w3lib.html import strip_html5_whitespace from w3lib.url import canonicalize_url, safe_url_string +from scrapy import Selector +from scrapy.http import TextResponse from scrapy.link import Link -from scrapy.linkextractors import ( - IGNORED_EXTENSIONS, - _is_valid_url, - _matches, - _re_type, - re, -) +from scrapy.linkextractors import IGNORED_EXTENSIONS, _is_valid_url, _matches, re from scrapy.utils.misc import arg_to_iter, rel_has_nofollow from scrapy.utils.python import unique as unique_list from scrapy.utils.response import get_base_url @@ -33,41 +31,56 @@ XHTML_NAMESPACE = "http://www.w3.org/1999/xhtml" _collect_string_content = etree.XPath("string()") -def _nons(tag): +def _nons(tag: Any) -> Any: if isinstance(tag, str): if tag[0] == "{" and tag[1 : len(XHTML_NAMESPACE) + 1] == XHTML_NAMESPACE: return tag.split("}")[-1] return tag -def _identity(x): +def _identity(x: Any) -> Any: return x -def _canonicalize_link_url(link): +def _canonicalize_link_url(link: Link) -> str: return canonicalize_url(link.url, keep_fragments=True) class LxmlParserLinkExtractor: def __init__( self, - tag="a", - attr="href", - process=None, - unique=False, - strip=True, - canonicalized=False, + tag: Union[str, Callable[[str], bool]] = "a", + attr: Union[str, Callable[[str], bool]] = "href", + process: Optional[Callable[[Any], Any]] = None, + unique: bool = False, + strip: bool = True, + canonicalized: bool = False, ): - self.scan_tag = tag if callable(tag) else partial(operator.eq, tag) - self.scan_attr = attr if callable(attr) else partial(operator.eq, attr) - self.process_attr = process if callable(process) else _identity - self.unique = unique - self.strip = strip - self.link_key = ( - operator.attrgetter("url") if canonicalized else _canonicalize_link_url + # mypy doesn't infer types for operator.* and also for partial() + self.scan_tag: Callable[[str], bool] = ( + tag + if callable(tag) + else cast(Callable[[str], bool], partial(operator.eq, tag)) + ) + self.scan_attr: Callable[[str], bool] = ( + attr + if callable(attr) + else cast(Callable[[str], bool], partial(operator.eq, attr)) + ) + self.process_attr: Callable[[Any], Any] = ( + process if callable(process) else _identity + ) + self.unique: bool = unique + self.strip: bool = strip + self.link_key: Callable[[Link], str] = ( + cast(Callable[[Link], str], operator.attrgetter("url")) + if canonicalized + else _canonicalize_link_url ) - def _iter_links(self, document): + def _iter_links( + self, document: HtmlElement + ) -> Iterable[Tuple[HtmlElement, str, str]]: for el in document.iter(etree.Element): if not self.scan_tag(_nons(el.tag)): continue @@ -75,10 +88,16 @@ class LxmlParserLinkExtractor: for attrib in attribs: if not self.scan_attr(attrib): continue - yield (el, attrib, attribs[attrib]) + yield el, attrib, attribs[attrib] - def _extract_links(self, selector, response_url, response_encoding, base_url): - links = [] + def _extract_links( + self, + selector: Selector, + response_url: str, + response_encoding: str, + base_url: str, + ) -> List[Link]: + links: List[Link] = [] # hacky way to get the underlying lxml parsed document for el, attr, attr_val in self._iter_links(selector.root): # pseudo lxml.html.HtmlElement.make_links_absolute(base_url) @@ -108,44 +127,48 @@ class LxmlParserLinkExtractor: links.append(link) return self._deduplicate_if_needed(links) - def extract_links(self, response): + def extract_links(self, response: TextResponse) -> List[Link]: base_url = get_base_url(response) return self._extract_links( response.selector, response.url, response.encoding, base_url ) - def _process_links(self, links): + def _process_links(self, links: List[Link]) -> List[Link]: """Normalize and filter extracted links The subclass should override it if necessary """ return self._deduplicate_if_needed(links) - def _deduplicate_if_needed(self, links): + def _deduplicate_if_needed(self, links: List[Link]) -> List[Link]: if self.unique: return unique_list(links, key=self.link_key) return links +_RegexT = Union[str, re.Pattern[str]] +_RegexOrSeveralT = Union[_RegexT, Iterable[_RegexT]] + + class LxmlLinkExtractor: _csstranslator = HTMLTranslator() def __init__( self, - allow=(), - deny=(), - allow_domains=(), - deny_domains=(), - restrict_xpaths=(), - tags=("a", "area"), - attrs=("href",), - canonicalize=False, - unique=True, - process_value=None, - deny_extensions=None, - restrict_css=(), - strip=True, - restrict_text=None, + allow: _RegexOrSeveralT = (), + deny: _RegexOrSeveralT = (), + allow_domains: Union[str, Iterable[str]] = (), + deny_domains: Union[str, Iterable[str]] = (), + restrict_xpaths: Union[str, Iterable[str]] = (), + tags: Union[str, Iterable[str]] = ("a", "area"), + attrs: Union[str, Iterable[str]] = ("href",), + canonicalize: bool = False, + unique: bool = True, + process_value: Optional[Callable[[Any], Any]] = None, + deny_extensions: Union[str, Iterable[str], None] = None, + restrict_css: Union[str, Iterable[str]] = (), + strip: bool = True, + restrict_text: Optional[_RegexOrSeveralT] = None, ): tags, attrs = set(arg_to_iter(tags)), set(arg_to_iter(attrs)) self.link_extractor = LxmlParserLinkExtractor( @@ -156,31 +179,31 @@ class LxmlLinkExtractor: strip=strip, canonicalized=not canonicalize, ) - self.allow_res = [ - x if isinstance(x, _re_type) else re.compile(x) for x in arg_to_iter(allow) - ] - self.deny_res = [ - x if isinstance(x, _re_type) else re.compile(x) for x in arg_to_iter(deny) - ] + self.allow_res: List[re.Pattern[str]] = self._compile_regexes(allow) + self.deny_res: List[re.Pattern[str]] = self._compile_regexes(deny) - self.allow_domains = set(arg_to_iter(allow_domains)) - self.deny_domains = set(arg_to_iter(deny_domains)) + self.allow_domains: Set[str] = set(arg_to_iter(allow_domains)) + self.deny_domains: Set[str] = set(arg_to_iter(deny_domains)) - self.restrict_xpaths = tuple(arg_to_iter(restrict_xpaths)) + self.restrict_xpaths: Tuple[str, ...] = tuple(arg_to_iter(restrict_xpaths)) self.restrict_xpaths += tuple( map(self._csstranslator.css_to_xpath, arg_to_iter(restrict_css)) ) if deny_extensions is None: deny_extensions = IGNORED_EXTENSIONS - self.canonicalize = canonicalize - self.deny_extensions = {"." + e for e in arg_to_iter(deny_extensions)} - self.restrict_text = [ - x if isinstance(x, _re_type) else re.compile(x) - for x in arg_to_iter(restrict_text) + self.canonicalize: bool = canonicalize + self.deny_extensions: Set[str] = {"." + e for e in arg_to_iter(deny_extensions)} + self.restrict_text: List[re.Pattern[str]] = self._compile_regexes(restrict_text) + + @staticmethod + def _compile_regexes(value: Optional[_RegexOrSeveralT]) -> List[re.Pattern[str]]: + return [ + x if isinstance(x, re.Pattern) else re.compile(x) + for x in arg_to_iter(value) ] - def _link_allowed(self, link): + def _link_allowed(self, link: Link) -> bool: if not _is_valid_url(link.url): return False if self.allow_res and not _matches(link.url, self.allow_res): @@ -202,7 +225,7 @@ class LxmlLinkExtractor: return False return True - def matches(self, url): + def matches(self, url: str) -> bool: if self.allow_domains and not url_is_from_any_domain(url, self.allow_domains): return False if self.deny_domains and url_is_from_any_domain(url, self.deny_domains): @@ -216,7 +239,7 @@ class LxmlLinkExtractor: denied = (regex.search(url) for regex in self.deny_res) if self.deny_res else [] return any(allowed) and not any(denied) - def _process_links(self, links): + def _process_links(self, links: List[Link]) -> List[Link]: links = [x for x in links if self._link_allowed(x)] if self.canonicalize: for link in links: @@ -224,10 +247,10 @@ class LxmlLinkExtractor: links = self.link_extractor._process_links(links) return links - def _extract_links(self, *args, **kwargs): + def _extract_links(self, *args: Any, **kwargs: Any) -> List[Link]: return self.link_extractor._extract_links(*args, **kwargs) - def extract_links(self, response): + def extract_links(self, response: TextResponse) -> List[Link]: """Returns a list of :class:`~scrapy.link.Link` objects from the specified :class:`response `. From 40e4a5960477299e36a0f7363db4a3c03576f64e Mon Sep 17 00:00:00 2001 From: Andrey Rakhmatullin Date: Sun, 5 May 2024 23:00:16 +0500 Subject: [PATCH 2/2] Fix Python 3.8. --- scrapy/linkextractors/__init__.py | 4 ++-- scrapy/linkextractors/lxmlhtml.py | 23 +++++++++++++++++------ 2 files changed, 19 insertions(+), 8 deletions(-) diff --git a/scrapy/linkextractors/__init__.py b/scrapy/linkextractors/__init__.py index 38dbe8135..d59005edd 100644 --- a/scrapy/linkextractors/__init__.py +++ b/scrapy/linkextractors/__init__.py @@ -7,7 +7,7 @@ For more info see docs/topics/link-extractors.rst """ import re -from typing import Iterable +from typing import Iterable, Pattern # common file extensions that are not followed if they occur in links IGNORED_EXTENSIONS = [ @@ -111,7 +111,7 @@ IGNORED_EXTENSIONS = [ ] -def _matches(url: str, regexs: Iterable[re.Pattern[str]]) -> bool: +def _matches(url: str, regexs: Iterable[Pattern[str]]) -> bool: return any(r.search(url) for r in regexs) diff --git a/scrapy/linkextractors/lxmlhtml.py b/scrapy/linkextractors/lxmlhtml.py index 3fa7d1e3a..33a10cd6c 100644 --- a/scrapy/linkextractors/lxmlhtml.py +++ b/scrapy/linkextractors/lxmlhtml.py @@ -5,7 +5,18 @@ Link extractor based on lxml.html import logging import operator from functools import partial -from typing import Any, Callable, Iterable, List, Optional, Set, Tuple, Union, cast +from typing import ( + Any, + Callable, + Iterable, + List, + Optional, + Pattern, + Set, + Tuple, + Union, + cast, +) from urllib.parse import urljoin, urlparse from lxml import etree # nosec @@ -146,7 +157,7 @@ class LxmlParserLinkExtractor: return links -_RegexT = Union[str, re.Pattern[str]] +_RegexT = Union[str, Pattern[str]] _RegexOrSeveralT = Union[_RegexT, Iterable[_RegexT]] @@ -179,8 +190,8 @@ class LxmlLinkExtractor: strip=strip, canonicalized=not canonicalize, ) - self.allow_res: List[re.Pattern[str]] = self._compile_regexes(allow) - self.deny_res: List[re.Pattern[str]] = self._compile_regexes(deny) + self.allow_res: List[Pattern[str]] = self._compile_regexes(allow) + self.deny_res: List[Pattern[str]] = self._compile_regexes(deny) self.allow_domains: Set[str] = set(arg_to_iter(allow_domains)) self.deny_domains: Set[str] = set(arg_to_iter(deny_domains)) @@ -194,10 +205,10 @@ class LxmlLinkExtractor: deny_extensions = IGNORED_EXTENSIONS self.canonicalize: bool = canonicalize self.deny_extensions: Set[str] = {"." + e for e in arg_to_iter(deny_extensions)} - self.restrict_text: List[re.Pattern[str]] = self._compile_regexes(restrict_text) + self.restrict_text: List[Pattern[str]] = self._compile_regexes(restrict_text) @staticmethod - def _compile_regexes(value: Optional[_RegexOrSeveralT]) -> List[re.Pattern[str]]: + def _compile_regexes(value: Optional[_RegexOrSeveralT]) -> List[Pattern[str]]: return [ x if isinstance(x, re.Pattern) else re.compile(x) for x in arg_to_iter(value)