feat(spiders system): a prototype of the new spiders system
- 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.
This commit is contained in:
Karim shoair
@@ -0,0 +1,20 @@
|
||||
from .spider import Spider, SessionConfigurationError, LogCounterHandler
|
||||
from .request import Request
|
||||
from .result import CrawlStats, CrawlResult
|
||||
from .engine import CrawlerEngine
|
||||
from .session import SessionManager
|
||||
from .scheduler import Scheduler
|
||||
from scrapling.engines.toolbelt.custom import Response
|
||||
|
||||
__all__ = [
|
||||
"Spider",
|
||||
"SessionConfigurationError",
|
||||
"LogCounterHandler",
|
||||
"Request",
|
||||
"CrawlerEngine",
|
||||
"CrawlStats",
|
||||
"CrawlResult",
|
||||
"SessionManager",
|
||||
"Scheduler",
|
||||
"Response",
|
||||
]
|
||||
@@ -0,0 +1,153 @@
|
||||
"""Crawler engine that orchestrates the crawling process."""
|
||||
|
||||
import json
|
||||
|
||||
import anyio
|
||||
from anyio import create_task_group, CapacityLimiter
|
||||
|
||||
from scrapling.core.utils import log
|
||||
from scrapling.spiders.request import Request
|
||||
from scrapling.spiders.result import CrawlStats
|
||||
from scrapling.spiders.scheduler import Scheduler
|
||||
from scrapling.spiders.session import SessionManager
|
||||
from scrapling.core._types import Dict, TYPE_CHECKING, Any
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from scrapling.spiders.spider import Spider
|
||||
|
||||
|
||||
def _dump(obj: Dict) -> str:
|
||||
return json.dumps(obj, indent=4)
|
||||
|
||||
|
||||
class CrawlerEngine:
|
||||
"""Orchestrates the crawling process."""
|
||||
|
||||
def __init__(self, spider: "Spider", session_manager: SessionManager, scheduler: Scheduler | None = None):
|
||||
self.spider = spider
|
||||
self.session_manager = session_manager
|
||||
self.scheduler = scheduler or Scheduler()
|
||||
self.stats = CrawlStats()
|
||||
|
||||
self._global_limiter = CapacityLimiter(spider.concurrent_requests)
|
||||
self._domain_limiters: dict[str, CapacityLimiter] = {}
|
||||
self._allowed_domains: set[str] = spider.allowed_domains or set()
|
||||
|
||||
self._active_tasks: int = 0
|
||||
self._running: bool = False
|
||||
self._items: list[dict[str, Any]] = []
|
||||
|
||||
def _is_domain_allowed(self, request: Request) -> bool:
|
||||
"""Check if the request's domain is in allowed_domains."""
|
||||
if not self._allowed_domains:
|
||||
return True
|
||||
|
||||
domain = request.domain
|
||||
for allowed in self._allowed_domains:
|
||||
if domain == allowed or domain.endswith("." + allowed):
|
||||
return True
|
||||
return False
|
||||
|
||||
def _rate_limiter(self, domain: str) -> CapacityLimiter:
|
||||
"""Get or create a per-domain concurrency limiter if enabled, otherwise use the global limiter."""
|
||||
if self.spider.concurrent_requests_per_domain:
|
||||
if domain not in self._domain_limiters:
|
||||
self._domain_limiters[domain] = CapacityLimiter(self.spider.concurrent_requests_per_domain)
|
||||
return self._domain_limiters[domain]
|
||||
return self._global_limiter
|
||||
|
||||
async def _process_request(self, request: Request) -> None:
|
||||
"""Download and process a single request."""
|
||||
async with self._rate_limiter(request.domain):
|
||||
if self.spider.download_delay:
|
||||
await anyio.sleep(self.spider.download_delay)
|
||||
|
||||
if request._session_kwargs.get("proxy"):
|
||||
self.stats.proxies.append(request._session_kwargs["proxy"])
|
||||
if request._session_kwargs.get("proxies"):
|
||||
self.stats.proxies.append(dict(request._session_kwargs["proxies"]))
|
||||
try:
|
||||
response = await self.session_manager.fetch(request)
|
||||
self.stats.increment_requests_count(request.sid)
|
||||
self.stats.increment_response_bytes(request.domain, len(response.body))
|
||||
self.stats.increment_status(response.status)
|
||||
|
||||
except Exception as e:
|
||||
self.stats.failed_requests_count += 1
|
||||
await self.spider.on_error(request, e)
|
||||
return
|
||||
|
||||
# Process response through callback
|
||||
callback = request.callback if request.callback else self.spider.parse
|
||||
try:
|
||||
async for result in callback(response):
|
||||
if isinstance(result, Request):
|
||||
if self._is_domain_allowed(result):
|
||||
await self.scheduler.enqueue(result)
|
||||
else:
|
||||
self.stats.offsite_requests_count += 1
|
||||
log.debug(f"Filtered offsite request to: {result.url}")
|
||||
elif isinstance(result, dict):
|
||||
await self._handle_item(result)
|
||||
log.debug(f"Scraped from {str(response)}\n{result}")
|
||||
except Exception as e:
|
||||
await self.spider.on_error(request, e)
|
||||
|
||||
async def _handle_item(self, item: dict[str, Any]) -> None:
|
||||
"""Handle a scraped item. Override or extend for item pipelines."""
|
||||
self.stats.items_scraped += 1
|
||||
self._items.append(item)
|
||||
|
||||
async def _task_wrapper(self, request: Request) -> None:
|
||||
"""Wrapper to track active task count."""
|
||||
try:
|
||||
await self._process_request(request)
|
||||
finally:
|
||||
self._active_tasks -= 1
|
||||
|
||||
async def crawl(self) -> CrawlStats:
|
||||
"""Run the spider and return CrawlStats."""
|
||||
self._running = True
|
||||
self._items.clear()
|
||||
self.stats = CrawlStats(start_time=anyio.current_time())
|
||||
|
||||
async with self.session_manager:
|
||||
self.stats.concurrent_requests = self.spider.concurrent_requests
|
||||
self.stats.concurrent_requests_per_domain = self.spider.concurrent_requests_per_domain
|
||||
self.stats.download_delay = self.spider.download_delay
|
||||
await self.spider.on_start()
|
||||
|
||||
try:
|
||||
async for request in self.spider.start_requests():
|
||||
await self.scheduler.enqueue(request)
|
||||
|
||||
# Process queue
|
||||
async with create_task_group() as tg:
|
||||
while self._running:
|
||||
if self.scheduler.is_empty:
|
||||
# Empty queue + no active tasks = done
|
||||
if self._active_tasks == 0:
|
||||
self._running = False
|
||||
log.debug("Spider idle")
|
||||
break
|
||||
|
||||
# Brief wait for callbacks to enqueue new requests
|
||||
await anyio.sleep(0.05)
|
||||
continue
|
||||
|
||||
request = await self.scheduler.dequeue()
|
||||
self._active_tasks += 1
|
||||
tg.start_soon(self._task_wrapper, request)
|
||||
|
||||
finally:
|
||||
await self.spider.on_close()
|
||||
|
||||
self.stats.log_levels_counter = self.spider._log_counter.get_counts()
|
||||
self.stats.end_time = anyio.current_time()
|
||||
log.info(_dump(self.stats.to_dict()))
|
||||
return self.stats
|
||||
|
||||
@property
|
||||
def items(self) -> list[dict[str, Any]]:
|
||||
"""Access scraped items."""
|
||||
return self._items
|
||||
@@ -0,0 +1,59 @@
|
||||
from urllib.parse import urlparse
|
||||
|
||||
from scrapling.engines.toolbelt.custom import Response
|
||||
from scrapling.core._types import Any, AsyncGenerator, Callable, Dict, Union
|
||||
|
||||
|
||||
class Request:
|
||||
def __init__(
|
||||
self,
|
||||
url: str,
|
||||
sid: str = "",
|
||||
callback: Callable[[Response], AsyncGenerator[Union[Dict[str, Any], "Request", None], None]] | None = None,
|
||||
priority: int = 0,
|
||||
dont_filter: bool = False,
|
||||
meta: dict[str, Any] | None = None,
|
||||
**kwargs: Any,
|
||||
) -> None:
|
||||
self.url: str = url
|
||||
self.sid: str = sid
|
||||
self.callback = callback
|
||||
self.priority: int = priority
|
||||
self.dont_filter: bool = dont_filter
|
||||
self.meta: dict[str, Any] = meta if meta else {}
|
||||
self._session_kwargs = kwargs if kwargs else {}
|
||||
|
||||
@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 __repr__(self) -> str:
|
||||
callback_name = getattr(self.callback, "__name__", None) or "None"
|
||||
return f"<Request({self.url}) priority={self.priority} callback={callback_name}>"
|
||||
|
||||
def __str__(self) -> str:
|
||||
return self.url
|
||||
|
||||
def __lt__(self, other: object) -> bool:
|
||||
"""Compare requests by priority"""
|
||||
if not isinstance(other, Request):
|
||||
return NotImplemented
|
||||
return self.priority < other.priority
|
||||
|
||||
def __gt__(self, other: object) -> bool:
|
||||
"""Compare requests by priority"""
|
||||
if not isinstance(other, Request):
|
||||
return NotImplemented
|
||||
return self.priority > other.priority
|
||||
|
||||
def __eq__(self, other: object) -> bool:
|
||||
"""Requests are equal if they have the same fingerprint."""
|
||||
if not isinstance(other, Request):
|
||||
return NotImplemented
|
||||
return self._fp == other._fp
|
||||
@@ -0,0 +1,82 @@
|
||||
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)
|
||||
@@ -0,0 +1,45 @@
|
||||
import asyncio
|
||||
from itertools import count
|
||||
|
||||
from scrapling.core.utils import log
|
||||
from scrapling.spiders.request import Request
|
||||
|
||||
|
||||
class Scheduler:
|
||||
"""
|
||||
Priority queue with URL deduplication. (heapq)
|
||||
|
||||
Higher priority requests are processed first.
|
||||
Duplicate URLs are filtered unless dont_filter=True.
|
||||
"""
|
||||
|
||||
def __init__(self):
|
||||
self._queue: asyncio.PriorityQueue[tuple[int, int, Request]] = asyncio.PriorityQueue()
|
||||
self._seen: set[str] = set()
|
||||
self._counter = count()
|
||||
|
||||
async def enqueue(self, request: Request) -> bool:
|
||||
"""Add a request to the queue."""
|
||||
fingerprint = request._fp
|
||||
|
||||
if not request.dont_filter and fingerprint in self._seen:
|
||||
log.debug("Dropped duplicate request: %s", request)
|
||||
return False
|
||||
|
||||
self._seen.add(fingerprint)
|
||||
|
||||
# Negative priority so higher priority = dequeued first
|
||||
await self._queue.put((-request.priority, next(self._counter), request))
|
||||
return True
|
||||
|
||||
async def dequeue(self) -> Request:
|
||||
"""Get the next request to process."""
|
||||
_, _, request = await self._queue.get()
|
||||
return request
|
||||
|
||||
def __len__(self) -> int:
|
||||
return self._queue.qsize()
|
||||
|
||||
@property
|
||||
def is_empty(self) -> bool:
|
||||
return self._queue.empty()
|
||||
@@ -0,0 +1,129 @@
|
||||
from scrapling.spiders.request import Request
|
||||
from scrapling.engines.static import _ASyncSessionLogic
|
||||
from scrapling.engines.toolbelt.convertor import Response
|
||||
from scrapling.core._types import cast, SUPPORTED_HTTP_METHODS
|
||||
from scrapling.fetchers import AsyncDynamicSession, AsyncStealthySession, FetcherSession
|
||||
|
||||
Session = FetcherSession | AsyncDynamicSession | AsyncStealthySession
|
||||
|
||||
|
||||
class SessionManager:
|
||||
"""Manages pre-configured session instances."""
|
||||
|
||||
def __init__(self):
|
||||
self._sessions: dict[str, Session] = {}
|
||||
self._default_session_id: str | None = None
|
||||
self._started: bool = False
|
||||
|
||||
def add(self, session_id: str, session: Session, *, default: bool = False) -> "SessionManager":
|
||||
"""Register a session instance.
|
||||
|
||||
:param session_id: Name to reference this session in requests
|
||||
:param session: Your pre-configured session instance
|
||||
:param default: If True, this becomes the default session
|
||||
"""
|
||||
if session_id in self._sessions:
|
||||
raise ValueError(f"Session '{session_id}' already registered")
|
||||
|
||||
self._sessions[session_id] = session
|
||||
|
||||
if default or self._default_session_id is None:
|
||||
self._default_session_id = session_id
|
||||
|
||||
return self
|
||||
|
||||
def remove(self, session_id: str) -> None:
|
||||
"""Removes a session.
|
||||
|
||||
:param session_id: ID of session to remove
|
||||
"""
|
||||
_ = self.pop(session_id)
|
||||
|
||||
def pop(self, session_id: str) -> Session:
|
||||
"""Remove and returns a session.
|
||||
|
||||
:param session_id: ID of session to remove
|
||||
"""
|
||||
if session_id not in self._sessions:
|
||||
raise KeyError(f"Session '{session_id}' not found")
|
||||
|
||||
session = self._sessions.pop(session_id)
|
||||
|
||||
if session and self._default_session_id == session_id:
|
||||
self._default_session_id = next(iter(self._sessions), None)
|
||||
|
||||
return session
|
||||
|
||||
@property
|
||||
def default_session_id(self) -> str:
|
||||
if self._default_session_id is None:
|
||||
raise RuntimeError("No sessions registered")
|
||||
return self._default_session_id
|
||||
|
||||
@property
|
||||
def session_ids(self) -> list[str]:
|
||||
return list(self._sessions.keys())
|
||||
|
||||
def get(self, session_id: str) -> Session:
|
||||
if session_id not in self._sessions:
|
||||
available = ", ".join(self._sessions.keys())
|
||||
raise KeyError(f"Session '{session_id}' not found. Available: {available}")
|
||||
return self._sessions[session_id]
|
||||
|
||||
async def start(self) -> None:
|
||||
"""Start all sessions that aren't already alive."""
|
||||
if self._started:
|
||||
return
|
||||
|
||||
for session in self._sessions.values():
|
||||
if not session._is_alive:
|
||||
await session.__aenter__()
|
||||
|
||||
self._started = True
|
||||
|
||||
async def close(self) -> None:
|
||||
"""Close all registered sessions."""
|
||||
for session in self._sessions.values():
|
||||
_ = await session.__aexit__(None, None, None)
|
||||
|
||||
self._started = False
|
||||
|
||||
async def fetch(self, request: Request) -> Response:
|
||||
sid = request.sid if request.sid else self.default_session_id
|
||||
session = self.get(sid)
|
||||
|
||||
if session:
|
||||
if isinstance(session, FetcherSession):
|
||||
session = session._client
|
||||
|
||||
if isinstance(session, _ASyncSessionLogic):
|
||||
response = await session._make_request(
|
||||
method=cast(SUPPORTED_HTTP_METHODS, request._session_kwargs.pop("method", "GET")),
|
||||
url=request.url,
|
||||
**request._session_kwargs,
|
||||
)
|
||||
else:
|
||||
# Sync session or other types - shouldn't happen in async context
|
||||
raise TypeError(f"Session type {type(session)} not supported for async fetch")
|
||||
else:
|
||||
response = await session.fetch(url=request.url, **request._session_kwargs)
|
||||
|
||||
response.request = request
|
||||
response.meta = request.meta
|
||||
return response
|
||||
raise RuntimeError("No session found with the request session id")
|
||||
|
||||
async def __aenter__(self) -> "SessionManager":
|
||||
await self.start()
|
||||
return self
|
||||
|
||||
async def __aexit__(self, *exc) -> None:
|
||||
await self.close()
|
||||
|
||||
def __contains__(self, session_id: str) -> bool:
|
||||
"""Check if a session ID is registered."""
|
||||
return session_id in self._sessions
|
||||
|
||||
def __len__(self) -> int:
|
||||
"""Number of registered sessions."""
|
||||
return len(self._sessions)
|
||||
@@ -0,0 +1,206 @@
|
||||
import logging
|
||||
from pathlib import Path
|
||||
from abc import ABC
|
||||
|
||||
import anyio
|
||||
|
||||
from scrapling.spiders.request import Request
|
||||
from scrapling.spiders.result import CrawlResult
|
||||
from scrapling.spiders.engine import CrawlerEngine
|
||||
from scrapling.spiders.session import SessionManager
|
||||
from scrapling.core.utils import set_logger, reset_logger
|
||||
from scrapling.core._types import Set, Any, Dict, Optional, TYPE_CHECKING, AsyncGenerator
|
||||
|
||||
BLOCKED_CODES = {401, 403, 407, 429, 444, 500, 502, 503, 504}
|
||||
if TYPE_CHECKING:
|
||||
from scrapling.engines.toolbelt.custom import Response
|
||||
|
||||
|
||||
class LogCounterHandler(logging.Handler):
|
||||
"""A logging handler that counts log messages by level."""
|
||||
|
||||
def __init__(self):
|
||||
super().__init__()
|
||||
self.counts = {
|
||||
logging.DEBUG: 0,
|
||||
logging.INFO: 0,
|
||||
logging.WARNING: 0,
|
||||
logging.ERROR: 0,
|
||||
logging.CRITICAL: 0,
|
||||
}
|
||||
|
||||
def emit(self, record: logging.LogRecord) -> None:
|
||||
level = record.levelno
|
||||
# Map to the closest standard level
|
||||
if level >= logging.CRITICAL:
|
||||
self.counts[logging.CRITICAL] += 1
|
||||
elif level >= logging.ERROR:
|
||||
self.counts[logging.ERROR] += 1
|
||||
elif level >= logging.WARNING:
|
||||
self.counts[logging.WARNING] += 1
|
||||
elif level >= logging.INFO:
|
||||
self.counts[logging.INFO] += 1
|
||||
else:
|
||||
self.counts[logging.DEBUG] += 1
|
||||
|
||||
def get_counts(self) -> Dict[str, int]:
|
||||
"""Return counts as a dictionary with string keys."""
|
||||
return {
|
||||
"debug": self.counts[logging.DEBUG],
|
||||
"info": self.counts[logging.INFO],
|
||||
"warning": self.counts[logging.WARNING],
|
||||
"error": self.counts[logging.ERROR],
|
||||
"critical": self.counts[logging.CRITICAL],
|
||||
}
|
||||
|
||||
|
||||
class SessionConfigurationError(Exception):
|
||||
"""Raised when session configuration fails."""
|
||||
|
||||
pass
|
||||
|
||||
|
||||
class Spider(ABC):
|
||||
"""An abstract base class for creating web spiders.
|
||||
|
||||
Check the documentation website for more information.
|
||||
"""
|
||||
|
||||
name: Optional[str] = None
|
||||
start_urls: list[str] = []
|
||||
allowed_domains: Set[str] = set()
|
||||
|
||||
# Concurrency settings
|
||||
concurrent_requests: int = 16
|
||||
concurrent_requests_per_domain: int = 0
|
||||
download_delay: float = 0.0
|
||||
|
||||
# Logging settings
|
||||
logging_level: int = logging.DEBUG
|
||||
log_file: Optional[str] = None
|
||||
|
||||
def __init__(self):
|
||||
if self.name is None:
|
||||
raise ValueError(f"{self.__class__.__name__} must have a name.")
|
||||
|
||||
self.logger = logging.getLogger(f"scrapling.spiders.{self.name}")
|
||||
self.logger.setLevel(self.logging_level)
|
||||
self.logger.handlers.clear()
|
||||
self.logger.propagate = False # Don't propagate to parent 'scrapling' logger
|
||||
|
||||
formatter = logging.Formatter(
|
||||
fmt=f"[%(asctime)s]:({self.name}) %(levelname)s: %(message)s", datefmt="%Y-%m-%d %H:%M:%S"
|
||||
)
|
||||
|
||||
# Add a log counter handler to track log counts by level
|
||||
self._log_counter = LogCounterHandler()
|
||||
self.logger.addHandler(self._log_counter)
|
||||
|
||||
console_handler = logging.StreamHandler()
|
||||
console_handler.setFormatter(formatter)
|
||||
self.logger.addHandler(console_handler)
|
||||
|
||||
if self.log_file:
|
||||
Path(self.log_file).parent.mkdir(parents=True, exist_ok=True)
|
||||
file_handler = logging.FileHandler(self.log_file)
|
||||
file_handler.setFormatter(formatter)
|
||||
self.logger.addHandler(file_handler)
|
||||
|
||||
self._session_manager = SessionManager()
|
||||
try:
|
||||
self.configure_sessions(self._session_manager)
|
||||
except Exception as e:
|
||||
raise SessionConfigurationError(f"Error in {self.__class__.__name__}.configure_sessions(): {e}") from e
|
||||
|
||||
if len(self._session_manager) == 0:
|
||||
raise SessionConfigurationError(f"{self.__class__.__name__}.configure_sessions() did not add any sessions")
|
||||
|
||||
self.logger.info("Spider initialized")
|
||||
|
||||
async def start_requests(self) -> AsyncGenerator[Request, None]:
|
||||
"""Generate initial requests to start the crawl.
|
||||
|
||||
By default, this generates Request objects for each URL in `start_urls`
|
||||
using the session manager's default session and `parse()` as callback.
|
||||
|
||||
Override this method for more control over initial requests
|
||||
(e.g., to add custom headers, use different callbacks, etc.)
|
||||
"""
|
||||
if not self.start_urls:
|
||||
raise RuntimeError(
|
||||
"Spider has no starting point, either set `start_urls` or override `start_requests` function."
|
||||
)
|
||||
|
||||
for url in self.start_urls:
|
||||
yield Request(url, sid=self._session_manager.default_session_id)
|
||||
|
||||
async def parse(self, response: "Response") -> AsyncGenerator[Dict[str, Any] | Request | None, None]:
|
||||
"""Default callback for processing responses"""
|
||||
raise NotImplementedError(f"{self.__class__.__name__} must implement parse() method")
|
||||
yield # Make this a generator
|
||||
|
||||
async def on_start(self) -> None:
|
||||
"""Called before crawling starts. Override for setup logic."""
|
||||
self.logger.debug("Starting spider")
|
||||
|
||||
async def on_close(self) -> None:
|
||||
"""Called after crawling finishes. Override for cleanup logic."""
|
||||
self.logger.debug("Spider closed")
|
||||
|
||||
async def on_error(self, request: Request, error: Exception) -> None:
|
||||
"""
|
||||
Handle request errors for all spider requests.
|
||||
|
||||
Override for custom error handling.
|
||||
"""
|
||||
self.logger.error(error, exc_info=error)
|
||||
|
||||
@staticmethod
|
||||
async def is_blocked(response: "Response") -> bool:
|
||||
"""Check if the response is blocked."""
|
||||
# TODO
|
||||
if response.status in BLOCKED_CODES:
|
||||
return True
|
||||
return False
|
||||
|
||||
def __repr__(self) -> str:
|
||||
"""String representation of the spider."""
|
||||
return f"<{self.__class__.__name__} '{self.name}'>"
|
||||
|
||||
def configure_sessions(self, manager: SessionManager) -> None:
|
||||
"""Configure sessions for this spider.
|
||||
|
||||
Override this method to add custom sessions.
|
||||
The default implementation creates a FetcherSession session.
|
||||
|
||||
The first session added becomes the default for `start_requests()` unless specified otherwise.
|
||||
|
||||
:param manager: SessionManager to configure
|
||||
"""
|
||||
from scrapling.fetchers import FetcherSession
|
||||
|
||||
manager.add("default", FetcherSession())
|
||||
|
||||
async def __run(self) -> CrawlResult:
|
||||
token = set_logger(self.logger)
|
||||
try:
|
||||
engine = CrawlerEngine(self, session_manager=self._session_manager)
|
||||
stats = await engine.crawl()
|
||||
return CrawlResult(stats=stats, items=engine.items)
|
||||
finally:
|
||||
reset_logger(token)
|
||||
# Close any file handlers to release file resources.
|
||||
if self.log_file:
|
||||
for handler in self.logger.handlers:
|
||||
if isinstance(handler, logging.FileHandler):
|
||||
handler.close()
|
||||
|
||||
def start(self, backend_options: Dict[str, Any] | None = None) -> CrawlResult:
|
||||
"""Run the spider and return results.
|
||||
|
||||
This is the main entry point for running a spider.
|
||||
Handles async execution internally via anyio.
|
||||
|
||||
:param backend_options: Asyncio backend options to be used with `anyio.run`
|
||||
"""
|
||||
return anyio.run(self.__run, backend="asyncio", backend_options=backend_options)
|
||||
Reference in New Issue
Block a user