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

Merge pull request #6337 from wRAR/typing-linkextractors 

Full typing for scrapy/linkextractors.
This commit is contained in:
Andrey Rakhmatullin 2024-05-06 14:02:03 +04:00 committed by GitHub
parent 07e31b9c93 40e4a59604
commit 8a083fb684
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
3 changed files with 102 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, Pattern
# 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[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"}

160
scrapy/linkextractors/lxmlhtml.py
View File

@ -5,21 +5,30 @@ Link extractor based on lxml.html
import logging
import operator
from functools import partial
from typing import (
Any,
Callable,
Iterable,
List,
Optional,
Pattern,
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 +42,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 +99,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 +138,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, 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 +190,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[Pattern[str]] = self._compile_regexes(allow)
self.deny_res: List[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[Pattern[str]] = self._compile_regexes(restrict_text)
@staticmethod
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)
]
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 +236,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 +250,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 +258,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>`.