RomanNum3ral
/
scrapy
mirror of https://github.com/scrapy/scrapy.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Full typing for scrapy/linkextractors.

This commit is contained in:
Andrey Rakhmatullin 2024-05-05 22:52:15 +05:00
parent d7da298e06
commit c76dfc383f
3 changed files with 91 additions and 69 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
scrapy/commands/bench.py
View File

@ -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)

8
scrapy/linkextractors/__init__.py
View File

@ -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"}

149
scrapy/linkextractors/lxmlhtml.py
View File

@ -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 <scrapy.http.Response>`.