"""
Scrapy extension for collecting scraping stats
"""
from __future__ import annotations
import logging
import pprint
from typing import TYPE_CHECKING, Any, TypeAlias
from scrapy.utils.decorators import _warn_spider_arg
if TYPE_CHECKING:
from scrapy import Spider
from scrapy.crawler import Crawler
logger = logging.getLogger(__name__)
StatsT: TypeAlias = dict[str, Any]
[docs]
class StatsCollector:
"""The base stats collector that other stats collectors are based on."""
def __init__(self, crawler: Crawler):
self._dump: bool = crawler.settings.getbool("STATS_DUMP")
self._stats: StatsT = {}
self._crawler: Crawler = crawler
def __getattribute__(self, name: str) -> Any:
cached_name = f"_cached_{name}"
try:
return super().__getattribute__(cached_name)
except AttributeError:
pass
original_attr = super().__getattribute__(name)
if name in {
"get_value",
"get_stats",
"set_value",
"set_stats",
"inc_value",
"max_value",
"min_value",
"clear_stats",
"open_spider",
"close_spider",
} and callable(original_attr):
wrapped = _warn_spider_arg(original_attr)
setattr(self, cached_name, wrapped)
return wrapped
return original_attr
[docs]
def get_value(
self, key: str, default: Any = None, spider: Spider | None = None
) -> Any:
"""Return the value of the *key* stat, or *default* if it is not set."""
return self._stats.get(key, default)
[docs]
def get_stats(self, spider: Spider | None = None) -> StatsT:
"""Return all stats as a dict."""
return self._stats
[docs]
def set_value(self, key: str, value: Any, spider: Spider | None = None) -> None:
"""Set the *key* stat to *value*."""
self._stats[key] = value
[docs]
def set_stats(self, stats: StatsT, spider: Spider | None = None) -> None:
"""Replace all stats with *stats*."""
self._stats = stats
[docs]
def inc_value(
self, key: str, count: int = 1, start: int = 0, spider: Spider | None = None
) -> None:
"""Increment the *key* stat by *count*, or set it to *start* if it is
not set."""
d = self._stats
d[key] = d.setdefault(key, start) + count
[docs]
def max_value(self, key: str, value: Any, spider: Spider | None = None) -> None:
"""Set the *key* stat to *value* if it is not set or lower than
*value*."""
self._stats[key] = max(self._stats.setdefault(key, value), value)
[docs]
def min_value(self, key: str, value: Any, spider: Spider | None = None) -> None:
"""Set the *key* stat to *value* if it is not set or higher than
*value*."""
self._stats[key] = min(self._stats.setdefault(key, value), value)
[docs]
def clear_stats(self, spider: Spider | None = None) -> None:
"""Clear all stats."""
self._stats.clear()
[docs]
def open_spider(self, spider: Spider | None = None) -> None:
"""Called when the spider is opened."""
[docs]
def close_spider(
self, spider: Spider | None = None, reason: str | None = None
) -> None:
"""Called when the spider is closed."""
if self._dump:
logger.info(
f"Dumping Scrapy stats:\n{self}",
extra={"spider": self._crawler.spider},
)
self._persist_stats(self._stats)
def _persist_stats(self, stats: StatsT) -> None:
pass
def __str__(self) -> str:
return pprint.pformat(self._stats)
[docs]
class MemoryStatsCollector(StatsCollector):
"""A simple stats collector that keeps the stats of the last scraping run
(for each spider) in memory, after they're closed. The stats can be
accessed through the :attr:`spider_stats` attribute, which is a dict keyed
by spider name.
This is the default stats collector used in Scrapy.
"""
spider_stats: dict[str, StatsT]
"""A dict of dicts (keyed by spider name) containing the stats of the last
scraping run for each spider."""
def __init__(self, crawler: Crawler):
super().__init__(crawler)
self.spider_stats = {}
def _persist_stats(self, stats: StatsT) -> None:
if self._crawler.spider:
self.spider_stats[self._crawler.spider.name] = stats
[docs]
class DummyStatsCollector(StatsCollector):
"""A stats collector which does nothing but is very efficient (because it
does nothing). This stats collector can be set via the
:setting:`STATS_CLASS` setting, to disable stats collection in order to
improve performance. However, the performance penalty of stats collection
is usually marginal compared to other Scrapy workload like parsing pages.
"""
def get_value(
self, key: str, default: Any = None, spider: Spider | None = None
) -> Any:
return default
def set_value(self, key: str, value: Any, spider: Spider | None = None) -> None:
pass
def set_stats(self, stats: StatsT, spider: Spider | None = None) -> None:
pass
def inc_value(
self, key: str, count: int = 1, start: int = 0, spider: Spider | None = None
) -> None:
pass
def max_value(self, key: str, value: Any, spider: Spider | None = None) -> None:
pass
def min_value(self, key: str, value: Any, spider: Spider | None = None) -> None:
pass