← back to Paul Conrad Archive
src/conrad/crawlers/base.py
344 lines
"""Polite HTTP client + crawler base: robots.txt, per-host delay, cache, retries, checkpoints."""
from __future__ import annotations
import hashlib
import logging
import json
import re
import threading
import time
import traceback
from urllib.parse import urlparse
import requests
try: # use the OS trust store (some library hosts serve incomplete chains that certifi rejects)
import truststore
truststore.inject_into_ssl()
# urllib3 cannot introspect truststore's SSLContext and emits a spurious InsecureRequestWarning; verification is
# real (expired/self-signed/wrong-host badssl.com endpoints all raise SSLError under this setup — checked 2026-09-24).
import warnings
import urllib3
warnings.filterwarnings("ignore", category=urllib3.exceptions.InsecureRequestWarning)
except ImportError: # pragma: no cover
pass
from tenacity import retry, retry_if_exception_type, stop_after_attempt, wait_exponential
from .. import config, db
log = logging.getLogger("conrad.http")
# copyright rail: this project never requests an image (Conrad cartoons are in copyright) — refused before any I/O
BINARY_CTYPES = ("image/", "application/pdf", "video/", "application/octet-stream")
IMAGE_MAGIC = (b"\xff\xd8\xff", b"\x89PNG", b"GIF8", b"%PDF", b"II*\x00", b"MM\x00*", b"RIFF")
IMAGE_URL = re.compile(r"\.(jpe?g|gif|png|tiff?|webp|bmp|jp2|svg)(?:[?#]|$)", re.I)
class RobotRules:
"""robots.txt matcher with Google/RFC 9309 semantics (urllib.robotparser does NOT support them):
* the group for the most specific matching user-agent token wins, else '*';
* '*' in a path matches any run of characters, a trailing '$' anchors the end;
* the LONGEST matching rule wins; on a tie Allow beats Disallow; no match -> allowed."""
def __init__(self):
self.groups: list[tuple[list[str], list[tuple[bool, str]], float | None]] = []
def parse(self, lines) -> None:
groups, agents, rules, delay, last_was_agent = [], [], [], None, False
for raw in lines:
line = raw.split("#", 1)[0].strip()
if ":" not in line:
continue
k, v = (x.strip() for x in line.split(":", 1))
k = k.lower()
if k == "user-agent":
if not last_was_agent and agents:
groups.append((agents, rules, delay))
agents, rules, delay = [], [], None
agents.append(v.lower())
last_was_agent = True
continue
last_was_agent = False
if not agents:
continue
if k in ("allow", "disallow"):
if v:
rules.append((k == "allow", v))
elif k == "crawl-delay":
try:
delay = float(v)
except ValueError:
pass
if agents:
groups.append((agents, rules, delay))
self.groups = groups
def _group(self, useragent: str):
# match on the PRODUCT token only ('paul-conrad-archive' of 'paul-conrad-archive/0.1 research ...'), so words in
# the comment part ('research', 'archive') can never select someone else's group
ua = (useragent.lower().split("/")[0].split() or [""])[0]
best, best_len = None, -1
for agents, rules, delay in self.groups:
for a in agents:
if a != "*" and ua.startswith(a) and len(a) > best_len:
best, best_len = (rules, delay), len(a)
if best is None:
best = next(((rules, delay) for agents, rules, delay in self.groups if "*" in agents), None)
if best is None:
return [], None
# several '*' (or same-agent) groups are merged, as Google does
if best_len < 0:
rules = [r for agents, rs, _ in self.groups if "*" in agents for r in rs]
return rules, best[1]
return best
@staticmethod
def _match(pattern: str, path: str) -> bool:
anchored = pattern.endswith("$")
pat = pattern[:-1] if anchored else pattern
rx = "".join(".*" if ch == "*" else re.escape(ch) for ch in pat)
return re.match(rx + ("$" if anchored else ""), path) is not None
def can_fetch(self, useragent: str, url: str) -> bool:
p = urlparse(url)
path = (p.path or "/") + (("?" + p.query) if p.query else "")
rules, _ = self._group(useragent)
best_len, allowed = -1, True
for allow, pat in rules:
if self._match(pat, path):
n = len(pat)
if n > best_len or (n == best_len and allow):
best_len, allowed = n, allow
return allowed
def crawl_delay(self, useragent: str):
return self._group(useragent)[1]
class Blocked(Exception):
"""Robots disallow, WAF/JS challenge, 401/403, paywall — we stop, record, and move on."""
class Transient(Exception):
pass
_host_locks: dict[str, threading.Lock] = {}
_host_last: dict[str, float] = {}
_robots: dict[str, RobotRules | None] = {}
_glock = threading.Lock()
def _lock_for(host: str) -> threading.Lock:
with _glock:
return _host_locks.setdefault(host, threading.Lock())
def looks_like_challenge(status: int, body: str) -> bool:
b = body[:5000].lower()
return (status == 202 and len(body.strip()) == 0) or "awswafintegration" in b or "challenge-platform" in b \
or ("just a moment" in b and "cloudflare" in b) or "captcha" in b and status != 200
class Http:
def __init__(self, use_cache: bool = True):
self.s = requests.Session()
self.s.headers.update({"User-Agent": config.USER_AGENT, "Accept": "*/*"})
self.use_cache = use_cache
self.cache_dir = config.CACHE_DIR / "http"
self.cache_dir.mkdir(parents=True, exist_ok=True)
self.requests_made = 0
# ---------------- robots
def robots(self, url: str) -> RobotRules | None:
p = urlparse(url)
base = f"{p.scheme}://{p.netloc}"
with _glock:
if base in _robots:
return _robots[base]
rp = RobotRules()
try:
r = self.s.get(base + "/robots.txt", timeout=config.TIMEOUT)
if r.status_code == 403 and "amazonaws.com" in p.netloc and "<Code>AccessDenied</Code>" in r.text:
rp.parse([]) # S3 answers 403 for any MISSING key in a non-listable bucket: no robots.txt exists
elif r.status_code in (401, 403) or looks_like_challenge(r.status_code, r.text):
rp.parse(["User-agent: *", "Disallow: /"]) # conservative: cannot read robots -> treat as disallowed
rp._unreadable = True # type: ignore[attr-defined]
elif r.status_code >= 400:
rp.parse([]) # no robots.txt -> allowed
else:
rp.parse(r.text.splitlines())
except requests.RequestException:
rp = None
with _glock:
_robots[base] = rp
return rp
def allowed(self, url: str) -> bool:
rp = self.robots(url)
return True if rp is None else rp.can_fetch(config.USER_AGENT, url)
def delay_for(self, url: str) -> float:
rp = self.robots(url)
cd = None
if rp is not None:
try:
cd = rp.crawl_delay(config.USER_AGENT)
except Exception: # noqa: BLE001
cd = None
return max(config.REQUEST_DELAY, float(cd or 0))
# ---------------- fetch
def _cache_path(self, url: str):
return self.cache_dir / (hashlib.sha1(url.encode()).hexdigest() + ".json")
def get(self, url: str, *, params: dict | None = None, check_robots: bool = True, refresh: bool = False,
secret_params: dict | None = None) -> tuple[int, str]:
"""GET text. secret_params (API keys) are sent but never cached or logged."""
full = requests.Request("GET", url, params=params).prepare().url
if IMAGE_URL.search(full.split("?", 1)[0]) or IMAGE_URL.search(full):
raise Blocked(f"copyright rail: image URL refused, never fetched: {full}")
cp = self._cache_path(full)
if self.use_cache and not refresh and cp.exists():
d = json.loads(cp.read_text())
return d["status"], d["body"]
if check_robots and not self.allowed(full):
rp = self.robots(full)
if getattr(rp, "_unreadable", False):
raise Blocked(f"robots.txt unreadable (HTTP 401/403 or WAF challenge) — treated as disallowed: {full}")
raise Blocked(f"robots.txt disallows {full}")
status, body = self._fetch(full, secret_params)
if looks_like_challenge(status, body):
raise Blocked(f"WAF/JS challenge (HTTP {status}) at {full}")
if status in (401, 403):
raise Blocked(f"HTTP {status} (access denied) at {full}")
if status == 200 and self.use_cache:
cp.write_text(json.dumps({"url": full, "status": status, "body": body, "at": db.now()}))
return status, body
def get_json(self, url: str, **kw):
status, body = self.get(url, **kw)
if status != 200:
raise Transient(f"HTTP {status} at {url}")
return json.loads(body)
@retry(retry=retry_if_exception_type(Transient), stop=stop_after_attempt(4),
wait=wait_exponential(multiplier=2, min=4, max=60), reraise=True)
def _fetch(self, url: str, secret_params: dict | None) -> tuple[int, str]:
host = urlparse(url).netloc
with _lock_for(host):
wait = self.delay_for(url) - (time.time() - _host_last.get(host, 0))
if wait > 0:
time.sleep(wait)
t0 = time.time()
try:
r = self.s.get(url, params=secret_params, timeout=config.TIMEOUT, stream=True)
# copyright rail, second layer: an image/scan served from an extensionless URL is
# refused on its Content-Type or magic bytes BEFORE the body is read or cached.
ctype = r.headers.get("content-type", "").lower()
if ctype.startswith(BINARY_CTYPES):
r.close()
raise Blocked(f"copyright rail: {ctype or 'binary'} response refused, body never read: {url}")
# ONE generator for the whole body: a second iter_content() call loses whatever the first
# generator's gzip decoder had already buffered (bodies were silently truncated to ~10 bytes).
chunks = r.iter_content(65536)
head = next(chunks, b"")
if head.startswith(IMAGE_MAGIC):
r.close()
raise Blocked(f"copyright rail: image bytes refused, body never read: {url}")
r._content = head + b"".join(chunks)
r._content_consumed = True
log.info("GET %s -> %s (%.1fs, %d bytes)", url, r.status_code, time.time() - t0, len(r.content))
except requests.RequestException as e:
log.warning("GET %s failed: %s", url, type(e).__name__)
raise Transient(str(e).replace(str((secret_params or {}).get("api_key", "\0")), "***")) from None
finally:
_host_last[host] = time.time()
self.requests_made += 1
if r.status_code == 429 or r.status_code >= 500:
raise Transient(f"HTTP {r.status_code} at {url}")
if "charset" not in r.headers.get("content-type", "").lower():
try:
return r.status_code, r.content.decode("utf-8")
except UnicodeDecodeError:
pass
return r.status_code, r.text
class Checkpoint:
def __init__(self, name: str):
self.path = config.CHECKPOINT_DIR / f"{name}.json"
self.data = json.loads(self.path.read_text()) if self.path.exists() else {}
def get(self, k, default=None):
return self.data.get(k, default)
def set(self, k, v):
self.data[k] = v
self.path.write_text(json.dumps(self.data, indent=1))
class Crawler:
source_id = "base"
name = "Base"
repository: str | None = None
url: str | None = None
classification: str | None = None
access_notes: str | None = None
def __init__(self, conn=None, http: Http | None = None):
self.conn = conn or db.connect()
self.http = http or Http()
self.stats = {"pages": 0, "seen": 0, "added": 0, "updated": 0, "errors": 0}
self.status = "worked"
self.notes: list[str] = []
self.run_id: int | None = None
def register(self, status=None, notes=None):
db.upsert_source(self.conn, self.source_id, self.name, repository=self.repository, url=self.url,
classification=self.classification, status=status, access_notes=self.access_notes,
notes=notes)
self.conn.commit()
def save(self, rec) -> None:
self.stats["seen"] += 1
st = db.save_record(self.conn, rec, self.source_id)
if st in ("added", "updated"):
self.stats[st] += 1
def error(self, url, err) -> None:
self.stats["errors"] += 1
db.log_error(self.conn, self.run_id, self.source_id, url, str(err))
def blocked(self, url, err) -> None:
self.error(url, f"BLOCKED: {err}")
self.status = "blocked"
self.notes.append(f"blocked: {err}")
def crawl(self) -> None: # pragma: no cover - implemented by subclasses
raise NotImplementedError
def run(self) -> dict:
self.register()
self.run_id = db.start_run(self.conn, self.source_id)
try:
self.crawl()
self.conn.commit()
except Blocked as e:
self.blocked(self.url, e)
except Exception as e: # noqa: BLE001 — never stall the whole crawl
self.conn.rollback()
self.error(self.url, f"{type(e).__name__}: {e}\n{traceback.format_exc()[-1500:]}")
self.status = "failed"
if self.status == "worked" and self.stats["errors"] and not self.stats["seen"]:
self.status = "failed"
note = "; ".join(self.notes) or None
db.finish_run(self.conn, self.run_id, self.stats, self.status, note)
self.register(status=self.status, notes=note)
return {"source": self.source_id, "status": self.status, **self.stats, "notes": note}