Web Crawling › Implementing a Focused Crawler
Language AI HandbookAll chapters Web Crawling › Implementing a Focused CrawlerOne section · Sections in this chapter6 / 13 ↓
Reading tools Definitions & inline links
Choose your expertise level to adjust how many terms are explained. Beginners see more tooltips, experts see fewer to maintain reading flow. Hover over underlined terms for instant definitions.
Article links
Make inline references clickable
Let us build a functional focused crawler that demonstrates the core concepts. Our crawler will target a specific topic, respect robots.txt, implement rate limiting, and extract structured text from crawled pages. This is deliberately simplified for clarity; production crawlers add distributed execution, persistent frontiers, and much more sophisticated scheduling.
Setting Up the Environment
We need a few Python libraries for HTTP fetching, HTML parsing, and robots.txt handling:
# uv pip install requests beautifulsoup4 lxmlCore Data Structures
A clean crawler separates concerns between the frontier (URL management), the fetcher (HTTP I/O), and the extractor (content parsing). Let us define the key data structures first:
from dataclasses import dataclass, field
@dataclass
class CrawledPage:
url: str
title: str
text: str
links: list[str]
status_code: int
content_length: int
crawl_timestamp: float
@dataclass
class CrawlConfig:
seed_urls: list[str]
max_pages: int = 100
max_depth: int = 3
crawl_delay: float = 1.0 # Seconds between requests to same host
request_timeout: int = 10 # Seconds before HTTP timeout
user_agent: str = "LanguageAIHandbookCrawler/1.0 (+https://example.com)"
respect_robots: bool = True
keywords: list[str] = field(default_factory=list) # For focused crawlingThe CrawlConfig bundles all tunable parameters. The user_agent string identifies our crawler so site operators can contact us. The keywords list enables focused crawling: we will score pages and URLs by relevance to these keywords and prioritize higher-scoring URLs. By separating configuration from crawler state, we can reuse the same crawler logic with different configurations for different crawl targets.
Robots.txt Cache
Checking robots.txt for every URL would be expensive. We cache the parsed robots.txt for each host, fetching it once per domain and reusing the parsed rules for all subsequent URLs on that domain:
import urllib.parse
import urllib.robotparser
from typing import Optional
class RobotsCache:
"""Cache robots.txt rules for each domain to avoid repeated fetches."""
def __init__(self, user_agent: str, timeout: int = 5):
self._cache: dict[str, urllib.robotparser.RobotFileParser] = {}
self.user_agent = user_agent
self.timeout = timeout
def can_fetch(self, url: str) -> bool:
"""Return True if robots.txt permits fetching this URL."""
parsed = urllib.parse.urlparse(url)
base = f"{parsed.scheme}://{parsed.netloc}"
if base not in self._cache:
rp = urllib.robotparser.RobotFileParser()
rp.set_url(f"{base}/robots.txt")
try:
rp.read()
except Exception:
# If we cannot fetch robots.txt, assume access is allowed
rp.allow_all = True
self._cache[base] = rp
return self._cache[base].can_fetch(self.user_agent, url)
def get_crawl_delay(self, url: str) -> Optional[float]:
"""Return the crawl delay specified in robots.txt, if any."""
parsed = urllib.parse.urlparse(url)
base = f"{parsed.scheme}://{parsed.netloc}"
if base in self._cache:
return self._cache[base].crawl_delay(self.user_agent)
return NoneThe urllib.robotparser.RobotFileParser class handles parsing robots.txt syntax, including User-agent, Disallow, Allow, and Crawl-delay directives. We extend it with a per-host cache so each domain's rules are fetched only once. The fallback when robots.txt is unreachable, to assume access is allowed, follows the convention adopted by most major crawlers: a missing robots.txt is not equivalent to a disallow-all directive.
The Focused Crawler
Now let us build the main crawler class. The scoring function is the heart of focused crawling: it assigns higher priority to URLs whose path and surrounding anchor text suggest relevance to our target keywords:
import time
import urllib.parse
from collections import defaultdict
import requests
from bs4 import BeautifulSoup
class FocusedCrawler:
"""A focused web crawler that prioritizes pages matching specified keywords."""
def __init__(self, config: CrawlConfig):
self.config = config
self.robots = RobotsCache(config.user_agent)
self.visited: set[str] = set()
self.results: list[CrawledPage] = []
# Track last fetch time per host for rate limiting
self._last_fetch: dict[str, float] = defaultdict(float)
# Frontier: list of (score, depth, url) tuples
self.frontier: list[tuple[float, int, str]] = []
def _score_url(self, url: str, text_context: str = "") -> float:
"""Score a URL by keyword relevance. Higher = more relevant."""
if not self.config.keywords:
return 1.0
url_lower = url.lower()
context_lower = text_context.lower()
score = sum(
2.0 if kw in url_lower else (1.0 if kw in context_lower else 0.0)
for kw in self.config.keywords
)
return score
def _normalize_url(self, url: str, base_url: str) -> Optional[str]:
"""Normalize and validate a URL relative to its page."""
try:
abs_url = urllib.parse.urljoin(base_url, url)
parsed = urllib.parse.urlparse(abs_url)
# Only HTTP/HTTPS
if parsed.scheme not in ("http", "https"):
return None
# Remove fragment
normalized = parsed._replace(fragment="").geturl()
return normalized
except Exception:
return None
def _wait_for_host(self, url: str):
"""Enforce rate limiting per host."""
parsed = urllib.parse.urlparse(url)
host = parsed.netloc
elapsed = time.time() - self._last_fetch[host]
# Check for crawl-delay from robots.txt
robots_delay = self.robots.get_crawl_delay(url) or 0.0
delay = max(self.config.crawl_delay, robots_delay)
if elapsed < delay:
time.sleep(delay - elapsed)
self._last_fetch[host] = time.time()
def _fetch(self, url: str) -> Optional[requests.Response]:
"""Fetch a URL, returning the response or None on failure."""
self._wait_for_host(url)
try:
headers = {"User-Agent": self.config.user_agent}
response = requests.get(
url,
headers=headers,
timeout=self.config.request_timeout,
allow_redirects=True,
)
return response
except requests.RequestException:
return None
def _extract(self, url: str, response: requests.Response) -> CrawledPage:
"""Extract title, text, and links from an HTML response."""
soup = BeautifulSoup(response.content, "lxml")
# Extract title
title_tag = soup.find("title")
title = title_tag.get_text(strip=True) if title_tag else ""
# Remove script and style elements
for tag in soup(["script", "style", "nav", "footer", "header"]):
tag.decompose()
# Extract main text
text = soup.get_text(separator=" ", strip=True)
# Clean excessive whitespace
text = re.sub(r"\s+", " ", text).strip()
# Extract all links
links = []
for anchor in soup.find_all("a", href=True):
normalized = self._normalize_url(anchor["href"], url)
if normalized and normalized not in self.visited:
links.append(normalized)
return CrawledPage(
url=url,
title=title,
text=text,
links=links,
status_code=response.status_code,
content_length=len(response.content),
crawl_timestamp=time.time(),
)
def crawl(self) -> list[CrawledPage]:
"""Run the crawl and return all successfully fetched pages."""
# Seed the frontier
for url in self.config.seed_urls:
score = self._score_url(url)
self.frontier.append((score, 0, url))
while self.frontier and len(self.results) < self.config.max_pages:
# Sort by score descending to visit best URLs first
self.frontier.sort(key=lambda x: x[0], reverse=True)
score, depth, url = self.frontier.pop(0)
if url in self.visited:
continue
self.visited.add(url)
# Check robots.txt
if self.config.respect_robots and not self.robots.can_fetch(url):
continue
# Fetch the page
response = self._fetch(url)
if response is None or response.status_code != 200:
continue
# Extract content
content_type = response.headers.get("Content-Type", "")
if "text/html" not in content_type:
continue
page = self._extract(url, response)
self.results.append(page)
# Expand frontier if within depth limit
if depth < self.config.max_depth:
for link in page.links:
link_score = self._score_url(link, page.text)
self.frontier.append((link_score, depth + 1, link))
return self.resultsThe scoring function rewards URLs that contain relevant keywords in the URL path itself (score 2.0 per keyword match) or in the surrounding anchor text context (score 1.0 per match). This distinction matters because a URL like /python-debugging-guide strongly suggests relevant content, while an anchor link that says "click here" but happens to be on a Python page provides weaker evidence. By sorting the frontier by score before each fetch, the crawler naturally gravitates toward the highest-relevance content, spending its bandwidth budget where it is most likely to find useful pages.
Running a Simulated Crawl
For demonstration purposes, we simulate what a crawl would produce without hitting live servers. The simulation uses realistic distributions for content lengths, domain distributions, and page link structures:
import random
import string
import time
# Simulate crawl results to show statistics
random.seed(42)
def simulate_crawl_results(n_pages: int = 150) -> list[CrawledPage]:
"""Simulate realistic crawl output for analysis."""
domains = [
"nlp-papers.org",
"machinelearning.wiki",
"ai-research.io",
"deep-learning-blog.com",
"transformer-models.net",
"linguistics.edu",
"corpus-research.org",
"text-mining.io",
"spam-site-123.xyz",
"thin-content-456.click",
]
# Power-law weights: top domains get disproportionately more pages
# Reflects web's link-following concentration effect
domain_weights = [30, 22, 14, 9, 6, 4, 7, 3, 3, 2]
pages = []
for i in range(n_pages):
domain = random.choices(domains, weights=domain_weights)[0]
url = f"https://{domain}/page-{i:04d}"
# Realistic content length distribution (log-normal)
mean_log = 8.5 # log(~4900 chars)
std_log = 1.2
content_len = int(random.lognormvariate(mean_log, std_log))
content_len = max(50, min(content_len, 200_000))
# Generate text with varying quality
is_quality = domain not in [
"spam-site-123.xyz",
"thin-content-456.click",
]
word_count = content_len // 6
text = " ".join(
random.choice(string.ascii_lowercase * 3)
for _ in range(min(word_count, 500))
)
# Vary status codes (most succeed)
status = random.choices(
[200, 200, 200, 200, 200, 301, 404, 429, 503],
weights=[70, 5, 5, 5, 5, 4, 3, 2, 1],
)[0]
pages.append(
CrawledPage(
url=url,
title=f"Page {i} on {domain}",
text=text,
links=[
f"https://{domain}/page-{random.randint(0, n_pages)}"
for _ in range(random.randint(2, 15))
],
status_code=status if status == 200 else status,
content_length=content_len,
crawl_timestamp=time.time() - random.uniform(0, 3600),
)
)
# Only keep successful fetches (status 200)
return [p for p in pages if p.status_code == 200]
crawl_results = simulate_crawl_results(150)
# Compute statistics
total_pages = len(crawl_results)
total_bytes = sum(p.content_length for p in crawl_results)
avg_content_len = total_bytes / total_pages if total_pages > 0 else 0
domain_counts: dict[str, int] = defaultdict(int)
for page in crawl_results:
domain = urllib.parse.urlparse(page.url).netloc
domain_counts[domain] += 1
top_domains = sorted(domain_counts.items(), key=lambda x: x[1], reverse=True)[
:5
]Crawl Summary ======================================== Pages successfully fetched: 136 Total content downloaded: 1323.1 KB Average content length: 9,962 bytes Top domains by page count: nlp-papers.org 40 pages (29.4%) machinelearning.wiki 32 pages (23.5%) ai-research.io 21 pages (15.4%) deep-learning-blog.com 17 pages (12.5%) transformer-models.net 7 pages (5.1%)
The distribution of pages across domains illustrates a fundamental characteristic of web crawls: content is not uniformly distributed. Some domains contribute disproportionately many pages, especially if the crawler follows many internal links. This domain imbalance is one reason deduplication and domain-level filtering are critical steps in the data processing pipeline.
Analyzing Content Length Distribution
Content length is a simple but powerful proxy for content quality. Very short pages, under 200 bytes of text, are often spam, error pages, or thin content with no informational value. Very long pages occasionally contain high-quality reference material but can also be terms-of-service documents, cookie consent dialogs, or concatenated low-quality content. The relationship between length and quality is real but imperfect: a 200-word blog post may be more useful than a 5,000-word spun article that repeats the same idea ten different ways.
import numpy as np
content_lengths = np.array([p.content_length for p in crawl_results])
# Compute percentile statistics
p10 = np.percentile(content_lengths, 10)
p25 = np.percentile(content_lengths, 25)
p50 = np.percentile(content_lengths, 50)
p75 = np.percentile(content_lengths, 75)
p90 = np.percentile(content_lengths, 90)
# Filter thresholds used in practice
min_threshold = 200 # bytes; below this is almost certainly junk
max_threshold = 100_000 # bytes; above this is often boilerplate-heavy
filtered_pages = [
p
for p in crawl_results
if min_threshold <= p.content_length <= max_threshold
]
filter_rate = 1 - len(filtered_pages) / len(crawl_results)Content Length Distribution ======================================== 10th percentile: 1,292 bytes 25th percentile: 2,288 bytes Median: 6,118 bytes 75th percentile: 12,633 bytes 90th percentile: 24,074 bytes After length filtering (200–100,000 bytes): Pages retained: 136 (100.0%) Pages filtered: 0 (0.0%)
The log-normal distribution of content lengths means most pages cluster around a moderate size, with long tails at both extremes. Quality filtering based on length typically removes 10 to 30 percent of crawled pages in practice, depending on how aggressively the crawler followed low-quality domains. This filtering step is fast and cheap, making it an excellent first pass before the more expensive operations like language detection and quality classification.