2979 lines
101 KiB
Python
2979 lines
101 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
JonasSearch Crawler Gamma 0.1
|
||
|
||
PostgreSQL-based parallel crawler.
|
||
|
||
Features:
|
||
- PostgreSQL queue with row-level locking (FOR UPDATE SKIP LOCKED)
|
||
- Multiple crawler workers can safely share one queue
|
||
- Stores title, headings, cleaned text, links and image URLs
|
||
- Image URLs from <img src/srcset>, <picture>/<source>, og:image,
|
||
twitter:image and link rel="image_src" are resolved to absolute
|
||
http(s) URLs, deduplicated and stored in pages.pictures
|
||
(migration: add_pages_pictures.sql)
|
||
- Search vector is maintained by the PostgreSQL trigger
|
||
- URL deduplication via UNIQUE(url)
|
||
- Deadlock-safe parallel queue inserts (deterministic ordering,
|
||
bounded retries with rollback + backoff)
|
||
- Smart priority queue: prefers never-crawled URLs, new domains and
|
||
new hosts, with a spread bonus for little-explored territory
|
||
- Host-aware scheduling: a worker never waits on a busy host - when
|
||
the current domain is rate-limited or blocked, the worker
|
||
automatically picks another domain and returns once the host is
|
||
crawlable again
|
||
- Per-host crawl delay (robots.txt) and per-host block backoff:
|
||
HTTP 429/5xx/403 and network errors pause the host with an
|
||
exponential backoff, Retry-After is honored
|
||
- Transient failures are requeued (bounded by JONASSEARCH_MAX_ATTEMPTS)
|
||
instead of being lost
|
||
- Non-crawlable targets (localhost, raw IPs, non-real domains) are
|
||
rejected before they reach the queue
|
||
- robots.txt support with in-memory cache
|
||
- Graceful shutdown
|
||
- Basic crawler statistics
|
||
- Generation-based rolling updates: the crawler remembers the
|
||
deployment generation it started with, checks it before every
|
||
relevant commit (safe point) and exits with exit code 42 when a new
|
||
generation requires an update - the open transaction is rolled
|
||
back, never half-committed
|
||
- Launcher support: --version, --self-update N (verify a staged
|
||
release) and --update-deployment N (bump generation + required_version)
|
||
|
||
Expected PostgreSQL schema:
|
||
pages (incl. pictures text[], see add_pages_pictures.sql)
|
||
crawl_queue
|
||
crawler_stats
|
||
|
||
Connection defaults can be overridden with environment variables:
|
||
JONASSEARCH_DB_HOST
|
||
JONASSEARCH_DB_PORT
|
||
JONASSEARCH_DB_NAME
|
||
JONASSEARCH_DB_USER
|
||
JONASSEARCH_DB_PASSWORD
|
||
JONASSEARCH_UPDATE_CHECK_SECONDS
|
||
JONASSEARCH_RELEASES_DIR
|
||
|
||
Example:
|
||
python3 crawler.py
|
||
"""
|
||
|
||
import os
|
||
import re
|
||
import sys
|
||
import math
|
||
import signal
|
||
import time
|
||
import logging
|
||
import argparse
|
||
import ipaddress
|
||
import subprocess
|
||
import threading
|
||
import multiprocessing as mp
|
||
from collections import defaultdict
|
||
from email.utils import parsedate_to_datetime
|
||
from html import unescape
|
||
from pathlib import Path
|
||
from urllib.parse import parse_qsl, urljoin, urlparse, urldefrag
|
||
from urllib.robotparser import RobotFileParser
|
||
from dotenv import load_dotenv
|
||
|
||
import psycopg2
|
||
from psycopg2.extras import execute_values
|
||
from urllib.request import Request, urlopen
|
||
from urllib.error import HTTPError, URLError
|
||
|
||
try:
|
||
from bs4 import BeautifulSoup
|
||
except ImportError:
|
||
BeautifulSoup = None
|
||
|
||
def _load_env_files():
|
||
"""
|
||
.env zuerst aus dem Arbeitsverzeichnis laden (Standard), zusätzlich
|
||
aus dem Projektordner – so findet der Prozess die Konfiguration
|
||
auch dann, wenn er von woanders gestartet wird (z. B. via systemd).
|
||
|
||
Eine nicht lesbare .env darf den Crawler nicht abstürzen lassen
|
||
(z. B. falscher Benutzer unter systemd): dann laufen die bereits
|
||
gesetzten Umgebungsvariablen unverändert weiter.
|
||
"""
|
||
for dotenv_path in (None, Path(__file__).resolve().parent / ".env"):
|
||
try:
|
||
load_dotenv(dotenv_path)
|
||
except OSError as exc:
|
||
logging.warning(
|
||
"[CONFIG] %s not readable (%s); continuing with the "
|
||
"current environment.",
|
||
dotenv_path or ".env (working directory)",
|
||
exc,
|
||
)
|
||
|
||
|
||
_load_env_files()
|
||
|
||
|
||
# ============================================================
|
||
# CONFIG
|
||
# ============================================================
|
||
|
||
DB_HOST = os.getenv("JONASSEARCH_DB_HOST", "127.0.0.1")
|
||
DB_PORT = int(os.getenv("JONASSEARCH_DB_PORT", "25432"))
|
||
DB_NAME = os.getenv("JONASSEARCH_DB_NAME", "jonassearch")
|
||
DB_USER = os.getenv("JONASSEARCH_DB_USER", "jonassearch")
|
||
DB_PASSWORD = os.getenv("JONASSEARCH_DB_PASSWORD", "")
|
||
|
||
START_URL = os.getenv("JONASSEARCH_START_URL", "https://github.com/")
|
||
|
||
WORKERS = int(os.getenv("JONASSEARCH_WORKERS", "3"))
|
||
|
||
# Minimum delay between requests to the same host, per worker.
|
||
# robots.txt can specify a larger crawl delay.
|
||
DEFAULT_DELAY = float(os.getenv("JONASSEARCH_DELAY", "0.0"))
|
||
|
||
ROBOTS_CACHE_SECONDS = int(
|
||
os.getenv("JONASSEARCH_ROBOTS_CACHE_SECONDS", str(24 * 60 * 60))
|
||
)
|
||
|
||
REQUEST_TIMEOUT = float(os.getenv("JONASSEARCH_TIMEOUT", "20"))
|
||
|
||
MAX_CONTENT_LENGTH = int(
|
||
os.getenv("JONASSEARCH_MAX_CONTENT_LENGTH", str(5 * 1024 * 1024))
|
||
)
|
||
|
||
USER_AGENT = os.getenv(
|
||
"JONASSEARCH_USER_AGENT",
|
||
"JonasSearchBot/0.1 (+https://jonassearch.local/bot)"
|
||
)
|
||
|
||
# ============================================================
|
||
# DEPLOYMENT / UPDATE CONFIG
|
||
# ============================================================
|
||
|
||
# Version of THIS crawler code. Increased with every released crawler
|
||
# change; completely independent from the deployment generation (the
|
||
# generation is a global deployment counter in PostgreSQL, the version
|
||
# identifies the crawler code).
|
||
CRAWLER_VERSION = int(os.getenv("JONASSEARCH_CRAWLER_VERSION", "5"))
|
||
|
||
# Exit codes. 42 is reserved exclusively for "I must be updated" so
|
||
# the launcher can distinguish update, crash and normal exit.
|
||
EXIT_OK = 0
|
||
EXIT_ERROR = 1
|
||
EXIT_UPDATE_REQUIRED = 42
|
||
|
||
# How often the background checker re-reads the generation. It only
|
||
# sets an internal flag; the actual exit happens at the next safe
|
||
# commit point, never in the middle of a transaction.
|
||
UPDATE_CHECK_INTERVAL_SECONDS = float(
|
||
os.getenv("JONASSEARCH_UPDATE_CHECK_SECONDS", "30")
|
||
)
|
||
|
||
# Staging directory for updates: the launcher stages a new version in
|
||
# releases/<VERSION>/crawler.py, verifies it with --self-update and
|
||
# only then activates it. Relative paths resolve against the crawler
|
||
# file's directory, so they work from any working directory.
|
||
_RELEASES_RAW = os.getenv("JONASSEARCH_RELEASES_DIR", "releases")
|
||
CRAWLER_RELEASES_DIR = Path(_RELEASES_RAW)
|
||
if not CRAWLER_RELEASES_DIR.is_absolute():
|
||
CRAWLER_RELEASES_DIR = (
|
||
Path(__file__).resolve().parent / CRAWLER_RELEASES_DIR
|
||
)
|
||
|
||
# How long the launcher waits before restarting the crawler after it
|
||
# exited with EXIT_UPDATE_REQUIRED.
|
||
LAUNCHER_RESTART_DELAY = float(
|
||
os.getenv("JONASSEARCH_LAUNCHER_RESTART_DELAY", "2")
|
||
)
|
||
|
||
# Only crawl normal HTTP(S) pages.
|
||
ALLOWED_SCHEMES = {"http", "https"}
|
||
|
||
# Host suffixes reserved for loopback/private networks (never delegated
|
||
# in public DNS). Together with the localhost/IP checks in
|
||
# is_non_crawlable_host() they keep non-real domains out of the queue.
|
||
NON_CRAWLABLE_HOST_SUFFIXES = (
|
||
".internal",
|
||
".local",
|
||
".localdomain",
|
||
".localhost",
|
||
)
|
||
|
||
# Ignore obvious non-HTML resources.
|
||
BLOCKED_EXTENSIONS = {
|
||
".7z", ".avi", ".bin", ".bmp", ".bz2", ".class", ".csv", ".doc",
|
||
".docx", ".gif", ".gz", ".ico", ".iso", ".jpeg", ".jpg", ".m4a",
|
||
".mkv", ".mov", ".mp3", ".mp4", ".mpeg", ".pdf", ".png", ".ppt",
|
||
".pptx", ".rar", ".svg", ".tar", ".tgz", ".tif", ".tiff", ".webm",
|
||
".webp", ".xls", ".xlsx", ".xml", ".zip"
|
||
}
|
||
|
||
# Do not follow these URL schemes.
|
||
BLOCKED_PREFIXES = (
|
||
"mailto:",
|
||
"javascript:",
|
||
"tel:",
|
||
"data:",
|
||
"ftp:",
|
||
"file:",
|
||
)
|
||
|
||
# Pages with no search-index value that only burn bandwidth, CPU and
|
||
# index space: pkg.go.dev's reverse-dependency tabs are megabyte-sized
|
||
# link lists (10k+ links of pure package names). These URLs are
|
||
# skipped before fetching; the queue item is marked done so it is not
|
||
# retried endlessly.
|
||
SKIP_URL_PATTERNS = tuple(
|
||
re.compile(pattern)
|
||
for pattern in (
|
||
# Query part (not fragment), then the tab value, then only a
|
||
# fragment or the end may follow.
|
||
r"^https?://pkg\.go\.dev/[^#]*[?&]tab=importedby(#|$)",
|
||
)
|
||
)
|
||
|
||
# Keep generated pages reasonably useful for search.
|
||
MAX_TITLE_LENGTH = 1000
|
||
MAX_HEADINGS = 200
|
||
MAX_HEADING_LENGTH = 1000
|
||
MAX_TEXT_LENGTH = 2_000_000
|
||
MAX_LINKS = 10_000
|
||
|
||
# Image URLs per page. A page with thousands of <img>/srcset entries
|
||
# would otherwise bloat pages.pictures for no search value; the first
|
||
# images in document order are kept.
|
||
MAX_PICTURES = 1_000
|
||
|
||
# Hard cap for one image URL (text[] has no B-tree limit, but absurd
|
||
# data: URLs / tracking parameters are not worth storing).
|
||
MAX_IMAGE_URL_LENGTH = 2_000
|
||
|
||
# Cap for the text that feeds the full-text index. PostgreSQL
|
||
# hard-limits a single tsvector value to 1,048,575 bytes (not
|
||
# configurable), and the trigger pages_search_vector_update() builds
|
||
# the vector from the content column. Content near MAX_TEXT_LENGTH
|
||
# would overflow that limit and abort the whole INSERT with
|
||
# ProgramLimitExceeded. 200,000 characters are always safe: even with
|
||
# 4-byte UTF-8 characters this stays well below 1,048,575 bytes.
|
||
CONTENT_INDEX_MAX_CHARS = 200_000
|
||
|
||
# Maximum size of a URL (UTF-8 encoded) that may be inserted into the
|
||
# crawl queue. PostgreSQL B-tree index entries are limited to about 2704
|
||
# bytes, so extremely long URLs would abort the INSERT with
|
||
# ProgramLimitExceeded and crash the worker. Filtering them out before
|
||
# the INSERT keeps the crawl running without touching the DB schema.
|
||
MAX_QUEUE_URL_BYTES = 2000
|
||
|
||
# Rows per INSERT statement when filling the queue. Short statements
|
||
# keep the lock window on the UNIQUE index small.
|
||
QUEUE_INSERT_PAGE_SIZE = 500
|
||
|
||
# Deadlock handling for queue inserts: several independent crawler
|
||
# instances write overlapping URL batches into the same crawl_queue.
|
||
# Depending on insert order, PostgreSQL may then abort one transaction
|
||
# with DeadlockDetected (SQLSTATE 40P01). Instead of letting the worker
|
||
# crash, the insert is retried a few times with a short, growing
|
||
# backoff. The deterministic ordering in queue_urls() makes such
|
||
# deadlocks rare in the first place - without requiring any change to
|
||
# the database schema or to the existing index.
|
||
QUEUE_INSERT_MAX_ATTEMPTS = 3
|
||
QUEUE_INSERT_RETRY_BASE_DELAY = 0.5
|
||
|
||
# ============================================================
|
||
# HOST SCHEDULING / BLOCK HANDLING CONFIG
|
||
# ============================================================
|
||
# The queue claim examines candidates in priority order and picks the
|
||
# first URL whose host is currently crawlable (not rate-limited, not
|
||
# blocked). Hosts that would have to wait are skipped, so a worker
|
||
# automatically crawls another domain while the blocked one cools
|
||
# down - no worker sits idle in front of a 429.
|
||
|
||
# Pending URLs examined per claim round (FOR UPDATE SKIP LOCKED, so
|
||
# several crawler processes stay safe on one shared queue).
|
||
QUEUE_CLAIM_BATCH = int(os.getenv("JONASSEARCH_CLAIM_BATCH", "25"))
|
||
|
||
# Maximum number of candidates examined per claim (batch * rounds).
|
||
# Keeps the common case (a ready host in the first batch) cheap while
|
||
# still finding crawlable hosts behind a small wall of blocked ones.
|
||
# Larger walls are handled in SQL: hosts that are known to be busy are
|
||
# excluded from the candidate query itself (see busy_hosts()), so a
|
||
# blocked host with thousands of queued URLs cannot hide the rest of
|
||
# the queue behind it.
|
||
QUEUE_CLAIM_SCAN_LIMIT = int(
|
||
os.getenv("JONASSEARCH_CLAIM_SCAN_LIMIT", "200")
|
||
)
|
||
|
||
# Upper bound for the host list that is passed into the claim query.
|
||
# Longest-running blocks are kept first, because those form the wall at
|
||
# the head of the priority order; hosts beyond the cap are still
|
||
# skipped by the per-candidate check, they are just not pre-filtered.
|
||
QUEUE_CLAIM_BUSY_HOST_LIMIT = int(
|
||
os.getenv("JONASSEARCH_CLAIM_BUSY_HOST_LIMIT", "1000")
|
||
)
|
||
|
||
# Cap for the "nothing is crawlable right now" sleep, so a host blocked
|
||
# for a long time cannot park the whole worker; after the cap the queue
|
||
# is simply re-scanned.
|
||
QUEUE_MAX_IDLE_SLEEP = float(os.getenv("JONASSEARCH_MAX_IDLE_SLEEP", "30"))
|
||
|
||
# A URL may be attempted this many times (transient errors are
|
||
# requeued) before it is marked failed for good, so permanently broken
|
||
# hosts cannot spin forever.
|
||
QUEUE_MAX_ATTEMPTS = int(os.getenv("JONASSEARCH_MAX_ATTEMPTS", "5"))
|
||
|
||
# Rate waits longer than this are NOT slept inline: the item is put
|
||
# back and the worker immediately looks for another domain instead of
|
||
# blocking on a slow host (e.g. a large robots.txt crawl-delay).
|
||
MAX_INLINE_RATE_WAIT = float(os.getenv("JONASSEARCH_MAX_INLINE_WAIT", "10"))
|
||
|
||
# Backoff for blocked hosts, per HTTP status: (base seconds, cap
|
||
# seconds). The pause doubles with every consecutive block of the same
|
||
# host, so persistent blocks fade out of the schedule while a single
|
||
# hiccup costs only the base delay.
|
||
HOST_BLOCK_RULES = {
|
||
401: (600.0, 7200.0), # auth required: probably permanent, long pause
|
||
403: (600.0, 7200.0), # forbidden
|
||
408: (30.0, 600.0), # request timeout
|
||
425: (30.0, 600.0), # too early
|
||
429: (60.0, 1800.0), # rate limited (Retry-After is honored on top)
|
||
500: (30.0, 600.0),
|
||
502: (30.0, 600.0),
|
||
503: (30.0, 900.0),
|
||
504: (30.0, 600.0),
|
||
}
|
||
|
||
# Network-level failures (DNS, timeout, connection reset) get a short
|
||
# backoff as well - a dead host must not be retried in a tight loop.
|
||
HOST_BLOCK_NETWORK_BASE = 30.0
|
||
HOST_BLOCK_NETWORK_MAX = 300.0
|
||
|
||
# Absolute upper bound for one host pause (also caps Retry-After).
|
||
HOST_BLOCK_HARD_CAP = 7200.0
|
||
|
||
# HTTP statuses whose queue items are requeued (retried later) instead
|
||
# of being failed terminally.
|
||
TRANSIENT_HTTP_STATUSES = {408, 425, 429, 500, 502, 503, 504}
|
||
|
||
# ============================================================
|
||
# URL PRIORITY CONFIG
|
||
# ============================================================
|
||
# The queue is consumed with ORDER BY priority DESC, but so far every
|
||
# URL was inserted with priority 0, so the sort never differentiated.
|
||
# The constants below feed calculate_url_priority() (see the "URL
|
||
# PRIORITY" section): the score prefers the unknown - never-crawled
|
||
# URLs, new domains, new hosts - and gives hosts with little presence
|
||
# in the crawl so far a spread bonus, WITHOUT ever blocking a domain.
|
||
# A heavily crawled host merely loses priority, so crawl capacity
|
||
# flows to new territory first.
|
||
|
||
# Base score every URL starts with (the old "everything equal" level).
|
||
PRIORITY_BASE = 100
|
||
|
||
# URL is not yet present in crawl_queue/pages (strongest single signal).
|
||
PRIORITY_BONUS_NEW_URL = 200
|
||
|
||
# Registrable domain has no pages and no queue entries yet.
|
||
PRIORITY_BONUS_NEW_DOMAIN = 120
|
||
|
||
# Host is unknown although the domain may already be known (subdomain
|
||
# or www variant of an existing site).
|
||
PRIORITY_BONUS_NEW_HOST = 60
|
||
|
||
# Spread bonus: log2-falloff boost for hosts with a small presence in
|
||
# the crawl, so many different domains/hosts are reached before the
|
||
# crawl digs deeper into a single ecosystem. 0 for hosts that do not
|
||
# exist in the DB yet (those get the explicit new-host bonuses above).
|
||
PRIORITY_SPREAD_BONUS_MAX = 80
|
||
PRIORITY_SPREAD_BONUS_DIVISOR = 20.0
|
||
|
||
# Source preference, integrated into the new system instead of
|
||
# replacing it: links out of little-explored territory are promising,
|
||
# links from a heavily crawled source are pulled slightly back.
|
||
PRIORITY_SOURCE_NOVELTY_BONUS = 20
|
||
PRIORITY_SOURCE_PENALTY_DIVISOR = 10.0
|
||
|
||
# Same-host link floods (big navigation lists, archives) are flattened,
|
||
# not blocked: a host with this many known URLs gets the penalty.
|
||
PRIORITY_CLUSTER_HOST_THRESHOLD = 25
|
||
PRIORITY_CLUSTER_PENALTY = 60
|
||
|
||
# URL variants of already known pages (trailing slash, tracking,
|
||
# session or pagination parameters) lose against genuinely new pages.
|
||
PRIORITY_VARIANT_PENALTY = 30
|
||
|
||
# How many known URLs are sampled for the variant check, so batches
|
||
# with many known URLs cannot turn into a runtime blow-up.
|
||
VARIANT_CHECK_SAMPLE_LIMIT = 200
|
||
|
||
# Query-string keys that mark trivial variants of the same page.
|
||
VARIANT_QUERY_KEYS = {
|
||
"session", "sessionid", "sid", "phpsessid", "jsessionid",
|
||
"aspsessionid", "utm_source", "utm_medium", "utm_campaign",
|
||
"utm_term", "utm_content", "fbclid", "gclid", "ref", "referer",
|
||
"page", "p", "start", "offset", "from",
|
||
}
|
||
|
||
# How long a crawler process reuses its aggregate host/domain counts
|
||
# before reloading them from the database. Each process keeps its own
|
||
# view; slightly stale counts only shift priorities a little, never
|
||
# correctness, so no cross-process locking is needed.
|
||
HOST_STATS_CACHE_SECONDS = 30
|
||
|
||
# Row cap for the statistics queries in _load_host_stats(). These are
|
||
# arbitrary samples over tables without host columns (the schema must
|
||
# not be extended), not complete counts - they only steer priorities,
|
||
# never block URLs. Missed hosts are treated as unexplored, which errs
|
||
# in favor of spread.
|
||
HOST_STATS_SAMPLE_LIMIT = 20000
|
||
|
||
|
||
# ============================================================
|
||
# DEPLOYMENT / UPDATE (generation-based rolling updates)
|
||
# ============================================================
|
||
#
|
||
# The deployment_state table is the global switch for rolling
|
||
# updates (exactly one row, id = 1):
|
||
#
|
||
# CREATE TABLE IF NOT EXISTS deployment_state (
|
||
# id INTEGER PRIMARY KEY,
|
||
# generation BIGINT NOT NULL,
|
||
# required_version INTEGER NOT NULL
|
||
# )
|
||
#
|
||
# - generation: global deployment counter. The launcher bumps
|
||
# it when a new crawler version goes live.
|
||
# - required_version: the crawler version this generation requires.
|
||
#
|
||
# Every crawler instance reads the generation ONCE at startup ("my
|
||
# generation is N"). Before every relevant PostgreSQL commit
|
||
# (safe_commit) and additionally in a periodic background check it
|
||
# compares the current generation with its own:
|
||
#
|
||
# - identical -> everything fine, continue.
|
||
# - different -> the open transaction is rolled back cleanly and the
|
||
# process exits with EXIT_UPDATE_REQUIRED (42). The
|
||
# launcher recognizes that exit code, provisions the
|
||
# new version and starts it; the new process reads the
|
||
# new generation at startup and runs on.
|
||
#
|
||
# No process is ever killed mid-transaction: exits happen only at the
|
||
# crawler's own safe commit points.
|
||
#
|
||
# CRAWLER_VERSION (CONFIG section) is the version of THIS crawler
|
||
# code. It is NOT the generation and must be increased with every
|
||
# released crawler change.
|
||
|
||
|
||
class UpdateRequired(Exception):
|
||
"""
|
||
Raised at a safe point (before a commit) when a new deployment
|
||
generation is active. safe_commit() rolls the open transaction
|
||
back BEFORE raising, so no work is ever committed half-done.
|
||
"""
|
||
|
||
|
||
class UpdateSignal:
|
||
"""
|
||
Per-process update state (every crawler worker process owns one).
|
||
|
||
generation: the deployment generation this process started with.
|
||
pending: True once a check noticed a new generation; the
|
||
process stops at the next safe commit point.
|
||
stop_checker: tells the background checker thread to finish.
|
||
"""
|
||
|
||
def __init__(self):
|
||
self.generation = None
|
||
self.pending = False
|
||
self.stop_checker = False
|
||
|
||
def check(self):
|
||
"""Raise UpdateRequired if an update is pending."""
|
||
if self.pending:
|
||
raise UpdateRequired(
|
||
"new deployment generation active; update required"
|
||
)
|
||
|
||
|
||
update_signal = UpdateSignal()
|
||
|
||
|
||
def ensure_deployment_state_table(conn):
|
||
"""
|
||
Idempotent bootstrap for deployment_state (fresh databases):
|
||
creates the table if needed and makes sure exactly one row with
|
||
id = 1 exists. Safe to run from several crawler instances at once
|
||
(CREATE TABLE IF NOT EXISTS + ON CONFLICT DO NOTHING).
|
||
"""
|
||
with conn.cursor() as cur:
|
||
cur.execute("""
|
||
CREATE TABLE IF NOT EXISTS deployment_state (
|
||
id INTEGER PRIMARY KEY,
|
||
generation BIGINT NOT NULL,
|
||
required_version INTEGER NOT NULL
|
||
)
|
||
""")
|
||
cur.execute("""
|
||
INSERT INTO deployment_state (id, generation, required_version)
|
||
VALUES (1, 1, %s)
|
||
ON CONFLICT (id) DO NOTHING
|
||
""", (CRAWLER_VERSION,))
|
||
conn.commit()
|
||
|
||
|
||
def _fetch_generation_row(conn):
|
||
"""
|
||
Read the deployment_state row (id = 1). If the table does not exist
|
||
yet, it is bootstrapped (see ensure_deployment_state_table) and the
|
||
row is read again. Returns the row or None (table without row).
|
||
"""
|
||
try:
|
||
with conn.cursor() as cur:
|
||
cur.execute(
|
||
"SELECT generation FROM deployment_state WHERE id = 1"
|
||
)
|
||
row = cur.fetchone()
|
||
conn.rollback() # read-only; leave no open transaction behind
|
||
return row
|
||
except psycopg2.errors.UndefinedTable:
|
||
conn.rollback()
|
||
logging.warning(
|
||
"deployment_state table missing; creating it with "
|
||
"generation 1 and required_version %d.",
|
||
CRAWLER_VERSION,
|
||
)
|
||
ensure_deployment_state_table(conn)
|
||
with conn.cursor() as cur:
|
||
cur.execute(
|
||
"SELECT generation FROM deployment_state WHERE id = 1"
|
||
)
|
||
row = cur.fetchone()
|
||
conn.rollback()
|
||
return row
|
||
|
||
|
||
def get_current_generation(conn=None):
|
||
"""
|
||
Read the current deployment generation. Called once per process at
|
||
startup: this becomes "my generation" (update_signal.generation).
|
||
"""
|
||
own_conn = conn is None
|
||
if own_conn:
|
||
conn = db_connect()
|
||
try:
|
||
row = _fetch_generation_row(conn)
|
||
if row is None:
|
||
logging.warning(
|
||
"deployment_state has no row with id = 1; inserting "
|
||
"generation 1 with required_version %d.",
|
||
CRAWLER_VERSION,
|
||
)
|
||
ensure_deployment_state_table(conn)
|
||
row = _fetch_generation_row(conn)
|
||
return int(row[0])
|
||
finally:
|
||
if own_conn:
|
||
conn.close()
|
||
|
||
|
||
def periodic_update_check():
|
||
"""
|
||
Regular update check OUTSIDE any transaction, so instances that
|
||
currently do not commit still notice a new generation.
|
||
|
||
Runs as a daemon thread per worker and uses its own short-lived
|
||
connection, so it can never interfere with the worker's open
|
||
transaction. A new generation only sets update_signal.pending;
|
||
the actual exit happens at the next safe commit point. A failed
|
||
check (e.g. short DB hiccup) is simply retried next interval.
|
||
"""
|
||
while not (
|
||
STOP.is_set()
|
||
or update_signal.pending
|
||
or update_signal.stop_checker
|
||
):
|
||
time.sleep(UPDATE_CHECK_INTERVAL_SECONDS)
|
||
if STOP.is_set() or update_signal.pending or update_signal.stop_checker:
|
||
break
|
||
|
||
try:
|
||
conn = db_connect()
|
||
except Exception:
|
||
continue
|
||
|
||
try:
|
||
with conn.cursor() as cur:
|
||
cur.execute(
|
||
"SELECT generation FROM deployment_state WHERE id = 1"
|
||
)
|
||
row = cur.fetchone()
|
||
conn.rollback()
|
||
except Exception:
|
||
try:
|
||
conn.rollback()
|
||
except Exception:
|
||
pass
|
||
continue
|
||
finally:
|
||
try:
|
||
conn.close()
|
||
except Exception:
|
||
pass
|
||
|
||
if row is None:
|
||
continue
|
||
|
||
try:
|
||
current = int(row[0])
|
||
except (TypeError, ValueError):
|
||
continue
|
||
|
||
if (
|
||
not update_signal.pending
|
||
and update_signal.generation is not None
|
||
and current != update_signal.generation
|
||
):
|
||
update_signal.pending = True
|
||
logging.warning(
|
||
"[UPDATE] Deployment generation changed: %d -> %d. "
|
||
"This worker (version %d) stops at the next safe "
|
||
"commit point.",
|
||
update_signal.generation,
|
||
current,
|
||
CRAWLER_VERSION,
|
||
)
|
||
|
||
|
||
def safe_commit(conn, operation=None):
|
||
"""
|
||
The ONLY commit path for crawling work.
|
||
|
||
1. If an update is pending, roll the open transaction back and
|
||
raise UpdateRequired: the caller's writes are discarded
|
||
cleanly, nothing is committed half-done.
|
||
2. Otherwise run the writes (operation) and commit them.
|
||
3. If the operation hits ProgramLimitExceeded (e.g. a tsvector
|
||
overflow - PostgreSQL hard-limits one tsvector value to
|
||
1,048,575 bytes), roll the aborted transaction back and
|
||
re-raise, so the caller can skip the work item cleanly
|
||
instead of dying on a poisoned connection.
|
||
|
||
Every crawler commit that does real work goes through here
|
||
(page saves, queue inserts, queue claim/finish, stats), so a
|
||
deployment switch can never cut a transaction in half. The
|
||
one-off startup commits (seeding, stats table) are routed
|
||
through this function as well.
|
||
"""
|
||
if update_signal.pending:
|
||
try:
|
||
conn.rollback()
|
||
except Exception:
|
||
logging.exception("Rollback before update exit failed.")
|
||
raise UpdateRequired(
|
||
"new deployment generation active; update required"
|
||
)
|
||
|
||
try:
|
||
result = operation() if operation is not None else None
|
||
except psycopg2.errors.ProgramLimitExceeded:
|
||
# The failed transaction has poisoned the connection ("current
|
||
# transaction is aborted, commands ignored until end of
|
||
# transaction block"); roll it back so the connection stays
|
||
# usable, then let the caller decide (worker: skip the page,
|
||
# queue_urls: retry the batch). Without the rollback the very
|
||
# next statement on this connection would fail as well.
|
||
try:
|
||
conn.rollback()
|
||
except Exception:
|
||
logging.exception(
|
||
"Rollback after ProgramLimitExceeded failed."
|
||
)
|
||
raise
|
||
conn.commit()
|
||
return result
|
||
|
||
|
||
# ============================================================
|
||
# LOGGING / SHUTDOWN
|
||
# ============================================================
|
||
|
||
logging.basicConfig(
|
||
level=logging.INFO,
|
||
format="%(asctime)s [%(processName)s] %(levelname)s %(message)s",
|
||
datefmt="%H:%M:%S",
|
||
)
|
||
|
||
STOP = mp.Event()
|
||
|
||
|
||
def handle_signal(signum, _frame):
|
||
logging.info("Shutdown signal received.")
|
||
STOP.set()
|
||
|
||
|
||
signal.signal(signal.SIGINT, handle_signal)
|
||
signal.signal(signal.SIGTERM, handle_signal)
|
||
|
||
|
||
# ============================================================
|
||
# DATABASE
|
||
# ============================================================
|
||
|
||
def db_connect():
|
||
return psycopg2.connect(
|
||
host=DB_HOST,
|
||
port=DB_PORT,
|
||
dbname=DB_NAME,
|
||
user=DB_USER,
|
||
password=DB_PASSWORD,
|
||
connect_timeout=10,
|
||
application_name="jonassearch-crawler",
|
||
)
|
||
|
||
|
||
def ensure_stats_table(conn):
|
||
"""Create crawler_stats. Writes only - commit happens in
|
||
safe_commit()."""
|
||
with conn.cursor() as cur:
|
||
cur.execute("""
|
||
CREATE TABLE IF NOT EXISTS crawler_stats (
|
||
id BIGSERIAL PRIMARY KEY,
|
||
recorded_at TIMESTAMPTZ NOT NULL DEFAULT now(),
|
||
crawled INTEGER NOT NULL DEFAULT 0,
|
||
queued INTEGER NOT NULL DEFAULT 0,
|
||
failed INTEGER NOT NULL DEFAULT 0
|
||
)
|
||
""")
|
||
|
||
|
||
def write_stats(conn, crawled, queued, failed):
|
||
"""Insert one stats row. Writes only - commit happens in
|
||
safe_commit()."""
|
||
with conn.cursor() as cur:
|
||
cur.execute("""
|
||
INSERT INTO crawler_stats (crawled, queued, failed)
|
||
VALUES (%s, %s, %s)
|
||
""", (crawled, queued, failed))
|
||
|
||
|
||
def claim_next_queue_item(conn, scheduler, robots):
|
||
"""
|
||
Atomically claim the best currently-crawlable pending URL.
|
||
|
||
Candidates are read in priority order (same ORDER BY as before) in
|
||
small batches with FOR UPDATE SKIP LOCKED, so several crawler
|
||
processes can still share one queue without waiting for each
|
||
other. Hosts that the HostScheduler already knows to be busy
|
||
(per-host crawl delay, active block backoff) are excluded by the
|
||
query itself. For the remaining candidates the scheduler decides
|
||
whether a host may be contacted right now. The first ready
|
||
candidate is claimed, so the worker automatically works on another
|
||
domain while a rate-limited or blocked host cools down.
|
||
|
||
The SQL-level host filter is what keeps a blocked host from
|
||
starving the whole queue: blocked rows stay 'pending' at the top
|
||
of the priority order, and without the filter a wall of them could
|
||
fill every scan window forever while tens of thousands of other
|
||
hosts wait behind it.
|
||
|
||
Returns (item, wait_hint):
|
||
item (id, url, source_url) of the claimed URL, or None
|
||
wait_hint seconds until the soonest examined host becomes
|
||
crawlable again (None when the queue is empty)
|
||
|
||
Writes only - the commit happens in safe_commit().
|
||
"""
|
||
now = time.monotonic()
|
||
seen_ids = []
|
||
scanned = 0
|
||
wait_hint = None
|
||
item = None
|
||
|
||
busy_hosts = scheduler.busy_hosts(now)
|
||
# Longest waits first: those hosts form the wall at the head of the
|
||
# priority order and are the most valuable ones to exclude.
|
||
excluded_hosts = [
|
||
host for host, _wait in sorted(
|
||
busy_hosts.items(), key=lambda pair: pair[1], reverse=True
|
||
)
|
||
][:QUEUE_CLAIM_BUSY_HOST_LIMIT]
|
||
|
||
while item is None and scanned < QUEUE_CLAIM_SCAN_LIMIT:
|
||
with conn.cursor() as cur:
|
||
cur.execute("""
|
||
SELECT id, url, source_url
|
||
FROM crawl_queue
|
||
WHERE status = 'pending'
|
||
AND NOT (id = ANY(%s::bigint[]))
|
||
AND lower(split_part(split_part(url, '://', 2), '/', 1))
|
||
<> ALL(%s::text[])
|
||
ORDER BY priority DESC, discovered_at ASC, id ASC
|
||
FOR UPDATE SKIP LOCKED
|
||
LIMIT %s
|
||
""", (seen_ids, excluded_hosts, QUEUE_CLAIM_BATCH))
|
||
candidates = cur.fetchall()
|
||
|
||
if not candidates:
|
||
break
|
||
|
||
for queue_id, url, source_url in candidates:
|
||
seen_ids.append(queue_id)
|
||
scanned += 1
|
||
|
||
wait_for = scheduler.wait_time(url, robots.delay(url), now)
|
||
if wait_for > 0.0:
|
||
if wait_hint is None or wait_for < wait_hint:
|
||
wait_hint = wait_for
|
||
continue
|
||
|
||
# Host is crawlable: claim it. The row is already locked by
|
||
# this transaction (FOR UPDATE SKIP LOCKED), so the
|
||
# status='pending' guard is just belt and braces against
|
||
# races with other claim paths.
|
||
with conn.cursor() as cur:
|
||
cur.execute("""
|
||
UPDATE crawl_queue
|
||
SET status = 'processing',
|
||
attempts = attempts + 1,
|
||
last_attempt_at = now()
|
||
WHERE id = %s AND status = 'pending'
|
||
RETURNING id, url, source_url
|
||
""", (queue_id,))
|
||
row = cur.fetchone()
|
||
|
||
if row:
|
||
item = row
|
||
break
|
||
|
||
if len(candidates) < QUEUE_CLAIM_BATCH:
|
||
break
|
||
|
||
if item is None and scanned == 0 and excluded_hosts:
|
||
# The busy-host filter removed every pending URL from the query
|
||
# before it could return a single candidate. Report when the
|
||
# soonest host is crawlable again - but only if there really are
|
||
# pending URLs; otherwise the queue is simply empty.
|
||
with conn.cursor() as cur:
|
||
cur.execute(
|
||
"SELECT 1 FROM crawl_queue WHERE status = 'pending' LIMIT 1"
|
||
)
|
||
if cur.fetchone() is not None:
|
||
wait_hint = min(busy_hosts.values())
|
||
|
||
return item, wait_hint
|
||
|
||
|
||
def defer_queue_item(conn, queue_id):
|
||
"""
|
||
Put a claimed item back WITHOUT burning an attempt: the host turned
|
||
out not to be crawlable at fetch time (e.g. a robots.txt crawl-delay
|
||
larger than MAX_INLINE_RATE_WAIT was discovered only after
|
||
claiming). The item stays pending and is retried when the host's
|
||
schedule allows it - the worker crawls other domains until then.
|
||
|
||
Writes only - the commit happens in safe_commit().
|
||
"""
|
||
with conn.cursor() as cur:
|
||
cur.execute("""
|
||
UPDATE crawl_queue
|
||
SET status = 'pending',
|
||
attempts = GREATEST(attempts - 1, 0)
|
||
WHERE id = %s
|
||
""", (queue_id,))
|
||
|
||
|
||
def requeue_queue_item(conn, queue_id):
|
||
"""
|
||
Requeue an item after a transient failure (HTTP 429/5xx, network
|
||
errors): the host got a backoff, the URL is retried afterwards.
|
||
After QUEUE_MAX_ATTEMPTS attempts the item is marked failed for
|
||
good, so a permanently broken host cannot spin forever.
|
||
|
||
Returns the new status ('pending' or 'failed'), or None.
|
||
|
||
Writes only - the commit happens in safe_commit().
|
||
"""
|
||
with conn.cursor() as cur:
|
||
cur.execute("""
|
||
UPDATE crawl_queue
|
||
SET status = CASE
|
||
WHEN attempts >= %s THEN 'failed'
|
||
ELSE 'pending'
|
||
END
|
||
WHERE id = %s
|
||
RETURNING status
|
||
""", (QUEUE_MAX_ATTEMPTS, queue_id))
|
||
row = cur.fetchone()
|
||
return row[0] if row else None
|
||
|
||
|
||
def finish_queue_item(conn, queue_id, success=True):
|
||
"""Mark a queue item done/failed. Writes only - commit happens in
|
||
safe_commit()."""
|
||
with conn.cursor() as cur:
|
||
cur.execute("""
|
||
UPDATE crawl_queue
|
||
SET status = %s
|
||
WHERE id = %s
|
||
""", ("done" if success else "failed", queue_id))
|
||
|
||
|
||
def requeue_stale_items(conn, older_than_seconds=3600):
|
||
"""
|
||
Recover URLs left in 'processing' after a crash.
|
||
|
||
Writes only - commit happens in safe_commit().
|
||
"""
|
||
with conn.cursor() as cur:
|
||
cur.execute("""
|
||
UPDATE crawl_queue
|
||
SET status = 'pending'
|
||
WHERE status = 'processing'
|
||
AND last_attempt_at < now() - (%s * interval '1 second')
|
||
""", (older_than_seconds,))
|
||
changed = cur.rowcount
|
||
if changed:
|
||
logging.info("Requeued %d stale processing URLs.", changed)
|
||
|
||
|
||
def _insert_queue_rows(conn, rows):
|
||
"""
|
||
Insert pre-ordered queue rows and return the number of newly
|
||
inserted rows.
|
||
|
||
All pages run inside ONE transaction (committed by safe_commit()
|
||
after this function returns), so the batch stays atomic as before.
|
||
rowcount is accumulated per page because execute_values() does not
|
||
aggregate rowcount across pages.
|
||
"""
|
||
inserted = 0
|
||
with conn.cursor() as cur:
|
||
for start in range(0, len(rows), QUEUE_INSERT_PAGE_SIZE):
|
||
page = rows[start:start + QUEUE_INSERT_PAGE_SIZE]
|
||
execute_values(
|
||
cur,
|
||
"""
|
||
INSERT INTO crawl_queue (url, source_url, priority, status)
|
||
VALUES %s
|
||
ON CONFLICT (url) DO NOTHING
|
||
""",
|
||
page,
|
||
template="(%s, %s, %s, %s)",
|
||
page_size=QUEUE_INSERT_PAGE_SIZE,
|
||
)
|
||
inserted += max(cur.rowcount, 0)
|
||
|
||
return inserted
|
||
|
||
|
||
def queue_urls(conn, links, source_url):
|
||
"""
|
||
Queue links for crawling, deadlock-safe for parallel crawlers that
|
||
share one crawl_queue and its UNIQUE index crawl_queue_url_key.
|
||
|
||
Strategy (no DB schema or index changes required):
|
||
- Clean and deduplicate the links within the batch.
|
||
- Compute priorities for all URLs (novelty + spread + source
|
||
preference - repetition penalties) with batched DB lookups only.
|
||
- Sort by priority, then deterministically by URL, so that
|
||
concurrent crawlers insert the same URLs in the same order into
|
||
the UNIQUE index. Lock acquisition on the index then happens in
|
||
a consistent order, which prevents most lock inversions
|
||
(deadlocks) at the source.
|
||
- ON CONFLICT (url) DO NOTHING keeps re-inserts harmless.
|
||
- If PostgreSQL still detects a deadlock, roll back the aborted
|
||
transaction and retry the whole batch a few times with a short
|
||
growing backoff instead of crashing the worker. Any other
|
||
database error is handled the same way.
|
||
"""
|
||
if not links:
|
||
return 0
|
||
|
||
# Link URLs also come from the web and can carry NUL bytes.
|
||
safe_links = clean_text_list(links)
|
||
safe_source = clean_text(source_url)
|
||
|
||
# Skip URLs that are too large for the UNIQUE B-tree index on
|
||
# crawl_queue.url (btree entries are limited to ~2704 bytes).
|
||
filtered_links = [
|
||
url for url in safe_links
|
||
if len(url.encode("utf-8")) <= MAX_QUEUE_URL_BYTES
|
||
]
|
||
|
||
skipped = len(safe_links) - len(filtered_links)
|
||
if skipped:
|
||
logging.warning(
|
||
"[QUEUE] %d URLs skipped because they exceed %d bytes",
|
||
skipped,
|
||
MAX_QUEUE_URL_BYTES,
|
||
)
|
||
|
||
if not filtered_links:
|
||
return 0
|
||
|
||
# Priorities are computed BEFORE the deterministic URL sort: the
|
||
# priority sort decides the crawl order, the final URL ordering
|
||
# keeps index lock acquisition consistent between crawlers.
|
||
try:
|
||
rows, stats = build_queue_rows(conn, filtered_links, safe_source)
|
||
except Exception:
|
||
# Priority computation must never take the worker down: fall
|
||
# back to the previous behavior (everything priority 0) and
|
||
# still queue the links.
|
||
logging.exception(
|
||
"[QUEUE] Priority computation failed; queuing without priorities."
|
||
)
|
||
try:
|
||
conn.rollback()
|
||
except Exception:
|
||
pass
|
||
unique_links = sorted(dict.fromkeys(filtered_links))
|
||
rows = [(url, safe_source, 0, "pending") for url in unique_links]
|
||
stats = {}
|
||
|
||
if stats:
|
||
log_queue_stats(safe_source, stats)
|
||
|
||
for attempt in range(1, QUEUE_INSERT_MAX_ATTEMPTS + 1):
|
||
try:
|
||
inserted = safe_commit(
|
||
conn, lambda: _insert_queue_rows(conn, rows)
|
||
)
|
||
if attempt > 1:
|
||
logging.info(
|
||
"[QUEUE] Insert succeeded on retry %d/%d.",
|
||
attempt,
|
||
QUEUE_INSERT_MAX_ATTEMPTS,
|
||
)
|
||
return inserted
|
||
except psycopg2.errors.DeadlockDetected as exc:
|
||
# The failed transaction has poisoned the connection's
|
||
# current transaction; roll it back so the connection stays
|
||
# usable for the retry and for everything after it.
|
||
try:
|
||
conn.rollback()
|
||
except Exception:
|
||
logging.exception("[QUEUE] Rollback after deadlock failed.")
|
||
|
||
if attempt < QUEUE_INSERT_MAX_ATTEMPTS:
|
||
delay = QUEUE_INSERT_RETRY_BASE_DELAY * (2 ** (attempt - 1))
|
||
logging.warning(
|
||
"[QUEUE] Deadlock while inserting %d URLs "
|
||
"(attempt %d/%d), retrying in %.1fs: %s",
|
||
len(rows),
|
||
attempt,
|
||
QUEUE_INSERT_MAX_ATTEMPTS,
|
||
delay,
|
||
exc,
|
||
)
|
||
time.sleep(delay)
|
||
except psycopg2.Error as exc:
|
||
# Any other database error (connection loss, constraint
|
||
# trouble, ...) must not crash the worker either. Same
|
||
# controlled retry with backoff as for deadlocks.
|
||
try:
|
||
conn.rollback()
|
||
except Exception:
|
||
logging.exception("[QUEUE] Rollback after DB error failed.")
|
||
|
||
if attempt < QUEUE_INSERT_MAX_ATTEMPTS:
|
||
delay = QUEUE_INSERT_RETRY_BASE_DELAY * (2 ** (attempt - 1))
|
||
logging.warning(
|
||
"[QUEUE] DB error while inserting %d URLs "
|
||
"(attempt %d/%d), retrying in %.1fs: %s",
|
||
len(rows),
|
||
attempt,
|
||
QUEUE_INSERT_MAX_ATTEMPTS,
|
||
delay,
|
||
exc,
|
||
)
|
||
time.sleep(delay)
|
||
|
||
# All retries exhausted. Log this clearly and keep the worker
|
||
# alive; the skipped URLs can be rediscovered from other pages.
|
||
logging.error(
|
||
"[QUEUE] Insert of %d URLs from %s failed after %d attempts. "
|
||
"Sample skipped URLs: %s",
|
||
len(rows),
|
||
safe_source or "-",
|
||
QUEUE_INSERT_MAX_ATTEMPTS,
|
||
", ".join(sorted(url for url, _s, _p, _st in rows)[:5])
|
||
+ ("..." if len(rows) > 5 else ""),
|
||
)
|
||
return 0
|
||
|
||
|
||
# ============================================================
|
||
# URL PRIORITY
|
||
# ============================================================
|
||
|
||
# Hosts known to serve near-infinite URL spaces (URL floods). Not a
|
||
# block list: links to these hosts are still queued, they just do not
|
||
# get preferential treatment for sheer quantity.
|
||
DIVERSITY_SOFT_HOSTS = {"github.com"}
|
||
|
||
# Cache for aggregate host/domain counts, per process. Multiple
|
||
# crawler processes/instances each maintain their own view; counts are
|
||
# advisory only (they shape priorities, never block), so a slightly
|
||
# stale view between reloads is harmless by design.
|
||
_HOST_STATS = {
|
||
"loaded_at": 0.0,
|
||
"host_counts": {}, # registrable domain -> crawled pages (pages)
|
||
"host_page_counts": {}, # hostname -> crawled pages (pages)
|
||
"queue_counts": {}, # hostname -> pending queue entries (crawl_queue)
|
||
}
|
||
|
||
|
||
def _registerable_domain(hostname):
|
||
"""
|
||
Registrable domain of a hostname, approximated with the last two
|
||
labels (example.com, sub.example.org -> example.com). Public Suffix
|
||
List handling would be more precise but is intentionally out of
|
||
scope here, matching the existing simple TLD logic in parse_html.
|
||
"""
|
||
if not hostname:
|
||
return ""
|
||
labels = [label for label in hostname.split(".") if label]
|
||
if len(labels) >= 2:
|
||
return ".".join(labels[-2:])
|
||
return "".join(labels)
|
||
|
||
|
||
def _split_url(url):
|
||
"""
|
||
Split a normalized URL into (hostname, registrable_domain) using
|
||
the same URL processing as the rest of the crawler (urlparse).
|
||
Falls back to the empty hostname rather than raising so that one
|
||
malformed URL can never break a batch.
|
||
"""
|
||
try:
|
||
hostname = (urlparse(url).hostname or "").lower().strip(".")
|
||
except ValueError:
|
||
return "", ""
|
||
return hostname, _registerable_domain(hostname)
|
||
|
||
|
||
def _strip_tracking_query(url):
|
||
"""
|
||
Remove tracking/session-style query parameters; returns None if the
|
||
URL has no such parameters at all.
|
||
"""
|
||
try:
|
||
parsed = urlparse(url)
|
||
except ValueError:
|
||
return None
|
||
if not parsed.query:
|
||
return None
|
||
|
||
query_pairs = [
|
||
(key, value) for key, value in parse_qsl(parsed.query, keep_blank_values=True)
|
||
if key.lower() not in VARIANT_QUERY_KEYS
|
||
]
|
||
if len(query_pairs) == len(parse_qsl(parsed.query)):
|
||
return None
|
||
|
||
query = "&".join(
|
||
f"{key}={value}" for key, value in query_pairs
|
||
)
|
||
netloc = parsed.netloc
|
||
stripped = f"{parsed.scheme}://{netloc}{parsed.path or '/'}"
|
||
if query:
|
||
stripped += "?" + query
|
||
return stripped
|
||
|
||
|
||
def _is_probable_variant(url, known_urls):
|
||
"""
|
||
Heuristic for "URL looks like a trivial variant of an already known
|
||
page": same address after dropping tracking/session/pagination
|
||
parameters, or a trailing-slash twin of a known URL. Purely local
|
||
computation - known_urls comes from a single batched DB lookup.
|
||
"""
|
||
if url in known_urls:
|
||
return True
|
||
|
||
stripped = _strip_tracking_query(url)
|
||
if stripped is not None and stripped in known_urls:
|
||
return True
|
||
|
||
if url.endswith("/") and url[:-1] in known_urls:
|
||
return True
|
||
if not url.endswith("/") and (url + "/") in known_urls:
|
||
return True
|
||
|
||
return False
|
||
|
||
|
||
def _load_host_stats(conn):
|
||
"""
|
||
Refresh the aggregate host/domain view if it is stale.
|
||
|
||
Two read-only grouped queries in ONE transaction, so this cannot
|
||
interact with concurrent queue inserts:
|
||
- pages.domain stores the crawl-time hostname; page counts are
|
||
folded to their registrable domain (and kept per host) in Python.
|
||
- crawl_queue hosts are extracted from the URL, so the queue counts
|
||
stay in sync with what is actually pending.
|
||
"""
|
||
now = time.monotonic()
|
||
if now - _HOST_STATS["loaded_at"] < HOST_STATS_CACHE_SECONDS:
|
||
return
|
||
|
||
try:
|
||
with conn.cursor() as cur:
|
||
cur.execute(
|
||
"SELECT domain FROM pages "
|
||
"ORDER BY domain LIMIT %s",
|
||
(HOST_STATS_SAMPLE_LIMIT,),
|
||
)
|
||
domain_counts = defaultdict(int)
|
||
host_page_counts = defaultdict(int)
|
||
for (host,) in cur.fetchall():
|
||
hostname = (host or "").lower().strip(".")
|
||
if not hostname:
|
||
continue
|
||
host_page_counts[hostname] += 1
|
||
domain_counts[_registerable_domain(hostname)] += 1
|
||
|
||
cur.execute(
|
||
"SELECT url FROM crawl_queue "
|
||
"WHERE status = 'pending' "
|
||
"ORDER BY url LIMIT %s",
|
||
(HOST_STATS_SAMPLE_LIMIT,),
|
||
)
|
||
queue_counts = defaultdict(int)
|
||
for (queued_url,) in cur.fetchall():
|
||
hostname, _domain = _split_url(queued_url)
|
||
if hostname:
|
||
queue_counts[hostname] += 1
|
||
except Exception:
|
||
# Statistics are an optimization. On failure keep the last view
|
||
# (or an empty one); the crawl continues with slightly worse
|
||
# priorities instead of losing URLs.
|
||
logging.exception("[QUEUE] Host statistics could not be loaded.")
|
||
try:
|
||
conn.rollback()
|
||
except Exception:
|
||
pass
|
||
_HOST_STATS["loaded_at"] = now
|
||
return
|
||
|
||
_HOST_STATS["host_counts"] = domain_counts
|
||
_HOST_STATS["host_page_counts"] = host_page_counts
|
||
_HOST_STATS["queue_counts"] = queue_counts
|
||
_HOST_STATS["loaded_at"] = now
|
||
|
||
|
||
def calculate_url_priority(
|
||
url,
|
||
*,
|
||
url_known,
|
||
domain_page_counts,
|
||
host_page_counts,
|
||
host_queued,
|
||
source_domain_pages,
|
||
batch_host_known,
|
||
):
|
||
"""
|
||
Central priority computation for one URL.
|
||
|
||
Concept (see the URL PRIORITY CONFIG constants):
|
||
|
||
priority = base
|
||
+ novelty bonuses for unknown URL / domain / host
|
||
+ spread bonus for little-represented hosts
|
||
+ source preference from the previously used rules
|
||
- repetition/clustering penalties
|
||
- variant penalty for near-duplicates of known pages
|
||
|
||
All inputs are plain integers looked up beforehand; this function
|
||
does no I/O of its own so it stays cheap and easy to extend.
|
||
"""
|
||
hostname, domain = _split_url(url)
|
||
|
||
priority = PRIORITY_BASE
|
||
|
||
# --- Novelty bonuses --------------------------------------
|
||
if not url_known:
|
||
priority += PRIORITY_BONUS_NEW_URL
|
||
|
||
domain_pages = domain_page_counts.get(domain, 0)
|
||
host_pages = host_page_counts.get(hostname, 0)
|
||
host_known = host_pages + host_queued
|
||
|
||
if domain_pages == 0:
|
||
# Never-crawled domain: the strongest exploration signal.
|
||
priority += PRIORITY_BONUS_NEW_DOMAIN
|
||
priority += PRIORITY_BONUS_NEW_HOST
|
||
elif host_pages == 0:
|
||
# New host within an already known domain (subdomain or www
|
||
# variant of an existing site).
|
||
priority += PRIORITY_BONUS_NEW_HOST
|
||
|
||
# --- Spread bonus -----------------------------------------
|
||
# Log2 falloff: a host with little presence in the crawl gets a
|
||
# strong boost, a huge host none at all. Unseen hosts do not get a
|
||
# spread bonus (they already receive the explicit new bonuses).
|
||
if host_known > 0:
|
||
spread = PRIORITY_SPREAD_BONUS_MAX - (
|
||
math.log2(1.0 + host_known / PRIORITY_SPREAD_BONUS_DIVISOR)
|
||
* PRIORITY_SPREAD_BONUS_MAX
|
||
/ math.log2(1.0 + PRIORITY_SPREAD_BONUS_DIVISOR)
|
||
)
|
||
priority += max(0.0, min(PRIORITY_SPREAD_BONUS_MAX, spread))
|
||
|
||
# --- Source preference ------------------------------------
|
||
# Integrated from the previous behavior: links out of little-
|
||
# explored territory are promising, links from an exhausted source
|
||
# are pulled back slightly.
|
||
if source_domain_pages == 0:
|
||
priority += PRIORITY_SOURCE_NOVELTY_BONUS
|
||
else:
|
||
priority -= (
|
||
math.log2(1.0 + source_domain_pages)
|
||
* PRIORITY_SOURCE_PENALTY_DIVISOR
|
||
)
|
||
|
||
# --- Repetition / clustering penalties ---------------------
|
||
if hostname in DIVERSITY_SOFT_HOSTS:
|
||
priority -= PRIORITY_CLUSTER_PENALTY
|
||
elif batch_host_known >= PRIORITY_CLUSTER_HOST_THRESHOLD:
|
||
# A flood of links to the same host within one page: flatten it
|
||
# instead of letting one host dominate the next queue window.
|
||
priority -= PRIORITY_CLUSTER_PENALTY
|
||
|
||
if url_known:
|
||
priority -= PRIORITY_VARIANT_PENALTY
|
||
|
||
# Keep the value in a sane integer range for the existing column.
|
||
return int(max(0, min(priority, 2 ** 31 - 1)))
|
||
|
||
|
||
def _find_known_urls(conn, urls):
|
||
"""
|
||
Look up which of the batch URLs already exist, in ONE query per
|
||
table using ANY(%s) instead of per-URL queries. Returns the set of
|
||
URLs that are already known (queued or crawled).
|
||
"""
|
||
known = set()
|
||
sample = urls[:VARIANT_CHECK_SAMPLE_LIMIT]
|
||
if not sample:
|
||
return known
|
||
|
||
try:
|
||
with conn.cursor() as cur:
|
||
cur.execute(
|
||
"SELECT url FROM crawl_queue WHERE url = ANY(%s)",
|
||
(sample,),
|
||
)
|
||
known.update(row[0] for row in cur.fetchall())
|
||
|
||
cur.execute(
|
||
"SELECT url FROM pages WHERE url = ANY(%s)",
|
||
(sample,),
|
||
)
|
||
known.update(row[0] for row in cur.fetchall())
|
||
except Exception:
|
||
logging.exception(
|
||
"[QUEUE] Known-URL lookup failed; assuming all URLs are new."
|
||
)
|
||
try:
|
||
conn.rollback()
|
||
except Exception:
|
||
pass
|
||
return known
|
||
|
||
return known
|
||
|
||
|
||
def build_queue_rows(conn, links, source_url):
|
||
"""
|
||
Turn sanitized, deduplicated links into queue rows ordered by the
|
||
computed priority (highest first).
|
||
|
||
Returns (rows, stats). Stats is a small dict for batch logging:
|
||
new_domains / new_hosts / known_hosts / spread_applied / min/max.
|
||
"""
|
||
unique_links = sorted(dict.fromkeys(links))
|
||
if not unique_links:
|
||
return [], {}
|
||
|
||
_load_host_stats(conn)
|
||
domain_page_counts = _HOST_STATS["host_counts"]
|
||
host_page_counts = _HOST_STATS["host_page_counts"]
|
||
queue_counts = _HOST_STATS["queue_counts"]
|
||
|
||
# Batched lookups only, never per-URL queries: one known-URL
|
||
# sample for variant detection plus per-batch host tallies.
|
||
known_urls = _find_known_urls(conn, unique_links)
|
||
host_known_in_batch = defaultdict(int)
|
||
for url in known_urls:
|
||
hostname, _domain = _split_url(url)
|
||
if hostname:
|
||
host_known_in_batch[hostname] += 1
|
||
|
||
_source_host, source_domain = _split_url(source_url or "")
|
||
source_domain_pages = domain_page_counts.get(source_domain, 0)
|
||
|
||
rows = []
|
||
new_domains = 0
|
||
new_hosts = 0
|
||
known_hosts = 0
|
||
spread_applied = 0
|
||
|
||
for url in unique_links:
|
||
hostname, domain = _split_url(url)
|
||
url_known = url in known_urls or _is_probable_variant(url, known_urls)
|
||
|
||
priority = calculate_url_priority(
|
||
url,
|
||
url_known=url_known,
|
||
domain_page_counts=domain_page_counts,
|
||
host_page_counts=host_page_counts,
|
||
host_queued=queue_counts.get(hostname, 0),
|
||
source_domain_pages=source_domain_pages,
|
||
batch_host_known=host_known_in_batch.get(hostname, 0),
|
||
)
|
||
|
||
# Batch statistics for the [QUEUE] log line.
|
||
domain_known = bool(domain_page_counts.get(domain, 0))
|
||
host_known = bool(
|
||
host_page_counts.get(hostname, 0)
|
||
or queue_counts.get(hostname, 0)
|
||
)
|
||
if domain and not domain_known:
|
||
new_domains += 1
|
||
if hostname:
|
||
if not host_known:
|
||
new_hosts += 1
|
||
elif not url_known:
|
||
known_hosts += 1
|
||
|
||
if priority > PRIORITY_BASE:
|
||
spread_applied += 1
|
||
|
||
rows.append((url, source_url, priority, "pending"))
|
||
|
||
# Highest priority first, URL as deterministic tiebreaker. The
|
||
# ordering also keeps concurrent batches inserting in a stable
|
||
# order, which is what keeps UNIQUE-index deadlocks rare.
|
||
rows.sort(key=lambda row: (-row[2], row[0]))
|
||
|
||
stats = {
|
||
"count": len(unique_links),
|
||
"new_domains": new_domains,
|
||
"new_hosts": new_hosts,
|
||
"known_hosts": known_hosts,
|
||
"spread_applied": spread_applied,
|
||
"priority_min": min((row[2] for row in rows), default=0),
|
||
"priority_max": max((row[2] for row in rows), default=0),
|
||
}
|
||
return rows, stats
|
||
|
||
|
||
def log_queue_stats(source_url, stats):
|
||
"""
|
||
One compact log line per batch so the journal later shows whether
|
||
the crawl actually spreads across more domains/hosts.
|
||
"""
|
||
if not stats:
|
||
return
|
||
logging.info(
|
||
"[QUEUE] %d URLs | new domains: %d | new hosts: %d | "
|
||
"known hosts: %d | priority range: %d-%d | "
|
||
"spread bonus applied: %d | source: %s",
|
||
stats.get("count", 0),
|
||
stats.get("new_domains", 0),
|
||
stats.get("new_hosts", 0),
|
||
stats.get("known_hosts", 0),
|
||
stats.get("priority_min", 0),
|
||
stats.get("priority_max", 0),
|
||
stats.get("spread_applied", 0),
|
||
source_url or "-",
|
||
)
|
||
|
||
|
||
# ============================================================
|
||
# TEXT SANITIZATION
|
||
# ============================================================
|
||
|
||
def clean_text(value):
|
||
"""
|
||
Remove NUL characters from a string.
|
||
|
||
Web content can contain a literal "\x00" anywhere in the extracted
|
||
text. PostgreSQL rejects NUL bytes inside ordinary text columns, so a
|
||
single dirty string would abort the INSERT and crash the worker.
|
||
Sanitizing centrally before every DB write keeps the crawl running.
|
||
"""
|
||
if isinstance(value, str):
|
||
return value.replace("\x00", "")
|
||
return value
|
||
|
||
|
||
def clean_text_list(values):
|
||
"""Sanitize every string of a list (headings, links, ...)."""
|
||
return [clean_text(v) for v in values]
|
||
|
||
|
||
# ============================================================
|
||
# URL NORMALIZATION
|
||
# ============================================================
|
||
|
||
def is_non_crawlable_host(hostname):
|
||
"""
|
||
True for hosts the crawler must never touch: loopback names,
|
||
private-use suffixes, raw IP literals (IPv4/IPv6) and other
|
||
non-resolvable (non-real) host names. A public web crawler can only
|
||
follow real, publicly resolving domains - everything else is a LAN
|
||
target, a loopback service or a literal address.
|
||
|
||
Applied centrally in normalize_url(), so neither the seed URL nor
|
||
any discovered link ever reaches the queue.
|
||
"""
|
||
if not hostname:
|
||
return True
|
||
host = hostname.strip(".").lower()
|
||
if not host:
|
||
return True
|
||
if host == "localhost" or host == "localhost.localdomain":
|
||
return True
|
||
if host.endswith(NON_CRAWLABLE_HOST_SUFFIXES):
|
||
return True
|
||
# Raw IP literals (IPv4 and IPv6) are not real domains. Note that
|
||
# urlparse already strips the brackets of IPv6 URLs, so "::1" is
|
||
# compared directly.
|
||
try:
|
||
ipaddress.ip_address(host)
|
||
return True
|
||
except ValueError:
|
||
pass
|
||
# Single-label names (no dot) can never be public domains.
|
||
if "." not in host:
|
||
return True
|
||
return False
|
||
|
||
|
||
def is_non_crawlable_url(url):
|
||
"""
|
||
Convenience wrapper for queue entries: True when the URL points at
|
||
a non-crawlable host (localhost, raw IP, non-real domain) or cannot
|
||
be parsed at all. Unparseable URLs are treated as non-crawlable so
|
||
one malformed entry can never crash the worker.
|
||
"""
|
||
try:
|
||
hostname = (urlparse(url).hostname or "").lower()
|
||
except ValueError:
|
||
return True
|
||
return is_non_crawlable_host(hostname)
|
||
|
||
|
||
def normalize_url(url, base_url=None):
|
||
if not url:
|
||
return None
|
||
|
||
url = unescape(url).strip()
|
||
|
||
if base_url:
|
||
url = urljoin(base_url, url)
|
||
|
||
if any(url.lower().startswith(prefix) for prefix in BLOCKED_PREFIXES):
|
||
return None
|
||
|
||
url, _fragment = urldefrag(url)
|
||
|
||
parsed = urlparse(url)
|
||
|
||
if parsed.scheme.lower() not in ALLOWED_SCHEMES:
|
||
return None
|
||
|
||
if not parsed.netloc:
|
||
return None
|
||
|
||
# Lowercase scheme and hostname.
|
||
scheme = parsed.scheme.lower()
|
||
hostname = (parsed.hostname or "").lower()
|
||
|
||
if not hostname:
|
||
return None
|
||
|
||
# Never queue loopback, LAN or raw-IP targets (localhost, IPs,
|
||
# non-real domains).
|
||
if is_non_crawlable_host(hostname):
|
||
return None
|
||
|
||
# Drop default ports.
|
||
port = parsed.port
|
||
if (scheme == "http" and port == 80) or (scheme == "https" and port == 443):
|
||
port = None
|
||
|
||
if port:
|
||
netloc = f"{hostname}:{port}"
|
||
else:
|
||
netloc = hostname
|
||
|
||
# Preserve path/query, but remove empty query marker.
|
||
path = parsed.path or "/"
|
||
query = parsed.query
|
||
|
||
normalized = f"{scheme}://{netloc}{path}"
|
||
if query:
|
||
normalized += "?" + query
|
||
|
||
# Avoid obvious files.
|
||
path_lower = path.lower()
|
||
for ext in BLOCKED_EXTENSIONS:
|
||
if path_lower.endswith(ext):
|
||
return None
|
||
|
||
return normalized
|
||
|
||
|
||
def normalize_image_url(raw, base_url):
|
||
"""
|
||
Turn an image reference from HTML into an absolute http(s) URL.
|
||
|
||
Unlike normalize_url() this does NOT reject image file extensions
|
||
(that list exists to keep binary resources out of the page queue).
|
||
Relative paths and protocol-relative URLs are resolved against the
|
||
page URL; fragments are dropped because they do not identify a
|
||
different image resource. data:, javascript:, ftp: and other
|
||
non-http(s) sources are rejected, as are empty/whitespace values,
|
||
non-public hosts and URLs that are unreasonably long.
|
||
"""
|
||
if not isinstance(raw, str):
|
||
return None
|
||
|
||
raw = unescape(raw).strip()
|
||
if not raw:
|
||
return None
|
||
|
||
try:
|
||
absolute = urljoin(base_url, raw)
|
||
parsed = urlparse(absolute)
|
||
except ValueError:
|
||
return None
|
||
|
||
if parsed.scheme.lower() not in ALLOWED_SCHEMES:
|
||
return None
|
||
if not parsed.hostname:
|
||
return None
|
||
if is_non_crawlable_host(parsed.hostname):
|
||
return None
|
||
|
||
# Fragments never select a different image file.
|
||
clean, _fragment = urldefrag(absolute)
|
||
if any(char.isspace() for char in clean):
|
||
return None
|
||
if len(clean) > MAX_IMAGE_URL_LENGTH:
|
||
return None
|
||
|
||
return clean
|
||
|
||
|
||
# ============================================================
|
||
# ROBOTS
|
||
# ============================================================
|
||
|
||
class RobotsCache:
|
||
def __init__(self):
|
||
self.cache = {}
|
||
self.last_checked = {}
|
||
|
||
def allowed(self, url):
|
||
parsed = urlparse(url)
|
||
host = parsed.netloc.lower()
|
||
|
||
now = time.monotonic()
|
||
|
||
if (
|
||
host not in self.cache
|
||
or now - self.last_checked.get(host, 0) > ROBOTS_CACHE_SECONDS
|
||
):
|
||
self._load(host, parsed.scheme)
|
||
|
||
rp = self.cache[host]
|
||
try:
|
||
return rp.can_fetch(USER_AGENT, url)
|
||
except Exception:
|
||
return True
|
||
|
||
def delay(self, url):
|
||
parsed = urlparse(url)
|
||
host = parsed.netloc.lower()
|
||
|
||
rp = self.cache.get(host)
|
||
if not rp:
|
||
return DEFAULT_DELAY
|
||
|
||
# RobotFileParser does not expose Crawl-delay consistently across
|
||
# all Python versions, so use it when available.
|
||
try:
|
||
delay = rp.crawl_delay(USER_AGENT)
|
||
if delay is None:
|
||
delay = rp.crawl_delay("*")
|
||
if delay is not None:
|
||
return max(DEFAULT_DELAY, float(delay))
|
||
except Exception:
|
||
pass
|
||
|
||
return DEFAULT_DELAY
|
||
|
||
def _load(self, host, scheme):
|
||
robots_url = f"{scheme}://{host}/robots.txt"
|
||
|
||
rp = RobotFileParser()
|
||
rp.set_url(robots_url)
|
||
|
||
try:
|
||
req = Request(
|
||
robots_url,
|
||
headers={"User-Agent": USER_AGENT}
|
||
)
|
||
with urlopen(req, timeout=REQUEST_TIMEOUT) as response:
|
||
data = response.read(512 * 1024).decode(
|
||
"utf-8", errors="replace"
|
||
)
|
||
rp.parse(data.splitlines())
|
||
except Exception:
|
||
# If robots.txt cannot be retrieved, default to allowing the URL.
|
||
# This is deliberately conservative with respect to availability.
|
||
rp.parse([])
|
||
|
||
self.cache[host] = rp
|
||
self.last_checked[host] = time.monotonic()
|
||
|
||
|
||
# ============================================================
|
||
# HTTP
|
||
# ============================================================
|
||
|
||
class HostScheduler:
|
||
"""
|
||
Per-process host availability tracker - the heart of the automatic
|
||
domain switching.
|
||
|
||
The scheduler keeps, per host:
|
||
last_request: monotonic time of the last request start, so
|
||
per-host delays (DEFAULT_DELAY / robots.txt
|
||
crawl-delay) are honored.
|
||
blocked_until: monotonic time until which the host is considered
|
||
blocked (HTTP 429/5xx/403, network errors,
|
||
Retry-After).
|
||
block_count: consecutive blocks of the host; the pause doubles
|
||
with every consecutive block and resets on a
|
||
successful fetch.
|
||
|
||
claim_next_queue_item() asks wait_time() for every queue candidate:
|
||
a host that would have to wait is skipped, so the worker crawls
|
||
another domain instead and returns to the blocked host when its
|
||
backoff has expired. No worker ever sleeps in front of a
|
||
rate-limited host while other domains are crawlable.
|
||
"""
|
||
|
||
def __init__(self):
|
||
self.last_request = {}
|
||
self.blocked_until = {}
|
||
self.block_count = {}
|
||
# Delay last used for a host (robots.txt crawl-delay or
|
||
# DEFAULT_DELAY), so busy_hosts() can report the host's schedule
|
||
# without looking at any URL again.
|
||
self.host_delay = {}
|
||
|
||
@staticmethod
|
||
def _host(url):
|
||
try:
|
||
return (urlparse(url).netloc or "").lower()
|
||
except ValueError:
|
||
return ""
|
||
|
||
def wait_time(self, url, delay, now=None):
|
||
"""Seconds until url's host may be contacted again (0.0 = now)."""
|
||
if now is None:
|
||
now = time.monotonic()
|
||
host = self._host(url)
|
||
if not host:
|
||
return 0.0
|
||
self.host_delay[host] = delay
|
||
last = self.last_request.get(host)
|
||
# A host that was never contacted has no pending crawl delay:
|
||
# monotonic() has an arbitrary origin, so treating "never" as 0
|
||
# and adding a large robots delay would wrongly block the host
|
||
# for a long time on a freshly booted machine.
|
||
rate_until = last + delay if last is not None else 0.0
|
||
next_allowed = max(
|
||
rate_until,
|
||
self.blocked_until.get(host, 0.0),
|
||
)
|
||
return max(0.0, next_allowed - now)
|
||
|
||
def busy_hosts(self, now=None):
|
||
"""
|
||
Map of hosts that cannot be contacted right now to the number
|
||
of seconds they still have to wait.
|
||
|
||
The queue claim passes these hosts into SQL: skipping busy hosts
|
||
only within the small candidate window would let a wall of
|
||
blocked URLs at the head of the priority order hide every other
|
||
crawlable host in the queue (the wall stays pending at the top,
|
||
so each rescan looks at the very same rows forever).
|
||
|
||
Expired entries are pruned here because this runs once per
|
||
claim - the per-process state then cannot grow without bound on
|
||
a queue with millions of distinct hosts.
|
||
"""
|
||
if now is None:
|
||
now = time.monotonic()
|
||
|
||
busy = {}
|
||
|
||
for host, last in list(self.last_request.items()):
|
||
next_allowed = last + self.host_delay.get(host, 0.0)
|
||
if next_allowed > now:
|
||
busy[host] = next_allowed - now
|
||
else:
|
||
del self.last_request[host]
|
||
|
||
for host, until in list(self.blocked_until.items()):
|
||
if until > now:
|
||
busy[host] = max(busy.get(host, 0.0), until - now)
|
||
else:
|
||
del self.blocked_until[host]
|
||
|
||
return busy
|
||
|
||
def is_ready(self, url, delay, now=None):
|
||
return self.wait_time(url, delay, now) <= 0.0
|
||
|
||
def wait(self, url, delay):
|
||
"""
|
||
Enforce the per-host delay directly before a request.
|
||
|
||
Returns True when the request may start now (the wait was slept
|
||
inline). Returns False when the wait would exceed
|
||
MAX_INLINE_RATE_WAIT (e.g. a large robots.txt crawl-delay that
|
||
claim_next_queue_item could not know yet, because robots.txt is
|
||
only loaded during the first fetch): the caller then defers the
|
||
item instead of parking the worker, and the host's schedule
|
||
stays untouched, so the item is retried exactly when the delay
|
||
has passed.
|
||
"""
|
||
host = self._host(url)
|
||
self.host_delay[host] = delay
|
||
wait_for = self.wait_time(url, delay)
|
||
if wait_for > MAX_INLINE_RATE_WAIT:
|
||
return False
|
||
if wait_for > 0:
|
||
if wait_for >= 1.0:
|
||
logging.info("[RATE ] %s: %.1fs", host, wait_for)
|
||
time.sleep(wait_for)
|
||
self.last_request[host] = time.monotonic()
|
||
return True
|
||
|
||
def _register_block(self, host, seconds):
|
||
count = self.block_count.get(host, 0) + 1
|
||
self.block_count[host] = count
|
||
blocked = min(seconds, HOST_BLOCK_HARD_CAP)
|
||
self.blocked_until[host] = time.monotonic() + blocked
|
||
return count, blocked
|
||
|
||
def note_success(self, url):
|
||
"""Successful fetch: the host's consecutive-block counter resets."""
|
||
self.block_count.pop(self._host(url), None)
|
||
|
||
def note_http_error(self, url, status, retry_after=None):
|
||
"""
|
||
Pause the host after a block-ish HTTP status (429, 5xx, 403,
|
||
...). An exponential backoff applies; a Retry-After header is
|
||
honored on top of it.
|
||
"""
|
||
rule = HOST_BLOCK_RULES.get(status)
|
||
if rule is None:
|
||
return
|
||
host = self._host(url)
|
||
base, cap = rule
|
||
seconds = min(cap, base * (2 ** self.block_count.get(host, 0)))
|
||
if retry_after is not None:
|
||
seconds = max(seconds, retry_after)
|
||
count, blocked = self._register_block(host, seconds)
|
||
logging.info(
|
||
"[BLOCK] %s: HTTP %d -> host paused for %.0fs (block #%d)",
|
||
host, status, blocked, count,
|
||
)
|
||
|
||
def note_network_error(self, url):
|
||
"""Short backoff after DNS/timeout/connection failures."""
|
||
host = self._host(url)
|
||
seconds = min(
|
||
HOST_BLOCK_NETWORK_MAX,
|
||
HOST_BLOCK_NETWORK_BASE * (2 ** self.block_count.get(host, 0)),
|
||
)
|
||
count, blocked = self._register_block(host, seconds)
|
||
logging.info(
|
||
"[BLOCK] %s: network error -> host paused for %.0fs (block #%d)",
|
||
host, blocked, count,
|
||
)
|
||
|
||
|
||
def _retry_after_seconds(headers):
|
||
"""
|
||
Parse a Retry-After header (delta-seconds or HTTP date). Returns
|
||
the number of seconds, or None when absent/unparseable.
|
||
"""
|
||
try:
|
||
value = headers.get("Retry-After") if headers is not None else None
|
||
except Exception:
|
||
return None
|
||
if not value:
|
||
return None
|
||
value = value.strip()
|
||
if value.isdigit():
|
||
return float(value)
|
||
try:
|
||
retry_at = parsedate_to_datetime(value)
|
||
return max(0.0, retry_at.timestamp() - time.time())
|
||
except Exception:
|
||
return None
|
||
|
||
|
||
def is_transient_error(error):
|
||
"""
|
||
True for fetch errors worth retrying later (the item is requeued
|
||
and the host gets a backoff): network-level failures and the HTTP
|
||
statuses in TRANSIENT_HTTP_STATUSES. Everything else (404, robots,
|
||
too large, not HTML, ...) stays a terminal failure.
|
||
"""
|
||
if not error:
|
||
return False
|
||
if error.startswith("network:"):
|
||
return True
|
||
match = re.match(r"^http_(\d{3})$", error)
|
||
return bool(match) and int(match.group(1)) in TRANSIENT_HTTP_STATUSES
|
||
|
||
|
||
def is_skipped_url(url):
|
||
"""
|
||
True for URLs whose pages bring no search-index value but burn
|
||
significant crawl resources (bandwidth, parsing time, index
|
||
size). See SKIP_URL_PATTERNS.
|
||
"""
|
||
return any(pattern.match(url) for pattern in SKIP_URL_PATTERNS)
|
||
|
||
|
||
def fetch(url, scheduler, robots):
|
||
parsed = urlparse(url)
|
||
|
||
if not robots.allowed(url):
|
||
return None, None, "robots"
|
||
|
||
delay = robots.delay(url)
|
||
if not scheduler.wait(url, delay):
|
||
# The host needs a longer break than we are willing to sleep
|
||
# for inline (e.g. a large robots.txt crawl-delay discovered
|
||
# only now, because robots.txt is loaded on the first fetch):
|
||
# report it so the worker defers the item and crawls another
|
||
# domain in the meantime.
|
||
return None, None, "host_busy"
|
||
|
||
req = Request(
|
||
url,
|
||
headers={
|
||
"User-Agent": USER_AGENT,
|
||
"Accept": (
|
||
"text/html,application/xhtml+xml,"
|
||
"application/xml;q=0.9,*/*;q=0.1"
|
||
),
|
||
"Accept-Language": "en,de;q=0.8",
|
||
},
|
||
)
|
||
|
||
try:
|
||
with urlopen(req, timeout=REQUEST_TIMEOUT) as response:
|
||
status = getattr(response, "status", 200)
|
||
content_type = response.headers.get("Content-Type", "")
|
||
content_length = response.headers.get("Content-Length")
|
||
|
||
if content_length:
|
||
try:
|
||
if int(content_length) > MAX_CONTENT_LENGTH:
|
||
return status, None, "too_large"
|
||
except ValueError:
|
||
pass
|
||
|
||
# We only parse HTML/XHTML.
|
||
ctype = content_type.lower()
|
||
if ctype and not (
|
||
"text/html" in ctype
|
||
or "application/xhtml+xml" in ctype
|
||
):
|
||
return status, None, "not_html"
|
||
|
||
data = response.read(MAX_CONTENT_LENGTH + 1)
|
||
|
||
if len(data) > MAX_CONTENT_LENGTH:
|
||
return status, None, "too_large"
|
||
|
||
charset = response.headers.get_content_charset() or "utf-8"
|
||
text = data.decode(charset, errors="replace")
|
||
|
||
return status, text, None
|
||
|
||
except HTTPError as e:
|
||
# Blocked or rate-limited? Put the host on backoff so the next
|
||
# claims automatically prefer other domains.
|
||
scheduler.note_http_error(
|
||
url,
|
||
e.code,
|
||
retry_after=_retry_after_seconds(
|
||
getattr(e, "headers", None)
|
||
),
|
||
)
|
||
return e.code, None, f"http_{e.code}"
|
||
except (URLError, TimeoutError, OSError) as e:
|
||
scheduler.note_network_error(url)
|
||
return None, None, f"network:{e}"
|
||
except Exception as e:
|
||
return None, None, f"error:{e}"
|
||
|
||
|
||
# ============================================================
|
||
# HTML PARSING
|
||
# ============================================================
|
||
|
||
def clean_text_bs4(soup):
|
||
for tag in soup([
|
||
"script", "style", "noscript", "template",
|
||
"svg", "canvas", "iframe", "object",
|
||
"embed", "form"
|
||
]):
|
||
tag.decompose()
|
||
|
||
# Remove navigation/boilerplate-ish elements where possible.
|
||
for tag in soup.find_all([
|
||
"nav", "footer", "header", "aside"
|
||
]):
|
||
tag.decompose()
|
||
|
||
text = soup.get_text(" ", strip=True)
|
||
|
||
# Normalize whitespace.
|
||
text = re.sub(r"\s+", " ", text)
|
||
return text[:MAX_TEXT_LENGTH]
|
||
|
||
|
||
# Common attributes used for lazy-loaded images. They are not
|
||
# standardized, but these three are reliable enough in practice.
|
||
IMAGE_LAZY_SRC_ATTRS = ("data-src", "data-lazy-src", "data-original")
|
||
|
||
# srcset appears on <img> and on <source> inside <picture>.
|
||
IMAGE_SRCSET_ATTRS = ("srcset", "data-srcset")
|
||
|
||
# Social/social-preview image metadata.
|
||
IMAGE_META_KEYS = frozenset({
|
||
"og:image",
|
||
"og:image:url",
|
||
"og:image:secure_url",
|
||
"twitter:image",
|
||
"twitter:image:src",
|
||
})
|
||
|
||
|
||
def _srcset_candidates(value):
|
||
"""
|
||
Yield the URL part of every srcset candidate.
|
||
|
||
srcset format: "img.jpg 1x, img@2x.jpg 2x" (or a width descriptor).
|
||
Parsed like the HTML spec: whitespace separates URL and descriptor,
|
||
a trailing comma ends a candidate. This keeps commas inside URLs
|
||
(including data: URLs, which are rejected later) from being split
|
||
into bogus relative candidates.
|
||
"""
|
||
if not value:
|
||
return
|
||
|
||
tokens = str(value).split()
|
||
index = 0
|
||
while index < len(tokens):
|
||
candidate = tokens[index]
|
||
index += 1
|
||
if not candidate:
|
||
continue
|
||
|
||
ends_candidate = candidate.endswith(",")
|
||
candidate = candidate.rstrip(",").strip()
|
||
if not candidate:
|
||
continue
|
||
|
||
if not ends_candidate:
|
||
# The next token is the descriptor ("2x", "640w"), if any.
|
||
index += 1
|
||
|
||
yield candidate
|
||
|
||
|
||
def extract_pictures(soup, base_url):
|
||
"""
|
||
All absolute, deduplicated image URLs of a parsed document.
|
||
|
||
Sources: <img src>, lazy-loading data-* attributes, <img srcset>,
|
||
<picture> via <source srcset>, Open-Graph/Twitter preview images and
|
||
<link rel="image_src">. Order of first appearance is preserved.
|
||
"""
|
||
pictures = []
|
||
seen = set()
|
||
|
||
def add(raw):
|
||
if len(pictures) >= MAX_PICTURES:
|
||
return
|
||
normalized = normalize_image_url(raw, base_url)
|
||
if not normalized or normalized in seen:
|
||
return
|
||
seen.add(normalized)
|
||
pictures.append(normalized)
|
||
|
||
for tag in soup.find_all("img"):
|
||
add(tag.get("src"))
|
||
for attr in IMAGE_LAZY_SRC_ATTRS:
|
||
add(tag.get(attr))
|
||
for attr in IMAGE_SRCSET_ATTRS:
|
||
for candidate in _srcset_candidates(tag.get(attr)):
|
||
add(candidate)
|
||
|
||
for tag in soup.find_all("source"):
|
||
for attr in IMAGE_SRCSET_ATTRS:
|
||
for candidate in _srcset_candidates(tag.get(attr)):
|
||
add(candidate)
|
||
|
||
for tag in soup.find_all("meta"):
|
||
key = (tag.get("property") or tag.get("name") or "").strip().lower()
|
||
if key in IMAGE_META_KEYS:
|
||
add(tag.get("content"))
|
||
|
||
for tag in soup.find_all("link"):
|
||
rel = tag.get("rel") or []
|
||
if isinstance(rel, str):
|
||
rel = rel.lower().split()
|
||
if any(str(value).lower() == "image_src" for value in rel):
|
||
add(tag.get("href"))
|
||
|
||
return pictures
|
||
|
||
|
||
def parse_html(html, base_url):
|
||
if BeautifulSoup is None:
|
||
raise RuntimeError(
|
||
"BeautifulSoup fehlt. Installiere python3-bs4."
|
||
)
|
||
|
||
soup = BeautifulSoup(html, "html.parser")
|
||
|
||
title = ""
|
||
if soup.title:
|
||
title = soup.title.get_text(" ", strip=True)
|
||
|
||
title = re.sub(r"\s+", " ", title)[:MAX_TITLE_LENGTH]
|
||
|
||
headings = []
|
||
for tag in soup.find_all(re.compile(r"^h[1-6]$")):
|
||
value = re.sub(r"\s+", " ", tag.get_text(" ", strip=True))
|
||
if value:
|
||
headings.append(value[:MAX_HEADING_LENGTH])
|
||
if len(headings) >= MAX_HEADINGS:
|
||
break
|
||
|
||
links = []
|
||
seen = set()
|
||
|
||
for tag in soup.find_all("a", href=True):
|
||
normalized = normalize_url(tag.get("href"), base_url)
|
||
if not normalized:
|
||
continue
|
||
|
||
if normalized in seen:
|
||
continue
|
||
|
||
seen.add(normalized)
|
||
links.append(normalized)
|
||
|
||
if len(links) >= MAX_LINKS:
|
||
break
|
||
|
||
# Extract images BEFORE clean_text_bs4() mutates the tree (it
|
||
# removes navigation, noscript, svg, ... elements).
|
||
pictures = extract_pictures(soup, base_url)
|
||
|
||
content = clean_text_bs4(soup)
|
||
|
||
domain = urlparse(base_url).hostname or ""
|
||
domain = domain.lower()
|
||
|
||
# TLD is intentionally simple here; the DB can later be extended with
|
||
# Public Suffix List handling if needed.
|
||
tld = ""
|
||
if "." in domain:
|
||
tld = "." + domain.rsplit(".", 1)[-1]
|
||
|
||
return {
|
||
"title": title,
|
||
"headings": headings,
|
||
"content": content,
|
||
"links": links,
|
||
"pictures": pictures,
|
||
"domain": domain,
|
||
"tld": tld,
|
||
}
|
||
|
||
|
||
# ============================================================
|
||
# PAGE STORAGE
|
||
# ============================================================
|
||
|
||
def save_page(conn, url, source_url, status, content_type, parsed, crawl_ms):
|
||
"""
|
||
Store/refresh one page. Writes only - commit happens in
|
||
safe_commit().
|
||
"""
|
||
# Sanitize everything that came from the web before it reaches
|
||
# PostgreSQL. NUL bytes are legal in HTML but illegal in text columns.
|
||
url = clean_text(url)
|
||
source_url = clean_text(source_url)
|
||
content_type = clean_text(content_type)
|
||
|
||
safe_parsed = {
|
||
"title": clean_text(parsed["title"]),
|
||
"headings": clean_text_list(parsed["headings"]),
|
||
"content": clean_text(parsed["content"]),
|
||
"links": clean_text_list(parsed["links"]),
|
||
# parse_html() already deduplicates; repeat it here so a caller
|
||
# with a hand-built parsed dict can never write duplicates.
|
||
"pictures": list(dict.fromkeys(
|
||
clean_text_list(parsed.get("pictures") or [])
|
||
))[:MAX_PICTURES],
|
||
"domain": clean_text(parsed["domain"]),
|
||
"tld": clean_text(parsed["tld"]),
|
||
}
|
||
|
||
# Cap the text that feeds the full-text index. The DB trigger
|
||
# pages_search_vector_update() builds a tsvector from the content
|
||
# column and PostgreSQL hard-limits one tsvector value to
|
||
# 1,048,575 bytes (not configurable). Pages near MAX_TEXT_LENGTH
|
||
# overflow it and abort the whole INSERT (ProgramLimitExceeded),
|
||
# which used to kill the worker. Capping here keeps oversize
|
||
# pages saveable (with a truncated index); the trigger-side cap
|
||
# (fix_pages_search_vector_trigger.sql) adds the same protection
|
||
# for every other insert path, independent of this code.
|
||
index_content = safe_parsed["content"][:CONTENT_INDEX_MAX_CHARS]
|
||
if len(index_content) < len(safe_parsed["content"]):
|
||
logging.warning(
|
||
"[SAVE ] %s: content capped for the search index "
|
||
"(%d -> %d chars).",
|
||
url,
|
||
len(safe_parsed["content"]),
|
||
len(index_content),
|
||
)
|
||
|
||
with conn.cursor() as cur:
|
||
cur.execute("""
|
||
INSERT INTO pages (
|
||
url,
|
||
title,
|
||
headings,
|
||
content,
|
||
links,
|
||
pictures,
|
||
domain,
|
||
tld,
|
||
source_url,
|
||
http_status,
|
||
content_type,
|
||
discovered_at,
|
||
crawled_at,
|
||
crawl_time_ms
|
||
)
|
||
VALUES (
|
||
%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s,
|
||
COALESCE(
|
||
(SELECT discovered_at
|
||
FROM crawl_queue
|
||
WHERE url = %s),
|
||
now()
|
||
),
|
||
now(),
|
||
%s
|
||
)
|
||
ON CONFLICT (url) DO UPDATE SET
|
||
title = EXCLUDED.title,
|
||
headings = EXCLUDED.headings,
|
||
content = EXCLUDED.content,
|
||
links = EXCLUDED.links,
|
||
pictures = EXCLUDED.pictures,
|
||
domain = EXCLUDED.domain,
|
||
tld = EXCLUDED.tld,
|
||
source_url = EXCLUDED.source_url,
|
||
http_status = EXCLUDED.http_status,
|
||
content_type = EXCLUDED.content_type,
|
||
crawled_at = now(),
|
||
crawl_time_ms = EXCLUDED.crawl_time_ms
|
||
""", (
|
||
url,
|
||
safe_parsed["title"],
|
||
safe_parsed["headings"],
|
||
index_content,
|
||
safe_parsed["links"],
|
||
safe_parsed["pictures"],
|
||
safe_parsed["domain"],
|
||
safe_parsed["tld"],
|
||
source_url,
|
||
status,
|
||
content_type,
|
||
url,
|
||
crawl_ms,
|
||
))
|
||
|
||
|
||
# ============================================================
|
||
# WORKER
|
||
# ============================================================
|
||
|
||
def worker(worker_id):
|
||
process_name = f"crawler-{worker_id}"
|
||
mp.current_process().name = process_name
|
||
|
||
conn = None
|
||
robots = RobotsCache()
|
||
scheduler = HostScheduler()
|
||
|
||
crawled = 0
|
||
failed = 0
|
||
retried = 0
|
||
last_stats = time.monotonic()
|
||
current_queue_id = None
|
||
update_requested = False
|
||
|
||
try:
|
||
conn = db_connect()
|
||
|
||
# Remember the generation this instance started with. Read ONCE
|
||
# per process: five crawler instances all remember their own
|
||
# generation (here: the same one, until a deployment bump).
|
||
update_signal.generation = get_current_generation(conn)
|
||
|
||
logging.info(
|
||
"Started. DB=%s:%s/%s version=%d generation=%d",
|
||
DB_HOST, DB_PORT, DB_NAME,
|
||
CRAWLER_VERSION, update_signal.generation,
|
||
)
|
||
|
||
# Regular update check OUTSIDE the transactions (own short-lived
|
||
# connection, daemon thread). It only sets update_signal.pending;
|
||
# the exit itself happens at the next safe commit point, never
|
||
# in the middle of a transaction.
|
||
checker_thread = threading.Thread(
|
||
target=periodic_update_check,
|
||
name=f"{process_name}-update-check",
|
||
daemon=True,
|
||
)
|
||
checker_thread.start()
|
||
|
||
while not STOP.is_set():
|
||
# Safe point between page cycles: if the periodic check
|
||
# noticed a new generation, stop before starting new work.
|
||
if update_signal.pending:
|
||
update_requested = True
|
||
break
|
||
|
||
item, wait_hint = safe_commit(
|
||
conn,
|
||
lambda: claim_next_queue_item(conn, scheduler, robots),
|
||
)
|
||
|
||
if not item:
|
||
if wait_hint:
|
||
# Every candidate host is rate-limited or blocked
|
||
# right now; sleep until the soonest one becomes
|
||
# crawlable (capped, then the queue is re-scanned).
|
||
sleep_for = min(wait_hint, QUEUE_MAX_IDLE_SLEEP)
|
||
logging.info(
|
||
"[WAIT ] All candidate hosts busy or blocked; "
|
||
"soonest host free in %.1fs, worker sleeps %.1fs",
|
||
wait_hint,
|
||
sleep_for,
|
||
)
|
||
else:
|
||
sleep_for = 5
|
||
logging.info("Queue empty. Worker waiting...")
|
||
time.sleep(sleep_for)
|
||
continue
|
||
|
||
queue_id, url, source_url = item
|
||
current_queue_id = queue_id
|
||
|
||
logging.info("=" * 70)
|
||
logging.info("[CRAWL] %s", url)
|
||
logging.info("[FROM ] %s", source_url or "-")
|
||
|
||
# Known no-value pages (e.g. pkg.go.dev "?tab=importedby":
|
||
# megabyte-sized reverse-dependency link lists) are
|
||
# skipped before fetching. Marked 'done' so they are not
|
||
# retried endlessly.
|
||
if is_skipped_url(url):
|
||
logging.info("[SKIP ] %s (no search value)", url)
|
||
safe_commit(
|
||
conn,
|
||
lambda: finish_queue_item(
|
||
conn, queue_id, success=True
|
||
),
|
||
)
|
||
continue
|
||
|
||
# Hosts that may never be crawled (localhost, raw IPs,
|
||
# non-real domains) can still sit in the queue from before
|
||
# this rule existed: drop them here so they are not
|
||
# fetched and not retried endlessly.
|
||
if is_non_crawlable_url(url):
|
||
logging.info("[SKIP ] %s (non-crawlable host)", url)
|
||
safe_commit(
|
||
conn,
|
||
lambda: finish_queue_item(
|
||
conn, queue_id, success=True
|
||
),
|
||
)
|
||
continue
|
||
|
||
started = time.monotonic()
|
||
|
||
status, html, error = fetch(url, scheduler, robots)
|
||
|
||
if error:
|
||
logging.info("[HTTP ] %s -> %s", url, error)
|
||
|
||
if error == "host_busy":
|
||
# Host needs a longer break than we sleep for
|
||
# inline: put the item back (without burning an
|
||
# attempt) and crawl another domain instead.
|
||
safe_commit(
|
||
conn,
|
||
lambda: defer_queue_item(conn, queue_id),
|
||
)
|
||
continue
|
||
|
||
if is_transient_error(error):
|
||
# The host is already on backoff (set in fetch);
|
||
# requeue the URL for a later attempt.
|
||
new_status = safe_commit(
|
||
conn,
|
||
lambda: requeue_queue_item(conn, queue_id),
|
||
)
|
||
retried += 1
|
||
logging.info(
|
||
"[RETRY] %s -> %s (max %d attempts)",
|
||
url,
|
||
new_status or "requeued",
|
||
QUEUE_MAX_ATTEMPTS,
|
||
)
|
||
continue
|
||
|
||
safe_commit(
|
||
conn,
|
||
lambda: finish_queue_item(
|
||
conn, queue_id, success=False
|
||
),
|
||
)
|
||
failed += 1
|
||
continue
|
||
|
||
logging.info("[HTTP ] %s -> HTTP %s", url, status)
|
||
scheduler.note_success(url)
|
||
|
||
# A redirect normally arrives as the final URL in urllib.
|
||
# The requested URL remains the canonical queue identity.
|
||
if html is None:
|
||
safe_commit(
|
||
conn,
|
||
lambda: finish_queue_item(
|
||
conn, queue_id, success=False
|
||
),
|
||
)
|
||
failed += 1
|
||
continue
|
||
|
||
try:
|
||
parsed = parse_html(html, url)
|
||
except Exception as exc:
|
||
logging.exception("[PARSE] %s: %s", url, exc)
|
||
safe_commit(
|
||
conn,
|
||
lambda: finish_queue_item(
|
||
conn, queue_id, success=False
|
||
),
|
||
)
|
||
failed += 1
|
||
continue
|
||
|
||
crawl_ms = int((time.monotonic() - started) * 1000)
|
||
|
||
logging.info("[TITLE] %s", parsed["title"][:160])
|
||
logging.info("[DOMAIN] %s", parsed["domain"])
|
||
logging.info("[HEADS ] %d", len(parsed["headings"]))
|
||
logging.info("[LINKS ] %d gefunden", len(parsed["links"]))
|
||
logging.info("[IMGS ] %d gefunden", len(parsed["pictures"]))
|
||
|
||
try:
|
||
safe_commit(
|
||
conn,
|
||
lambda: save_page(
|
||
conn=conn,
|
||
url=url,
|
||
source_url=source_url,
|
||
status=status,
|
||
content_type="text/html",
|
||
parsed=parsed,
|
||
crawl_ms=crawl_ms,
|
||
),
|
||
)
|
||
except psycopg2.errors.ProgramLimitExceeded as exc:
|
||
# Defense-in-depth: save_page() caps the content before
|
||
# the insert, so this can only fire if the DB-side
|
||
# trigger overflows anyway (e.g. because
|
||
# fix_pages_search_vector_trigger.sql has not been
|
||
# applied yet). safe_commit() has already rolled the
|
||
# transaction back; mark the queue item failed so the
|
||
# page is not retried endlessly, and keep the worker
|
||
# alive.
|
||
logging.warning(
|
||
"[SAVE ] %s: not indexable, skipped (%s)",
|
||
url,
|
||
exc,
|
||
)
|
||
safe_commit(
|
||
conn,
|
||
lambda: finish_queue_item(
|
||
conn, queue_id, success=False
|
||
),
|
||
)
|
||
failed += 1
|
||
continue
|
||
|
||
inserted = queue_urls(
|
||
conn,
|
||
parsed["links"],
|
||
url
|
||
)
|
||
|
||
logging.info(
|
||
"[SORT ] %s -> %d Links, %d neue URLs",
|
||
url,
|
||
len(parsed["links"]),
|
||
inserted,
|
||
)
|
||
|
||
safe_commit(
|
||
conn,
|
||
lambda: finish_queue_item(conn, queue_id, success=True),
|
||
)
|
||
|
||
crawled += 1
|
||
current_queue_id = None
|
||
|
||
# Periodic stats; not on every page to avoid unnecessary DB load.
|
||
if time.monotonic() - last_stats >= 30:
|
||
try:
|
||
with conn.cursor() as cur:
|
||
cur.execute(
|
||
"SELECT COUNT(*) FROM crawl_queue WHERE status='pending'"
|
||
)
|
||
queued = cur.fetchone()[0]
|
||
|
||
safe_commit(
|
||
conn,
|
||
lambda: write_stats(conn, crawled, queued, failed),
|
||
)
|
||
last_stats = time.monotonic()
|
||
except UpdateRequired:
|
||
raise
|
||
except Exception:
|
||
conn.rollback()
|
||
|
||
except UpdateRequired:
|
||
# A safe point (safe_commit) noticed the new generation and
|
||
# rolled back the open transaction BEFORE raising. Nothing was
|
||
# committed half-done.
|
||
update_requested = True
|
||
|
||
# Best effort: give the claimed item back to the queue so the
|
||
# next version can crawl it immediately. (Should this fail,
|
||
# requeue_stale_items() recovers it later anyway.) This commit
|
||
# is wanted shutdown bookkeeping, not crawled content.
|
||
try:
|
||
if current_queue_id is not None:
|
||
with conn.cursor() as cur:
|
||
cur.execute(
|
||
"UPDATE crawl_queue "
|
||
"SET status = 'pending' "
|
||
"WHERE id = %s AND status = 'processing'",
|
||
(current_queue_id,),
|
||
)
|
||
conn.commit()
|
||
logging.info(
|
||
"[UPDATE] Queue item %d requeued for the next version.",
|
||
current_queue_id,
|
||
)
|
||
except Exception:
|
||
try:
|
||
conn.rollback()
|
||
except Exception:
|
||
pass
|
||
|
||
logging.warning(
|
||
"[UPDATE] %s stops for update (started with generation %s, "
|
||
"version %d).",
|
||
process_name,
|
||
update_signal.generation,
|
||
CRAWLER_VERSION,
|
||
)
|
||
except Exception:
|
||
logging.exception("Worker crashed.")
|
||
raise
|
||
finally:
|
||
update_signal.stop_checker = True
|
||
if conn is not None:
|
||
conn.close()
|
||
|
||
logging.info(
|
||
"Stopped. crawled=%d failed=%d retried=%d",
|
||
crawled, failed, retried
|
||
)
|
||
|
||
if update_requested:
|
||
# Reserved exit code, exclusively for "update required".
|
||
sys.exit(EXIT_UPDATE_REQUIRED)
|
||
|
||
|
||
# ============================================================
|
||
# STARTUP
|
||
# ============================================================
|
||
|
||
def main():
|
||
print("=" * 70)
|
||
print(" JonasSearch Crawler Gamma 0.1")
|
||
print("=" * 70)
|
||
print(f"Start: {START_URL}")
|
||
print(f"Workers: {WORKERS}")
|
||
print(f"Delay: {DEFAULT_DELAY:.1f} Sekunden")
|
||
print("Auto-Switch: aktiv (blockierte Domains werden übersprungen)")
|
||
print(f"Robots-Cache: {ROBOTS_CACHE_SECONDS // 3600} Stunden")
|
||
print(f"DB: {DB_HOST}:{DB_PORT}/{DB_NAME}")
|
||
print(f"Version: {CRAWLER_VERSION}")
|
||
print()
|
||
|
||
conn = db_connect()
|
||
|
||
ensure_deployment_state_table(conn)
|
||
ensure_stats_table(conn)
|
||
safe_commit(conn)
|
||
|
||
safe_commit(conn, lambda: requeue_stale_items(conn))
|
||
|
||
# Seed URL. ON CONFLICT makes restarting harmless.
|
||
seed_url = normalize_url(START_URL)
|
||
if seed_url is None:
|
||
# Either an invalid URL or a non-crawlable host (localhost,
|
||
# raw IP, non-real domain): crawling can never work, so fail
|
||
# clearly instead of spinning on an empty queue.
|
||
conn.close()
|
||
logging.error(
|
||
"Start URL %s is not crawlable (invalid or non-public "
|
||
"host such as localhost or an IP address).",
|
||
START_URL,
|
||
)
|
||
return EXIT_ERROR
|
||
queue_urls(conn, [seed_url], None)
|
||
|
||
conn.close()
|
||
|
||
workers = []
|
||
|
||
for i in range(WORKERS):
|
||
p = mp.Process(target=worker, args=(i,), name=f"crawler-{i}")
|
||
p.start()
|
||
workers.append(p)
|
||
|
||
update_requested = False
|
||
worker_crashed = False
|
||
|
||
try:
|
||
while any(p.is_alive() for p in workers):
|
||
time.sleep(1)
|
||
|
||
for p in workers:
|
||
if p.exitcode == EXIT_UPDATE_REQUIRED:
|
||
# This worker reached its safe point and wants the
|
||
# update. Stop the remaining workers as well (they
|
||
# also exit at their own safe points, never
|
||
# mid-transaction) and propagate exit code 42.
|
||
if not update_requested:
|
||
logging.warning(
|
||
"[UPDATE] Worker %s requests update; the "
|
||
"other workers stop at their safe points.",
|
||
p.name,
|
||
)
|
||
update_requested = True
|
||
STOP.set()
|
||
elif p.exitcode not in (None, 0):
|
||
logging.error(
|
||
"Worker %s exited with code %s.",
|
||
p.name,
|
||
p.exitcode,
|
||
)
|
||
worker_crashed = True
|
||
STOP.set()
|
||
|
||
except KeyboardInterrupt:
|
||
STOP.set()
|
||
|
||
finally:
|
||
STOP.set()
|
||
|
||
for p in workers:
|
||
p.join(timeout=10)
|
||
|
||
for p in workers:
|
||
if p.is_alive():
|
||
p.terminate()
|
||
p.join()
|
||
|
||
if update_requested:
|
||
# Reserved exit code: "update required". Only the launcher may
|
||
# interpret this code; normal errors use EXIT_ERROR (1).
|
||
print("Crawler stopped: update required.")
|
||
sys.exit(EXIT_UPDATE_REQUIRED)
|
||
|
||
if worker_crashed:
|
||
# A normal error must NOT look like a normal end (exit 0) or an
|
||
# update (42): the launcher restarts on this code unchanged.
|
||
print("Crawler stopped: worker error.")
|
||
sys.exit(EXIT_ERROR)
|
||
|
||
print("Crawler stopped.")
|
||
return EXIT_OK
|
||
|
||
|
||
# ============================================================
|
||
# CLI (version / deployment / update checks)
|
||
# ============================================================
|
||
|
||
def build_arg_parser():
|
||
parser = argparse.ArgumentParser(
|
||
prog="crawler.py",
|
||
description="JonasSearch PostgreSQL crawler with "
|
||
"generation-based rolling updates.",
|
||
)
|
||
parser.add_argument(
|
||
"--version",
|
||
action="store_true",
|
||
help="print the crawler version (just the number) and exit",
|
||
)
|
||
parser.add_argument(
|
||
"--self-update",
|
||
type=int,
|
||
metavar="VERSION",
|
||
help="verify a staged crawler release in "
|
||
"%s/<VERSION>/crawler.py without activating it "
|
||
"(exit 0 = startable, 1 = broken/missing)" % CRAWLER_RELEASES_DIR,
|
||
)
|
||
parser.add_argument(
|
||
"--update-deployment",
|
||
type=int,
|
||
metavar="VERSION",
|
||
help="bump the deployment generation and set required_version "
|
||
"to VERSION. Run ONLY after the new version has been "
|
||
"provisioned and verified - from this moment all running "
|
||
"crawlers stop at their next safe commit point.",
|
||
)
|
||
parser.add_argument(
|
||
"--init-deployment",
|
||
action="store_true",
|
||
help="create deployment_state if needed and seed the initial "
|
||
"row (id=1, generation 1, required_version = this "
|
||
"crawler's version)",
|
||
)
|
||
return parser
|
||
|
||
|
||
def cmd_version():
|
||
# Deliberately machine-readable: the launcher parses this output.
|
||
print(CRAWLER_VERSION)
|
||
return EXIT_OK
|
||
|
||
|
||
def _basic_self_check():
|
||
"""
|
||
Minimal smoke test used by --self-update when no releases directory
|
||
exists yet: this file must at least compile.
|
||
"""
|
||
try:
|
||
with open(__file__, "r", encoding="utf-8") as fh:
|
||
compile(fh.read(), __file__, "exec")
|
||
except Exception as exc:
|
||
logging.error("[UPDATE] Self-check failed: %s", exc)
|
||
return EXIT_ERROR
|
||
print("[UPDATE] Basic self-check passed.")
|
||
return EXIT_OK
|
||
|
||
|
||
def cmd_self_update(version):
|
||
"""
|
||
Verify that a staged release for `version` exists and is startable
|
||
- WITHOUT activating it. The launcher runs this with the CURRENT
|
||
(old) interpreter before swapping anything, so a broken release can
|
||
never become active (fallback guarantee).
|
||
|
||
Checks:
|
||
1. releases/<VERSION>/crawler.py exists,
|
||
2. `python3 releases/<VERSION>/crawler.py --version` prints exactly
|
||
`version` and exits with EXIT_OK (imports, syntax and the
|
||
version constant of the new file are all checked).
|
||
"""
|
||
release_file = CRAWLER_RELEASES_DIR / str(version) / "crawler.py"
|
||
|
||
if not CRAWLER_RELEASES_DIR.is_dir():
|
||
logging.warning(
|
||
"[UPDATE] No releases directory (%s); running the basic "
|
||
"self-check only.",
|
||
CRAWLER_RELEASES_DIR,
|
||
)
|
||
return _basic_self_check()
|
||
|
||
if not release_file.is_file():
|
||
logging.error(
|
||
"[UPDATE] Staged release missing: %s", release_file
|
||
)
|
||
return EXIT_ERROR
|
||
|
||
print(f"[UPDATE] Checking staged release: {release_file}")
|
||
try:
|
||
result = subprocess.run(
|
||
[sys.executable, str(release_file), "--version"],
|
||
capture_output=True,
|
||
text=True,
|
||
timeout=60,
|
||
)
|
||
except Exception as exc:
|
||
logging.error("[UPDATE] Release check could not run: %s", exc)
|
||
return EXIT_ERROR
|
||
|
||
printed = (result.stdout or "").strip()
|
||
if result.returncode != EXIT_OK or printed != str(version):
|
||
logging.error(
|
||
"[UPDATE] Staged release check FAILED (exit code %d, "
|
||
"version output %r, stderr: %s). The release is NOT "
|
||
"activated; the old version stays in place.",
|
||
result.returncode,
|
||
printed,
|
||
(result.stderr or "").strip()[-500:],
|
||
)
|
||
return EXIT_ERROR
|
||
|
||
print(f"[UPDATE] Release {version} verified and startable.")
|
||
return EXIT_OK
|
||
|
||
|
||
def cmd_update_deployment(required_version):
|
||
"""
|
||
The global switch point: atomically raise the deployment generation
|
||
and set the required crawler version in ONE statement.
|
||
|
||
Run only AFTER the new version has been provisioned and verified on
|
||
the machines (see --self-update). From this moment every running
|
||
old crawler sees the new generation at its next safe commit point,
|
||
rolls its open transaction back and exits with EXIT_UPDATE_REQUIRED
|
||
- no process is ever killed mid-transaction.
|
||
"""
|
||
conn = None
|
||
try:
|
||
conn = db_connect()
|
||
ensure_deployment_state_table(conn)
|
||
with conn.cursor() as cur:
|
||
cur.execute(
|
||
"""
|
||
UPDATE deployment_state
|
||
SET generation = generation + 1,
|
||
required_version = %s
|
||
WHERE id = 1
|
||
RETURNING generation, required_version
|
||
""",
|
||
(required_version,),
|
||
)
|
||
row = cur.fetchone()
|
||
if row is None:
|
||
logging.error(
|
||
"deployment_state row (id = 1) missing; nothing updated."
|
||
)
|
||
conn.rollback()
|
||
return EXIT_ERROR
|
||
conn.commit()
|
||
generation, activated_version = row
|
||
print(
|
||
f"[UPDATE] Deployment generation -> {generation}, "
|
||
f"required_version -> {activated_version}. Running crawlers "
|
||
"stop at their next safe commit point with exit code "
|
||
f"{EXIT_UPDATE_REQUIRED}."
|
||
)
|
||
return EXIT_OK
|
||
except Exception:
|
||
logging.exception("Deployment update failed.")
|
||
if conn is not None:
|
||
try:
|
||
conn.rollback()
|
||
except Exception:
|
||
pass
|
||
return EXIT_ERROR
|
||
finally:
|
||
if conn is not None:
|
||
conn.close()
|
||
|
||
|
||
def cmd_init_deployment():
|
||
"""
|
||
One-time initialization of deployment_state: create the table and
|
||
the single current-state row (id = 1) with generation 1 and the
|
||
currently running crawler version. Restarting is harmless.
|
||
"""
|
||
conn = None
|
||
try:
|
||
conn = db_connect()
|
||
ensure_deployment_state_table(conn)
|
||
row = _fetch_generation_row(conn)
|
||
generation = int(row[0]) if row else 1
|
||
print(
|
||
f"[UPDATE] deployment_state ready: generation {generation}, "
|
||
f"required_version {CRAWLER_VERSION}."
|
||
)
|
||
return EXIT_OK
|
||
except Exception:
|
||
logging.exception("Initializing deployment_state failed.")
|
||
return EXIT_ERROR
|
||
finally:
|
||
if conn is not None:
|
||
conn.close()
|
||
|
||
|
||
def main_cli(argv=None):
|
||
args = build_arg_parser().parse_args(argv)
|
||
|
||
if args.version:
|
||
return cmd_version()
|
||
if args.self_update is not None:
|
||
return cmd_self_update(args.self_update)
|
||
if args.update_deployment is not None:
|
||
return cmd_update_deployment(args.update_deployment)
|
||
if args.init_deployment:
|
||
return cmd_init_deployment()
|
||
|
||
return main()
|
||
|
||
|
||
if __name__ == "__main__":
|
||
mp.set_start_method("spawn", force=True)
|
||
sys.exit(main_cli())
|