""" Functions related to custom types or type checking """ from functools import lru_cache from scrapling.core.utils import log from scrapling.core._types import ( Any, Dict, cast, List, Tuple, Union, Optional, Callable, Sequence, TYPE_CHECKING, AsyncGenerator, ) from scrapling.core.custom_types import MappingProxyType from scrapling.parser import Selector, SQLiteStorageSystem if TYPE_CHECKING: from scrapling.spiders import Request class Response(Selector): """This class is returned by all engines as a way to unify the response type between different libraries. :param status: HTTP status code. :param reason: HTTP status message. :param cookies: Response cookies. :param headers: Response headers. :param request_headers: Request headers sent with the request. :param history: List of redirect responses, if any. :param meta: Metadata dictionary (e.g., proxy used). :param request: Associated spider Request object (set by crawler, in the spiders framework). :param captured_xhr: List of captured XHR/fetch ``Response`` objects. Populated when ``capture_xhr`` is set on a browser session. """ def __init__( self, url: str, content: str | bytes, status: int, reason: str, cookies: Tuple[Dict[str, str], ...] | Dict[str, str], headers: Dict, request_headers: Dict, encoding: str = "utf-8", method: str = "GET", history: List | None = None, meta: Dict[str, Any] | None = None, **selector_config: Any, ): if isinstance(content, str): content = content.encode("utf-8") adaptive_domain: str = cast(str, selector_config.pop("adaptive_domain", "")) self.status = status self.reason = reason self.cookies = cookies self.headers = headers self.request_headers = request_headers self.history = history or [] super().__init__( content=content, url=adaptive_domain or url, encoding=encoding, **selector_config, ) # For easier debugging while working from a Python shell log.info(f"Fetched ({status}) <{method} {url}> (referer: {request_headers.get('referer')})") if meta and not isinstance(meta, dict): raise TypeError(f"Response meta should be dictionary but got {type(meta).__name__} instead!") self.meta: Dict[str, Any] = meta or {} self.request: Optional["Request"] = None # Will be set by crawler self.captured_xhr: List["Response"] = [] @property def body(self) -> bytes: """Return the raw body of the response as bytes.""" return cast(bytes, cast(Sequence, self._raw_body)) def follow( self, url: str, sid: str = "", callback: Callable[["Response"], AsyncGenerator[Union[Dict[str, Any], "Request", None], None]] | None = None, priority: int | None = None, dont_filter: bool = False, meta: dict[str, Any] | None = None, referer_flow: bool = True, **kwargs: Any, ) -> Any: """Create a Request to follow a URL. This is a helper method for spiders to easily follow links found in pages. **IMPORTANT**: The below arguments if left empty, the corresponding value from the previous request will be used. The only exception is `dont_filter`. :param url: The URL to follow (can be relative, will be joined with current URL) :param sid: The session id to use :param callback: Spider callback method to use :param priority: The priority number to use, the higher the number, the higher priority to be processed first. :param dont_filter: If this request has been done before, disable the filter to allow it again. :param meta: Additional meta data to included in the request :param referer_flow: Enabled by default, set the current response url as referer for the new request url. :param kwargs: Additional Request arguments :return: Request object ready to be yielded """ from scrapling.spiders import Request if not self.request or not isinstance(self.request, Request): raise TypeError("This response has no request set yet.") # Merge original session kwargs with new kwargs (new takes precedence) session_kwargs = {**self.request._session_kwargs, **kwargs} if referer_flow: # For requests headers = session_kwargs.get("headers", {}) headers["referer"] = self.url session_kwargs["headers"] = headers # For browsers extra_headers = session_kwargs.get("extra_headers", {}) extra_headers["referer"] = self.url session_kwargs["extra_headers"] = extra_headers session_kwargs["google_search"] = False return Request( url=self.urljoin(url), sid=sid or self.request.sid, callback=callback or self.request.callback, priority=priority if priority is not None else self.request.priority, dont_filter=dont_filter, meta={**(self.meta or {}), **(meta or {})}, **session_kwargs, ) def __str__(self) -> str: return f"<{self.status} {self.url}>" class BaseFetcher: __slots__ = () huge_tree: bool = True adaptive: Optional[bool] = False storage: Any = SQLiteStorageSystem keep_cdata: Optional[bool] = False storage_args: Optional[Dict] = None keep_comments: Optional[bool] = False adaptive_domain: str = "" parser_keywords: Tuple = ( "huge_tree", "adaptive", "storage", "keep_cdata", "storage_args", "keep_comments", "adaptive_domain", ) # Left open for the user def __init__(self, *args, **kwargs): # For backward-compatibility before 0.2.99 args_str = ", ".join(args) or "" kwargs_str = ", ".join(f"{k}={v}" for k, v in kwargs.items()) or "" if args_str: args_str += ", " log.warning( f"This logic is deprecated now, and have no effect; It will be removed with v0.3. Use `{self.__class__.__name__}.configure({args_str}{kwargs_str})` instead before fetching" ) pass @classmethod def display_config(cls): return dict( huge_tree=cls.huge_tree, keep_comments=cls.keep_comments, keep_cdata=cls.keep_cdata, adaptive=cls.adaptive, storage=cls.storage, storage_args=cls.storage_args, adaptive_domain=cls.adaptive_domain, ) @classmethod def configure(cls, **kwargs): """Set multiple arguments for the parser at once globally :param kwargs: The keywords can be any arguments of the following: huge_tree, keep_comments, keep_cdata, adaptive, storage, storage_args, adaptive_domain """ for key, value in kwargs.items(): key = key.strip().lower() if hasattr(cls, key): if key in cls.parser_keywords: setattr(cls, key, value) else: # Yup, no fun allowed LOL raise AttributeError(f'Unknown parser argument: "{key}"; maybe you meant {cls.parser_keywords}?') else: raise ValueError(f'Unknown parser argument: "{key}"; maybe you meant {cls.parser_keywords}?') if not kwargs: raise AttributeError(f"You must pass a keyword to configure, current keywords: {cls.parser_keywords}?") @classmethod def _generate_parser_arguments(cls) -> Dict: # Selector class parameters # I won't validate Selector's class parameters here again, I will leave it to be validated later parser_arguments = dict( huge_tree=cls.huge_tree, keep_comments=cls.keep_comments, keep_cdata=cls.keep_cdata, adaptive=cls.adaptive, storage=cls.storage, storage_args=cls.storage_args, adaptive_domain=cls.adaptive_domain, ) return parser_arguments class StatusText: """A class that gets the status text of the response status code. Reference: https://developer.mozilla.org/en-US/docs/Web/HTTP/Status """ _phrases = MappingProxyType( { 100: "Continue", 101: "Switching Protocols", 102: "Processing", 103: "Early Hints", 200: "OK", 201: "Created", 202: "Accepted", 203: "Non-Authoritative Information", 204: "No Content", 205: "Reset Content", 206: "Partial Content", 207: "Multi-Status", 208: "Already Reported", 226: "IM Used", 300: "Multiple Choices", 301: "Moved Permanently", 302: "Found", 303: "See Other", 304: "Not Modified", 305: "Use Proxy", 307: "Temporary Redirect", 308: "Permanent Redirect", 400: "Bad Request", 401: "Unauthorized", 402: "Payment Required", 403: "Forbidden", 404: "Not Found", 405: "Method Not Allowed", 406: "Not Acceptable", 407: "Proxy Authentication Required", 408: "Request Timeout", 409: "Conflict", 410: "Gone", 411: "Length Required", 412: "Precondition Failed", 413: "Payload Too Large", 414: "URI Too Long", 415: "Unsupported Media Type", 416: "Range Not Satisfiable", 417: "Expectation Failed", 418: "I'm a teapot", 421: "Misdirected Request", 422: "Unprocessable Entity", 423: "Locked", 424: "Failed Dependency", 425: "Too Early", 426: "Upgrade Required", 428: "Precondition Required", 429: "Too Many Requests", 431: "Request Header Fields Too Large", 451: "Unavailable For Legal Reasons", 500: "Internal Server Error", 501: "Not Implemented", 502: "Bad Gateway", 503: "Service Unavailable", 504: "Gateway Timeout", 505: "HTTP Version Not Supported", 506: "Variant Also Negotiates", 507: "Insufficient Storage", 508: "Loop Detected", 510: "Not Extended", 511: "Network Authentication Required", } ) @classmethod @lru_cache(maxsize=128) def get(cls, status_code: int) -> str: """Get the phrase for a given HTTP status code.""" return cls._phrases.get(status_code, "Unknown Status Code")