wehub-resource-sync
247153575d
Tests / tests (map[TOXENV:py310], macos-latest, 3.10) (push) Has been cancelled
Tests / tests (map[TOXENV:py311], macos-latest, 3.11) (push) Has been cancelled
Tests / tests (map[TOXENV:py312], macos-latest, 3.12) (push) Has been cancelled
Tests / tests (map[TOXENV:py313], macos-latest, 3.13) (push) Has been cancelled
324 lines
12 KiB
Python
324 lines
12 KiB
Python
import signal
|
|
import logging
|
|
from pathlib import Path
|
|
from abc import ABC, abstractmethod
|
|
|
|
import anyio
|
|
from anyio import Path as AsyncPath
|
|
|
|
from scrapling.spiders.request import Request
|
|
from scrapling.spiders.engine import CrawlerEngine
|
|
from scrapling.spiders.session import SessionManager
|
|
from scrapling.core.utils import set_logger, reset_logger
|
|
from scrapling.spiders.result import CrawlResult, CrawlStats
|
|
from scrapling.core._types import Set, Any, Dict, Optional, Union, 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()
|
|
|
|
# Robots.txt compliance
|
|
robots_txt_obey: bool = False
|
|
|
|
# Development mode
|
|
development_mode: bool = False
|
|
development_cache_dir: Optional[str] = None
|
|
|
|
# Concurrency settings
|
|
concurrent_requests: int = 4
|
|
concurrent_requests_per_domain: int = 0
|
|
download_delay: float = 0.0
|
|
max_blocked_retries: int = 3
|
|
|
|
# Fingerprint adjustments
|
|
fp_include_kwargs: bool = False
|
|
fp_keep_fragments: bool = False
|
|
fp_include_headers: bool = False
|
|
|
|
# Logging settings
|
|
logging_level: int = logging.DEBUG
|
|
logging_format: str = "[%(asctime)s]:({spider_name}) %(levelname)s: %(message)s"
|
|
logging_date_format: str = "%Y-%m-%d %H:%M:%S"
|
|
log_file: Optional[str] = None
|
|
|
|
def __init__(self, crawldir: Optional[Union[str, Path, AsyncPath]] = None, interval: float = 300.0):
|
|
"""Initialize the spider.
|
|
|
|
:param crawldir: Directory for checkpoint files. If provided, enables pause/resume.
|
|
:param interval: Seconds between periodic checkpoint saves (default 5 minutes).
|
|
"""
|
|
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=self.logging_format.format(spider_name=self.name), datefmt=self.logging_date_format
|
|
)
|
|
|
|
# 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.crawldir: Optional[Path] = Path(crawldir) if crawldir else None
|
|
self._interval = interval
|
|
self._engine: Optional[CrawlerEngine] = None
|
|
self._original_sigint_handler: Any = None
|
|
|
|
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)
|
|
|
|
@abstractmethod
|
|
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 for type checkers
|
|
|
|
async def on_start(self, resuming: bool = False) -> None:
|
|
"""Called before crawling starts. Override for setup logic.
|
|
|
|
:param resuming: It's enabled if the spider is resuming from a checkpoint, left for the user to use.
|
|
"""
|
|
if resuming:
|
|
self.logger.debug("Resuming spider from checkpoint")
|
|
else:
|
|
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.
|
|
"""
|
|
pass
|
|
|
|
async def on_scraped_item(self, item: Dict[str, Any]) -> Dict[str, Any] | None:
|
|
"""A hook to be overridden by users to do some processing on scraped items, return `None` to drop the item silently."""
|
|
return item
|
|
|
|
async def is_blocked(self, response: "Response") -> bool:
|
|
"""Check if the response is blocked. Users should override this for custom detection logic."""
|
|
if response.status in BLOCKED_CODES:
|
|
return True
|
|
return False
|
|
|
|
async def retry_blocked_request(self, request: Request, response: "Response") -> Request:
|
|
"""Users should override this to prepare the blocked request before retrying, if needed."""
|
|
return request
|
|
|
|
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())
|
|
|
|
def pause(self):
|
|
"""Request graceful shutdown of the crawling process."""
|
|
if self._engine:
|
|
self._engine.request_pause()
|
|
else:
|
|
raise RuntimeError("No active crawl to stop")
|
|
|
|
def _setup_signal_handler(self) -> None:
|
|
"""Set up SIGINT handler for graceful pause."""
|
|
|
|
def handler(_signum: int, _frame: Any) -> None:
|
|
if self._engine:
|
|
self._engine.request_pause()
|
|
else:
|
|
# No engine yet, just raise KeyboardInterrupt
|
|
raise KeyboardInterrupt
|
|
|
|
try:
|
|
self._original_sigint_handler = signal.signal(signal.SIGINT, handler)
|
|
except ValueError:
|
|
self._original_sigint_handler = None
|
|
|
|
def _restore_signal_handler(self) -> None:
|
|
"""Restore original SIGINT handler."""
|
|
if self._original_sigint_handler is not None:
|
|
try:
|
|
signal.signal(signal.SIGINT, self._original_sigint_handler)
|
|
except ValueError:
|
|
pass
|
|
|
|
async def __run(self) -> CrawlResult:
|
|
token = set_logger(self.logger)
|
|
try:
|
|
self._engine = CrawlerEngine(self, self._session_manager, self.crawldir, self._interval)
|
|
stats = await self._engine.crawl()
|
|
paused = self._engine.paused
|
|
return CrawlResult(stats=stats, items=self._engine.items, paused=paused)
|
|
finally:
|
|
self._engine = None
|
|
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, use_uvloop: bool = False, **backend_options: Any) -> CrawlResult:
|
|
"""Run the spider and return results.
|
|
|
|
This is the main entry point for running a spider.
|
|
Handles async execution internally via anyio.
|
|
|
|
Pressing Ctrl+C will initiate graceful shutdown (waits for active tasks to complete).
|
|
Pressing Ctrl+C a second time will force immediate stop.
|
|
|
|
If crawldir is set, a checkpoint will also be saved on graceful shutdown,
|
|
allowing you to resume the crawl later by running the spider again.
|
|
|
|
:param use_uvloop: Whether to use the faster uvloop/winloop event loop implementation, if available.
|
|
:param backend_options: Asyncio backend options to be used with `anyio.run`
|
|
"""
|
|
backend_options = backend_options or {}
|
|
if use_uvloop:
|
|
backend_options.update({"use_uvloop": True})
|
|
|
|
# Set up SIGINT handler for graceful shutdown
|
|
self._setup_signal_handler()
|
|
try:
|
|
return anyio.run(self.__run, backend="asyncio", backend_options=backend_options)
|
|
finally:
|
|
self._restore_signal_handler()
|
|
|
|
async def stream(self) -> AsyncGenerator[Dict[str, Any], None]:
|
|
"""Stream items as they're scraped. Ideal for long-running spiders or building applications on top of the spiders.
|
|
|
|
Must be called from an async context. Yields items one by one as they are scraped.
|
|
Access `spider.stats` during iteration for real-time statistics.
|
|
|
|
Note: SIGINT handling for pause/resume is not available in stream mode.
|
|
"""
|
|
token = set_logger(self.logger)
|
|
try:
|
|
self._engine = CrawlerEngine(self, self._session_manager, self.crawldir, self._interval)
|
|
async for item in self._engine:
|
|
yield item
|
|
finally:
|
|
self._engine = None
|
|
reset_logger(token)
|
|
if self.log_file:
|
|
for handler in self.logger.handlers:
|
|
if isinstance(handler, logging.FileHandler):
|
|
handler.close()
|
|
|
|
@property
|
|
def stats(self) -> CrawlStats:
|
|
"""Access current crawl stats (works during streaming)."""
|
|
if self._engine:
|
|
return self._engine.stats
|
|
raise RuntimeError("No active crawl. Use this property inside `async for item in spider.stream():`")
|