feat(spiders/requests): Use a reliable method for fingerprint generation

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:

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()")

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]:

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]

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"