1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
|
"""
Scrapy extension for collecting scraping stats
"""
from __future__ import annotations
import logging
import pprint
from typing import TYPE_CHECKING, Any
if TYPE_CHECKING:
from scrapy import Spider
from scrapy.crawler import Crawler
logger = logging.getLogger(__name__)
StatsT = dict[str, Any]
class StatsCollector:
def __init__(self, crawler: Crawler):
self._dump: bool = crawler.settings.getbool("STATS_DUMP")
self._stats: StatsT = {}
def get_value(
self, key: str, default: Any = None, spider: Spider | None = None
) -> Any:
return self._stats.get(key, default)
def get_stats(self, spider: Spider | None = None) -> StatsT:
return self._stats
def set_value(self, key: str, value: Any, spider: Spider | None = None) -> None:
self._stats[key] = value
def set_stats(self, stats: StatsT, spider: Spider | None = None) -> None:
self._stats = stats
def inc_value(
self, key: str, count: int = 1, start: int = 0, spider: Spider | None = None
) -> None:
d = self._stats
d[key] = d.setdefault(key, start) + count
def max_value(self, key: str, value: Any, spider: Spider | None = None) -> None:
self._stats[key] = max(self._stats.setdefault(key, value), value)
def min_value(self, key: str, value: Any, spider: Spider | None = None) -> None:
self._stats[key] = min(self._stats.setdefault(key, value), value)
def clear_stats(self, spider: Spider | None = None) -> None:
self._stats.clear()
def open_spider(self, spider: Spider) -> None:
pass
def close_spider(self, spider: Spider, reason: str) -> None:
if self._dump:
logger.info(
"Dumping Scrapy stats:\n" + pprint.pformat(self._stats),
extra={"spider": spider},
)
self._persist_stats(self._stats, spider)
def _persist_stats(self, stats: StatsT, spider: Spider) -> None:
pass
class MemoryStatsCollector(StatsCollector):
def __init__(self, crawler: Crawler):
super().__init__(crawler)
self.spider_stats: dict[str, StatsT] = {}
def _persist_stats(self, stats: StatsT, spider: Spider) -> None:
self.spider_stats[spider.name] = stats
class DummyStatsCollector(StatsCollector):
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
|
