commit local changes

.env.example

@@ -16,6 +16,21 @@ JONASSEARCH_WORKERS=4
 # Mindestabstand zwischen Requests zur gleichen Domain
 JONASSEARCH_DELAY=3.0
 
+# Host-Scheduling: blockierte/rate-limitierte Domains werden automatisch
+# übersprungen und später erneut versucht (mit exponentiellem Backoff).
+# Kandidaten, die pro Claim-Runde in der Queue geprüft werden
+#JONASSEARCH_CLAIM_BATCH=25
+# Maximale Anzahl geprüfter Kandidaten pro Claim
+#JONASSEARCH_CLAIM_SCAN_LIMIT=200
+# Maximale Versuche pro URL (transiente Fehler werden erneut eingereiht),
+# bevor der Eintrag endgültig failed wird
+#JONASSEARCH_MAX_ATTEMPTS=5
+# Wartezeiten über dieser Grenze (Sekunden) werden nicht inline geschlafen:
+# das Item wird zurückgestellt und eine andere Domain gecrawlt
+#JONASSEARCH_MAX_INLINE_WAIT=10
+# Obergrenze für die Wartezeit, wenn gerade kein Host crawlbar ist
+#JONASSEARCH_MAX_IDLE_SLEEP=30
+
 # robots.txt Cache
 JONASSEARCH_ROBOTS_CACHE_SECONDS=86400
 
@@ -32,7 +47,7 @@ JONASSEARCH_USER_AGENT=JonasSearchBot/0.1 (+https://jonassearch.local/bot)
 
 # Version dieser Crawler-Codebasis (NICHT die Deployment-Generation).
 # Muss bei jedem verteilten Crawler-Änderung erhöht werden.
-JONASSEARCH_CRAWLER_VERSION=1
+JONASSEARCH_CRAWLER_VERSION=3
 
 # Wie oft jeder Crawler außerhalb der Transaktionen nach einer neuen
 # Deployment-Generation sieht (Sekunden). Der Check setzt nur ein
@@ -55,16 +70,43 @@ JONASSEARCH_UPDATE_CHECK_SECONDS=30
 #JONASSEARCH_LAUNCHER_RESTART_DELAY=2
 
 # ============================================================
-# Suchmaschine (search/)
+# Suchmaschine (search/) – Port 8083
 # ============================================================
 
-# PostgreSQL-Textsuch-Konfiguration (Stemming/Stopwörter)
-JONASSEARCH_TS_CONFIG=german
+# PostgreSQL-Textsuch-Konfiguration. MUSS zur Konfiguration passen,
+# mit der der DB-Trigger den Suchvektor (pages.search_vector) baut –
+# laut DB-Doku: 'simple'. Ansonsten finden die Queries nichts.
+JONASSEARCH_TS_CONFIG=simple
 
 # tsvector-Spalte der Tabelle `pages` erzwingen (optional,
 # wird automatisch erkannt)
 #JONASSEARCH_VECTOR_COLUMN=search_vector
 
-# HTTP-API (search/api.py)
+# HTTP-API + Weboberfläche (search/api.py)
 JONASSEARCH_API_HOST=0.0.0.0
-JONASSEARCH_API_PORT=8082
+JONASSEARCH_API_PORT=8083
+
+# Ranking: ts_rank_cd-Gewichte (Titel=A, Überschriften=B, Inhalt=C)
+#JONASSEARCH_RANK_WEIGHTS=A=1.0,B=0.6,C=0.3,D=0.1
+
+# Ranking: Aktualitätsbonus (Faktor auf den Text-Rang)
+#JONASSEARCH_FRESHNESS_WEIGHT=0.25
+#JONASSEARCH_FRESHNESS_HALF_LIFE_DAYS=180
+
+# Ranking: ts_rank_cd-Normalisierung (Bitmaske). 8 = Rang /
+# (1 + log Dokumentlänge) – riesige Seiten dominieren sonst automatisch.
+# 0 = keine Normalisierung.
+#JONASSEARCH_RANK_NORMALIZATION=8
+
+# Fallback bei sehr häufigen Begriffen: volles Ranking wird bei
+# Zeitüberschreitung auf die ersten N Kandidaten (Indexreihenfolge)
+# begrenzt, damit die Antwort immer kommt.
+#JONASSEARCH_MAX_RANK_CANDIDATES=5000
+
+# Schutz der großen DB: Statement-Timeouts (ms)
+#JONASSEARCH_STATEMENT_TIMEOUT_MS=10000
+#JONASSEARCH_COUNT_TIMEOUT_MS=4000
+
+# Paginierung
+#JONASSEARCH_RESULTS_PER_PAGE=10
+#JONASSEARCH_MAX_RESULTS_PER_PAGE=50

README.md

@@ -4,8 +4,12 @@ das komplette serverjonas search
 
 - `crawler.py` – PostgreSQL-Crawler, baut den Suchindex auf
 - `launcher.py` – startet/überwacht den Crawler und rollt Updates aus (siehe unten)
-- `search/` – Suchmaschine darüber (CLI + HTTP-API), siehe [search/README.md](search/README.md)
-- `web/` – Statistik-Webseite
+- `search/` – Suchmaschine darüber (CLI + HTTP-API + Weboberfläche, Port 8083),
+  siehe [search/README.md](search/README.md)
+- `web/` – Statistik-Webseite (Port 8081, `jonassearch-stats.service`)
+- `priorityweb/` – Vorschlagsverfeinerer (Port 8082, `jonassearch-priority.service`)
+- `search/jonassearch-engine.service` – systemd-Unit für die Suchmaschine
+  (Port 8083, `jonassearch-engine.service`)
 - `fix_pages_search_vector_trigger.sql` – einmalig laufen lassen: kappelt den Text,
   der im DB-Trigger `pages_search_vector_update()` in `to_tsvector()` einfließt
   (Postgres hartes Limit: 1.048.575 Bytes pro `tsvector`)
@@ -38,6 +42,45 @@ Zusätzlich werden bekannt wertlose URLs (`pkg.go.dev/...?tab=importedby` –
 Megabyte-große reine Rückverweis-Listen) über `SKIP_URL_PATTERNS` schon vor dem
 Fetch übersprungen und als `done` markiert.
 
+## Domain-Switching / Block-Handling
+
+Der Crawler wartet nie auf einer langsamen oder blockierten Domain:
+
+1. **Claim mit Host-Check:** Beim Holen des nächsten Queue-Eintrags prüft der
+   Worker pro Kandidat (Prioritätsreihenfolge, bis zu
+   `JONASSEARCH_CLAIM_SCAN_LIMIT` Kandidaten), ob der Host gerade crawlbar ist
+   (Per-Host-Delay aus robots.txt bzw. `JONASSEARCH_DELAY`, aktive
+   Backoff-Sperre). Nicht crawlbare Hosts werden übersprungen – der Worker
+   arbeitet automatisch auf einer anderen Domain weiter und kehrt zurück,
+   sobald der Host wieder darf. Ist gerade gar kein Host crawlbar, schläft
+   der Worker bis zum frühesten freigegebenen Host (gekappet bei
+   `JONASSEARCH_MAX_IDLE_SLEEP`, dann wird neu gescannt).
+2. **Backoff bei Blocks:** HTTP 429/5xx/403/401 und Netzwerkfehler setzen den
+   Host auf exponentielles Backoff (z. B. 429: 60 s, sich verdoppelnd bis
+   30 min; 403/401: 10 min bis 2 h). Ein `Retry-After`-Header wird zusätzlich
+   beachtet. Bei einem erfolgreichen Fetch zählt der Zähler wieder von vorn.
+3. **Retry statt Verlust:** Transiente Fehler (429/5xx/Netzwerk) werden
+   wieder in die Queue gestellt (`pending`) statt endgültig `failed` –
+   begrenzt durch `JONASSEARCH_MAX_ATTEMPTS` (Standard 5), damit dauerhaft
+   kaputte Hosts nicht ewig rotieren. Wartezeiten größer als
+   `JONASSEARCH_MAX_INLINE_WAIT` (z. B. großer robots-crawl-delay) werden
+   nicht inline geschlafen: das Item wird ohne Versuchs-Zählung zurückgestellt
+   und solange eine andere Domain gecrawlt.
+
+## Nicht crawlbare Ziele (localhost, IPs, nicht-echte Domains)
+
+Der Crawler folgt nur echten, öffentlich auflösbaren Domains. Bereits in
+`normalize_url()` werden abgelehnt und nie in die Queue gestellt:
+
+- `localhost`, `*.localhost`, `localhost.localdomain`, `*.local`, `*.internal`
+- reine IP-Adressen als Host (IPv4 und IPv6, z. B. `127.0.0.1`, `http://[::1]/`)
+- Hostnamen ohne Punkt (Einzel-Labels wie `http://myserver/` – LAN-Namen)
+
+Alte Queue-Einträge mit solchen Hosts werden beim Zugriff als `done`
+markiert und nicht mehr geholt. Zeigt der Start-URL auf so einen Host,
+beendet sich der Crawler mit einem klaren Fehler (Exit 1) statt endlos
+zu rotieren.
+
 ## Rolling Updates (Generation-Mechanismus)
 
 Zentrale Tabelle (eine Zeile, `id = 1`):

clear_domain_suggestions.sh

@@ -0,0 +1,142 @@
+#!/usr/bin/env bash
+#
+# clear_domain_suggestions.sh
+#
+# Loescht ALLE Eintraege aus der Tabelle `domain_suggestions` — der
+# Buchhaltungs-Tabelle des Vorschlagsverfeinerers (priorityweb/app.py,
+# Port 8082). Die Liste "Letzte Vorschlaege" im Web-UI ist danach leer.
+#
+# WICHTIG: Die crawl_queue wird NICHT angefasst. Bereits priorisierte
+# Domains bleiben mit Prioritaet 1000000 (SUGGEST_PRIORITY) vorne in
+# der Queue und werden weiterhin als Naechstes gecrawlt. Falls auch
+# diese Boosts aufgehoben werden sollen, zusaetzlich ausfuehren:
+#
+#   DELETE FROM crawl_queue
+#    WHERE status = 'pending' AND priority >= 1000000;
+#
+# (Nur pending-Zeilen: laufende Verarbeitungen werden nicht abgeschossen.)
+#
+# Verwendung:
+#   ./clear_domain_suggestions.sh          mit Rueckfrage
+#   ./clear_domain_suggestions.sh --yes    ohne Rueckfrage (z. B. fuer Cron)
+#   ./clear_domain_suggestions.sh --help   diesen Text anzeigen
+#
+# DB-Zugangsdaten kommen wie bei allen Komponenten aus der .env
+# (JONASSEARCH_DB_*). Ist dort kein Passwort gesetzt oder die direkte
+# Verbindung fehlgeschlagen, wird lokal per `sudo -u postgres psql`
+# ausgefuehrt (gleicher Weg wie in web/app.py).
+
+set -euo pipefail
+cd "$(dirname "$0")"
+
+# Muss zu SUGGEST_PRIORITY in priorityweb/app.py passen.
+SUGGEST_PRIORITY=1000000
+
+usage() {
+    sed -n '2,27p' "$0" | sed 's/^# \{0,1\}//'
+    exit 0
+}
+
+YES=0
+for arg in "$@"; do
+    case "$arg" in
+        -y|--yes)  YES=1 ;;
+        -h|--help) usage ;;
+        *) echo "Unbekanntes Argument: $arg (--help fuer Nutzung)" >&2; exit 2 ;;
+    esac
+done
+
+# ---------------------------------------------------------------
+# Zugangsdaten aus .env laden (nur JONASSEARCH_DB_*-Zeilen)
+# ---------------------------------------------------------------
+if [[ -f .env ]]; then
+    set -a
+    # shellcheck disable=SC1091
+    source <(grep -E '^JONASSEARCH_DB_[A-Z]+=' .env | tr -d '\r')
+    set +a
+fi
+
+DB_HOST="${JONASSEARCH_DB_HOST:-127.0.0.1}"
+DB_PORT="${JONASSEARCH_DB_PORT:-25432}"
+DB_NAME="${JONASSEARCH_DB_NAME:-jonassearch}"
+DB_USER="${JONASSEARCH_DB_USER:-jonassearch}"
+
+# ---------------------------------------------------------------
+# psql-Wrapper: zuerst direkt mit Passwort, sonst als postgres-Superuser
+# ---------------------------------------------------------------
+psql_run() {
+    if [[ -n "${JONASSEARCH_DB_PASSWORD:-}" ]] && command -v psql >/dev/null 2>&1; then
+        if PGPASSWORD="${JONASSEARCH_DB_PASSWORD}" \
+            psql -h "$DB_HOST" -p "$DB_PORT" -U "$DB_USER" -d "$DB_NAME" \
+                 -v ON_ERROR_STOP=1 "$@"; then
+            return 0
+        fi
+        echo "Direkte DB-Verbindung fehlgeschlagen — Fallback: sudo -u postgres psql" >&2
+    fi
+    sudo -u postgres psql -p "$DB_PORT" -d "$DB_NAME" -v ON_ERROR_STOP=1 "$@"
+}
+
+die() { echo "FEHLER: $*" >&2; exit 3; }
+
+# ---------------------------------------------------------------
+# 1) Existiert die Tabelle ueberhaupt?
+#    (Die App legt sie beim ersten Vorschlag selbst an.)
+# ---------------------------------------------------------------
+EXISTS="$(psql_run -t -A -c \
+    "SELECT to_regclass('public.domain_suggestions') IS NOT NULL;")" \
+    || die "keine Verbindung zur DB ${DB_HOST}:${DB_PORT}/${DB_NAME}"
+
+if [[ "$EXISTS" != "t" ]]; then
+    echo "Tabelle 'domain_suggestions' existiert nicht — es gibt nichts zu loeschen."
+    exit 0
+fi
+
+# ---------------------------------------------------------------
+# 2) Zaehlen und bestaetigen lassen
+# ---------------------------------------------------------------
+BEFORE="$(psql_run -t -A -c "SELECT COUNT(*) FROM domain_suggestions;")" \
+    || die "Konnte Anzahl der Eintraege nicht lesen."
+
+echo "DB        : ${DB_USER}@${DB_HOST}:${DB_PORT}/${DB_NAME}"
+echo "Eintraege : ${BEFORE}"
+
+if [[ "$BEFORE" == "0" ]]; then
+    echo "Tabelle ist bereits leer — nichts zu tun."
+    exit 0
+fi
+
+if (( ! YES )); then
+    read -r -p "Wirklich ALLE ${BEFORE} Eintraege aus 'domain_suggestions' loeschen? (yes/no) " answer
+    if [[ "$answer" != "yes" ]]; then
+        echo "Abgebrochen — nichts geloescht."
+        exit 1
+    fi
+fi
+
+# ---------------------------------------------------------------
+# 3) Loeschen (DELETE statt TRUNCATE: kein ACCESS EXCLUSIVE Lock,
+#    vertraeglich mit gleichzeitig laufenden Vorschlaegen im UI)
+# ---------------------------------------------------------------
+DELETED="$(psql_run -t -A -c \
+    "WITH d AS (DELETE FROM domain_suggestions RETURNING 1) \
+     SELECT COUNT(*) FROM d;")" \
+    || die "DELETE fehlgeschlagen."
+
+echo "Geloescht : ${DELETED} Eintraege (vorher: ${BEFORE})."
+echo "Die Liste 'Letzte Vorschlaege' im Web-UI leert sich beim naechsten Refresh (max. 30 s)."
+
+# ---------------------------------------------------------------
+# 4) Nur Hinweis: stecken in der crawl_queue noch priorisierte Domains?
+#    (gleicher schneller Index-Range wie /api/queue in der App)
+# ---------------------------------------------------------------
+BOOSTED="$(psql_run -t -A -c \
+    "SELECT COUNT(*) FROM crawl_queue \
+     WHERE status = 'pending' AND priority >= ${SUGGEST_PRIORITY};" 2>/dev/null)" \
+    || BOOSTED="?"
+
+if [[ "$BOOSTED" != "0" ]]; then
+    echo "Hinweis: ${BOOSTED} priorisierte Domain(s) stecken noch in der crawl_queue"
+    echo "         (dort bleiben sie auch nach diesem Skript vorne)."
+    echo "         Aufheben mit:"
+    echo "           DELETE FROM crawl_queue WHERE status = 'pending' AND priority >= ${SUGGEST_PRIORITY};"
+fi

crawler.py

@@ -14,7 +14,17 @@ Features:
   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
+- 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
@@ -52,10 +62,12 @@ 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
@@ -113,7 +125,7 @@ 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"))
+DEFAULT_DELAY = float(os.getenv("JONASSEARCH_DELAY", "0.0"))
 
 ROBOTS_CACHE_SECONDS = int(
     os.getenv("JONASSEARCH_ROBOTS_CACHE_SECONDS", str(24 * 60 * 60))
@@ -138,7 +150,7 @@ USER_AGENT = os.getenv(
 # 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"))
+CRAWLER_VERSION = int(os.getenv("JONASSEARCH_CRAWLER_VERSION", "3"))
 
 # Exit codes. 42 is reserved exclusively for "I must be updated" so
 # the launcher can distinguish update, crash and normal exit.
@@ -173,6 +185,16 @@ LAUNCHER_RESTART_DELAY = float(
 # 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",
@@ -244,6 +266,69 @@ QUEUE_INSERT_PAGE_SIZE = 500
 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 wall of blocked ones.
+QUEUE_CLAIM_SCAN_LIMIT = int(
+    os.getenv("JONASSEARCH_CLAIM_SCAN_LIMIT", "200")
+)
+
+# 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
 # ============================================================
@@ -646,35 +731,135 @@ def write_stats(conn, crawled, queued, failed):
         """, (crawled, queued, failed))
 
 
-def get_next_queue_item(conn):
+def claim_next_queue_item(conn, scheduler, robots):
     """
-    Atomically claim one pending URL.
+    Atomically claim the best currently-crawlable pending URL.
 
-    SKIP LOCKED is the important part: several crawler processes can
-    consume the same PostgreSQL queue without waiting for each other.
+    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. For every candidate the HostScheduler decides whether its
+    host may be contacted right now (per-host crawl delay, active
+    block backoff). The first ready candidate is claimed; hosts that
+    would have to wait are skipped, so the worker automatically works
+    on another domain while a rate-limited or blocked host cools down.
+
+    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
+
+    while item is None and scanned < QUEUE_CLAIM_SCAN_LIMIT:
+        with conn.cursor() as cur:
+            if seen_ids:
+                cur.execute("""
+                    SELECT id, url, source_url
+                    FROM crawl_queue
+                    WHERE status = 'pending'
+                      AND NOT (id = ANY(%s))
+                    ORDER BY priority DESC, discovered_at ASC, id ASC
+                    FOR UPDATE SKIP LOCKED
+                    LIMIT %s
+                """, (seen_ids, QUEUE_CLAIM_BATCH))
+            else:
+                cur.execute("""
+                    SELECT id, url, source_url
+                    FROM crawl_queue
+                    WHERE status = 'pending'
+                    ORDER BY priority DESC, discovered_at ASC, id ASC
+                    FOR UPDATE SKIP LOCKED
+                    LIMIT %s
+                """, (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
+
+    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("""
-            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
-        """)
+            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
+    return row[0] if row else None
 
 
 def finish_queue_item(conn, queue_id, success=True):
@@ -1290,6 +1475,54 @@ def clean_text_list(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
@@ -1319,6 +1552,11 @@ def normalize_url(url, base_url=None):
     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):
@@ -1423,20 +1661,159 @@ class RobotsCache:
 # HTTP
 # ============================================================
 
-class RateLimiter:
+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 = defaultdict(float)
+        self.last_request = {}
+        self.blocked_until = {}
+        self.block_count = {}
+
+    @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
+        next_allowed = max(
+            self.last_request.get(host, 0.0) + delay,
+            self.blocked_until.get(host, 0.0),
+        )
+        return max(0.0, next_allowed - now)
+
+    def is_ready(self, url, delay, now=None):
+        return self.wait_time(url, delay, now) <= 0.0
 
     def wait(self, url, delay):
-        host = urlparse(url).netloc.lower()
-        now = time.monotonic()
-        wait_for = delay - (now - self.last_request[host])
+        """
+        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.
+        """
+        wait_for = self.wait_time(url, delay)
+        if wait_for > MAX_INLINE_RATE_WAIT:
+            return False
         if wait_for > 0:
-            logging.info("[RATE] %s: %.1fs", host, wait_for)
+            if wait_for >= 1.0:
+                logging.info("[RATE ] %s: %.1fs", self._host(url), wait_for)
             time.sleep(wait_for)
+        self.last_request[self._host(url)] = time.monotonic()
+        return True
 
-        self.last_request[host] = time.monotonic()
+    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):
@@ -1448,14 +1825,20 @@ def is_skipped_url(url):
     return any(pattern.match(url) for pattern in SKIP_URL_PATTERNS)
 
 
-def fetch(url, limiter, robots):
+def fetch(url, scheduler, robots):
     parsed = urlparse(url)
 
     if not robots.allowed(url):
         return None, None, "robots"
 
     delay = robots.delay(url)
-    limiter.wait(url, delay)
+    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,
@@ -1501,8 +1884,18 @@ def fetch(url, limiter, robots):
             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}"
@@ -1702,10 +2095,11 @@ def worker(worker_id):
 
     conn = None
     robots = RobotsCache()
-    limiter = RateLimiter()
+    scheduler = HostScheduler()
 
     crawled = 0
     failed = 0
+    retried = 0
     last_stats = time.monotonic()
     current_queue_id = None
     update_requested = False
@@ -1742,11 +2136,26 @@ def worker(worker_id):
                 update_requested = True
                 break
 
-            item = safe_commit(conn, lambda: get_next_queue_item(conn))
+            item, wait_hint = safe_commit(
+                conn,
+                lambda: claim_next_queue_item(conn, scheduler, robots),
+            )
 
             if not item:
-                logging.info("Queue empty. Worker waiting...")
-                time.sleep(5)
+                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; "
+                        "worker sleeps %.1fs",
+                        sleep_for,
+                    )
+                else:
+                    sleep_for = 5
+                    logging.info("Queue empty. Worker waiting...")
+                time.sleep(sleep_for)
                 continue
 
             queue_id, url, source_url = item
@@ -1770,12 +2179,53 @@ def worker(worker_id):
                 )
                 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, limiter, robots)
+            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(
@@ -1786,6 +2236,7 @@ def worker(worker_id):
                 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.
@@ -1941,8 +2392,8 @@ def worker(worker_id):
             conn.close()
 
         logging.info(
-            "Stopped. crawled=%d failed=%d",
-            crawled, failed
+            "Stopped. crawled=%d failed=%d retried=%d",
+            crawled, failed, retried
         )
 
     if update_requested:
@@ -1961,6 +2412,7 @@ def main():
     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}")
@@ -1975,7 +2427,19 @@ def main():
     safe_commit(conn, lambda: requeue_stale_items(conn))
 
     # Seed URL. ON CONFLICT makes restarting harmless.
-    queue_urls(conn, [normalize_url(START_URL)], None)
+    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()
 
@@ -2043,6 +2507,7 @@ def main():
         sys.exit(EXIT_ERROR)
 
     print("Crawler stopped.")
+    return EXIT_OK
 
 
 # ============================================================
@@ -2257,8 +2722,7 @@ def main_cli(argv=None):
     if args.init_deployment:
         return cmd_init_deployment()
 
-    main()
-    return EXIT_OK
+    return main()
 
 
 if __name__ == "__main__":

priorityweb/jonassearch-priority.service

@@ -19,11 +19,11 @@ Type=simple
 User=jonassearch
 Group=jonassearch
 
-WorkingDirectory=/srv/serverjonas/jonassearch
+WorkingDirectory=/srv/serverjonas/search
 
 EnvironmentFile=/srv/serverjonas/search/.env
 
-ExecStart=/usr/bin/python3 /srv/serverjonas/jonassearch/priorityweb/app.py
+ExecStart=/usr/bin/python3 /srv/serverjonas/search/priorityweb/app.py
 
 Restart=on-failure
 RestartSec=10

search/README.md

@@ -1,74 +1,178 @@
-# JonasSearch – Suchmaschine
+# JonasSearch Suchmaschine (`search/`)
 
-Volltextsuche über den Index, den `crawler.py` in PostgreSQL aufgebaut hat.
-Die Engine sucht in der Tabelle `pages` und nutzt dabei PostgreSQLs
-Volltextsuche (`tsvector` / `tsquery`).
+Die eigentliche Websuchmaschine über der PostgreSQL-Datenbank `jonassearch`:
+Volltextsuche auf dem vorhandenen `tsvector`-Index (`pages.search_vector`,
+GIN-Index `idx_pages_search`) mit mehrstufigem Ranking, Filtern, Snippets,
+Weboberfläche, HTTP-API und Kommandozeile.
 
-## Starten
-
-CLI:
-
-```bash
-python3 search/cli.py "raspberry pi"
-python3 search/cli.py "postgres" --domain example.com --page 2
-python3 search/cli.py "amplifier" --json
-```
-
-HTTP-API + Suchseite:
-
-```bash
-python3 search/api.py
-# Suchseite:  http://localhost:8082/
-# JSON-API:   http://localhost:8082/search?q=raspberry%20pi
-```
-
-Als Bibliothek:
-
-```python
-from search import SearchEngine
-
-engine = SearchEngine()
-result = engine.search("postgres", page=1, per_page=10)
-
-for hit in result.hits:
-    print(hit.position, hit.title, hit.url)
-```
-
-## Funktionsweise
-
-- **Index-Nutzung:** Die Engine sucht automatisch die tsvector-Spalte der
-  Tabelle `pages` (z. B. `search_vector`, die ein Trigger pflegt) und
-  durchsucht sie per `@@`-Operator. Fehlt eine solche Spalte, wird der
-  Suchvektor zur Laufzeit aus `title` + `headings` + `content` berechnet –
-  die Suche funktioniert dann genauso, nur langsamer.
-- **Ranking:** Sortierung nach `ts_rank` absteigend, danach nach
-  Crawl-Zeitpunkt.
-- **Snippets:** `ts_headline` erzeugt Ausschnitte, in denen die Suchbegriffe
-  mit `<mark>` hervorgehoben sind.
-- **Präfix-Suche:** Jeder Begriff wird mit `:*` gesucht, „raspberr" findet
-  also auch „raspberry".
-- **UND-Verknüpfung:** Mehrere Begriffe müssen alle vorkommen.
-- **Paginierung:** `--page` / `--per-page` (max. 50).
-- **Filter:** `--domain` (exakte Domain) und `--tld` (z. B. `.de`).
-
-## Umgebungsvariablen
-
-Wie der Crawler (siehe `.env.example`):
-
-| Variable | Bedeutung | Standard |
+| Baustein | Zweck | Port |
 |---|---|---|
-| `JONASSEARCH_DB_HOST` | DB-Host | `127.0.0.1` |
-| `JONASSEARCH_DB_PORT` | DB-Port | `25432` |
-| `JONASSEARCH_DB_NAME` | DB-Name | `jonassearch` |
-| `JONASSEARCH_DB_USER` | DB-User | `jonassearch` |
-| `JONASSEARCH_DB_PASSWORD` | DB-Passwort | leer |
-| `JONASSEARCH_TS_CONFIG` | PostgreSQL-Textsuch-Konfiguration | `german` |
-| `JONASSEARCH_VECTOR_COLUMN` | tsvector-Spalte erzwingen (optional) | automatisch |
-| `JONASSEARCH_API_HOST` | Bind-Adresse der API | `0.0.0.0` |
-| `JONASSEARCH_API_PORT` | Port der API | `8082` |
+| `search/api.py` | HTTP-API + Weboberfläche | **8083** |
+| `search/cli.py` | Kommandozeile (ohne Webserver) | – |
+| `search/engine.py` | Suchlogik (von api/cli genutzt) | – |
+| `search/jonassearch-engine.service` | systemd-Dienst | 8083 |
 
-## Abhängigkeiten
+Bestehende Dienste: Statserver 8081 (`web/app.py`), Vorschlagsverfeinerer
+8082 (`priorityweb/app.py`), Suchmaschine **8083**.
 
-- `psycopg2` (DB-Zugriff)
-- `python-dotenv` (`.env`-Datei)
-- `flask` (nur für `search/api.py`)
+## Wie gesucht wird
+
+1. **Retrieval:** Die Anfrage wird per `websearch_to_tsquery` geparst und
+   per `@@` gegen `pages.search_vector` über den GIN-Index ausgewertet.
+   Nutzersyntax: `wort1 wort2` (UND), `"exakte phrase"`, `OR`,
+   `-ausschluss`. Da die Suchvektoren mit der Textsuch-Konfiguration
+   `simple` gebaut werden (DB-Trigger), matched die Suche ohne
+   Stemming – exakte Lexeme, Groß-/Kleinschreibung egal.
+2. **Ranking (in SQL):**
+
+   ```text
+   score = ts_rank_cd(Gewichte, search_vector, tsquery, Norm=8)
+                                                      – Titel=A, Überschriften=B, Inhalt=C
+         * (1 + 0.25 · exp(−Alter_in_Tagen / 180))    – Aktualität (crawled_at/discovered_at)
+   ```
+
+   Die Gewichte sind über `JONASSEARCH_RANK_WEIGHTS` einstellbar
+   (`A=1.0,B=0.6,C=0.3,D=0.1`). Die ts_rank_cd-Normalisierung (Bitmaske 8
+   = Rang / (1 + log Dokumentlänge), `JONASSEARCH_RANK_NORMALIZATION`)
+   verhindert, dass riesige Seiten kurze, präzise Treffer dominieren.
+   Nicht in die Formel eingeflossen (noch): Link-Autorität aus
+   `pages.links`/`source_url` – dafür liegt mit dem gespeicherten
+   Webgraph die Grundlage bereit.
+
+   **Fallback bei sehr häufigen Begriffen:** Ist das volle Ranking über
+   alle Kandidaten nach `JONASSEARCH_STATEMENT_TIMEOUT_MS` nicht fertig
+   (z. B. Einzelwort mit zigtausend Treffern), wird automatisch ein
+   zweiter Versuch über die ersten `JONASSEARCH_MAX_RANK_CANDIDATES`
+   (Default 5000) Kandidaten gefahren – die Antwort kommt dann garantiert,
+   die Antwortmarke `ranking_capped: true` zeigt die Einschränkung an.
+
+3. **Filter** (kombinierbar): `site=` (Domain inkl. Subdomains), `tld=`,
+   `after=`/`before=` (ISO-Datum oder `24h`/`7d`/`30d`/`365d`), nur
+   `http_status = 200` und nur HTML-Inhalt (jeweils abschaltbar).
+4. **Snippets:** `ts_headline` wird nur für die Treffer der angeforderten
+   Seite berechnet – `content` (TOAST-lastig) wird nie für die ganze
+   Kandidatenmenge gelesen. Markierungen im Snippet: `⟦…⟧`.
+
+## Schutz der Produktivdatenbank (~10 GB)
+
+- Jede Operation nutzt eine eigene Verbindung mit `statement_timeout`
+  (Default 10 s); Zeitüberschreitung → HTTP 504 bzw. CLI-Hinweis zum
+  Eingrenzen, niemals eine blockierte Datenbank.
+- Die Gesamt-Trefferzahl läuft mit eigenem, kürzerem Timeout (4 s) und
+  ist **best effort**: bei Überschreitung liefert die API `total: null`
+  („viele Treffer“), die Ergebnisse bleiben unbeeinflusst.
+- `/api/stats` nutzt nur Schätzwerte aus `pg_class`/`pg_stats`, keine
+  Full-Scans.
+
+## HTTP-API (Port 8083)
+
+| Endpunkt | Beschreibung |
+|---|---|
+| `GET /` | Weboberfläche (Suchseite, Deutsch, Dark/Light) |
+| `GET /api/search` | Suche – Parameter siehe unten |
+| `GET /api/suggest?q=begr` | Autocomplete (Präfix auf letztem Wort, `to_tsquery 'begr:*'`) |
+| `GET /api/page?id=42` oder `?url=…` | gespeicherte Seite (Inhalt gekürzt) |
+| `GET /api/stats` | Index-Statistik (Seiten, Domains, Queue, DB-Größe) |
+| `GET /health` | Healthcheck inkl. DB-Status |
+
+Parameter von `/api/search`:
+
+| Parameter | Bedeutung | Default |
+|---|---|---|
+| `q` | Suchbegriffe (Pflicht) | – |
+| `page`, `per_page` | Paginierung (max 50/Seite) | 1, 10 |
+| `site` | Domain-Filter, mehrfach möglich (`site=example.com`) | – |
+| `tld` | TLD-Filter, mehrfach möglich (`tld=de`) | – |
+| `after` / `before` | Zeitfilter: ISO oder `24h`/`7d`/`2w` | – |
+| `sort` | `relevance` (Default) oder `freshness` | relevance |
+| `include_errors=1` | auch Seiten ohne HTTP 200 | aus |
+| `include_non_html=1` | auch Nicht-HTML-Dokumente | aus |
+| `no_snippets=1` | Snippets weglassen (schneller) | aus |
+
+Beispiel:
+
+```bash
+curl "http://localhost:8083/api/search?q=volltextsuche&site=postgresql.org&after=30d"
+```
+
+Antwort (Auszug):
+
+```json
+{
+  "query": "volltextsuche",
+  "parsed_query": "'volltextsuche'",
+  "page": 1, "per_page": 10,
+  "total": 1234,
+  "has_more": true,
+  "results": [
+    {
+      "id": 42, "url": "https://…", "title": "…",
+      "domain": "…", "tld": ".de", "http_status": 200,
+      "crawled_at": "2026-09-30T18:12:03+00:00",
+      "score": 0.081203, "text_rank": 0.078341, "freshness": 1.074,
+      "snippet": "… PostgreSQL ⟦Volltextsuche⟧ mit tsvector …"
+    }
+  ],
+  "took_ms": 31.4
+}
+```
+
+Fehler: `400` ungültige Parameter · `404` Seite unbekannt ·
+`503` Datenbank nicht erreichbar · `504` Zeitlimit (bitte eingrenzen).
+
+## Kommandozeile
+
+```bash
+python3 search/cli.py "postgresql volltextsuche"
+python3 search/cli.py "crawler" --site example.com --sort freshness --limit 20
+python3 search/cli.py "python" --after 7d
+python3 search/cli.py --suggest "suchmasch"
+python3 search/cli.py --stats
+python3 search/cli.py --page-id 42
+python3 search/cli.py "test" --json
+```
+
+## Installation als Dienst
+
+```bash
+sudo cp /srv/serverjonas/search/search/jonassearch-engine.service \
+        /etc/systemd/system/
+sudo systemctl daemon-reload
+sudo systemctl enable --now jonassearch-engine.service
+journalctl -u jonassearch-engine -f
+```
+
+Port-Auflösung: `--port` (manueller Start) > `JONASSEARCH_API_PORT`
+(`.env`) > **8083**. Der systemd-Dienst pinnt `JONASSEARCH_API_PORT=8083`
+*nach* dem EnvironmentFile und überschreibt damit einen eventuell
+abweichenden Wert aus `.env` – der Dienst läuft garantiert auf 8083.
+
+## Konfiguration (.env)
+
+| Variable | Default | Bedeutung |
+|---|---|---|
+| `JONASSEARCH_DB_HOST/PORT/NAME/USER/PASSWORD` | wie crawler.py | Datenbankzugriff |
+| `JONASSEARCH_API_HOST` | `0.0.0.0` | Bind-Adresse |
+| `JONASSEARCH_API_PORT` | `8083` | Port (Dienst pinnt 8083) |
+| `JONASSEARCH_TS_CONFIG` | `simple` | Muss zur Trigger-Konfiguration des Suchvektors passen! |
+| `JONASSEARCH_VECTOR_COLUMN` | auto | tsvector-Spalte erzwingen (sonst automatische Erkennung) |
+| `JONASSEARCH_RANK_WEIGHTS` | `A=1.0,B=0.6,C=0.3,D=0.1` | ts_rank_cd-Gewichte |
+| `JONASSEARCH_FRESHNESS_WEIGHT` | `0.25` | Stärke des Aktualitätsbonus |
+| `JONASSEARCH_FRESHNESS_HALF_LIFE_DAYS` | `180` | Halbwertszeit der Aktualität |
+| `JONASSEARCH_RANK_NORMALIZATION` | `8` | ts_rank_cd-Normalisierung (8 = 1+log Länge, 0 = aus) |
+| `JONASSEARCH_MAX_RANK_CANDIDATES` | `5000` | Kandidaten-Fenster des Ranking-Fallbacks |
+| `JONASSEARCH_STATEMENT_TIMEOUT_MS` | `10000` | Timeout pro Suchstatement |
+| `JONASSEARCH_COUNT_TIMEOUT_MS` | `4000` | Timeout der Trefferzählung |
+| `JONASSEARCH_RESULTS_PER_PAGE` | `10` | Treffer pro Seite |
+| `JONASSEARCH_MAX_RESULTS_PER_PAGE` | `50` | Maximum pro Seite |
+| `JONASSEARCH_HEADLINE_SOURCE_CHARS` | `100000` | `content`-Anteil für ts_headline |
+
+## Tests
+
+```bash
+python3 search/test_engine.py
+```
+
+Getestet werden die DB-freien Teile: Parameter-Normalisierung, Gewichte,
+Zeitfilter, SQL-Zusammensetzung, Query-Parsing. Alles ohne
+Datenbankzugriff.

search/__init__.py

@@ -1,22 +1,10 @@
 """
-JonasSearch – Suchmaschine über den PostgreSQL-Index des Crawlers.
+JonasSearch Suchmaschine (Package `search`).
 
-Public API:
-    from search import SearchEngine, SearchHit, SearchResponse
+Module:
+    engine  – Suchlogik über PostgreSQL-Volltextindex (pages.search_vector)
+    api     – HTTP-API + Weboberfläche (Flask, Port 8083)
+    cli     – Kommandozeilen-Werkzeug (python3 search/cli.py)
 """
 
-from .engine import (
-    SearchEngine,
-    SearchHit,
-    SearchResponse,
-    parse_terms,
-)
-
-__all__ = [
-    "SearchEngine",
-    "SearchHit",
-    "SearchResponse",
-    "parse_terms",
-]
-
-__version__ = "1.0.0"
+__version__ = "1.0.0"

search/api.py

@@ -1,395 +1,591 @@
 #!/usr/bin/env python3
 """
-JonasSearch – Suchmaschine (HTTP-API).
+JonasSearch Suchmaschine – HTTP-API + Weboberfläche (api.py)
 
-Einfacher Suchserver über den vom Crawler aufgebauten Index:
+Läuft auf PORT = Vorschlagsverfeinerer (8082) + 1 = 8083.
 
-    python3 search/api.py
+Endpunkte:
+    GET /               Weboberfläche (Suchseite)
+    GET /api/search     Volltextsuche (q, page, per_page, site, tld,
+                        after, before, sort, include_errors,
+                        include_non_html, snippets)
+    GET /api/suggest    Autocomplete (q, limit)
+    GET /api/page       einzelne Seite (id oder url)
+    GET /api/stats      Index-Statistiken
+    GET /health         Healthcheck (DB-Status)
 
-    curl 'http://localhost:8082/search?q=raspberry%20pi'
-
-Unter / gibt es eine kleine Suchseite, unter /search die JSON-API.
-
-Konfiguration über Umgebungsvariablen:
-    JONASSEARCH_DB_HOST / PORT / NAME / USER / PASSWORD
-    JONASSEARCH_TS_CONFIG
-    JONASSEARCH_API_HOST   (Standard: 0.0.0.0)
-    JONASSEARCH_API_PORT   (Standard: 8082)
+Port-Auflösung: --port > JONASSEARCH_API_PORT (.env) > 8083.
+Der systemd-Dienst (search/jonassearch-engine.service) pinnt
+JONASSEARCH_API_PORT=8083 NACH dem EnvironmentFile und überschreibt
+damit einen eventuell abweichenden Wert aus .env.
 """
 
 import logging
 import os
+import sys
+import traceback
+from pathlib import Path
 
-from flask import Flask, jsonify, render_template_string, request
+from dotenv import load_dotenv
 
-try:
-    from .engine import SearchEngine
-except ImportError:  # python3 search/api.py (ohne -m)
-    from engine import SearchEngine
+# .env laden BEVOR search.engine importiert wird (liest die
+# JONASSEARCH_*-Variablen bei Importzeit): erst Working Directory,
+# dann .env neben dem Projektstamm – wie launcher.py.
+for _dotenv_path in (None, Path(__file__).resolve().parent.parent / ".env"):
+    try:
+        load_dotenv(_dotenv_path)
+    except OSError:
+        pass
 
-API_HOST = os.getenv("JONASSEARCH_API_HOST", "0.0.0.0")
-API_PORT = int(os.getenv("JONASSEARCH_API_PORT", "8082"))
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
 
-PAGE_HTML = """\
+import psycopg2  # noqa: E402
+from flask import Flask, jsonify, render_template_string, request  # noqa: E402
+
+try:  # Paketkontext (python3 -m search.api)
+    from . import engine  # noqa: E402
+except ImportError:  # Direktstart: python3 search/api.py
+    from search import engine  # noqa: E402
+
+logging.basicConfig(
+    level=logging.INFO,
+    format="%(asctime)s [search-api] %(levelname)s %(message)s",
+    datefmt="%H:%M:%S",
+)
+log = logging.getLogger("jonassearch.search-api")
+
+DEFAULT_PORT = 8083
+
+app = Flask(__name__)
+
+
+# ============================================================
+# HELFER
+# ============================================================
+
+def _truthy(value):
+    return str(value or "").strip().lower() in ("1", "true", "yes", "on", "ja")
+
+
+def _page_param(name, default):
+    raw = request.args.get(name)
+    if raw is None or str(raw).strip() == "":
+        return default
+    try:
+        return int(raw)
+    except ValueError:
+        raise engine.QueryError(f"{name} muss eine Zahl sein.")
+
+
+def _search_params():
+    """Query-Parameter der Request -> engine.search()-Argumente."""
+    q = request.args.get("q", "").strip()
+    if not q:
+        raise engine.QueryError(
+            "Parameter 'q' (Suchbegriffe) fehlt oder ist leer."
+        )
+    return dict(
+        query=q,
+        page=_page_param("page", 1),
+        per_page=_page_param("per_page", None),
+        sites=request.args.getlist("site") + request.args.getlist("sites"),
+        tlds=request.args.getlist("tld") + request.args.getlist("tlds"),
+        after=request.args.get("after"),
+        before=request.args.get("before"),
+        sort=request.args.get("sort", "relevance"),
+        include_errors=_truthy(request.args.get("include_errors")),
+        include_non_html=_truthy(request.args.get("include_non_html")),
+        with_snippets=not _truthy(request.args.get("no_snippets")),
+    )
+
+
+# ============================================================
+# API-ENDPUNKTE
+# ============================================================
+
+@app.route("/api/search")
+def api_search():
+    try:
+        result = engine.search(**_search_params())
+        return jsonify(result)
+    except engine.QueryError as exc:
+        return jsonify({"error": str(exc)}), 400
+    except engine.SearchTimeout as exc:
+        return jsonify({"error": str(exc)}), 504
+    except psycopg2.OperationalError as exc:
+        log.error("Datenbank nicht erreichbar: %s", exc)
+        return jsonify({"error": "Datenbank nicht erreichbar."}), 503
+    except engine.SearchEngineError as exc:
+        log.error("Suchfehler: %s", exc)
+        return jsonify({"error": str(exc)}), 500
+    except Exception:
+        log.error("Unerwarteter Fehler in /api/search:\n%s",
+                  traceback.format_exc())
+        return jsonify({"error": "Interner Fehler."}), 500
+
+
+@app.route("/api/suggest")
+def api_suggest():
+    try:
+        limit = _page_param("limit", engine.MAX_SUGGEST)
+        return jsonify({
+            "query": request.args.get("q", ""),
+            "suggestions": engine.suggest(request.args.get("q", ""), limit),
+        })
+    except engine.QueryError as exc:
+        return jsonify({"error": str(exc)}), 400
+    except psycopg2.OperationalError:
+        # Vorschläge sind optional: still leere Liste liefern.
+        return jsonify({"query": request.args.get("q", ""),
+                        "suggestions": []})
+
+
+@app.route("/api/page")
+def api_page():
+    try:
+        page_id = request.args.get("id")
+        url = request.args.get("url")
+        page = engine.get_page(
+            page_id=page_id,
+            url=url,
+            max_content_chars=_page_param(
+                "max_content_chars", engine.PAGE_CONTENT_CHARS
+            ),
+        )
+        if page is None:
+            return jsonify({"error": "Seite nicht gefunden."}), 404
+        return jsonify(page)
+    except engine.QueryError as exc:
+        return jsonify({"error": str(exc)}), 400
+    except psycopg2.OperationalError as exc:
+        log.error("Datenbank nicht erreichbar: %s", exc)
+        return jsonify({"error": "Datenbank nicht erreichbar."}), 503
+
+
+@app.route("/api/stats")
+def api_stats():
+    try:
+        return jsonify(engine.stats())
+    except psycopg2.OperationalError as exc:
+        log.error("Datenbank nicht erreichbar: %s", exc)
+        return jsonify({"error": "Datenbank nicht erreichbar."}), 503
+
+
+@app.route("/health")
+def health():
+    status = engine.health()
+    return jsonify(status), (200 if status.get("ok") else 503)
+
+
+# ============================================================
+# WEBOBERFLÄCHE
+# ============================================================
+
+HTML = r"""
 <!DOCTYPE html>
 <html lang="de">
 <head>
-    <meta charset="UTF-8">
-    <meta name="viewport" content="width=device-width, initial-scale=1.0">
-    <title>JonasSearch</title>
-
-    <style>
-        body {
-            margin: 0;
-            background: #111;
-            color: #eee;
-            font-family: Arial, sans-serif;
-        }
-
-        .container {
-            max-width: 800px;
-            margin: auto;
-            padding: 30px;
-        }
-
-        h1 {
-            margin: 0 0 4px;
-        }
-
-        .subtitle {
-            color: #888;
-            margin-bottom: 24px;
-        }
-
-        form.search {
-            display: flex;
-            gap: 8px;
-            margin-bottom: 16px;
-        }
-
-        input[type="text"] {
-            flex: 1;
-            padding: 12px 14px;
-            font-size: 1rem;
-            border-radius: 10px;
-            border: 1px solid #333;
-            background: #1b1b1b;
-            color: #eee;
-        }
-
-        button {
-            padding: 12px 20px;
-            font-size: 1rem;
-            border: none;
-            border-radius: 10px;
-            background: #2563eb;
-            color: #fff;
-            cursor: pointer;
-        }
-
-        button:hover {
-            background: #1d4ed8;
-        }
-
-        .meta {
-            color: #aaa;
-            margin-bottom: 20px;
-        }
-
-        .result {
-            background: #1b1b1b;
-            border-radius: 12px;
-            padding: 16px 18px;
-            margin-bottom: 12px;
-        }
-
-        .result h3 {
-            margin: 0 0 4px;
-            font-size: 1.05rem;
-        }
-
-        .result h3 a {
-            color: #60a5fa;
-            text-decoration: none;
-        }
-
-        .result h3 a:hover {
-            text-decoration: underline;
-        }
-
-        .result .url {
-            color: #4ade80;
-            font-size: 0.85rem;
-            margin-bottom: 6px;
-            overflow-wrap: anywhere;
-        }
-
-        .result .snippet {
-            color: #ccc;
-            line-height: 1.5;
-        }
-
-        .result .snippet mark {
-            background: #ca8a04;
-            color: #111;
-            border-radius: 3px;
-            padding: 0 2px;
-        }
-
-        .result .meta {
-            color: #777;
-            font-size: 0.8rem;
-            margin-top: 8px;
-            margin-bottom: 0;
-        }
-
-        .pagination {
-            display: flex;
-            gap: 10px;
-            margin-top: 20px;
-        }
-
-        .pagination a {
-            color: #60a5fa;
-            text-decoration: none;
-        }
-
-        .empty {
-            color: #888;
-            text-align: center;
-            margin-top: 40px;
-        }
-    </style>
+<meta charset="UTF-8">
+<meta name="viewport" content="width=device-width, initial-scale=1.0">
+<title>JonasSearch – Websuche</title>
+<style>
+:root {
+    --bg: #0f1115;
+    --card: #1b1e24;
+    --border: #2c313a;
+    --text: #e8eaed;
+    --muted: #9aa0a6;
+    --accent: #4f8cff;
+    --mark: #f5a623;
+    --link: #8ab4f8;
+}
+html[data-theme="light"] {
+    --bg: #f4f6fa;
+    --card: #ffffff;
+    --border: #d8dde5;
+    --text: #1a1d23;
+    --muted: #6b7280;
+    --accent: #2f6fe4;
+    --mark: #b45309;
+    --link: #1a4fc0;
+}
+* { box-sizing: border-box; }
+body {
+    margin: 0;
+    background: var(--bg);
+    color: var(--text);
+    font-family: -apple-system, BlinkMacSystemFont, 'Segoe UI', Roboto,
+                 Arial, sans-serif;
+}
+.wrap { max-width: 860px; margin: 0 auto; padding: 40px 18px 60px; }
+header {
+    display: flex; align-items: baseline; gap: 12px; margin-bottom: 26px;
+}
+h1 { margin: 0; font-size: 2rem; letter-spacing: .3px; }
+h1 a { color: var(--text); text-decoration: none; }
+h1 a:hover { color: var(--accent); }
+.badge {
+    background: var(--card); border: 1px solid var(--border);
+    color: var(--muted); font-size: .78rem; padding: 4px 10px;
+    border-radius: 999px; white-space: nowrap;
+}
+.searchbar { position: relative; display: flex; gap: 10px; }
+#q {
+    flex: 1; font-size: 1.05rem; padding: 13px 16px; color: var(--text);
+    background: var(--card); border: 1px solid var(--border);
+    border-radius: 12px; outline: none;
+}
+#q:focus { border-color: var(--accent); }
+button {
+    background: var(--accent); border: none; color: #fff;
+    font-size: .95rem; font-weight: 600; padding: 12px 22px;
+    border-radius: 12px; cursor: pointer;
+}
+button:hover { filter: brightness(1.1); }
+button:disabled { opacity: .45; cursor: default; filter: none; }
+.filters { display: flex; flex-wrap: wrap; gap: 10px; margin-top: 12px; }
+.filters input, .filters select {
+    background: var(--card); color: var(--text);
+    border: 1px solid var(--border); border-radius: 999px;
+    font-size: .82rem; padding: 7px 14px; outline: none;
+}
+.filters input:focus, .filters select:focus { border-color: var(--accent); }
+.suggest {
+    position: absolute; top: 54px; left: 0; right: 90px; z-index: 20;
+    background: var(--card); border: 1px solid var(--border);
+    border-radius: 12px; overflow: hidden; box-shadow: 0 10px 30px rgba(0,0,0,.35);
+}
+.suggest.hidden { display: none; }
+.suggest-item {
+    padding: 9px 14px; cursor: pointer; font-size: .92rem;
+    display: flex; justify-content: space-between; gap: 12px;
+}
+.suggest-item:hover { background: rgba(128,128,128,.12); }
+.suggest-item .dom { color: var(--muted); font-size: .78rem; }
+.info { color: var(--muted); font-size: .85rem; margin: 20px 0 14px; }
+.info.error { color: #ff6b6b; }
+.result { margin-bottom: 26px; }
+.result .title {
+    display: block; color: var(--link); font-size: 1.08rem;
+    font-weight: 600; text-decoration: none; margin-bottom: 2px;
+    overflow-wrap: anywhere;
+}
+.result .title:hover { text-decoration: underline; }
+.result .url { color: #2ecc71; font-size: .8rem; margin-bottom: 4px;
+               overflow-wrap: anywhere; }
+html[data-theme="light"] .result .url { color: #157a3c; }
+.result .snippet { font-size: .9rem; line-height: 1.55;
+                   color: var(--text); overflow-wrap: anywhere; }
+.result .meta { color: var(--muted); font-size: .76rem; margin-top: 4px; }
+mark { background: none; color: var(--mark); font-weight: 700;
+       padding: 0; }
+.pager { display: flex; align-items: center; gap: 16px; margin: 34px 0 0; }
+.pager .pageinfo { color: var(--muted); font-size: .85rem; }
+.pager button { background: var(--card); border: 1px solid var(--border);
+                color: var(--text); padding: 8px 16px; }
+.pager button:hover { border-color: var(--accent); }
+.hint { color: var(--muted); margin-top: 40px; font-size: .9rem;
+        line-height: 1.7; }
+.hint code { background: var(--card); border: 1px solid var(--border);
+             border-radius: 6px; padding: 1px 7px; font-size: .82rem; }
+footer { color: var(--muted); font-size: .76rem; margin-top: 60px;
+         border-top: 1px solid var(--border); padding-top: 14px; }
+@media (max-width: 600px) {
+    .wrap { padding: 24px 12px 40px; }
+    .searchbar { flex-direction: column; }
+    .suggest { right: 0; top: 106px; }
+}
+</style>
 </head>
-
 <body>
-<div class="container">
-    <h1>JonasSearch</h1>
-    <div class="subtitle">Suchmaschine über den Crawler-Index</div>
+<div class="wrap">
+  <header>
+    <h1><a href="/">JonasSearch</a></h1>
+    <span class="badge" id="stats-badge" style="display:none"></span>
+  </header>
 
-    <form class="search" id="searchform">
-        <input
-            type="text"
-            id="q"
-            placeholder="Suchen…"
-            autofocus
-            autocomplete="off"
-        >
-        <button type="submit">Suchen</button>
-    </form>
+  <form id="f" autocomplete="off">
+    <div class="searchbar">
+      <input id="q" type="search" placeholder="Suchbegriffe – &quot;exakte Phrase&quot;, OR, -ausschluss" autofocus>
+      <button type="submit">Suchen</button>
+      <div id="suggest" class="suggest hidden"></div>
+    </div>
+    <div class="filters">
+      <input id="site" placeholder="site: example.com" size="18">
+      <select id="range">
+        <option value="">Beliebige Zeit</option>
+        <option value="24h">Letzte 24 Stunden</option>
+        <option value="7d">Letzte 7 Tage</option>
+        <option value="30d">Letzte 30 Tage</option>
+        <option value="365d">Letztes Jahr</option>
+      </select>
+      <select id="sort">
+        <option value="relevance">Relevanz</option>
+        <option value="freshness">Aktualität</option>
+      </select>
+    </div>
+  </form>
 
-    <div class="meta" id="meta"></div>
-    <div id="results"></div>
-    <div class="pagination" id="pagination"></div>
+  <div id="info" class="info"></div>
+  <div id="results"></div>
+
+  <div id="pager" class="pager" style="display:none">
+    <button id="prev">&lsaquo; Zurück</button>
+    <span id="pageinfo" class="pageinfo"></span>
+    <button id="next">Weiter &rsaquo;</button>
+  </div>
+
+  <div class="hint" id="hint">
+    Beispiele: <code>postgreSQL volltextsuche</code> &middot;
+    <code>"exacte phrase"</code> &middot;
+    <code>crawler -python</code> &middot;
+    <code>postgres OR sqlite</code>
+  </div>
+
+  <footer>JonasSearch Suchmaschine &middot; PostgreSQL-Volltextsuche
+  über gecrawlte Webseiten &middot; Port 8083</footer>
 </div>
 
 <script>
-const form = document.getElementById("searchform");
-const input = document.getElementById("q");
-const meta = document.getElementById("meta");
-const results = document.getElementById("results");
-const pagination = document.getElementById("pagination");
+'use strict';
+const $ = id => document.getElementById(id);
+const state = { q: "", site: "", range: "", sort: "relevance", page: 1 };
+let suggestTimer = null;
 
-let currentQuery = "";
-let currentPage = 1;
-
-function urlFor(page) {
-    return "/search?q=" + encodeURIComponent(currentQuery) + "&page=" + page;
+function esc(s) {
+    return String(s == null ? "" : s)
+        .replaceAll("&", "&amp;").replaceAll("<", "&lt;")
+        .replaceAll(">", "&gt;").replaceAll('"', "&quot;");
+}
+function fmtInt(n) { return Number(n).toLocaleString("de-DE"); }
+function prettyUrl(u) {
+    let s = String(u || "").replace(/^https?:\/\//, "");
+    if (s.length > 100) s = s.slice(0, 100) + "…";
+    return s;
+}
+function fmtDate(iso) {
+    if (!iso) return "–";
+    const d = new Date(iso);
+    return isNaN(d) ? "–"
+        : d.toLocaleDateString("de-DE",
+            { day: "2-digit", month: "2-digit", year: "numeric" });
 }
 
-function render(data) {
-    const query = data.query || "";
-    meta.textContent = data.total
-        ? data.total + " Ergebnisse in " + data.took_ms + " ms"
-            + (data.pages > 1 ? " – Seite " + data.page + "/" + data.pages : "")
-        : "Keine Ergebnisse für „" + query + "“";
+function fromHash() {
+    const p = new URLSearchParams(location.hash.slice(1));
+    state.q = p.get("q") || "";
+    state.site = p.get("site") || "";
+    state.range = p.get("after") || "";
+    state.sort = p.get("sort") || "relevance";
+    state.page = Math.max(1, parseInt(p.get("page") || "1", 10) || 1);
+    $("q").value = state.q;
+    $("site").value = state.site;
+    $("range").value = state.range;
+    $("sort").value = state.sort;
+}
+function toHash() {
+    const p = new URLSearchParams();
+    if (state.q) p.set("q", state.q);
+    if (state.site) p.set("site", state.site);
+    if (state.range) p.set("after", state.range);
+    if (state.sort && state.sort !== "relevance") p.set("sort", state.sort);
+    if (state.page > 1) p.set("page", String(state.page));
+    history.replaceState(null, "", p.toString() ? "#" + p.toString() : "#");
+}
 
-    results.innerHTML = "";
+function readInputs() {
+    state.q = $("q").value.trim();
+    state.site = $("site").value.trim();
+    state.range = $("range").value;
+    state.sort = $("sort").value;
+    state.page = 1;
+}
 
-    for (const hit of data.hits) {
-        const box = document.createElement("div");
-        box.className = "result";
-
-        const title = document.createElement("h3");
-        const link = document.createElement("a");
-        link.href = hit.url;
-        link.target = "_blank";
-        link.rel = "noopener";
-        link.textContent = hit.title;
-        title.appendChild(link);
-        box.appendChild(title);
-
-        const url = document.createElement("div");
-        url.className = "url";
-        url.textContent = hit.url;
-        box.appendChild(url);
-
-        // Das Snippet enthält vom Server bereits <mark>-Tags;
-        // der Text selbst ist serverseitig escaped (XSS-sicher).
-        const snippet = document.createElement("div");
-        snippet.className = "snippet";
-        snippet.innerHTML = hit.snippet;
-        box.appendChild(snippet);
-
-        const metaLine = document.createElement("div");
-        metaLine.className = "meta";
-        const parts = [];
-        if (hit.domain) parts.push(hit.domain);
-        if (hit.crawled_at) parts.push(hit.crawled_at.slice(0, 10));
-        metaLine.textContent = parts.join(" · ");
-        if (metaLine.textContent) box.appendChild(metaLine);
-
-        results.appendChild(box);
+async function run() {
+    hideSuggest();
+    if (!state.q) {
+        $("info").textContent = "";
+        $("results").innerHTML = "";
+        $("pager").style.display = "none";
+        $("hint").style.display = "";
+        return;
     }
+    $("hint").style.display = "none";
+    $("info").className = "info";
+    $("info").textContent = "Suche …";
+    $("results").innerHTML = "";
 
-    pagination.innerHTML = "";
-    if (data.pages > 1) {
-        if (data.page > 1) {
-            const prev = document.createElement("a");
-            prev.href = "#";
-            prev.textContent = "← Zurück";
-            prev.addEventListener("click", e => {
-                e.preventDefault();
-                loadPage(data.page - 1);
-            });
-            pagination.appendChild(prev);
-        }
+    const p = new URLSearchParams({ q: state.q, page: state.page });
+    if (state.site) p.set("site", state.site);
+    if (state.range) p.set("after", state.range);
+    if (state.sort) p.set("sort", state.sort);
 
-        const label = document.createElement("span");
-        label.textContent = "Seite " + data.page + " / " + data.pages;
-        pagination.appendChild(label);
-
-        if (data.page < data.pages) {
-            const next = document.createElement("a");
-            next.href = "#";
-            next.textContent = "Weiter →";
-            next.addEventListener("click", e => {
-                e.preventDefault();
-                loadPage(data.page + 1);
-            });
-            pagination.appendChild(next);
-        }
-    }
-}
-
-function showError(message) {
-    meta.textContent = message;
-    results.innerHTML = "";
-    pagination.innerHTML = "";
-}
-
-async function loadPage(page) {
-    let response;
+    let data;
     try {
-        response = await fetch(urlFor(page));
+        const res = await fetch("/api/search?" + p.toString());
+        data = await res.json();
+        if (!res.ok) {
+            $("info").className = "info error";
+            $("info").textContent = data.error ||
+                ("Fehler " + res.status);
+            return;
+        }
     } catch (e) {
-        showError("Suche fehlgeschlagen – Server nicht erreichbar.");
+        $("info").className = "info error";
+        $("info").textContent = "Suchdienst nicht erreichbar.";
         return;
     }
 
-    if (!response.ok) {
-        let message = "Suche fehlgeschlagen (HTTP " + response.status + ").";
-        try {
-            const err = await response.json();
-            if (err && err.error) message = err.error;
-        } catch (e) { /* Antwort war kein JSON */ }
-        showError(message);
+    const totalTxt = data.total != null
+        ? "Etwa " + fmtInt(data.total) + " Treffer"
+        : "Viele Treffer";
+    const emptyHint = data.parsed_query ? "" :
+        " · keine durchsuchbaren Begriffe gefunden";
+    const capHint = data.ranking_capped ?
+        " · Ranking auf Top-Kandidaten eingegrenzt" : "";
+    const degradedHint = data.ranking_degraded ?
+        " · Trefferreihenfolge vereinfacht (Suchbegriffe sehr häufig)" : "";
+    $("info").textContent = totalTxt + " · Seite " + data.page +
+        " · " + (data.took_ms / 1000).toFixed(2) + " s" + emptyHint +
+        capHint + degradedHint;
+
+    if (!data.results.length) {
+        $("results").innerHTML =
+            '<div class="info">Keine Treffer. Andere Begriffe oder ' +
+            'weniger Filter probieren.</div>';
+        $("pager").style.display = "none";
         return;
     }
 
-    const data = await response.json();
-    currentPage = data.page;
-    render(data);
-    window.scrollTo({ top: 0, behavior: "smooth" });
+    $("results").innerHTML = data.results.map(r => {
+        const snippet = esc(r.snippet)
+            .split("⟦").join("<mark>").split("⟧").join("</mark>");
+        return '<div class="result">' +
+            '<a class="title" href="' + esc(r.url) + '" target="_blank" ' +
+            'rel="noopener">' + esc(r.title) + "</a>" +
+            '<div class="url">' + esc(prettyUrl(r.url)) + "</div>" +
+            (snippet ? '<div class="snippet">' + snippet + "</div>" : "") +
+            '<div class="meta">' + esc(r.domain || prettyUrl(r.url)) +
+            " · " + fmtDate(r.crawled_at) +
+            " · Score " + r.score.toFixed(4) + "</div></div>";
+    }).join("");
+
+    $("pager").style.display = "";
+    $("pageinfo").textContent = "Seite " + data.page;
+    $("prev").disabled = data.page <= 1;
+    $("next").disabled = !data.has_more;
 }
 
-form.addEventListener("submit", async event => {
-    event.preventDefault();
-    const query = input.value.trim();
-    if (!query) return;
+/* ---------- Vorschläge (Autocomplete) ---------- */
+function hideSuggest() { $("suggest").classList.add("hidden"); }
+$("q").addEventListener("input", () => {
+    clearTimeout(suggestTimer);
+    suggestTimer = setTimeout(async () => {
+        const q = $("q").value.trim();
+        if (q.length < 2) { hideSuggest(); return; }
+        try {
+            const res = await fetch("/api/suggest?q=" + encodeURIComponent(q));
+            const data = await res.json();
+            const items = data.suggestions || [];
+            if (!items.length) { hideSuggest(); return; }
+            $("suggest").innerHTML = items.map(s =>
+                '<div class="suggest-item" data-title="' +
+                esc(s.title) + '"><span>' + esc(s.title) + "</span>" +
+                '<span class="dom">' + esc(s.domain) + "</span></div>"
+            ).join("");
+            $("suggest").classList.remove("hidden");
+        } catch (e) { hideSuggest(); }
+    }, 250);
+});
+$("suggest").addEventListener("mousedown", e => {
+    const item = e.target.closest(".suggest-item");
+    if (item) {
+        $("q").value = item.dataset.title;
+        hideSuggest();
+        readInputs(); toHash(); run();
+    }
+});
+$("q").addEventListener("blur", () => setTimeout(hideSuggest, 180));
 
-    currentQuery = query;
-    await loadPage(1);
-
-    history.replaceState(null, "", "/?q=" + encodeURIComponent(query));
+/* ---------- Ereignisse ---------- */
+$("f").addEventListener("submit", e => {
+    e.preventDefault();
+    readInputs(); toHash(); run();
+});
+$("prev").addEventListener("click", () => {
+    if (state.page > 1) { state.page--; toHash(); run(); }
+});
+$("next").addEventListener("click", () => { state.page++; toHash(); run(); });
+document.addEventListener("keydown", e => {
+    if (e.key === "/" && document.activeElement !== $("q") &&
+        document.activeElement !== $("site")) {
+        e.preventDefault(); $("q").focus();
+    }
 });
 
-// Beim Laden vorbefüllen (z. B. nach Reload mit ?q=)
-const initial = new URLSearchParams(location.search).get("q");
-if (initial) {
-    input.value = initial;
-    currentQuery = initial;
-    loadPage(1);
-}
+/* ---------- Start ---------- */
+fromHash();
+if (state.q) run();
+(async () => {
+    try {
+        const res = await fetch("/api/stats");
+        if (!res.ok) return;
+        const s = await res.json();
+        if (s.pages) {
+            const b = $("stats-badge");
+            b.textContent = "≈ " + fmtInt(s.pages) + " Seiten indexiert";
+            b.style.display = "";
+        }
+    } catch (e) { /* Statistik optional */ }
+})();
 </script>
-
 </body>
 </html>
 """
 
 
-def create_app() -> Flask:
-    app = Flask(__name__)
+@app.route("/")
+def index():
+    return render_template_string(HTML)
 
-    # Umlaute nicht als \uXXXX escapen.
-    try:
-        app.json.ensure_ascii = False
-    except AttributeError:
-        pass
-
-    engine = SearchEngine()
-
-    @app.route("/")
-    def index():
-        return render_template_string(PAGE_HTML)
-
-    @app.route("/search")
-    def search():
-        query = request.args.get("q", "").strip()
-
-        if not query:
-            return jsonify({"error": "Parameter 'q' fehlt."}), 400
-
-        page_arg = request.args.get("page", "1")
-        per_page_arg = request.args.get("per_page", "10")
-
-        try:
-            page = max(1, int(page_arg))
-        except ValueError:
-            return jsonify({"error": "Parameter 'page' muss eine Zahl sein."}), 400
-
-        try:
-            per_page = max(1, min(int(per_page_arg), 50))
-        except ValueError:
-            return jsonify({"error": "Parameter 'per_page' muss eine Zahl sein."}), 400
-
-        domain = request.args.get("domain") or None
-        tld = request.args.get("tld") or None
-
-        try:
-            response = engine.search(
-                query=query,
-                page=page,
-                per_page=per_page,
-                domain=domain,
-                tld=tld,
-            )
-        except Exception as exc:
-            logging.exception("Suche fehlgeschlagen")
-            return jsonify({"error": f"Suche fehlgeschlagen: {exc}"}), 500
-
-        return jsonify(response.to_dict())
-
-    return app
 
+# ============================================================
+# START
+# ============================================================
 
 if __name__ == "__main__":
-    logging.basicConfig(level=logging.INFO)
+    import argparse
 
-    print("=" * 60)
-    print(" JonasSearch – Suchmaschine (HTTP-API)")
-    print(f" URL:     http://{API_HOST}:{API_PORT}/")
-    print(f" JSON:    http://{API_HOST}:{API_PORT}/search?q=...")
-    print("=" * 60)
+    parser = argparse.ArgumentParser(
+        prog="search/api.py",
+        description="JonasSearch Suchmaschine (HTTP-API + Weboberfläche)",
+    )
+    parser.add_argument("--host", default=None,
+                        help="Bind-Adresse (Default: JONASSEARCH_API_HOST "
+                             "oder 0.0.0.0)")
+    parser.add_argument("--port", type=int, default=None,
+                        help="Port (Default: JONASSEARCH_API_PORT oder 8083)")
+    args = parser.parse_args()
 
-    create_app().run(host=API_HOST, port=API_PORT, threaded=True)
+    host = args.host or os.getenv("JONASSEARCH_API_HOST",
+                                  os.getenv("HOST", "0.0.0.0"))
+    port = args.port or int(os.getenv("JONASSEARCH_API_PORT",
+                                      str(DEFAULT_PORT)))
+
+    print("=" * 70)
+    print("     JonasSearch Suchmaschine")
+    print("=" * 70)
+    print(f"Statserver : Port 8081 (jonassearch-stats.service)")
+    print(f"Priorität  : Port 8082 (jonassearch-priority.service)")
+    print(f"Suchmaschine: http://{host}:{port}")
+    print(f"DB         : {engine.DB_HOST}:{engine.DB_PORT}/{engine.DB_NAME}")
+    print(f"TS-Config  : {engine.TS_CONFIG}")
+    print("=" * 70)
+
+    app.run(
+        host=host,
+        port=port,
+        debug=False,
+        threaded=True,
+        load_dotenv=False,  # .env wurde oben schon geladen
+    )

search/cli.py

@@ -1,153 +1,259 @@
 #!/usr/bin/env python3
 """
-JonasSearch – Suchmaschine (Kommandozeile).
+JonasSearch Suchmaschine – Kommandozeile (cli.py)
 
-Suche im vom Crawler aufgebauten Index direkt vom Terminal:
-
-    python3 search/cli.py "raspberry pi"
-    python3 search/cli.py "postgres" --domain example.com --page 2
-    python3 search/cli.py "amplifier" --json
-
-Konfiguration über Umgebungsvariablen (siehe .env.example):
-    JONASSEARCH_DB_HOST / PORT / NAME / USER / PASSWORD
-    JONASSEARCH_TS_CONFIG
+Beispiele:
+    python3 search/cli.py "postgresql volltextsuche"
+    python3 search/cli.py "crawler" --site example.com --sort freshness
+    python3 search/cli.py "python" --after 7d --limit 20
+    python3 search/cli.py --suggest "suchmasch"
+    python3 search/cli.py --stats
+    python3 search/cli.py --page-id 42
+    python3 search/cli.py --url https://example.com/seite
+    python3 search/cli.py "test" --json
 """
 
 import argparse
-import html
 import json
-import re
 import sys
+import textwrap
+from pathlib import Path
 
-try:
-    from .engine import SearchEngine
-except ImportError:  # python3 search/cli.py (ohne -m)
-    from engine import SearchEngine
+from dotenv import load_dotenv
 
-_TAG_RE = re.compile(r"<[^>]+>")
+# .env laden, bevor engine importiert wird (wie api.py).
+for _dotenv_path in (None, Path(__file__).resolve().parent.parent / ".env"):
+    try:
+        load_dotenv(_dotenv_path)
+    except OSError:
+        pass
+
+try:  # Paketkontext (python3 -m search.cli)
+    from . import engine
+except ImportError:  # Direktstart: python3 search/cli.py
+    sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+    from search import engine
 
 
-def strip_markup(text: str) -> str:
-    """Entfernt <mark>-Tags und dekodiert HTML-Entities fürs Terminal."""
-    text = _TAG_RE.sub("", text or "")
-    return html.unescape(text)
+# ------------------------------------------------------------
+# Ausgabe-Helfer
+# ------------------------------------------------------------
+
+class Style:
+    def __init__(self, enabled):
+        self.bold = "\033[1m" if enabled else ""
+        self.green = "\033[32m" if enabled else ""
+        self.blue = "\033[36m" if enabled else ""
+        self.dim = "\033[2m" if enabled else ""
+        self.reset = "\033[0m" if enabled else ""
 
 
-def format_plain(response, show_rank=True) -> str:
-    lines = []
-    total = response.total
-    unit = "Ergebnis" if total == 1 else "Ergebnisse"
-
-    header = f"{total} {unit} für „{response.query}“ ({response.took_ms} ms)"
-    if response.pages > 0:
-        header += f" – Seite {response.page}/{response.pages}"
-    lines.append(header)
-    lines.append("")
-
-    for hit in response.hits:
-        if show_rank:
-            lines.append(f"{hit.position:4d}. {hit.title}")
-        else:
-            lines.append(f"     {hit.title}")
-
-        lines.append(f"     {hit.url}")
-
-        if hit.snippet:
-            lines.append(f"     {strip_markup(hit.snippet)}")
-
-        meta = []
-        if hit.domain:
-            meta.append(hit.domain)
-        if hit.crawled_at:
-            meta.append(hit.crawled_at[:10])
-        if hit.rank:
-            meta.append(f"Relevanz {hit.rank:.4f}")
-        if meta:
-            lines.append(f"     {' · '.join(meta)}")
-
-        lines.append("")
-
-    if not response.hits:
-        lines.append("Keine Treffer. Versuche weniger oder andere Begriffe.")
-
-    return "\n".join(lines)
+def _highlight(snippet, style):
+    """⟦…⟧-Markierungen -> ANSI-Hervorhebung (oder Plaintext)."""
+    return (snippet
+            .replace(engine.SNIPPET_START,
+                     style.bold + style.green)
+            .replace(engine.SNIPPET_STOP, style.reset))
 
 
-def build_parser() -> argparse.ArgumentParser:
+def _fmt_date(iso):
+    if not iso:
+        return "–"
+    return str(iso)[:16].replace("T", " ")
+
+
+def print_results(data, style):
+    print(
+        f"{style.dim}"
+        f"{'Etwa ' if data['total'] is not None else ''}"
+        f"{data['total'] if data['total'] is not None else 'Viele'} "
+        f"Treffer · Seite {data['page']} · "
+        f"{data['took_ms'] / 1000:.2f} s"
+        f"{style.reset}"
+    )
+    if not data["parsed_query"]:
+        print(f"{style.dim}Hinweis: Anfrage enthält keine durchsuchbaren "
+              f"Begriffe.{style.reset}")
+    if not data["results"]:
+        print("Keine Treffer.")
+        return
+
+    for i, r in enumerate(data["results"], start=1):
+        print()
+        print(f"{style.bold}{i}. {r['title']}{style.reset}")
+        print(f"{style.blue}{r['url']}{style.reset}")
+        if r["snippet"]:
+            wrapped = textwrap.fill(
+                _highlight(r["snippet"], style), width=100,
+                initial_indent="   ", subsequent_indent="   ",
+            )
+            print(wrapped)
+        meta = (f"   {style.dim}{r['domain'] or '–'} · "
+                f"{_fmt_date(r['crawled_at'])} · "
+                f"Score {r['score']:.4f}")
+        if r["http_status"] not in (200, None):
+            meta += f" · HTTP {r['http_status']}"
+        print(meta + style.reset)
+
+    print()
+    nav = []
+    if data["page"] > 1:
+        nav.append(f"--page {data['page'] - 1}")
+    if data["has_more"]:
+        nav.append(f"--page {data['page'] + 1}")
+    if nav:
+        print(f"{style.dim}Weitere Seite: " + " | ".join(nav) + style.reset)
+
+
+def print_page(page, style, max_chars):
+    print(f"{style.bold}{page.get('title') or '(ohne Titel)'}{style.reset}")
+    print(f"{style.blue}{page['url']}{style.reset}")
+    print(
+        f"{style.dim}ID {page['id']} · {page['domain'] or '–'} · "
+        f"HTTP {page['http_status']} · {page['content_type'] or '–'} · "
+        f"{page['content_length']} Zeichen"
+        f"{' (gekürzt)' if page['content_truncated'] else ''}\n"
+        f"Gefunden am {_fmt_date(page.get('discovered_at'))} · "
+        f"Gecrawlt am {_fmt_date(page.get('crawled_at'))} · "
+        f"{page['crawl_time_ms'] or '–'} ms · "
+        f"{page['links_total']} Links{style.reset}"
+    )
+    if page.get("headings"):
+        print(f"\n{style.bold}Überschriften:{style.reset}")
+        for h in page["headings"][:15]:
+            print(f"  – {h}")
+    content = (page.get("content") or "")[:max_chars]
+    if content:
+        print(f"\n{style.bold}Inhalt (Auszug):{style.reset}")
+        print(textwrap.fill(content, width=100))
+
+
+def print_stats(s, style):
+    def fmt(value):
+        return "–" if value is None else f"{value:,}".replace(",", ".")
+
+    print(f"{style.bold}JonasSearch Index-Statistik{style.reset}")
+    print(f"  Seiten (geschätzt) : {fmt(s.get('pages'))}")
+    print(f"  Domains (geschätzt): {fmt(s.get('domains'))}")
+    print(f"  Queue gesamt       : {fmt(s.get('queue'))}")
+    print(f"  Queue ausstehend   : {fmt(s.get('queue_pending'))}")
+    print(f"  Letzter Crawl      : {_fmt_date(s.get('last_crawl'))}")
+    print(f"  Datenbankgröße     : {s.get('db_size') or '–'}")
+
+
+# ------------------------------------------------------------
+# Main
+# ------------------------------------------------------------
+
+def build_parser():
     parser = argparse.ArgumentParser(
-        prog="jonassearch",
-        description="JonasSearch – Suchmaschine (Kommandozeile).",
-        epilog=(
-            "Konfiguration über JONASSEARCH_* Umgebungsvariablen, "
-            "siehe .env.example."
-        ),
+        prog="search/cli.py",
+        description="JonasSearch Volltextsuche über der PostgreSQL-"
+                    "Datenbank (pages.search_vector).",
+        epilog='Query-Syntax: "exakte Phrase", OR, -ausschluss',
     )
-
-    parser.add_argument(
-        "query",
-        help="Suchabfrage, z. B. \"raspberry pi\"",
-    )
-    parser.add_argument(
-        "--page",
-        type=int,
-        default=1,
-        help="Seite der Ergebnisse (Standard: 1)",
-    )
-    parser.add_argument(
-        "--per-page",
-        type=int,
-        default=10,
-        help="Ergebnisse pro Seite, max. 50 (Standard: 10)",
-    )
-    parser.add_argument(
-        "--domain",
-        default=None,
-        help="Nur Seiten dieser exakten Domain (z. B. example.com)",
-    )
-    parser.add_argument(
-        "--tld",
-        default=None,
-        help="Nur Seiten dieser TLD (z. B. .de oder .com)",
-    )
-    parser.add_argument(
-        "--config",
-        default=None,
-        help="PostgreSQL-Textsuch-Konfiguration "
-             "(Standard: german, oder JONASSEARCH_TS_CONFIG)",
-    )
-    parser.add_argument(
-        "--json",
-        action="store_true",
-        help="Ergebnisse als JSON ausgeben",
-    )
-
+    parser.add_argument("query", nargs="?",
+                        help="Suchbegriffe (Websearch-Syntax)")
+    parser.add_argument("--site", action="append", default=[],
+                        help="Auf Domain einschränken (mehrfach möglich)")
+    parser.add_argument("--tld", action="append", default=[],
+                        help="Auf Top-Level-Domain einschränken (z. B. de)")
+    parser.add_argument("--after", help="frühestes Datum: 2026-01-31 oder "
+                                        "24h/7d/2w relativ")
+    parser.add_argument("--before", help="spätestes Datum (exklusiv) oder "
+                                         "24h/7d/2w relativ")
+    parser.add_argument("--sort", default="relevance",
+                        choices=["relevance", "freshness"],
+                        help="Relevanz (Default) oder Aktualität")
+    parser.add_argument("--page", type=int, default=1)
+    parser.add_argument("--limit", type=int, default=None,
+                        help="Treffer pro Seite (Default 10, max 50)")
+    parser.add_argument("--no-snippets", action="store_true",
+                        help="keine ts_headline-Snippets laden")
+    parser.add_argument("--include-errors", action="store_true",
+                        help="auch Seiten ohne HTTP 200 liefern")
+    parser.add_argument("--include-non-html", action="store_true",
+                        help="auch Nicht-HTML-Inhalte liefern")
+    parser.add_argument("--json", action="store_true",
+                        help="rohe JSON-Ausgabe")
+    parser.add_argument("--no-color", action="store_true")
+    parser.add_argument("--suggest", metavar="BEGRIFF",
+                        help="Autocomplete-Vorschläge anzeigen")
+    parser.add_argument("--stats", action="store_true",
+                        help="Index-Statistik anzeigen")
+    parser.add_argument("--page-id", type=int,
+                        help="gespeicherte Seite per ID anzeigen")
+    parser.add_argument("--url", help="gespeicherte Seite per URL anzeigen")
     return parser
 
 
-def main(argv=None) -> int:
+def main(argv=None):
     args = build_parser().parse_args(argv)
-
-    engine = SearchEngine(ts_config=args.config)
+    style = Style(sys.stdout.isatty() and not args.no_color)
 
     try:
-        response = engine.search(
-            query=args.query,
+        if args.stats:
+            print_stats(engine.stats(), style)
+            return 0
+
+        if args.suggest is not None:
+            suggestions = engine.suggest(args.suggest)
+            if not suggestions:
+                print("Keine Vorschläge.")
+                return 0
+            for s in suggestions:
+                print(f"  {s['title']}  {style.dim}({s['domain']})"
+                      f"{style.reset}")
+            return 0
+
+        if args.page_id is not None or args.url is not None:
+            page = engine.get_page(page_id=args.page_id, url=args.url)
+            if page is None:
+                print("Seite nicht gefunden.", file=sys.stderr)
+                return 1
+            if args.json:
+                print(json.dumps(page, ensure_ascii=False, indent=2))
+            else:
+                print_page(page, style, max_chars=3000)
+            return 0
+
+        if not args.query:
+            build_parser().error("Suchbegriffe fehlen (oder --stats / "
+                                 "--suggest / --page-id / --url nutzen).")
+
+        data = engine.search(
+            args.query,
             page=args.page,
-            per_page=args.per_page,
-            domain=args.domain,
-            tld=args.tld,
+            per_page=args.limit,
+            sites=args.site,
+            tlds=args.tld,
+            after=args.after,
+            before=args.before,
+            sort=args.sort,
+            include_errors=args.include_errors,
+            include_non_html=args.include_non_html,
+            with_snippets=not args.no_snippets,
         )
-    except Exception as exc:  # z. B. DB nicht erreichbar, Tabelle fehlt
-        print(f"Fehler bei der Suche: {exc}", file=sys.stderr)
+        if args.json:
+            print(json.dumps(data, ensure_ascii=False, indent=2))
+        else:
+            print_results(data, style)
+        return 0
+
+    except engine.QueryError as exc:
+        print(f"Fehlerhafte Anfrage: {exc}", file=sys.stderr)
+        return 2
+    except engine.SearchTimeout:
+        print("Die Suche dauert ungewöhnlich lange. Neuer Versuch oder ein "
+              "Filter (site=, nach=, weitere Begriffe) hilft.", file=sys.stderr)
+        return 4
+    except engine.SearchEngineError as exc:
+        print(f"Suchfehler: {exc}", file=sys.stderr)
+        return 3
+    except Exception as exc:
+        print(f"Fehler: {exc}", file=sys.stderr)
         return 1
 
-    if args.json:
-        print(json.dumps(response.to_dict(), ensure_ascii=False, indent=2))
-    else:
-        print(format_plain(response))
-
-    return 0
-
 
 if __name__ == "__main__":
-    sys.exit(main())
+    sys.exit(main())

search/jonassearch-engine.service

@@ -0,0 +1,42 @@
+# JonasSearch Suchmaschine (Volltextsuch-Weboberfläche + HTTP-API)
+#
+# Installieren (als root):
+#   cp /srv/serverjonas/search/search/jonassearch-engine.service \
+#      /etc/systemd/system/
+#   systemctl daemon-reload
+#   systemctl enable --now jonassearch-engine.service
+#
+# Ports: Statserver 8081 · Vorschlagsverfeinerer 8082 · Suchmaschine 8083
+#
+# Der Port ist unten über Environment= fest auf 8083 gepinnt; diese
+# Zuweisung gilt NACH dem EnvironmentFile und überschreibt damit einen
+# eventuell abweichenden JONASSEARCH_API_PORT-Wert aus .env.
+
+[Unit]
+Description=JonasSearch Suchmaschine (HTTP-API + Weboberfläche)
+After=network-online.target postgresql@17-jonassearch.service
+Wants=network-online.target
+
+[Service]
+Type=simple
+
+User=jonassearch
+Group=jonassearch
+
+WorkingDirectory=/srv/serverjonas/search
+
+EnvironmentFile=/srv/serverjonas/search/.env
+
+Environment=JONASSEARCH_API_PORT=8083
+
+ExecStart=/usr/bin/python3 /srv/serverjonas/search/search/api.py
+
+Restart=on-failure
+RestartSec=10
+
+# Sauberer Shutdown
+KillSignal=SIGTERM
+TimeoutStopSec=30
+
+[Install]
+WantedBy=multi-user.target

search/test_engine.py

@@ -0,0 +1,344 @@
+#!/usr/bin/env python3
+"""
+Local test suite for the search engine (no database required).
+
+Only the pure, DB-free parts are tested: parameter normalization,
+SQL assembly and the query-parsing helpers. Database-backed behaviour
+is covered by the statement timeouts in production.
+"""
+
+import os
+import re
+import sys
+import unittest
+from datetime import timedelta, timezone
+from unittest import mock
+
+sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
+
+
+def _load_env_file():
+    """
+    Load ../.env like the systemd EnvironmentFile does, so the tests
+    also run outside systemd. Must happen before engine.py is imported,
+    which reads the JONASSEARCH_* variables at import time.
+    """
+    env_path = os.path.join(
+        os.path.dirname(os.path.abspath(__file__)),
+        os.pardir,
+        ".env",
+    )
+
+    try:
+        with open(env_path) as handle:
+            for line in handle:
+                line = line.strip()
+                if not line or line.startswith("#") or "=" not in line:
+                    continue
+                key, _, value = line.partition("=")
+                os.environ.setdefault(key.strip(), value.strip())
+    except OSError:
+        pass
+
+
+_load_env_file()
+
+import engine  # noqa: E402
+
+
+class ParseRankWeightsTest(unittest.TestCase):
+
+    def test_defaults_map_to_pg_order(self):
+        # ts_rank_cd erwartet [D, C, B, A]
+        self.assertEqual(
+            engine.parse_rank_weights(None),
+            [0.1, 0.3, 0.6, 1.0],
+        )
+
+    def test_custom_spec(self):
+        self.assertEqual(
+            engine.parse_rank_weights("A=1.0,B=0.5,C=0.25,D=0.05"),
+            [0.05, 0.25, 0.5, 1.0],
+        )
+
+    def test_partial_spec_keeps_defaults(self):
+        self.assertEqual(
+            engine.parse_rank_weights("B=0.9"),
+            [0.1, 0.3, 0.9, 1.0],
+        )
+
+    def test_invalid_key(self):
+        with self.assertRaises(engine.QueryError):
+            engine.parse_rank_weights("E=1.0")
+
+    def test_invalid_value(self):
+        with self.assertRaises(engine.QueryError):
+            engine.parse_rank_weights("A=xyz")
+
+    def test_missing_equals(self):
+        with self.assertRaises(engine.QueryError):
+            engine.parse_rank_weights("A 1.0")
+
+
+class NormalizeSitesTest(unittest.TestCase):
+
+    def test_plain(self):
+        self.assertEqual(engine.normalize_sites(["example.com"]),
+                         ["example.com"])
+
+    def test_url_and_www(self):
+        self.assertEqual(
+            engine.normalize_sites(["https://www.Example.com/pfad?x=1"]),
+            ["example.com"],
+        )
+
+    def test_wildcard(self):
+        self.assertEqual(engine.normalize_sites("*.example.com"),
+                         ["example.com"])
+
+    def test_dedup_and_multiple(self):
+        self.assertEqual(
+            engine.normalize_sites(["example.com", "EXAMPLE.com", "de.example.org"]),
+            ["example.com", "de.example.org"],
+        )
+
+    def test_empty(self):
+        self.assertEqual(engine.normalize_sites(None), [])
+        self.assertEqual(engine.normalize_sites(["", "  "]), [])
+
+    def test_invalid_raises(self):
+        with self.assertRaises(engine.QueryError):
+            engine.normalize_sites(["not a domain!"])
+
+
+class NormalizeTldsTest(unittest.TestCase):
+
+    def test_plain_gets_leading_dot(self):
+        # Der Crawler speichert die TLD mit führendem Punkt ('.de').
+        self.assertEqual(engine.normalize_tlds(["de"]), [".de"])
+
+    def test_dot_and_case(self):
+        self.assertEqual(engine.normalize_tlds([".DE", "at"]), [".de", ".at"])
+
+    def test_invalid_raises(self):
+        with self.assertRaises(engine.QueryError):
+            engine.normalize_tlds(["de/)"])
+
+    def test_empty(self):
+        self.assertEqual(engine.normalize_tlds(None), [])
+
+
+class NormalizeSortTest(unittest.TestCase):
+
+    def test_defaults(self):
+        self.assertEqual(engine.normalize_sort(None), "relevance")
+        self.assertEqual(engine.normalize_sort("relevance"), "relevance")
+        self.assertEqual(engine.normalize_sort("freshness"), "freshness")
+
+    def test_aliases(self):
+        self.assertEqual(engine.normalize_sort("date"), "freshness")
+        self.assertEqual(engine.normalize_sort("score"), "relevance")
+
+    def test_invalid(self):
+        with self.assertRaises(engine.QueryError):
+            engine.normalize_sort("banana")
+
+
+class ParseTimeBoundTest(unittest.TestCase):
+
+    def test_empty_is_none(self):
+        self.assertIsNone(engine.parse_time_bound(None))
+        self.assertIsNone(engine.parse_time_bound(""))
+
+    def test_relative_days(self):
+        before = engine.parse_time_bound("7d", "after")
+        now = datetime_now()
+        delta = now - before
+        self.assertGreater(delta.total_seconds(), 7 * 86400 - 5)
+        self.assertLess(delta.total_seconds(), 7 * 86400 + 5)
+
+    def test_relative_hours_and_weeks(self):
+        before = engine.parse_time_bound("24h", "after")
+        self.assertAlmostEqual(
+            (datetime_now() - before).total_seconds(), 86400, delta=5
+        )
+        before = engine.parse_time_bound("2w", "after")
+        self.assertAlmostEqual(
+            (datetime_now() - before).total_seconds(), 14 * 86400, delta=5
+        )
+
+    def test_relative_zero_raises(self):
+        with self.assertRaises(engine.QueryError):
+            engine.parse_time_bound("0d")
+
+    def test_iso_date_becomes_utc(self):
+        dt = engine.parse_time_bound("2026-01-31", "after")
+        self.assertEqual(dt.tzinfo, timezone.utc)
+        self.assertEqual((dt.year, dt.month, dt.day), (2026, 1, 31))
+
+    def test_iso_datetime_with_z(self):
+        dt = engine.parse_time_bound("2026-01-31T12:00:00Z", "before")
+        self.assertEqual(dt.utcoffset(), timedelta(0))
+        self.assertEqual(dt.hour, 12)
+
+    def test_naive_datetime_becomes_utc(self):
+        dt = engine.parse_time_bound("2026-01-31 12:00:00", "after")
+        self.assertEqual(dt.tzinfo, timezone.utc)
+
+    def test_invalid_raises(self):
+        with self.assertRaises(engine.QueryError):
+            engine.parse_time_bound("gestern")
+
+
+def datetime_now():
+    return engine.datetime.now(timezone.utc)
+
+
+class PrefixLexemeTest(unittest.TestCase):
+
+    def test_last_word(self):
+        self.assertEqual(engine.prefix_lexeme("hallo welt"), "welt:*")
+
+    def test_lowercased(self):
+        self.assertEqual(engine.prefix_lexeme("SuchMASCHINE"), "suchmaschine:*")
+
+    def test_skips_short_tokens(self):
+        self.assertEqual(engine.prefix_lexeme("a b suchbegriff"),
+                         "suchbegriff:*")
+
+    def test_punctuation_only(self):
+        self.assertIsNone(engine.prefix_lexeme("!!! ..."))
+
+    def test_umlauts(self):
+        self.assertEqual(engine.prefix_lexeme("größen"), "größen:*")
+
+    def test_empty(self):
+        self.assertIsNone(engine.prefix_lexeme(""))
+        self.assertIsNone(engine.prefix_lexeme(None))
+
+
+class FilterClausesTest(unittest.TestCase):
+
+    def test_defaults_filter_errors_and_non_html(self):
+        sql, params = engine._filter_clauses([], [], None, None,
+                                             False, False)
+        self.assertIn("p.http_status = 200", sql)
+        self.assertIn("text/html%", sql)
+        self.assertEqual(params, {})
+
+    def test_include_errors(self):
+        sql, _ = engine._filter_clauses([], [], None, None, True, False)
+        self.assertNotIn("http_status", sql)
+
+    def test_sites_clause(self):
+        sql, params = engine._filter_clauses(
+            engine.normalize_sites(["example.com"]), [], None, None,
+            False, False,
+        )
+        self.assertIn("p.domain = ANY(%(sites)s)", sql)
+        self.assertIn("p.domain LIKE ANY(%(site_pats)s)", sql)
+        self.assertEqual(params["sites"], ["example.com"])
+        self.assertEqual(params["site_pats"], ["%.example.com"])
+
+    def test_time_bounds(self):
+        after = engine.datetime(2026, 1, 1, tzinfo=timezone.utc)
+        before = engine.datetime(2026, 2, 1, tzinfo=timezone.utc)
+        sql, params = engine._filter_clauses(
+            [], [], after, before, False, False,
+        )
+        self.assertIn("COALESCE(p.crawled_at, p.discovered_at) >= %(after)s",
+                      sql)
+        self.assertIn("COALESCE(p.crawled_at, p.discovered_at) < %(before)s",
+                      sql)
+        self.assertEqual(params["after"], after)
+        self.assertEqual(params["before"], before)
+
+
+class SearchSqlTest(unittest.TestCase):
+
+    def test_relevance_ordering(self):
+        sql = engine._search_sql("search_vector", "relevance", "")
+        self.assertIn(
+            "ts_rank_cd(%(weights)s::real[], p.search_vector, q.tsq,",
+            sql,
+        )
+        self.assertIn("%(norm)s::integer", sql)
+        self.assertIn("(r.text_rank * f.fresh) DESC", sql)
+        self.assertIn("websearch_to_tsquery(%(ts_config)s, %(query)s)", sql)
+        self.assertIn("LIMIT %(limit)s OFFSET %(offset)s", sql)
+        # content darf im Ranking-Query nicht gelesen werden (TOAST-Schonung)
+        self.assertIsNone(re.search(r"\bp\.content\b", sql),
+                          "content-Spalte im Ranking-Query gelesen")
+
+    def test_capped_fallback_sql(self):
+        sql = engine._search_sql("search_vector", "relevance", "",
+                                 capped=True)
+        self.assertIn("LIMIT %(rank_cap)s", sql)
+        self.assertIn("p.search_vector AS vector", sql)
+        self.assertIn("ts_rank_cd(%(weights)s::real[], p.vector,", sql)
+        self.assertIn("LIMIT %(limit)s OFFSET %(offset)s", sql)
+        # auch der gekappte Query darf kein content lesen
+        self.assertIsNone(re.search(r"\bp\.content\b", sql))
+
+    def test_freshness_ordering(self):
+        sql = engine._search_sql("search_vector", "freshness", " AND x")
+        self.assertIn(
+            "COALESCE(p.crawled_at, p.discovered_at) DESC NULLS LAST", sql
+        )
+        self.assertIn(" AND x", sql)
+
+    def test_vector_column_is_interpolated(self):
+        sql = engine._search_sql("my_vector", "relevance", "")
+        self.assertIn("p.my_vector", sql)
+        self.assertNotIn("p.search_vector", sql)
+
+
+class CountSqlTest(unittest.TestCase):
+
+    def test_count_uses_filters(self):
+        sql = engine._count_sql("search_vector", " AND p.http_status = 200")
+        self.assertIn("count(*) AS n", sql)
+        self.assertIn("p.search_vector @@ q.tsq AND p.http_status = 200", sql)
+
+
+class SetStatementTimeoutTest(unittest.TestCase):
+
+    def test_casts_to_int(self):
+        cur = mock.Mock()
+        engine._set_statement_timeout(cur, "4000")
+        cur.execute.assert_called_once_with("SET statement_timeout = 4000")
+
+
+class SearchValidationTest(unittest.TestCase):
+
+    def test_empty_query_raises(self):
+        with self.assertRaises(engine.QueryError):
+            engine.search("   ")
+
+    def test_bad_page_raises(self):
+        with self.assertRaises(engine.QueryError):
+            engine.search("test", page="x")
+
+    def test_per_page_capped(self):
+        with mock.patch.object(engine, "db_connect") as db:
+            conn = mock.MagicMock()
+            db.return_value = conn
+            # 1. fetchone: geparste tsquery, 2. fetchone: Trefferzahl
+            conn.cursor.return_value.__enter__.return_value\
+                .fetchone.side_effect = [{"tsq": "test"}, {"n": 0}]
+            conn.cursor.return_value.__enter__.return_value\
+                .fetchall.return_value = []
+            with mock.patch.object(engine, "resolve_vector_column",
+                                   return_value="search_vector"):
+                data = engine.search("test", per_page=10000)
+            self.assertEqual(data["per_page"], engine.MAX_PER_PAGE)
+
+    def test_query_error_for_bad_filters(self):
+        with mock.patch.object(engine, "db_connect") as db:
+            db.return_value = mock.MagicMock()
+            with self.assertRaises(engine.QueryError):
+                engine.search("test", sites=["ungültig!"])
+
+
+if __name__ == "__main__":
+    unittest.main(verbosity=2)

test_crawler.py

@@ -0,0 +1,455 @@
+#!/usr/bin/env python3
+"""
+Local test suite for the crawler (no database required).
+
+Only the pure, DB-free parts are tested: the HostScheduler (per-host
+delays, block backoff, Retry-After), the transient-error classification
+and the queue-claim decision logic (blocked hosts are skipped, another
+domain is picked) against an in-memory queue substitute.
+"""
+
+import os
+import sys
+import time
+import unittest
+from email import utils as email_utils
+from unittest import mock
+
+sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
+
+
+def _load_env_file():
+    """
+    Load .env like the systemd EnvironmentFile does, so the tests also
+    run outside systemd. Must happen before crawler.py is imported,
+    which reads the JONASSEARCH_* variables at import time.
+    """
+    env_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), ".env")
+
+    try:
+        with open(env_path) as handle:
+            for line in handle:
+                line = line.strip()
+                if not line or line.startswith("#") or "=" not in line:
+                    continue
+                key, _, value = line.partition("=")
+                os.environ.setdefault(key.strip(), value.strip())
+    except OSError:
+        pass
+
+
+_load_env_file()
+
+import crawler  # noqa: E402
+
+
+class StubRobots:
+    """robots.txt stub: constant delay, no network."""
+
+    def __init__(self, delay=3.0):
+        self.delay_value = delay
+
+    def delay(self, url):
+        return self.delay_value
+
+
+class FakeState:
+    """In-memory crawl_queue substitute for claim/defer/requeue."""
+
+    def __init__(self, rows=None):
+        self.rows = rows if rows is not None else []
+
+    def add(self, **kwargs):
+        row = {
+            "id": len(self.rows) + 1,
+            "url": "",
+            "source": None,
+            "status": "pending",
+            "priority": 0,
+            "attempts": 0,
+        }
+        row.update(kwargs)
+        self.rows.append(row)
+        return row["id"]
+
+
+class FakeCursor:
+    """Just enough SQL surface for the crawler's queue functions."""
+
+    def __init__(self, state):
+        self._state = state
+        self._rows = []
+
+    def execute(self, query, params=None):
+        query = " ".join(query.split())
+        params = params or ()
+
+        if "SELECT id, url, source_url" in query:
+            batch = params[-1]
+            exclude = set(params[0]) if len(params) == 2 else set()
+            candidates = [
+                row for row in self._state.rows
+                if row["status"] == "pending" and row["id"] not in exclude
+            ]
+            candidates.sort(key=lambda row: (-row["priority"], row["id"]))
+            self._rows = [
+                (row["id"], row["url"], row["source"])
+                for row in candidates[:batch]
+            ]
+        elif "SET status = 'processing'" in query:
+            (queue_id,) = params
+            self._rows = []
+            for row in self._state.rows:
+                if row["id"] == queue_id and row["status"] == "pending":
+                    row["status"] = "processing"
+                    row["attempts"] += 1
+                    self._rows = [(row["id"], row["url"], row["source"])]
+                    break
+        elif "GREATEST(attempts - 1, 0)" in query:
+            (queue_id,) = params
+            for row in self._state.rows:
+                if row["id"] == queue_id:
+                    row["status"] = "pending"
+                    row["attempts"] = max(row["attempts"] - 1, 0)
+        elif "WHEN attempts >=" in query:
+            max_attempts, queue_id = params
+            self._rows = []
+            for row in self._state.rows:
+                if row["id"] == queue_id:
+                    row["status"] = (
+                        "failed" if row["attempts"] >= max_attempts
+                        else "pending"
+                    )
+                    self._rows = [(row["status"],)]
+        else:
+            raise AssertionError("unexpected query: %s" % query[:120])
+
+    def fetchone(self):
+        return self._rows[0] if self._rows else None
+
+    def fetchall(self):
+        return self._rows
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, *_exc):
+        return False
+
+
+class FakeConnection:
+    def __init__(self, state):
+        self.state = state
+
+    def cursor(self):
+        return FakeCursor(self.state)
+
+    def commit(self):
+        pass
+
+    def rollback(self):
+        pass
+
+
+class HostSchedulerTest(unittest.TestCase):
+
+    def test_first_request_is_ready(self):
+        scheduler = crawler.HostScheduler()
+        self.assertEqual(
+            scheduler.wait_time("https://a.example/x", 3.0), 0.0
+        )
+
+    def test_delay_is_honored_after_request(self):
+        scheduler = crawler.HostScheduler()
+        scheduler.wait("https://a.example/x", 0.0)
+        remaining = scheduler.wait_time("https://a.example/y", 3.0)
+        self.assertGreater(remaining, 2.5)
+        self.assertLessEqual(remaining, 3.0)
+
+    def test_short_wait_is_slept_inline(self):
+        scheduler = crawler.HostScheduler()
+        scheduler.wait("https://a.example/x", 0.0)
+        with mock.patch("time.sleep") as sleep:
+            self.assertTrue(scheduler.wait("https://a.example/y", 3.0))
+        sleep.assert_called_once()
+
+    def test_long_wait_is_deferred_without_touching_schedule(self):
+        scheduler = crawler.HostScheduler()
+        scheduler.wait("https://a.example/x", 0.0)
+        recorded = scheduler.last_request["a.example"]
+        with mock.patch("time.sleep") as sleep:
+            self.assertFalse(scheduler.wait("https://a.example/y", 60.0))
+        sleep.assert_not_called()
+        self.assertEqual(scheduler.last_request["a.example"], recorded)
+
+    def test_block_backoff_doubles(self):
+        scheduler = crawler.HostScheduler()
+        url = "https://a.example/x"
+        scheduler.note_http_error(url, 429)
+        first = scheduler.wait_time("https://a.example/y", 3.0)
+        scheduler.note_http_error(url, 429)
+        second = scheduler.wait_time("https://a.example/y", 3.0)
+        self.assertGreater(first, crawler.HOST_BLOCK_RULES[429][0] * 0.9)
+        self.assertGreater(second, first * 1.5)
+
+    def test_retry_after_is_honored(self):
+        scheduler = crawler.HostScheduler()
+        scheduler.note_http_error(
+            "https://a.example/x", 429, retry_after=1200
+        )
+        remaining = scheduler.wait_time("https://a.example/y", 3.0)
+        self.assertGreater(remaining, 1000)
+        self.assertLessEqual(remaining, crawler.HOST_BLOCK_HARD_CAP)
+
+    def test_block_never_exceeds_hard_cap(self):
+        scheduler = crawler.HostScheduler()
+        url = "https://a.example/x"
+        for _ in range(40):
+            scheduler.note_http_error(url, 429)
+        remaining = scheduler.wait_time(url, 3.0)
+        self.assertLessEqual(remaining, crawler.HOST_BLOCK_HARD_CAP + 1)
+
+    def test_success_resets_block_counter(self):
+        scheduler = crawler.HostScheduler()
+        url = "https://a.example/x"
+        scheduler.note_http_error(url, 429)
+        scheduler.note_http_error(url, 429)
+        scheduler.note_success(url)
+        scheduler.note_http_error(url, 429)
+        remaining = scheduler.wait_time("https://a.example/y", 3.0)
+        self.assertLessEqual(
+            remaining, crawler.HOST_BLOCK_RULES[429][0] + 1
+        )
+
+    def test_non_block_status_is_ignored(self):
+        scheduler = crawler.HostScheduler()
+        scheduler.note_http_error("https://a.example/x", 404)
+        self.assertEqual(
+            scheduler.wait_time("https://a.example/y", 3.0), 0.0
+        )
+
+    def test_network_error_backs_off(self):
+        scheduler = crawler.HostScheduler()
+        scheduler.note_network_error("https://a.example/x")
+        remaining = scheduler.wait_time("https://a.example/y", 3.0)
+        self.assertGreater(remaining, 20)
+
+    def test_unrelated_host_is_not_blocked(self):
+        scheduler = crawler.HostScheduler()
+        scheduler.note_http_error("https://a.example/x", 429)
+        self.assertEqual(
+            scheduler.wait_time("https://b.example/y", 3.0), 0.0
+        )
+
+
+class ErrorClassificationTest(unittest.TestCase):
+
+    def test_network_errors_are_transient(self):
+        self.assertTrue(crawler.is_transient_error("network:timeout"))
+        self.assertTrue(crawler.is_transient_error("network:[Errno -2] x"))
+
+    def test_rate_limit_and_server_errors_are_transient(self):
+        for status in (408, 425, 429, 500, 502, 503, 504):
+            self.assertTrue(crawler.is_transient_error("http_%d" % status))
+
+    def test_terminal_errors_are_not_transient(self):
+        for error in (
+            "http_404", "http_410", "robots", "too_large",
+            "not_html", "error:boom", "host_busy", None, "",
+        ):
+            self.assertFalse(crawler.is_transient_error(error))
+
+    def test_retry_after_seconds_digit(self):
+        self.assertEqual(
+            crawler._retry_after_seconds({"Retry-After": "120"}), 120.0
+        )
+
+    def test_retry_after_http_date(self):
+        headers = {
+            "Retry-After": email_utils.formatdate(
+                time.time() + 120, usegmt=True
+            )
+        }
+        value = crawler._retry_after_seconds(headers)
+        self.assertGreater(value, 100)
+        self.assertLess(value, 140)
+
+    def test_retry_after_missing_or_garbage(self):
+        self.assertIsNone(crawler._retry_after_seconds({}))
+        self.assertIsNone(
+            crawler._retry_after_seconds({"Retry-After": "soon"})
+        )
+        self.assertIsNone(crawler._retry_after_seconds(None))
+
+
+class QueueClaimTest(unittest.TestCase):
+
+    def _state(self):
+        state = FakeState()
+        state.add(url="https://blocked.example/a", priority=500)
+        state.add(url="https://other.example/b", priority=100)
+        state.add(url="https://blocked.example/c", priority=400)
+        return state
+
+    def _claim(self, state, scheduler=None):
+        scheduler = scheduler or crawler.HostScheduler()
+        return crawler.claim_next_queue_item(
+            FakeConnection(state), scheduler, StubRobots()
+        )
+
+    def test_highest_priority_ready_host_wins(self):
+        item, wait_hint = self._claim(self._state())
+        self.assertEqual(item[1], "https://blocked.example/a")
+        self.assertIsNone(wait_hint)
+
+    def test_blocked_host_is_skipped_for_other_domain(self):
+        scheduler = crawler.HostScheduler()
+        scheduler.blocked_until["blocked.example"] = time.monotonic() + 900
+        item, wait_hint = self._claim(self._state(), scheduler)
+        self.assertEqual(item[1], "https://other.example/b")
+        self.assertGreater(wait_hint, 800)
+
+    def test_rate_limited_host_is_skipped(self):
+        scheduler = crawler.HostScheduler()
+        scheduler.wait("https://blocked.example/a", 0.0)
+        item, _wait_hint = self._claim(self._state(), scheduler)
+        self.assertEqual(item[1], "https://other.example/b")
+
+    def test_only_blocked_hosts_left_reports_wait_hint(self):
+        state = FakeState()
+        state.add(url="https://blocked.example/a", priority=500)
+        scheduler = crawler.HostScheduler()
+        scheduler.blocked_until["blocked.example"] = time.monotonic() + 900
+        item, wait_hint = self._claim(state, scheduler)
+        self.assertIsNone(item)
+        self.assertGreater(wait_hint, 800)
+
+    def test_empty_queue(self):
+        item, wait_hint = self._claim(FakeState())
+        self.assertIsNone(item)
+        self.assertIsNone(wait_hint)
+
+    def test_claim_marks_processing_and_counts_attempt(self):
+        state = self._state()
+        self._claim(state)
+        self.assertEqual(state.rows[0]["status"], "processing")
+        self.assertEqual(state.rows[0]["attempts"], 1)
+        self.assertEqual(state.rows[1]["status"], "pending")
+
+    def test_claim_finds_ready_host_behind_blocked_wall(self):
+        state = FakeState()
+        for index in range(30):
+            state.add(
+                url="https://blocked.example/%d" % index,
+                priority=300 - index,
+            )
+        state.add(url="https://ready.example/x", priority=10)
+        scheduler = crawler.HostScheduler()
+        scheduler.blocked_until["blocked.example"] = time.monotonic() + 900
+        item, _wait_hint = self._claim(state, scheduler)
+        self.assertEqual(item[1], "https://ready.example/x")
+
+    def test_defer_does_not_burn_an_attempt(self):
+        state = FakeState()
+        queue_id = state.add(
+            url="https://slow.example/a",
+            status="processing",
+            attempts=1,
+        )
+        crawler.defer_queue_item(FakeConnection(state), queue_id)
+        row = state.rows[0]
+        self.assertEqual(row["status"], "pending")
+        self.assertEqual(row["attempts"], 0)
+
+    def test_requeue_retries_below_attempt_cap(self):
+        state = FakeState()
+        queue_id = state.add(
+            url="https://x.example/a",
+            status="processing",
+            attempts=1,
+        )
+        status = crawler.requeue_queue_item(FakeConnection(state), queue_id)
+        self.assertEqual(status, "pending")
+        self.assertEqual(state.rows[0]["status"], "pending")
+
+    def test_requeue_fails_after_attempt_cap(self):
+        state = FakeState()
+        queue_id = state.add(
+            url="https://x.example/a",
+            status="processing",
+            attempts=crawler.QUEUE_MAX_ATTEMPTS,
+        )
+        status = crawler.requeue_queue_item(FakeConnection(state), queue_id)
+        self.assertEqual(status, "failed")
+        self.assertEqual(state.rows[0]["status"], "failed")
+
+
+class HostFilterTest(unittest.TestCase):
+
+    def test_localhost_and_private_suffixes_are_blocked(self):
+        for host in (
+            "localhost", "sub.localhost", "localhost.localdomain",
+            "myserver.local", "app.internal",
+        ):
+            self.assertTrue(
+                crawler.is_non_crawlable_host(host), host
+            )
+
+    def test_ip_literals_are_blocked(self):
+        for host in (
+            "127.0.0.1", "10.0.0.5", "192.168.1.10", "172.16.0.1",
+            "8.8.8.8", "::1", "fe80::1", "2001:db8::1",
+            "::ffff:127.0.0.1",
+        ):
+            self.assertTrue(
+                crawler.is_non_crawlable_host(host), host
+            )
+
+    def test_single_label_and_empty_names_are_blocked(self):
+        for host in ("mymachine", "", "   ", "."):
+            self.assertTrue(
+                crawler.is_non_crawlable_host(host), repr(host)
+            )
+
+    def test_real_domains_are_allowed(self):
+        for host in (
+            "example.com", "www.example.com",
+            "sub.domain.example.co.uk", "xn--bcher-kva.example",
+            "example.com.",
+        ):
+            self.assertFalse(
+                crawler.is_non_crawlable_host(host), host
+            )
+
+    def test_normalize_url_rejects_non_public_targets(self):
+        for url in (
+            "http://localhost/x",
+            "http://127.0.0.1/x",
+            "http://[::1]:8080/x",
+            "http://10.1.2.3/a?b=c",
+            "http://intranet.local/x",
+            "http://host/x",
+        ):
+            self.assertIsNone(crawler.normalize_url(url), url)
+
+    def test_normalize_url_keeps_real_domains(self):
+        self.assertEqual(
+            crawler.normalize_url("http://www.example.com/x?y=1"),
+            "http://www.example.com/x?y=1",
+        )
+
+    def test_is_non_crawlable_url_handles_malformed_urls(self):
+        self.assertTrue(crawler.is_non_crawlable_url("http://localhost/x"))
+        self.assertTrue(crawler.is_non_crawlable_url("http://[::1]/"))
+        self.assertFalse(crawler.is_non_crawlable_url("https://example.com/"))
+
+    def test_malformed_url_is_treated_as_non_crawlable(self):
+        # urlparse raises ValueError on malformed IPv6 brackets; the
+        # filter must not crash on it.
+        self.assertTrue(
+            crawler.is_non_crawlable_url("http://[::1")
+        )
+
+
+if __name__ == "__main__":
+    unittest.main()