Karim shoair
5ec929435b
**Resolved all 65 mypy errors across 14 files and added type annotations to all previously untyped function bodies. Final result: 0 errors with --check-untyped-defs enabled, all 454 tests pass.** `scrapling/core/_types.py` - Removed broken Self = object fallback — now requires typing_extensions for Python < 3.11 `scrapling/core/storage.py` - Fixed str/bytes mismatch in _get_hash() — used separate _identifier_bytes variable instead of reassigning from str to bytes `scrapling/core/custom_types.py` - split() return type: Union[List, "TextHandlers"] → list[Any] (avoids LSP violation with parent list[str]) - format() kwargs: **kwargs: str → **kwargs: object (matches parent str.format signature) - AttributesHandler.__init__: Added mapping: Any = None, **kwargs: Any and -> None - json_string property: Added -> bytes return type `scrapling/core/mixins.py` - Changed self: "Selector" to self: Any on all mixin methods (mypy can't handle forward-reference self types on non-subclass mixins) - Added Dict[str, int] annotation for counter variable - Removed unused TYPE_CHECKING / Selector imports `scrapling/parser.py (~30 errors)` - Added body: str | bytes pre-annotation for dual-type if/elif assignment - Used Dict[str, Any] kwargs dict for HTMLParser(...) to bypass incomplete lxml stubs missing default_doctype - Changed base_url=url or None → base_url=url or "" (avoids str | None vs str | bytes) - bool(adaptive) to guarantee bool type for __adaptive_enabled - Declared __text: Optional[TextHandler], __tag: Optional[str], __attributes: Optional[AttributesHandler] at top of __init__ - cast(List, ...) for all XPath() call results (_find_all_elements, _find_all_elements_with_spaces) - Added Dict[float, List[Any]] for score_table, Dict[str, Any] for attributes - Changed score, checks = 0, 0 → score: float = 0; checks: int = 0 (two locations) - Renamed target → target_element in save() to avoid variable redefinition with different types - Wrapped node_text.clean() / .lower() in TextHandler(...) to preserve type `scrapling/engines/_browsers/_page.py` - Added PageInfo[SyncPage] | PageInfo[AsyncPage] union type annotation to page_info variable `scrapling/engines/_browsers/_validators.py` - Convert method_kwargs (TypedDict) to plain Dict[str, Any] before dynamic key access `scrapling/engines/_browsers/_base.py` - Added _config declaration to BaseSessionMixin - Used cast(StealthConfig, self._config) in __generate_stealth_options to access stealth-only attributes - Added Tuple[str, ...] annotation for flags - Removed redundant narrower StealthConfig type annotation on self._config in StealthySessionMixin.__validate__ - Widened SyncSession and AsyncSession fields (playwright, context, browser) to Any to support both playwright and patchright types - Added -> None to both start() methods `scrapling/engines/_browsers/_stealth.py` - Added Optional, ProxyType imports - Annotated proxy: Optional[ProxyType] in both sync/async fetch loops - Annotated outer_box: Any at first declaration, removed duplicate type annotations in subsequent branches - Added -> None to sync and async start() - Added config: Any parameter type to _initialize_context - Removed redundant self.context: AsyncBrowserContext re-annotations in conditional branches `scrapling/engines/_browsers/_controllers.py` - Added Optional, ProxyType imports - Annotated proxy: Optional[ProxyType] in both sync/async fetch loops - Added -> None to async start() - Removed redundant self.context: AsyncBrowserContext re-annotations `scrapling/spiders/request.py` - Added Optional import, typed _fp: Optional[bytes] = None - Removed redundant body: bytes re-annotation `scrapling/spiders/session.py` - Used separate client variable instead of reassigning session = session._client (avoids type incompatibility and fixes a bug where session._make_request was called instead of client._make_request) - Added -> None to SessionManager.__init__ `scrapling/engines/toolbelt/convertor.py` - Added list[Response] annotation for history in both sync/async methods `scrapling/engines/static.py` - FetcherClient.__init__ and AsyncFetcherClient.__init__: Added **kwargs: Any and -> None `scrapling/core/shell.py` - Wrapped re_sub(...) result in TextHandler(...) to maintain correct type - Added -> None to CurlParser.__init__ - Added full type signature to create_wrapper, replaced wrapper.__signature__ = ... with setattr(wrapper, "__signature__", ...) to satisfy mypy - Added Callable to imports
307 lines
15 KiB
Python
307 lines
15 KiB
Python
from functools import lru_cache
|
|
from re import compile as re_compile
|
|
|
|
from curl_cffi.requests import Response as CurlResponse
|
|
from playwright._impl._errors import Error as PlaywrightError
|
|
from playwright.sync_api import Page as SyncPage, Response as SyncResponse
|
|
from playwright.async_api import Page as AsyncPage, Response as AsyncResponse
|
|
|
|
from scrapling.core.utils import log
|
|
from .custom import Response, StatusText
|
|
from scrapling.core._types import Dict, Optional
|
|
|
|
__CHARSET_RE__ = re_compile(r"charset=([\w-]+)")
|
|
|
|
|
|
class ResponseFactory:
|
|
"""
|
|
Factory class for creating `Response` objects from various sources.
|
|
|
|
This class provides multiple static and instance methods for building standardized `Response` objects
|
|
from diverse input sources such as Playwright responses, asynchronous Playwright responses,
|
|
and raw HTTP request responses. It supports handling response histories, constructing the proper
|
|
response objects, and managing encoding, headers, cookies, and other attributes.
|
|
"""
|
|
|
|
@classmethod
|
|
@lru_cache(maxsize=16)
|
|
def __extract_browser_encoding(cls, content_type: str | None, default: str = "utf-8") -> str:
|
|
"""Extract browser encoding from headers.
|
|
Ex: from header "content-type: text/html; charset=utf-8" -> "utf-8
|
|
"""
|
|
if content_type:
|
|
# Because Playwright can't do that by themselves like all libraries for some reason :3
|
|
match = __CHARSET_RE__.search(content_type)
|
|
return match.group(1) if match else default
|
|
return default
|
|
|
|
@classmethod
|
|
def _process_response_history(cls, first_response: SyncResponse, parser_arguments: Dict) -> list[Response]:
|
|
"""Process response history to build a list of `Response` objects"""
|
|
history: list[Response] = []
|
|
current_request = first_response.request.redirected_from
|
|
|
|
try:
|
|
while current_request:
|
|
try:
|
|
current_response = current_request.response()
|
|
history.insert(
|
|
0,
|
|
Response(
|
|
**{
|
|
"url": current_request.url,
|
|
# using current_response.text() will trigger "Error: Response.text: Response body is unavailable for redirect responses"
|
|
"content": "",
|
|
"status": current_response.status if current_response else 301,
|
|
"reason": (current_response.status_text or StatusText.get(current_response.status))
|
|
if current_response
|
|
else StatusText.get(301),
|
|
"encoding": cls.__extract_browser_encoding(
|
|
current_response.headers.get("content-type", "")
|
|
)
|
|
if current_response
|
|
else "utf-8",
|
|
"cookies": tuple(),
|
|
"headers": current_response.all_headers() if current_response else {},
|
|
"request_headers": current_request.all_headers(),
|
|
**parser_arguments,
|
|
}
|
|
),
|
|
)
|
|
except Exception as e: # pragma: no cover
|
|
log.error(f"Error processing redirect: {e}")
|
|
break
|
|
|
|
current_request = current_request.redirected_from
|
|
except Exception as e: # pragma: no cover
|
|
log.error(f"Error processing response history: {e}")
|
|
|
|
return history
|
|
|
|
@classmethod
|
|
def from_playwright_response(
|
|
cls,
|
|
page: SyncPage,
|
|
first_response: SyncResponse,
|
|
final_response: Optional[SyncResponse],
|
|
parser_arguments: Dict,
|
|
meta: Optional[Dict] = None,
|
|
) -> Response:
|
|
"""
|
|
Transforms a Playwright response into an internal `Response` object, encapsulating
|
|
the page's content, response status, headers, and relevant metadata.
|
|
|
|
The function handles potential issues, such as empty or missing final responses,
|
|
by falling back to the first response if necessary. Encoding and status text
|
|
are also derived from the provided response headers or reasonable defaults.
|
|
Additionally, the page content and cookies are extracted for further use.
|
|
|
|
:param page: A synchronous Playwright `Page` instance that represents the current browser page. Required to retrieve the page's URL, cookies, and content.
|
|
:param final_response: The last response received for the given request from the Playwright instance. Typically used as the main response object to derive status, headers, and other metadata.
|
|
:param first_response: An earlier or initial Playwright `Response` object that may serve as a fallback response in the absence of the final one.
|
|
:param parser_arguments: A dictionary containing additional arguments needed for parsing or further customization of the returned `Response`. These arguments are dynamically unpacked into
|
|
the `Response` object.
|
|
:param meta: Additional meta data to be saved with the response.
|
|
|
|
:return: A fully populated `Response` object containing the page's URL, content, status, headers, cookies, and other derived metadata.
|
|
:rtype: Response
|
|
"""
|
|
# In case we didn't catch a document type somehow
|
|
final_response = final_response if final_response else first_response
|
|
if not final_response:
|
|
raise ValueError("Failed to get a response from the page")
|
|
|
|
encoding = cls.__extract_browser_encoding(final_response.headers.get("content-type", ""))
|
|
# PlayWright API sometimes give empty status text for some reason!
|
|
status_text = final_response.status_text or StatusText.get(final_response.status)
|
|
|
|
history = cls._process_response_history(first_response, parser_arguments)
|
|
try:
|
|
if "html" in final_response.all_headers().get("content-type", ""):
|
|
page_content = cls._get_page_content(page).encode("utf-8")
|
|
else:
|
|
page_content = final_response.body()
|
|
except Exception as e: # pragma: no cover
|
|
log.error(f"Error getting page content: {e}")
|
|
page_content = b""
|
|
|
|
return Response(
|
|
**{
|
|
"url": page.url,
|
|
"content": page_content,
|
|
"status": final_response.status,
|
|
"reason": status_text,
|
|
"encoding": encoding,
|
|
"cookies": tuple(dict(cookie) for cookie in page.context.cookies()),
|
|
"headers": first_response.all_headers(),
|
|
"request_headers": first_response.request.all_headers(),
|
|
"history": history,
|
|
"meta": meta,
|
|
**parser_arguments,
|
|
}
|
|
)
|
|
|
|
@classmethod
|
|
async def _async_process_response_history(
|
|
cls, first_response: AsyncResponse, parser_arguments: Dict
|
|
) -> list[Response]:
|
|
"""Process response history to build a list of `Response` objects"""
|
|
history: list[Response] = []
|
|
current_request = first_response.request.redirected_from
|
|
|
|
try:
|
|
while current_request:
|
|
try:
|
|
current_response = await current_request.response()
|
|
history.insert(
|
|
0,
|
|
Response(
|
|
**{
|
|
"url": current_request.url,
|
|
# using current_response.text() will trigger "Error: Response.text: Response body is unavailable for redirect responses"
|
|
"content": "",
|
|
"status": current_response.status if current_response else 301,
|
|
"reason": (current_response.status_text or StatusText.get(current_response.status))
|
|
if current_response
|
|
else StatusText.get(301),
|
|
"encoding": cls.__extract_browser_encoding(
|
|
current_response.headers.get("content-type", "")
|
|
)
|
|
if current_response
|
|
else "utf-8",
|
|
"cookies": tuple(),
|
|
"headers": await current_response.all_headers() if current_response else {},
|
|
"request_headers": await current_request.all_headers(),
|
|
**parser_arguments,
|
|
}
|
|
),
|
|
)
|
|
except Exception as e: # pragma: no cover
|
|
log.error(f"Error processing redirect: {e}")
|
|
break
|
|
|
|
current_request = current_request.redirected_from
|
|
except Exception as e: # pragma: no cover
|
|
log.error(f"Error processing response history: {e}")
|
|
|
|
return history
|
|
|
|
@classmethod
|
|
def _get_page_content(cls, page: SyncPage) -> str:
|
|
"""
|
|
A workaround for the Playwright issue with `page.content()` on Windows. Ref.: https://github.com/microsoft/playwright/issues/16108
|
|
:param page: The page to extract content from.
|
|
:return:
|
|
"""
|
|
while True:
|
|
try:
|
|
return page.content() or ""
|
|
except PlaywrightError:
|
|
page.wait_for_timeout(500)
|
|
continue
|
|
return "" # pyright: ignore
|
|
|
|
@classmethod
|
|
async def _get_async_page_content(cls, page: AsyncPage) -> str:
|
|
"""
|
|
A workaround for the Playwright issue with `page.content()` on Windows. Ref.: https://github.com/microsoft/playwright/issues/16108
|
|
:param page: The page to extract content from.
|
|
:return:
|
|
"""
|
|
while True:
|
|
try:
|
|
return (await page.content()) or ""
|
|
except PlaywrightError:
|
|
await page.wait_for_timeout(500)
|
|
continue
|
|
return "" # pyright: ignore
|
|
|
|
@classmethod
|
|
async def from_async_playwright_response(
|
|
cls,
|
|
page: AsyncPage,
|
|
first_response: AsyncResponse,
|
|
final_response: Optional[AsyncResponse],
|
|
parser_arguments: Dict,
|
|
meta: Optional[Dict] = None,
|
|
) -> Response:
|
|
"""
|
|
Transforms a Playwright response into an internal `Response` object, encapsulating
|
|
the page's content, response status, headers, and relevant metadata.
|
|
|
|
The function handles potential issues, such as empty or missing final responses,
|
|
by falling back to the first response if necessary. Encoding and status text
|
|
are also derived from the provided response headers or reasonable defaults.
|
|
Additionally, the page content and cookies are extracted for further use.
|
|
|
|
:param page: An asynchronous Playwright `Page` instance that represents the current browser page. Required to retrieve the page's URL, cookies, and content.
|
|
:param final_response: The last response received for the given request from the Playwright instance. Typically used as the main response object to derive status, headers, and other metadata.
|
|
:param first_response: An earlier or initial Playwright `Response` object that may serve as a fallback response in the absence of the final one.
|
|
:param parser_arguments: A dictionary containing additional arguments needed for parsing or further customization of the returned `Response`. These arguments are dynamically unpacked into
|
|
the `Response` object.
|
|
:param meta: Additional meta data to be saved with the response.
|
|
|
|
:return: A fully populated `Response` object containing the page's URL, content, status, headers, cookies, and other derived metadata.
|
|
:rtype: Response
|
|
"""
|
|
# In case we didn't catch a document type somehow
|
|
final_response = final_response if final_response else first_response
|
|
if not final_response:
|
|
raise ValueError("Failed to get a response from the page")
|
|
|
|
encoding = cls.__extract_browser_encoding(final_response.headers.get("content-type", ""))
|
|
# PlayWright API sometimes give empty status text for some reason!
|
|
status_text = final_response.status_text or StatusText.get(final_response.status)
|
|
|
|
history = await cls._async_process_response_history(first_response, parser_arguments)
|
|
try:
|
|
if "html" in (await final_response.all_headers()).get("content-type", ""):
|
|
page_content = (await cls._get_async_page_content(page)).encode("utf-8")
|
|
else:
|
|
page_content = await final_response.body()
|
|
except Exception as e: # pragma: no cover
|
|
log.error(f"Error getting page content in async: {e}")
|
|
page_content = b""
|
|
|
|
return Response(
|
|
**{
|
|
"url": page.url,
|
|
"content": page_content,
|
|
"status": final_response.status,
|
|
"reason": status_text,
|
|
"encoding": encoding,
|
|
"cookies": tuple(dict(cookie) for cookie in await page.context.cookies()),
|
|
"headers": await first_response.all_headers(),
|
|
"request_headers": await first_response.request.all_headers(),
|
|
"history": history,
|
|
"meta": meta,
|
|
**parser_arguments,
|
|
}
|
|
)
|
|
|
|
@staticmethod
|
|
def from_http_request(response: CurlResponse, parser_arguments: Dict, meta: Optional[Dict] = None) -> Response:
|
|
"""Takes `curl_cffi` response and generates `Response` object from it.
|
|
|
|
:param response: `curl_cffi` response object
|
|
:param parser_arguments: Additional arguments to be passed to the `Response` object constructor.
|
|
:param meta: Optional metadata dictionary to attach to the Response.
|
|
:return: A `Response` object that is the same as `Selector` object except it has these added attributes: `status`, `reason`, `cookies`, `headers`, and `request_headers`
|
|
"""
|
|
return Response(
|
|
**{
|
|
"url": response.url,
|
|
"content": response.content,
|
|
"status": response.status_code,
|
|
"reason": response.reason,
|
|
"encoding": response.encoding or "utf-8",
|
|
"cookies": dict(response.cookies),
|
|
"headers": dict(response.headers),
|
|
"request_headers": dict(response.request.headers) if response.request else {},
|
|
"method": response.request.method if response.request else "GET",
|
|
"history": response.history, # https://github.com/lexiforest/curl_cffi/issues/82
|
|
"meta": meta,
|
|
**parser_arguments,
|
|
}
|
|
)
|