Karim shoair
059a708b6d
- A modern spider design that uses AnyIO and asyncio, yet it's very similar to Scrapy spiders API because it's the easiest design for users, and to make it easier for new users. - Spiders can have multiple sessions per crawl, and users decide which session to use with each request. - A scheduler system that uses heapq logic. - The user can set the number of concurrent requests for a spider globally or per domain. - The user can set a download delay to control the speed of the spider more. - There's a global function that can be overridden to handle errors for all requests. (Similar to errback in scrapy). - There's a spider argument to set the allowed domains for the spider to stay in. - Each spider has a very detailed crawl stats that can be accessed right away from the code after the crawl finishes. Same case with scraped items. - The whole spider as written as any other script and you just run it. No command-line arguments, and no need to run it from the terminal through the library like other known alternatives. - Each spider has its own logger that forces sessions to use it. - Each spider has functions to override that run before start and after close. - There's a spider argument to set the logging level and another one to make the spider write to a log file. - This is only the start. A lot more features are coming in the way.
83 lines
3.0 KiB
Python
83 lines
3.0 KiB
Python
from dataclasses import dataclass, field
|
|
|
|
from scrapling.core._types import Any, Iterator, Dict, List, Tuple
|
|
|
|
|
|
@dataclass
|
|
class CrawlStats:
|
|
"""Statistics for a crawl run."""
|
|
|
|
requests_count: int = 0
|
|
concurrent_requests: int = 0
|
|
concurrent_requests_per_domain: int = 0
|
|
failed_requests_count: int = 0
|
|
offsite_requests_count: int = 0
|
|
response_bytes: int = 0
|
|
items_scraped: int = 0
|
|
start_time: float = 0.0
|
|
end_time: float = 0.0
|
|
download_delay: float = 0.0
|
|
blocked_requests_count: int = 0
|
|
custom_stats: Dict = field(default_factory=dict)
|
|
response_status_count: Dict = field(default_factory=dict)
|
|
domains_response_bytes: Dict = field(default_factory=dict)
|
|
sessions_requests_count: Dict = field(default_factory=dict)
|
|
proxies: List[str | Dict | Tuple] = field(default_factory=list)
|
|
log_levels_counter: Dict = field(default_factory=dict)
|
|
|
|
@property
|
|
def elapsed_seconds(self) -> float:
|
|
return self.end_time - self.start_time
|
|
|
|
@property
|
|
def requests_per_second(self) -> float:
|
|
if self.elapsed_seconds == 0:
|
|
return 0.0
|
|
return self.requests_count / self.elapsed_seconds
|
|
|
|
def increment_status(self, status: int) -> None:
|
|
self.response_status_count[f"status_{status}"] = self.response_status_count.get(f"status_{status}", 0) + 1
|
|
|
|
def increment_response_bytes(self, domain: str, count: int) -> None:
|
|
self.response_bytes += count
|
|
self.domains_response_bytes[domain] = self.domains_response_bytes.get(domain, 0) + count
|
|
|
|
def increment_requests_count(self, sid: str) -> None:
|
|
self.requests_count += 1
|
|
self.sessions_requests_count[sid] = self.sessions_requests_count.get(sid, 0) + 1
|
|
|
|
def to_dict(self) -> dict[str, Any]:
|
|
return {
|
|
"items_scraped": self.items_scraped,
|
|
"elapsed_seconds": round(self.elapsed_seconds, 2),
|
|
"download_delay": round(self.download_delay, 2),
|
|
"concurrent_requests": self.concurrent_requests,
|
|
"concurrent_requests_per_domain": self.concurrent_requests_per_domain,
|
|
"requests_count": self.requests_count,
|
|
"requests_per_second": round(self.requests_per_second, 2),
|
|
"sessions_requests_count": self.sessions_requests_count,
|
|
"failed_requests_count": self.failed_requests_count,
|
|
"offsite_requests_count": self.offsite_requests_count,
|
|
"blocked_requests_count": self.blocked_requests_count,
|
|
"response_status_count": self.response_status_count,
|
|
"response_bytes": self.response_bytes,
|
|
"domains_response_bytes": self.domains_response_bytes,
|
|
"proxies": self.proxies,
|
|
"custom_stats": self.custom_stats,
|
|
"log_count": self.log_levels_counter,
|
|
}
|
|
|
|
|
|
@dataclass
|
|
class CrawlResult:
|
|
"""Complete result from a spider run."""
|
|
|
|
stats: CrawlStats
|
|
items: list[dict[str, Any]]
|
|
|
|
def __len__(self) -> int:
|
|
return len(self.items)
|
|
|
|
def __iter__(self) -> Iterator[dict[str, Any]]:
|
|
return iter(self.items)
|