Full typing for scrapy/robotstxt.py.

This commit is contained in:
Andrey Rakhmatullin 2024-04-29 23:17:59 +05:00
parent ad35ffdb0d
commit b749db92e5
1 changed files with 33 additions and 32 deletions

View File

@ -3,9 +3,10 @@ from __future__ import annotations
import logging
import sys
from abc import ABCMeta, abstractmethod
from typing import TYPE_CHECKING, Union
from typing import TYPE_CHECKING, Optional, Union
from warnings import warn
from scrapy import Spider
from scrapy.exceptions import ScrapyDeprecationWarning
from scrapy.utils.python import to_unicode
@ -18,12 +19,14 @@ if TYPE_CHECKING:
logger = logging.getLogger(__name__)
def decode_robotstxt(robotstxt_body, spider, to_native_str_type=False):
def decode_robotstxt(
robotstxt_body: bytes, spider: Optional[Spider], to_native_str_type: bool = False
) -> str:
try:
if to_native_str_type:
robotstxt_body = to_unicode(robotstxt_body)
body_decoded = to_unicode(robotstxt_body)
else:
robotstxt_body = robotstxt_body.decode("utf-8", errors="ignore")
body_decoded = robotstxt_body.decode("utf-8", errors="ignore")
except UnicodeDecodeError:
# If we found garbage or robots.txt in an encoding other than UTF-8, disregard it.
# Switch to 'allow all' state.
@ -33,8 +36,8 @@ def decode_robotstxt(robotstxt_body, spider, to_native_str_type=False):
exc_info=sys.exc_info(),
extra={"spider": spider},
)
robotstxt_body = ""
return robotstxt_body
body_decoded = ""
return body_decoded
class RobotParser(metaclass=ABCMeta):
@ -66,82 +69,80 @@ class RobotParser(metaclass=ABCMeta):
class PythonRobotParser(RobotParser):
def __init__(self, robotstxt_body, spider):
def __init__(self, robotstxt_body: bytes, spider: Optional[Spider]):
from urllib.robotparser import RobotFileParser
self.spider = spider
robotstxt_body = decode_robotstxt(
robotstxt_body, spider, to_native_str_type=True
)
self.rp = RobotFileParser()
self.rp.parse(robotstxt_body.splitlines())
self.spider: Optional[Spider] = spider
body_decoded = decode_robotstxt(robotstxt_body, spider, to_native_str_type=True)
self.rp: RobotFileParser = RobotFileParser()
self.rp.parse(body_decoded.splitlines())
@classmethod
def from_crawler(cls, crawler, robotstxt_body):
def from_crawler(cls, crawler: Crawler, robotstxt_body: bytes) -> Self:
spider = None if not crawler else crawler.spider
o = cls(robotstxt_body, spider)
return o
def allowed(self, url, user_agent):
def allowed(self, url: Union[str, bytes], user_agent: Union[str, bytes]) -> bool:
user_agent = to_unicode(user_agent)
url = to_unicode(url)
return self.rp.can_fetch(user_agent, url)
class ReppyRobotParser(RobotParser):
def __init__(self, robotstxt_body, spider):
def __init__(self, robotstxt_body: bytes, spider: Optional[Spider]):
warn("ReppyRobotParser is deprecated.", ScrapyDeprecationWarning, stacklevel=2)
from reppy.robots import Robots
self.spider = spider
self.spider: Optional[Spider] = spider
self.rp = Robots.parse("", robotstxt_body)
@classmethod
def from_crawler(cls, crawler, robotstxt_body):
def from_crawler(cls, crawler: Crawler, robotstxt_body: bytes) -> Self:
spider = None if not crawler else crawler.spider
o = cls(robotstxt_body, spider)
return o
def allowed(self, url, user_agent):
def allowed(self, url: Union[str, bytes], user_agent: Union[str, bytes]) -> bool:
return self.rp.allowed(url, user_agent)
class RerpRobotParser(RobotParser):
def __init__(self, robotstxt_body, spider):
def __init__(self, robotstxt_body: bytes, spider: Optional[Spider]):
from robotexclusionrulesparser import RobotExclusionRulesParser
self.spider = spider
self.rp = RobotExclusionRulesParser()
robotstxt_body = decode_robotstxt(robotstxt_body, spider)
self.rp.parse(robotstxt_body)
self.spider: Optional[Spider] = spider
self.rp: RobotExclusionRulesParser = RobotExclusionRulesParser()
body_decoded = decode_robotstxt(robotstxt_body, spider)
self.rp.parse(body_decoded)
@classmethod
def from_crawler(cls, crawler, robotstxt_body):
def from_crawler(cls, crawler: Crawler, robotstxt_body: bytes) -> Self:
spider = None if not crawler else crawler.spider
o = cls(robotstxt_body, spider)
return o
def allowed(self, url, user_agent):
def allowed(self, url: Union[str, bytes], user_agent: Union[str, bytes]) -> bool:
user_agent = to_unicode(user_agent)
url = to_unicode(url)
return self.rp.is_allowed(user_agent, url)
class ProtegoRobotParser(RobotParser):
def __init__(self, robotstxt_body, spider):
def __init__(self, robotstxt_body: bytes, spider: Optional[Spider]):
from protego import Protego
self.spider = spider
robotstxt_body = decode_robotstxt(robotstxt_body, spider)
self.rp = Protego.parse(robotstxt_body)
self.spider: Optional[Spider] = spider
body_decoded = decode_robotstxt(robotstxt_body, spider)
self.rp = Protego.parse(body_decoded)
@classmethod
def from_crawler(cls, crawler, robotstxt_body):
def from_crawler(cls, crawler: Crawler, robotstxt_body: bytes) -> Self:
spider = None if not crawler else crawler.spider
o = cls(robotstxt_body, spider)
return o
def allowed(self, url, user_agent):
def allowed(self, url: Union[str, bytes], user_agent: Union[str, bytes]) -> bool:
user_agent = to_unicode(user_agent)
url = to_unicode(url)
return self.rp.can_fetch(url, user_agent)