From 122923c80e48b0074a9ab239bdff55e190f12134 Mon Sep 17 00:00:00 2001 From: Karim shoair Date: Tue, 20 Jan 2026 20:21:42 +0200 Subject: [PATCH] feat(spiders/requests): Use a reliable method for fingerprint generation --- scrapling/spiders/checkpoint.py | 2 +- scrapling/spiders/engine.py | 18 +++++++- scrapling/spiders/request.py | 79 +++++++++++++++++++++++++++++---- scrapling/spiders/scheduler.py | 11 +++-- scrapling/spiders/spider.py | 5 +++ 5 files changed, 101 insertions(+), 14 deletions(-) diff --git a/scrapling/spiders/checkpoint.py b/scrapling/spiders/checkpoint.py index ae795dc..4c511ae 100644 --- a/scrapling/spiders/checkpoint.py +++ b/scrapling/spiders/checkpoint.py @@ -17,7 +17,7 @@ class CheckpointData: """Container for checkpoint state.""" requests: List["Request"] = field(default_factory=list) - seen: Set[str] = field(default_factory=set) + seen: Set[bytes] = field(default_factory=set) class CheckpointManager: diff --git a/scrapling/spiders/engine.py b/scrapling/spiders/engine.py index 2a812be..c57a4b4 100644 --- a/scrapling/spiders/engine.py +++ b/scrapling/spiders/engine.py @@ -34,7 +34,11 @@ class CrawlerEngine: ): self.spider = spider self.session_manager = session_manager - self.scheduler = Scheduler() + self.scheduler = Scheduler( + include_kwargs=spider.fp_include_kwargs, + include_headers=spider.fp_include_headers, + keep_fragments=spider.fp_keep_fragments, + ) self.stats = CrawlStats() self._global_limiter = CapacityLimiter(spider.concurrent_requests) @@ -72,6 +76,15 @@ class CrawlerEngine: return self._domain_limiters[domain] return self._global_limiter + def _normalize_request(self, request: Request) -> None: + """Normalize request fields before enqueueing. + + Resolves empty sid to the session manager's default session ID. + This ensures consistent fingerprinting for requests using the same session. + """ + if not request.sid: + request.sid = self.session_manager.default_session_id + async def _process_request(self, request: Request) -> None: """Download and process a single request.""" async with self._rate_limiter(request.domain): @@ -101,6 +114,7 @@ class CrawlerEngine: retry_request.priority -= 1 # Don't retry immediately retry_request.dont_filter = True new_request = await self.spider.retry_blocked_request(retry_request, response) + self._normalize_request(new_request) await self.scheduler.enqueue(new_request) log.info( f"Scheduled blocked request for retry ({retry_request._retry_count}/{self.spider.max_blocked_retries}): {request.url}" @@ -114,6 +128,7 @@ class CrawlerEngine: async for result in callback(response): if isinstance(result, Request): if self._is_domain_allowed(result): + self._normalize_request(result) await self.scheduler.enqueue(result) else: self.stats.offsite_requests_count += 1 @@ -223,6 +238,7 @@ class CrawlerEngine: try: if not resuming: async for request in self.spider.start_requests(): + self._normalize_request(request) await self.scheduler.enqueue(request) else: log.info("Resuming from checkpoint, skipping start_requests()") diff --git a/scrapling/spiders/request.py b/scrapling/spiders/request.py index e2fb588..977e8e8 100644 --- a/scrapling/spiders/request.py +++ b/scrapling/spiders/request.py @@ -1,12 +1,27 @@ -from urllib.parse import urlparse +import hashlib +from io import BytesIO +from functools import cached_property +from urllib.parse import urlparse, urlencode + +import orjson +from w3lib.url import canonicalize_url from scrapling.engines.toolbelt.custom import Response -from scrapling.core._types import Any, AsyncGenerator, Callable, Dict, Union, TYPE_CHECKING +from scrapling.core._types import Any, AsyncGenerator, Callable, Dict, Union, Tuple, TYPE_CHECKING if TYPE_CHECKING: from scrapling.spiders.spider import Spider +def _convert_to_bytes(value: str | bytes) -> bytes: + if isinstance(value, bytes): + return value + if not isinstance(value, str): + raise TypeError(f"Can't convert {type(value).__name__} to bytes") + + return value.encode(encoding="utf-8", errors="ignore") + + class Request: def __init__( self, @@ -27,6 +42,7 @@ class Request: self.meta: dict[str, Any] = meta if meta else {} self._retry_count: int = _retry_count self._session_kwargs = kwargs if kwargs else {} + self._fp = None def copy(self) -> "Request": """Create a copy of this request.""" @@ -41,15 +57,60 @@ class Request: **self._session_kwargs, ) - @property + @cached_property def domain(self) -> str: return urlparse(self.url).netloc - @property - def _fp(self) -> str: - """Generate a unique fingerprint for deduplication.""" - # TODO: Improve fingerprint - return f"{self.sid}:{self.url}" + def update_fingerprint( + self, + include_kwargs: bool = False, + include_headers: bool = False, + keep_fragments: bool = False, + ) -> bytes: + """Generate a unique fingerprint for deduplication. + + Caches the result in self._fp after first computation. + """ + + post_data = self._session_kwargs.get("data", {}) + body = b"" + if post_data: + if isinstance(post_data, dict | list | tuple): + body = urlencode(post_data).encode() + elif isinstance(post_data, str): + body = post_data.encode() + elif isinstance(post_data, BytesIO): + body = post_data.getvalue() + elif isinstance(post_data, bytes): + body = post_data + elif post_data is None: + body = b"" + else: + post_data = self._session_kwargs.get("json", {}) + body: bytes = orjson.dumps(post_data) if post_data else b"" + + data: Dict[str, str | Tuple] = { + "sid": self.sid, + "body": body.hex(), + "method": self._session_kwargs.get("method", "GET"), + "url": canonicalize_url(self.url, keep_fragments=keep_fragments), + } + + if include_kwargs: + kwargs = (key.lower() for key in self._session_kwargs.keys() if key.lower() not in ("data", "json")) + data["kwargs"] = "".join(set(_convert_to_bytes(key).hex() for key in kwargs)) + + if include_headers: + headers = self._session_kwargs.get("headers") or self._session_kwargs.get("extra_headers") or {} + processed_headers = {} + # Some header normalization + for key, value in headers.items(): + processed_headers[_convert_to_bytes(key.lower()).hex()] = _convert_to_bytes(value.lower()).hex() + data["headers"] = tuple(processed_headers.items()) + + fp = hashlib.sha1(orjson.dumps(data, option=orjson.OPT_SORT_KEYS), usedforsecurity=False).digest() + self._fp = fp + return fp def __repr__(self) -> str: callback_name = getattr(self.callback, "__name__", None) or "None" @@ -74,6 +135,8 @@ class Request: """Requests are equal if they have the same fingerprint.""" if not isinstance(other, Request): return NotImplemented + if self._fp is None or other._fp is None: + raise RuntimeError("Cannot compare requests before generating their fingerprints!") return self._fp == other._fp def __getstate__(self) -> dict[str, Any]: diff --git a/scrapling/spiders/scheduler.py b/scrapling/spiders/scheduler.py index e36472e..aca3277 100644 --- a/scrapling/spiders/scheduler.py +++ b/scrapling/spiders/scheduler.py @@ -17,16 +17,19 @@ class Scheduler: Duplicate URLs are filtered unless dont_filter=True. """ - def __init__(self): + def __init__(self, include_kwargs: bool = False, include_headers: bool = False, keep_fragments: bool = False): self._queue: asyncio.PriorityQueue[tuple[int, int, Request]] = asyncio.PriorityQueue() - self._seen: set[str] = set() + self._seen: set[bytes] = set() self._counter = count() # Mirror dict for snapshot without draining queue self._pending: dict[int, tuple[int, int, Request]] = {} + self._include_kwargs = include_kwargs + self._include_headers = include_headers + self._keep_fragments = keep_fragments async def enqueue(self, request: Request) -> bool: """Add a request to the queue.""" - fingerprint = request._fp + fingerprint = request.update_fingerprint(self._include_kwargs, self._include_headers, self._keep_fragments) if not request.dont_filter and fingerprint in self._seen: log.debug("Dropped duplicate request: %s", request) @@ -54,7 +57,7 @@ class Scheduler: def is_empty(self) -> bool: return self._queue.empty() - def snapshot(self) -> Tuple[List[Request], Set[str]]: + def snapshot(self) -> Tuple[List[Request], Set[bytes]]: """Create a snapshot of the current state for checkpoints.""" sorted_items = sorted(self._pending.values(), key=lambda x: (x[0], x[1])) # Maintain queue order requests = [item[2] for item in sorted_items] diff --git a/scrapling/spiders/spider.py b/scrapling/spiders/spider.py index 465bd4f..f14174f 100644 --- a/scrapling/spiders/spider.py +++ b/scrapling/spiders/spider.py @@ -78,6 +78,11 @@ class Spider(ABC): download_delay: float = 0.0 max_blocked_retries: int = 3 + # Fingerprint adjustments + fp_include_kwargs = False + fp_keep_fragments = False + fp_include_headers = False + # Logging settings logging_level: int = logging.DEBUG logging_format: str = "[%(asctime)s]:({spider_name}) %(levelname)s: %(message)s"