1"""Chapter 36 (case study: web crawler) deep-dive code.
21. URL frontier: priority on the way in, one queue per host on the way out, next-allowed times.
32. Politeness and robots.txt (Python's own urllib.robotparser).
43. De-duplication: URL normalisation, a Bloom filter of seen URLs, content checksums.
54. Distributed crawling: assign hosts (not URLs) to crawler nodes.
6Run: python3 ch36_web_crawler.py"""
7import hashlib
8import heapq
9import math
10from collections import defaultdict, deque
11from urllib.parse import urlsplit, urlunsplit, parse_qsl, urlencode
12from urllib.robotparser import RobotFileParser
13
14
15# ---- 1. frontier -------------------------------------------------------------------
16
17class Frontier:
18 """Front: URLs ordered by priority. Back: one FIFO per host + a heap of when each host is next allowed."""
19
20 def __init__(self, delay_s=1.0):
21 self.delay = delay_s
22 self.front = [] # (-priority, seq, url)
23 self.seq = 0
24 self.host_q = defaultdict(deque)
25 self.ready = [] # (next_allowed_time, host)
26 self.in_ready = set()
27
28 def add(self, url, priority=1.0):
29 heapq.heappush(self.front, (-priority, self.seq, url))
30 self.seq += 1
31
32 def _refill(self):
33 while self.front:
34 neg_p, _, url = heapq.heappop(self.front)
35 host = urlsplit(url).netloc
36 self.host_q[host].append((neg_p, url))
37 if host not in self.in_ready:
38 heapq.heappush(self.ready, (0.0, neg_p, host)) # ties on time: higher priority first
39 self.in_ready.add(host)
40
41 def next(self, now):
42 """The next URL we may fetch at `now`, or None (with the time we must wait until)."""
43 self._refill()
44 if not self.ready:
45 return None, None
46 t, _, host = self.ready[0]
47 if t > now:
48 return None, t
49 heapq.heappop(self.ready)
50 _, url = self.host_q[host].popleft()
51 if self.host_q[host]:
52 heapq.heappush(self.ready, (now + self.delay, self.host_q[host][0][0], host)) # politeness gap
53 else:
54 self.in_ready.discard(host)
55 return url, now
56
57
58def crawl_time(frontier, n_workers_per_s=100.0):
59 """Simulate fetching everything; returns finish time and per-host fetch times."""
60 now, fetched = 0.0, defaultdict(list)
61 while True:
62 url, t = frontier.next(now)
63 if url is None:
64 if t is None:
65 return now, fetched
66 now = t
67 continue
68 fetched[urlsplit(url).netloc].append(now)
69 now += 1.0 / n_workers_per_s
70
71
72# ---- 2. robots.txt -----------------------------------------------------------------
73
74ROBOTS = """
75User-agent: *
76Disallow: /cart
77Disallow: /search
78Allow: /search/about
79
80User-agent: BadBot
81Disallow: /
82"""
83
84
85def robots(text):
86 rp = RobotFileParser()
87 rp.parse(text.splitlines())
88 return rp
89
90
91def rfc9309_allowed(text, agent, path):
92 """RFC 9309: use the group for this agent (else *); the LONGEST matching rule wins; Allow wins ties."""
93 groups, current, agents = {}, None, []
94 for line in text.splitlines():
95 line = line.split("#")[0].strip()
96 if not line or ":" not in line:
97 continue
98 k, v = (x.strip() for x in line.split(":", 1))
99 if k.lower() == "user-agent":
100 if current is None or current: # a new group starts after any rules
101 current, agents = [], []
102 agents.append(v.lower())
103 groups.setdefault(v.lower(), current) # consecutive User-agent lines share one group
104 elif k.lower() in ("allow", "disallow") and current is not None:
105 current.append((k.lower(), v))
106 rules = groups.get(agent.lower(), groups.get("*", []))
107 best = None
108 for kind, prefix in rules:
109 if prefix and path.startswith(prefix):
110 key = (len(prefix), kind == "allow")
111 if best is None or key > best[0]:
112 best = (key, kind)
113 return best is None or best[1] == "allow"
114
115
116# ---- 3. de-duplication -------------------------------------------------------------
117
118TRACKING = ("utm_", "fbclid", "gclid", "sessionid")
119
120
121def normalise(url):
122 p = urlsplit(url)
123 scheme, host = p.scheme.lower(), p.netloc.lower()
124 if (scheme, host.rsplit(":", 1)[-1]) in (("http", "80"), ("https", "443")):
125 host = host.rsplit(":", 1)[0]
126 path = p.path or "/"
127 q = urlencode(sorted((k, v) for k, v in parse_qsl(p.query) if not k.lower().startswith(TRACKING)))
128 return urlunsplit((scheme, host, path, q, "")) # drop #fragment
129
130
131class Bloom:
132 def __init__(self, n, p):
133 self.m = math.ceil(-n * math.log(p) / math.log(2) ** 2)
134 self.k = max(1, round(self.m / n * math.log(2)))
135 self.bits = bytearray((self.m + 7) // 8)
136
137 def _h(self, s):
138 d = hashlib.sha256(s.encode()).digest()
139 a, b = int.from_bytes(d[:8], "big"), int.from_bytes(d[8:16], "big") | 1
140 return [(a + i * b) % self.m for i in range(self.k)]
141
142 def add(self, s):
143 for x in self._h(s):
144 self.bits[x // 8] |= 1 << (x % 8)
145
146 def __contains__(self, s):
147 return all(self.bits[x // 8] >> (x % 8) & 1 for x in self._h(s))
148
149
150def content_key(html):
151 text = " ".join(html.split()).lower() # ignore whitespace and case
152 return hashlib.sha256(text.encode()).hexdigest()
153
154
155# ---- 4. distributed ---------------------------------------------------------------
156
157def node_for(url, nodes):
158 host = urlsplit(url).netloc
159 return nodes[int(hashlib.md5(host.encode()).hexdigest(), 16) % len(nodes)]
160
161
162# ---- checks ------------------------------------------------------------------------
163
164def run_checks():
165 # 1+2. politeness: 1,000 URLs on 5 hosts, one fetch per host per second
166 f = Frontier(delay_s=1.0)
167 hosts = [f"https://site{i}.example" for i in range(5)]
168 for i in range(1_000):
169 f.add(f"{hosts[i % 5]}/book/{i}", priority=1.0)
170 done, fetched = crawl_time(f)
171 for times in fetched.values():
172 gaps = [b - a for a, b in zip(times, times[1:])]
173 assert min(gaps) >= 1.0 - 1e-9 # never faster than 1 a second per host
174 assert 199 <= done <= 201 # 200 URLs per host at 1/s ≈ 200 s
175 # a naive crawler with 100 fetches a second would hit each host 20 times a second
176 assert 100 / 5 == 20
177
178 # priority: an important page jumps the queue
179 f2 = Frontier()
180 f2.add("https://a.example/old", priority=0.1)
181 f2.add("https://b.example/new-bestseller", priority=0.9)
182 assert f2.next(0)[0] == "https://b.example/new-bestseller"
183
184 # robots.txt
185 rp = robots(ROBOTS)
186 assert rp.can_fetch("BookBot", "https://shop.example/books/dune")
187 assert not rp.can_fetch("BookBot", "https://shop.example/cart")
188 assert not rp.can_fetch("BookBot", "https://shop.example/search/about") # stdlib: first match wins
189 assert rfc9309_allowed(ROBOTS, "BookBot", "/search/about") # RFC 9309: longest match wins
190 assert not rfc9309_allowed(ROBOTS, "BookBot", "/search?q=dune")
191 assert not rfc9309_allowed(ROBOTS, "BadBot", "/books/dune")
192 shared = "User-agent: a\nUser-agent: b\nDisallow: /x\n"
193 assert not rfc9309_allowed(shared, "a", "/x/1") and not rfc9309_allowed(shared, "b", "/x/1")
194 assert not rp.can_fetch("BadBot", "https://shop.example/books/dune")
195
196 # 3. normalisation collapses variants
197 variants = ["HTTPS://Shop.Example:443/books/42?utm_source=x&b=2&a=1#reviews",
198 "https://shop.example/books/42?a=1&b=2",
199 "https://shop.example/books/42?b=2&a=1&fbclid=zz"]
200 assert len({normalise(u) for u in variants}) == 1
201
202 # Bloom filter of seen URLs: no misses, ~1% false positives
203 bf = Bloom(50_000, 0.01)
204 seen = [f"https://site{i % 97}.example/p/{i}" for i in range(50_000)]
205 for u in seen:
206 bf.add(u)
207 assert all(u in bf for u in seen)
208 fp = sum(f"https://new{i}.example/x" in bf for i in range(20_000)) / 20_000
209 assert fp < 0.015
210 # sizing for 10 billion URLs at 1%: ~9.6 bits each
211 assert round(10e9 * 9.585 / 8 / 1e9) == 12 # ≈ 12 GB
212
213 # content de-duplication (exact, after whitespace/case normalisation)
214 assert content_key("<p>Dune</p>\n <p>Frank Herbert</p>") == content_key("<P>Dune</P> <p>Frank Herbert</p>".replace("<P>", "<p>").replace("</P>", "</p>"))
215
216 # 4. distributed: every URL of a host goes to the same node
217 nodes = [f"crawler-{i}" for i in range(8)]
218 urls = [f"{h}/p/{i}" for h in hosts for i in range(50)]
219 owner = defaultdict(set)
220 for u in urls:
221 owner[urlsplit(u).netloc].add(node_for(u, nodes))
222 assert all(len(v) == 1 for v in owner.values())
223
224 # Step 2 arithmetic: 1 billion pages a month
225 pps = 1e9 / (30 * 86_400)
226 assert round(pps) == 386 and round(pps * 100e3 * 8 / 1e6) == 309 # pages/s, Mbit/s at 100 KB/page
227 return {"politeness_finish_s": done, "bloom_fp": fp, "pages_per_s": round(pps)}
228
229
230if __name__ == "__main__":
231 print("All checks pass.", run_checks())