Serverjonas/search
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
682c0f4971662b5b6b6d7df43eb6a470d5223149
search/crawler.py
T
jhkaktus 682c0f4971 add launcher
2026-09-13 01:57:46 +02:00

2159 lines
69 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
#!/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 and links
- 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
- Per-host crawl delay
- 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
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 subprocess
import threading
import multiprocessing as mp
from collections import defaultdict
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", "3.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", "1"))
# 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"}
# 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:",
)
# 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
# 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
# ============================================================
# 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.
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"
)
result = operation() if operation is not None else None
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 get_next_queue_item(conn):
"""
Atomically claim one pending URL.
SKIP LOCKED is the important part: several crawler processes can
consume the same PostgreSQL queue without waiting for each other.
Writes only - the commit happens in safe_commit().
"""
with conn.cursor() as cur:
cur.execute("""
WITH next_item AS (
SELECT id
FROM crawl_queue
WHERE status = 'pending'
ORDER BY priority DESC, discovered_at ASC, id ASC
FOR UPDATE SKIP LOCKED
LIMIT 1
)
UPDATE crawl_queue q
SET status = 'processing',
attempts = attempts + 1,
last_attempt_at = now()
FROM next_item
WHERE q.id = next_item.id
RETURNING q.id, q.url, q.source_url
""")
row = cur.fetchone()
return row
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 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
# 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
# ============================================================
# 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 RateLimiter:
def __init__(self):
self.last_request = defaultdict(float)
def wait(self, url, delay):
host = urlparse(url).netloc.lower()
now = time.monotonic()
wait_for = delay - (now - self.last_request[host])
if wait_for > 0:
logging.info("[RATE] %s: %.1fs", host, wait_for)
time.sleep(wait_for)
self.last_request[host] = time.monotonic()
def fetch(url, limiter, robots):
parsed = urlparse(url)
if not robots.allowed(url):
return None, None, "robots"
delay = robots.delay(url)
limiter.wait(url, delay)
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:
return e.code, None, f"http_{e.code}"
except (URLError, TimeoutError, OSError) as e:
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]
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
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,
"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"]),
"domain": clean_text(parsed["domain"]),
"tld": clean_text(parsed["tld"]),
}
with conn.cursor() as cur:
cur.execute("""
INSERT INTO pages (
url,
title,
headings,
content,
links,
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,
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,
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"],
safe_parsed["content"],
safe_parsed["links"],
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()
limiter = RateLimiter()
crawled = 0
failed = 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 = safe_commit(conn, lambda: get_next_queue_item(conn))
if not item:
logging.info("Queue empty. Worker waiting...")
time.sleep(5)
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 "-")
started = time.monotonic()
status, html, error = fetch(url, limiter, robots)
if error:
logging.info("[HTTP ] %s -> %s", url, error)
safe_commit(
conn,
lambda: finish_queue_item(
conn, queue_id, success=False
),
)
failed += 1
continue
logging.info("[HTTP ] %s -> HTTP %s", url, status)
# 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"]))
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,
),
)
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",
crawled, failed
)
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(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.
queue_urls(conn, [normalize_url(START_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.")
# ============================================================
# 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()
main()
return EXIT_OK
if __name__ == "__main__":
mp.set_start_method("spawn", force=True)
sys.exit(main_cli())
Reference in New Issue View Git Blame Copy Permalink