add stuff

.env.example

@@ -16,6 +16,26 @@ 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
+# Bekannte blockierte/rate-limitierte Hosts werden direkt in der
+# Claim-Query ausgeschlossen (verhindert, dass ein blocked Host mit
+# vielen URLs die restliche Queue verdeckt). Obergrenze für diese Liste;
+# die am längsten blockierten Hosts werden zuerst berücksichtigt.
+#JONASSEARCH_CLAIM_BUSY_HOST_LIMIT=1000
+# 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 +52,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=5
 
 # Wie oft jeder Crawler außerhalb der Transaktionen nach einer neuen
 # Deployment-Generation sieht (Sekunden). Der Check setzt nur ein
@@ -55,16 +75,51 @@ JONASSEARCH_UPDATE_CHECK_SECONDS=30
 #JONASSEARCH_LAUNCHER_RESTART_DELAY=2
 
 # ============================================================
-# Suchmaschine (search/)
+# Suchmaschine (search/) – Port 8083
 # ============================================================
 
-# PostgreSQL-Textsuch-Konfiguration (Stemming/Stopwörter)
+# PostgreSQL-Textsuch-Konfiguration. MUSS zur Konfiguration passen,
-JONASSEARCH_TS_CONFIG=german
+# 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 Websuche
+#JONASSEARCH_RESULTS_PER_PAGE=10
+#JONASSEARCH_MAX_RESULTS_PER_PAGE=50
+
+# Bildersuche (Tab "Bilder", /api/images)
+# Bildkacheln pro Ergebnisseite bzw. Maximum pro Seite
+#JONASSEARCH_IMAGE_RESULTS_PER_PAGE=40
+#JONASSEARCH_MAX_IMAGE_RESULTS_PER_PAGE=100
+# Höchstens so viele Bilder pro Quellseite anzeigen (verhindert, dass
+# Seiten mit vielen Bildern die Ergebnisse dominieren)
+#JONASSEARCH_IMAGES_PER_SOURCE_PAGE=4

README.md

@@ -4,11 +4,54 @@ 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)
+- `search/` – Suchmaschine darüber (CLI + HTTP-API + Weboberfläche, Port 8083),
-- `web/` – Statistik-Webseite
+  siehe [search/README.md](search/README.md)
+- `web/` – Statistik-Webseite (Port 8081, `jonassearch-stats.service`).
+  Seiten werden alle 30 Min. gezählt; Bilder alle 2 h exakt, dazwischen nur
+  geschätzt. Die Image-Diagramme zeigen standardmäßig nur exakte Messpunkte;
+  ein umschaltbarer, im Browser gespeicherter Schalter kann Schätzwerte in allen
+  Image-Diagrammen, Records und der Heatmap einblenden.
+  Mit jeder Messung wird zusätzlich die Anzahl der `pending`-Einträge in
+  `crawl_queue` erfasst (exakter `COUNT`, keine Schätzung) und mit dem
+  kompletten Diagrammsatz (Verlauf, Change/5 min, kumulativ, pro Stunde,
+  Tagesstunden, Records, Heatmap) dargestellt.
+  Die Hauptseite hat drei Tabs mit Symbolen (Pages, Images, Queue): Jeder Tab
+  zeigt nur die Kennzahlen und Diagramme seines Datentyps, die Auswahl wird im
+  Browser gespeichert. CSV/JSON-Export folgt dem aktiven Tab.
+  Über das Feld „Reach target“ lässt sich eine beliebige Zielzahl eingeben;
+  die Seite schätzt aus dem Trend der letzten zwei Stunden (des gewählten
+  Zeitraums), wann dieser Stand erreicht ist – inklusive Datum und Uhrzeit.
+  Das Ziel wird pro Tab im Browser gespeichert.
+- `web/tui.py` – Terminal-Dashboard für dieselben Statistiken (siehe unten)
+- `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`)
+- `add_pages_pictures.sql` – einmalig laufen lassen: legt `pages.pictures`
+  (`text[]`) für die Bild-URLs der gecrawlten Seiten an (idempotent)
+
+## Terminal-Dashboard (`web/tui.py`)
+
+Zeigt Seiten- und Bildanzahl, Wachstum, einen Sparkline-Verlauf und die
+letzten Messwerte direkt im Terminal. Liest dieselbe
+`web/page_counts.json` wie der Statserver – kein Webserver, keine
+zusätzlichen Pakete (nur `curses` aus der Standardlib).
+
+```bash
+python3 web/tui.py                 # interaktiv, Aktualisierung alle 2 s
+python3 web/tui.py --window 7d     # 24h (Default) | 7d | 30d | all
+python3 web/tui.py --once          # Momentaufnahme, z. B. für Logs/Skripte
+python3 web/tui.py --once --width 100 --window all
+```
+
+Tasten im interaktiven Modus: `q` beendet, `r` lädt die Datei sofort neu,
+`t` wechselt das Zeitfenster. `--ascii` ersetzt die Unicode-Blöcke der
+Sparkline. Läuft die Ausgabe nicht in einem Terminal (Pipe, Cron), wird
+automatisch die `--once`-Ausgabe verwendet. Fehlt die Datei, zeigt das
+Dashboard „No samples yet“; ist sie defekt, erscheint die Fehlermeldung
+im Dashboard (bei `--once` auf stderr, Exit-Code 1).
 
 ## Sehr große Seiten / tsvector-Limit
 
@@ -38,6 +81,83 @@ 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, 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).
+
+   **Wichtig bei großen Queues:** Die bekannten blockierten Hosts werden
+   direkt in der Claim-Query ausgeschlossen (nicht nur innerhalb des kleinen
+   Kandidatenfensters von `JONASSEARCH_CLAIM_SCAN_LIMIT`). Blockierte Zeilen
+   bleiben sonst als `pending` oben in der Prioritätsreihenfolge stehen: Ein
+   Host mit Tausenden wartenden URLs würde dann bei jedem Rescan dasselbe
+   Fenster füllen und alle anderen Hosts der Queue unsichtbar machen.
+   Höchstens `JONASSEARCH_CLAIM_BUSY_HOST_LIMIT` Hosts (die mit der längsten
+   Restwartezeit) gehen in die Query; weitere werden weiterhin pro Kandidat
+   übersprungen.
+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.
+
+## Bildersuche (`pages.pictures`)
+
+Der Crawler speichert zu jeder Seite die gefundenen Bild-URLs in
+`pages.pictures`: absolute http(s)-URLs, pro Seite dedupliziert, ohne
+URL-Fragmente. Erfasst werden `<img src>`, gängige Lazy-Loading-Attribute
+(`data-src`, `data-lazy-src`, `data-original`), `srcset` (auch in
+`<picture><source>`), `og:image`/`twitter:image` sowie
+`<link rel="image_src">`. Relative Pfade werden gegen die Seiten-URL
+aufgelöst; `data:`, `javascript:`, `ftp:` und nicht-öffentliche Hosts
+werden verworfen. Die Bilder selbst werden **nicht** heruntergeladen oder
+lokal gespeichert.
+
+Einmalige Migration der Datenbank:
+
+```bash
+psql -h 127.0.0.1 -p 25432 -U jonassearch -d jonassearch \
+    -f add_pages_pictures.sql
+```
+
+Die Migration ist idempotent (`ADD COLUMN IF NOT EXISTS`) und lässt
+bestehende Zeilen unverändert. Bereits gecrawlte Seiten erhalten ihre
+Bilder beim nächsten Crawl; bis dahin behandeln die Abfragen `NULL` und
+`'{}'` identisch als „keine Bilder“. Die Websuche ist davon unberührt.
+
+Die Suchmaschine bietet dafür den Tab **Bilder** (`GET /api/images`).
+Pro Quellseite werden höchstens `JONASSEARCH_IMAGES_PER_SOURCE_PAGE`
+Bilder angezeigt; die Kacheln laden die Bilder direkt von der
+Original-URL (Lazy Loading, defekte Bilder werden ausgeblendet). Details
+stehen in [search/README.md](search/README.md).
+
+## 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`):

add_pages_pictures.sql

@@ -0,0 +1,27 @@
+-- add_pages_pictures.sql
+--
+-- Adds the image URLs found on crawled pages to `pages.pictures`.
+--
+-- Purpose:
+--   The crawler stores every absolute image URL of a page (img src,
+--   srcset, <picture>/<source>, og:image, twitter:image, link
+--   rel="image_src") in `pages.pictures`, so the search UI can offer an
+--   image tab without downloading or storing the images themselves.
+--
+-- Run once on the jonassearch database as its owner, e.g.:
+--   psql -h 127.0.0.1 -p 25432 -U jonassearch -d jonassearch \
+--       -f add_pages_pictures.sql
+--
+-- Safe to run repeatedly (IF NOT EXISTS) and safe while the crawler is
+-- running: the column is added as a nullable text[] and existing rows
+-- are not rewritten. The crawler always writes an array (possibly
+-- empty) from version 5 on; queries treat NULL and '{}' the same.
+--
+-- If the column already exists (e.g. created by an earlier schema), the
+-- statement is a no-op and the stored data is left untouched.
+
+ALTER TABLE pages ADD COLUMN IF NOT EXISTS pictures text[];
+
+-- New rows without an explicit value get an empty array. This only
+-- changes table metadata; it does not rewrite existing rows.
+ALTER TABLE pages ALTER COLUMN pictures SET DEFAULT '{}';

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

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,225 @@
-# JonasSearch – Suchmaschine
+# JonasSearch Suchmaschine (`search/`)
 
-Volltextsuche über den Index, den `crawler.py` in PostgreSQL aufgebaut hat.
+Die eigentliche Websuchmaschine über der PostgreSQL-Datenbank `jonassearch`:
-Die Engine sucht in der Tabelle `pages` und nutzt dabei PostgreSQLs
+Volltextsuche auf dem vorhandenen `tsvector`-Index (`pages.search_vector`,
-Volltextsuche (`tsvector` / `tsquery`).
+GIN-Index `idx_pages_search`) mit mehrstufigem Ranking, Filtern, Snippets,
+Bildersuche (Tab „Bilder“), Weboberfläche, HTTP-API und Kommandozeile.
 
-## Starten
+| Baustein | Zweck | Port |
-
-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 |
 |---|---|---|
-| `JONASSEARCH_DB_HOST` | DB-Host | `127.0.0.1` |
+| `search/api.py` | HTTP-API + Weboberfläche | **8083** |
-| `JONASSEARCH_DB_PORT` | DB-Port | `25432` |
+| `search/cli.py` | Kommandozeile (ohne Webserver) | – |
-| `JONASSEARCH_DB_NAME` | DB-Name | `jonassearch` |
+| `search/engine.py` | Suchlogik (von api/cli genutzt) | – |
-| `JONASSEARCH_DB_USER` | DB-User | `jonassearch` |
+| `search/jonassearch-engine.service` | systemd-Dienst | 8083 |
-| `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` |
 
-## Abhängigkeiten
+Bestehende Dienste: Statserver 8081 (`web/app.py`), Vorschlagsverfeinerer
+8082 (`priorityweb/app.py`), Suchmaschine **8083**.
 
-- `psycopg2` (DB-Zugriff)
+## Wie gesucht wird
-- `python-dotenv` (`.env`-Datei)
+
-- `flask` (nur für `search/api.py`)
+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: `⟦…⟧`.
+
+## Bildersuche
+
+Der Tab **Bilder** (`GET /api/images`) nutzt denselben Suchvektor,
+dieselben Filter und dieselbe Relevanz-/Frischeformel wie die Websuche.
+Die Bild-URLs stammen aus `pages.pictures`, das der Crawler füllt
+(absolute http(s)-URLs aus `<img>`, `srcset`, `<picture>`, `og:image`,
+`twitter:image`, `link rel="image_src"`; Migration:
+`add_pages_pictures.sql` im Projektstamm). Kandidatenseiten werden also
+anhand von Titel/Überschriften/Inhalt gefunden, nicht anhand des
+Bildinhalts.
+
+Pro Quellseite werden höchstens `JONASSEARCH_IMAGES_PER_SOURCE_PAGE`
+Bilder geliefert (Default 4), damit Seiten mit vielen Bildern die
+Ergebnisse nicht dominieren. Die Gesamtzahl ist eine Schätzung
+(`SUMME(LEAST(cardinality(pictures), Limit))`; `total: null` bei
+Timeout). Bei einem Ranking-Timeout greift derselbe begrenzte
+Kandidaten-Fallback wie bei der Websuche. Die Bilder werden direkt von
+der Original-URL geladen – kein Proxy, kein Download, keine lokale
+Bildspeicherung.
+
+## 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/images` | Bildersuche – gleiche Filter wie `/api/search`, liefert `image_url`, `page_url`, `title`, `domain` |
+| `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
+}
+```
+
+Parameter von `/api/images` sind `q` (Pflicht), `page`, `per_page`
+(Default 40, max 100), `site`, `tld`, `after`/`before` und `sort`.
+
+Antwort von `/api/images` (Auszug):
+
+```json
+{
+  "query": "katze", "page": 1, "per_page": 40,
+  "total": 128, "has_more": true,
+  "results": [
+    {
+      "image_url": "https://…/katze.jpg",
+      "page_url": "https://…/katzen",
+      "title": "…", "domain": "…", "tld": ".de",
+      "crawled_at": "2026-09-30T18:12:03+00:00",
+      "score": 0.42, "text_rank": 0.40, "freshness": 1.05
+    }
+  ]
+}
+```
+
+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_IMAGE_RESULTS_PER_PAGE` | `40` | Bildkacheln pro Ergebnisseite |
+| `JONASSEARCH_MAX_IMAGE_RESULTS_PER_PAGE` | `100` | Maximum Bildkacheln pro Seite |
+| `JONASSEARCH_IMAGES_PER_SOURCE_PAGE` | `4` | Höchstens so viele Bilder pro Quellseite |
+| `JONASSEARCH_HEADLINE_SOURCE_CHARS` | `100000` | `content`-Anteil für ts_headline |
+
+## Tests
+
+```bash
+python3 search/test_engine.py
+python3 search/test_api.py
+```
+
+Getestet werden die DB-freien Teile: Parameter-Normalisierung, Gewichte,
+Zeitfilter, SQL-Zusammensetzung (Web- und Bildersuche), Query-Parsing
+sowie Routing/Fehler-Mapping des `/api/images`-Endpunkts und die
+Einbindung des Bilder-Tabs. Alles ohne Datenbankzugriff.

search/__init__.py

@@ -1,22 +1,10 @@
 """
-JonasSearch – Suchmaschine über den PostgreSQL-Index des Crawlers.
+JonasSearch Suchmaschine (Package `search`).
 
-Public API:
+Module:
-    from search import SearchEngine, SearchHit, SearchResponse
+    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"

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:
+Beispiele:
-
+    python3 search/cli.py "postgresql volltextsuche"
-    python3 search/cli.py "raspberry pi"
+    python3 search/cli.py "crawler" --site example.com --sort freshness
-    python3 search/cli.py "postgres" --domain example.com --page 2
+    python3 search/cli.py "python" --after 7d --limit 20
-    python3 search/cli.py "amplifier" --json
+    python3 search/cli.py --suggest "suchmasch"
-
+    python3 search/cli.py --stats
-Konfiguration über Umgebungsvariablen (siehe .env.example):
+    python3 search/cli.py --page-id 42
-    JONASSEARCH_DB_HOST / PORT / NAME / USER / PASSWORD
+    python3 search/cli.py --url https://example.com/seite
-    JONASSEARCH_TS_CONFIG
+    python3 search/cli.py "test" --json
 """
 
 import argparse
-import html
 import json
-import re
 import sys
+import textwrap
+from pathlib import Path
 
+from dotenv import load_dotenv
+
+# .env laden, bevor engine importiert wird (wie api.py).
+for _dotenv_path in (None, Path(__file__).resolve().parent.parent / ".env"):
     try:
-    from .engine import SearchEngine
+        load_dotenv(_dotenv_path)
-except ImportError:  # python3 search/cli.py (ohne -m)
+    except OSError:
-    from engine import SearchEngine
+        pass
 
-_TAG_RE = re.compile(r"<[^>]+>")
+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."""
+# Ausgabe-Helfer
-    text = _TAG_RE.sub("", text or "")
+# ------------------------------------------------------------
-    return html.unescape(text)
+
+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:
+def _highlight(snippet, style):
-    lines = []
+    """⟦…⟧-Markierungen -> ANSI-Hervorhebung (oder Plaintext)."""
-    total = response.total
+    return (snippet
-    unit = "Ergebnis" if total == 1 else "Ergebnisse"
+            .replace(engine.SNIPPET_START,
-
+                     style.bold + style.green)
-    header = f"{total} {unit} für „{response.query}“ ({response.took_ms} ms)"
+            .replace(engine.SNIPPET_STOP, style.reset))
-    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 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",
+        prog="search/cli.py",
-        description="JonasSearch – Suchmaschine (Kommandozeile).",
+        description="JonasSearch Volltextsuche über der PostgreSQL-"
-        epilog=(
+                    "Datenbank (pages.search_vector).",
-            "Konfiguration über JONASSEARCH_* Umgebungsvariablen, "
+        epilog='Query-Syntax: "exakte Phrase", OR, -ausschluss',
-            "siehe .env.example."
-        ),
     )
-
+    parser.add_argument("query", nargs="?",
-    parser.add_argument(
+                        help="Suchbegriffe (Websearch-Syntax)")
-        "query",
+    parser.add_argument("--site", action="append", default=[],
-        help="Suchabfrage, z. B. \"raspberry pi\"",
+                        help="Auf Domain einschränken (mehrfach möglich)")
-    )
+    parser.add_argument("--tld", action="append", default=[],
-    parser.add_argument(
+                        help="Auf Top-Level-Domain einschränken (z. B. de)")
-        "--page",
+    parser.add_argument("--after", help="frühestes Datum: 2026-01-31 oder "
-        type=int,
+                                        "24h/7d/2w relativ")
-        default=1,
+    parser.add_argument("--before", help="spätestes Datum (exklusiv) oder "
-        help="Seite der Ergebnisse (Standard: 1)",
+                                         "24h/7d/2w relativ")
-    )
+    parser.add_argument("--sort", default="relevance",
-    parser.add_argument(
+                        choices=["relevance", "freshness"],
-        "--per-page",
+                        help="Relevanz (Default) oder Aktualität")
-        type=int,
+    parser.add_argument("--page", type=int, default=1)
-        default=10,
+    parser.add_argument("--limit", type=int, default=None,
-        help="Ergebnisse pro Seite, max. 50 (Standard: 10)",
+                        help="Treffer pro Seite (Default 10, max 50)")
-    )
+    parser.add_argument("--no-snippets", action="store_true",
-    parser.add_argument(
+                        help="keine ts_headline-Snippets laden")
-        "--domain",
+    parser.add_argument("--include-errors", action="store_true",
-        default=None,
+                        help="auch Seiten ohne HTTP 200 liefern")
-        help="Nur Seiten dieser exakten Domain (z. B. example.com)",
+    parser.add_argument("--include-non-html", action="store_true",
-    )
+                        help="auch Nicht-HTML-Inhalte liefern")
-    parser.add_argument(
+    parser.add_argument("--json", action="store_true",
-        "--tld",
+                        help="rohe JSON-Ausgabe")
-        default=None,
+    parser.add_argument("--no-color", action="store_true")
-        help="Nur Seiten dieser TLD (z. B. .de oder .com)",
+    parser.add_argument("--suggest", metavar="BEGRIFF",
-    )
+                        help="Autocomplete-Vorschläge anzeigen")
-    parser.add_argument(
+    parser.add_argument("--stats", action="store_true",
-        "--config",
+                        help="Index-Statistik anzeigen")
-        default=None,
+    parser.add_argument("--page-id", type=int,
-        help="PostgreSQL-Textsuch-Konfiguration "
+                        help="gespeicherte Seite per ID anzeigen")
-             "(Standard: german, oder JONASSEARCH_TS_CONFIG)",
+    parser.add_argument("--url", help="gespeicherte Seite per URL anzeigen")
-    )
-    parser.add_argument(
-        "--json",
-        action="store_true",
-        help="Ergebnisse als JSON ausgeben",
-    )
-
     return parser
 
 
-def main(argv=None) -> int:
+def main(argv=None):
     args = build_parser().parse_args(argv)
-
+    style = Style(sys.stdout.isatty() and not args.no_color)
-    engine = SearchEngine(ts_config=args.config)
 
     try:
-        response = engine.search(
+        if args.stats:
-            query=args.query,
+            print_stats(engine.stats(), style)
-            page=args.page,
-            per_page=args.per_page,
-            domain=args.domain,
-            tld=args.tld,
-        )
-    except Exception as exc:  # z. B. DB nicht erreichbar, Tabelle fehlt
-        print(f"Fehler bei der Suche: {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 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.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,
+        )
+        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 __name__ == "__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_api.py

@@ -0,0 +1,102 @@
+#!/usr/bin/env python3
+"""
+Tests für die HTTP-Schnittstelle der Bildersuche (ohne Datenbank).
+
+Die Engine-Aufrufe werden gemockt; geprüft werden Routing,
+Parametervalidierung, Fehler-Mapping und die Einbindung des Tabs
+"Bilder" in die Weboberfläche.
+"""
+
+import os
+import sys
+import unittest
+from unittest import mock
+
+ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+sys.path.insert(0, ROOT)
+
+import search.api as api  # noqa: E402
+
+
+class ImageApiTest(unittest.TestCase):
+
+    def setUp(self):
+        api.app.config.update(TESTING=True)
+        self.client = api.app.test_client()
+
+    def _payload(self):
+        return {
+            "query": "katze",
+            "parsed_query": "katze",
+            "page": 1,
+            "per_page": api.engine.IMAGE_RESULTS_PER_PAGE,
+            "sort": "relevance",
+            "ranking_capped": False,
+            "total": 1,
+            "has_more": False,
+            "results": [{
+                "image_url": "https://example.com/katze.jpg",
+                "page_url": "https://example.com/katzen",
+                "title": "Katzen",
+                "domain": "example.com",
+                "tld": ".com",
+                "crawled_at": None,
+                "score": 1.0,
+                "text_rank": 1.0,
+                "freshness": 1.0,
+            }],
+            "took_ms": 1.2,
+        }
+
+    def test_missing_query_returns_400(self):
+        resp = self.client.get("/api/images")
+        self.assertEqual(resp.status_code, 400)
+        self.assertIn("q", resp.get_json()["error"])
+
+    def test_parameters_are_forwarded(self):
+        with mock.patch.object(
+            api.engine, "search_images", return_value=self._payload()
+        ) as mocked:
+            resp = self.client.get(
+                "/api/images?q=katze&page=2&per_page=10"
+                "&site=example.com&tld=de&sort=freshness"
+            )
+        self.assertEqual(resp.status_code, 200)
+        self.assertEqual(resp.get_json()["results"][0]["image_url"],
+                         "https://example.com/katze.jpg")
+        kwargs = mocked.call_args.kwargs
+        self.assertEqual(kwargs["query"], "katze")
+        self.assertEqual(kwargs["page"], 2)
+        self.assertEqual(kwargs["per_page"], 10)
+        self.assertEqual(kwargs["sites"], ["example.com"])
+        self.assertEqual(kwargs["tlds"], ["de"])
+        self.assertEqual(kwargs["sort"], "freshness")
+
+    def test_query_error_returns_400(self):
+        with mock.patch.object(
+            api.engine, "search_images",
+            side_effect=api.engine.QueryError("Ungültig."),
+        ):
+            resp = self.client.get("/api/images?q=x")
+        self.assertEqual(resp.status_code, 400)
+        self.assertEqual(resp.get_json()["error"], "Ungültig.")
+
+    def test_timeout_returns_504(self):
+        with mock.patch.object(
+            api.engine, "search_images",
+            side_effect=api.engine.SearchTimeout("Zu langsam."),
+        ):
+            resp = self.client.get("/api/images?q=x")
+        self.assertEqual(resp.status_code, 504)
+
+    def test_index_contains_images_tab(self):
+        resp = self.client.get("/")
+        self.assertEqual(resp.status_code, 200)
+        html = resp.get_data(as_text=True)
+        self.assertIn('id="tab-images"', html)
+        self.assertIn("/api/images", html)
+        self.assertIn('class="img-grid"', html)
+
+
+if __name__ == "__main__":
+    unittest.main(verbosity=2)

search/test_engine.py

@@ -0,0 +1,418 @@
+#!/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 ImageSearchSqlTest(unittest.TestCase):
+
+    def test_image_sql_uses_pictures_and_per_page_limit(self):
+        sql = engine._image_search_sql("search_vector", "relevance", "")
+        self.assertIn("unnest(p.pictures)", sql)
+        self.assertIn("LIMIT %(images_per_source)s", sql)
+        self.assertIn("i.image_url", sql)
+        self.assertIn("p.url AS page_url", sql)
+        self.assertIn("(r.text_rank * f.fresh) DESC", sql)
+        self.assertIn("LIMIT %(limit)s OFFSET %(offset)s", sql)
+        # Ranking-Query darf content nicht lesen (TOAST-Schonung).
+        self.assertIsNone(re.search(r"\bp\.content\b", sql))
+
+    def test_image_sql_capped_fallback(self):
+        sql = engine._image_search_sql(
+            "search_vector", "relevance", "", capped=True,
+        )
+        self.assertIn("LIMIT %(rank_cap)s", sql)
+        self.assertIn("p.pictures", sql)
+        self.assertIn("LIMIT %(limit)s OFFSET %(offset)s", sql)
+
+    def test_image_sql_freshness_order(self):
+        sql = engine._image_search_sql("search_vector", "freshness", "")
+        self.assertIn(
+            "COALESCE(p.crawled_at, p.discovered_at) DESC NULLS LAST",
+            sql,
+        )
+
+    def test_image_sql_uses_filters(self):
+        sql = engine._image_search_sql(
+            "search_vector", "relevance", " AND p.http_status = 200",
+        )
+        self.assertIn(" AND p.http_status = 200", sql)
+
+    def test_image_count_is_capped_per_source_page(self):
+        sql = engine._image_count_sql("search_vector", "")
+        self.assertIn("cardinality(p.pictures)", sql)
+        self.assertIn("%(images_per_source)s", sql)
+        self.assertIn("sum(", sql.lower())
+
+
+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!"])
+
+
+class ImageSearchValidationTest(unittest.TestCase):
+
+    def test_empty_query_raises(self):
+        with self.assertRaises(engine.QueryError):
+            engine.search_images("   ")
+
+    def test_bad_page_raises(self):
+        with self.assertRaises(engine.QueryError):
+            engine.search_images("test", page="x")
+
+    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_images("test", sites=["ungültig!"])
+
+    def test_per_page_capped(self):
+        with mock.patch.object(engine, "db_connect") as db:
+            conn = mock.MagicMock()
+            db.return_value = conn
+            cur = conn.cursor.return_value.__enter__.return_value
+            cur.fetchone.side_effect = [{"tsq": "test"}, {"n": 0}]
+            cur.fetchall.return_value = []
+            with mock.patch.object(engine, "resolve_vector_column",
+                                   return_value="search_vector"):
+                data = engine.search_images("test", per_page=10000)
+            self.assertEqual(
+                data["per_page"], engine.MAX_IMAGE_RESULTS_PER_PAGE
+            )
+            self.assertFalse(data["has_more"])
+            self.assertEqual(data["results"], [])
+
+
+if __name__ == "__main__":
+    unittest.main(verbosity=2)

test_crawler.py

@@ -0,0 +1,679 @@
+#!/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"]
+
+
+def _url_host(url):
+    """Mirror the SQL host expression used by the claim query."""
+    rest = url.split("://", 1)[1] if "://" in url else url
+    return rest.split("/", 1)[0].lower()
+
+
+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 1 FROM crawl_queue" in query:
+            self._rows = [
+                (1,) for row in self._state.rows
+                if row["status"] == "pending"
+            ][:1]
+        elif "SELECT id, url, source_url" in query:
+            batch = params[-1]
+            exclude = set(params[0]) if len(params) == 3 else set()
+            excluded_hosts = set(params[1]) if len(params) == 3 else set()
+            candidates = [
+                row for row in self._state.rows
+                if row["status"] == "pending"
+                and row["id"] not in exclude
+                and _url_host(row["url"]) not in excluded_hosts
+            ]
+            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
+        )
+
+    def test_busy_hosts_reports_blocked_and_rate_limited_hosts(self):
+        scheduler = crawler.HostScheduler()
+        scheduler.blocked_until["blocked.example"] = time.monotonic() + 900
+        scheduler.wait("https://rate.example/x", 60.0)
+
+        busy = scheduler.busy_hosts()
+        self.assertGreater(busy["blocked.example"], 800)
+        self.assertGreater(busy["rate.example"], 50)
+
+    def test_busy_hosts_prunes_expired_entries(self):
+        scheduler = crawler.HostScheduler()
+        scheduler.blocked_until["a.example"] = time.monotonic() - 1
+        scheduler.last_request["b.example"] = time.monotonic() - 100
+        scheduler.host_delay["b.example"] = 3.0
+
+        self.assertEqual(scheduler.busy_hosts(), {})
+        self.assertNotIn("a.example", scheduler.blocked_until)
+        self.assertNotIn("b.example", scheduler.last_request)
+
+    def test_never_contacted_host_is_ready_even_with_large_delay(self):
+        # monotonic() has an arbitrary origin: "never contacted" must not
+        # be treated as last_request = 0, which would block a fresh host
+        # for the whole robots.txt crawl-delay on a freshly booted host.
+        scheduler = crawler.HostScheduler()
+        self.assertEqual(
+            scheduler.wait_time("https://slow.example/x", 86400.0, now=10.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 PictureExtractionTest(unittest.TestCase):
+
+    BASE = "https://example.com/articles/page.html"
+
+    def _pictures(self, html):
+        return crawler.parse_html(html, self.BASE)["pictures"]
+
+    def test_img_src_relative_is_absolute_and_fragment_dropped(self):
+        self.assertEqual(
+            self._pictures('<img src="/images/a.jpg#frag">'),
+            ["https://example.com/images/a.jpg"],
+        )
+
+    def test_protocol_relative_external_image_and_dedupe(self):
+        pictures = self._pictures(
+            '<img src="//cdn.example.net/pic.png">'
+            '<img src="https://cdn.example.net/pic.png">'
+        )
+        self.assertEqual(
+            pictures, ["https://cdn.example.net/pic.png"]
+        )
+
+    def test_srcset_and_picture_sources(self):
+        pictures = self._pictures(
+            '<img src="/fallback.jpg" '
+            'srcset="/small.jpg 1x, /large.jpg 2x">'
+            '<picture><source srcset="/b.webp 1x, /b2.webp 2x"></picture>'
+        )
+        self.assertEqual(pictures, [
+            "https://example.com/fallback.jpg",
+            "https://example.com/small.jpg",
+            "https://example.com/large.jpg",
+            "https://example.com/b.webp",
+            "https://example.com/b2.webp",
+        ])
+
+    def test_srcset_data_url_does_not_create_bogus_candidates(self):
+        pictures = self._pictures(
+            '<img srcset="data:image/png;base64,AAAA 1x, /real.jpg 2x">'
+        )
+        self.assertEqual(pictures, ["https://example.com/real.jpg"])
+
+    def test_og_twitter_and_image_src(self):
+        pictures = self._pictures(
+            '<meta property="og:image" content="https://example.com/og.png">'
+            '<meta name="twitter:image" content="/tw.png">'
+            '<link rel="image_src" href="/link.png">'
+        )
+        self.assertEqual(pictures, [
+            "https://example.com/og.png",
+            "https://example.com/tw.png",
+            "https://example.com/link.png",
+        ])
+
+    def test_lazy_loading_attributes(self):
+        pictures = self._pictures(
+            '<img data-src="//cdn.example.org/lazy.webp" '
+            'data-lazy-src="/lazy2.jpg" data-original="/orig.jpg">'
+        )
+        self.assertEqual(pictures, [
+            "https://cdn.example.org/lazy.webp",
+            "https://example.com/lazy2.jpg",
+            "https://example.com/orig.jpg",
+        ])
+
+    def test_invalid_and_unsupported_sources_are_ignored(self):
+        pictures = self._pictures(
+            '<img src="data:image/png;base64,AAAA">'
+            '<img src="javascript:alert(1)">'
+            '<img src="ftp://example.com/x.png">'
+            '<img src="">'
+            '<img src="   ">'
+            '<img>'
+        )
+        self.assertEqual(pictures, [])
+
+    def test_non_public_image_hosts_are_ignored(self):
+        pictures = self._pictures(
+            '<img src="http://localhost/a.png">'
+            '<img src="http://127.0.0.1/b.png">'
+        )
+        self.assertEqual(pictures, [])
+
+    def test_page_without_images(self):
+        self.assertEqual(
+            self._pictures("<html><body><p>keine Bilder</p></body></html>"),
+            [],
+        )
+
+    def test_existing_fields_still_parsed(self):
+        parsed = crawler.parse_html(
+            '<title>Seite</title><h1>Überschrift</h1>'
+            '<a href="/x">x</a><img src="/i.png">',
+            self.BASE,
+        )
+        self.assertEqual(parsed["title"], "Seite")
+        self.assertEqual(parsed["headings"], ["Überschrift"])
+        self.assertEqual(parsed["links"], ["https://example.com/x"])
+        self.assertEqual(parsed["pictures"], ["https://example.com/i.png"])
+
+
+class SavePagePicturesTest(unittest.TestCase):
+
+    @staticmethod
+    def _parsed(pictures=None):
+        parsed = {
+            "title": "Titel",
+            "headings": [],
+            "content": "Inhalt",
+            "links": [],
+            "domain": "example.com",
+            "tld": ".com",
+        }
+        if pictures is not None:
+            parsed["pictures"] = pictures
+        return parsed
+
+    def _save(self, parsed):
+        conn = mock.MagicMock()
+        cur = conn.cursor.return_value.__enter__.return_value
+        crawler.save_page(
+            conn, "https://example.com/", None, 200, "text/html",
+            parsed, 12,
+        )
+        sql, params = cur.execute.call_args[0]
+        return sql, params
+
+    def test_pictures_are_written_and_deduplicated(self):
+        sql, params = self._save(self._parsed([
+            "https://example.com/a.png",
+            "https://example.com/a.png",
+            "https://example.com/b.png",
+        ]))
+        self.assertIn("pictures", sql)
+        self.assertEqual(params[5], [
+            "https://example.com/a.png",
+            "https://example.com/b.png",
+        ])
+
+    def test_missing_pictures_key_writes_empty_array(self):
+        sql, params = self._save(self._parsed())
+        self.assertIn("pictures = EXCLUDED.pictures", sql)
+        self.assertEqual(params[5], [])
+
+
+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")
+        # The blocked host is filtered out by the query itself, so the
+        # claim never had to look at it; no wait hint is needed.
+        self.assertIsNone(wait_hint)
+
+    def test_rate_limited_host_is_skipped(self):
+        scheduler = crawler.HostScheduler()
+        # Was just crawled with the 3s per-host delay of StubRobots.
+        scheduler.wait("https://blocked.example/a", 3.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_blocked_wall_larger_than_scan_window_does_not_starve_queue(self):
+        # Regression: 30000+ hosts were queued, but every worker kept
+        # reporting "All candidate hosts busy or blocked". A blocked
+        # host with more pending URLs than the scan window sits at the
+        # head of the priority order forever, hiding the rest of the
+        # queue. The busy host must be excluded in SQL, not just skipped
+        # within the small candidate window.
+        state = FakeState()
+        wall = crawler.QUEUE_CLAIM_SCAN_LIMIT + 50
+        for index in range(wall):
+            state.add(
+                url="https://blocked.example/%d" % index,
+                priority=500 - index % 500,
+            )
+        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")
+        self.assertIsNone(wait_hint)
+
+    def test_busy_hosts_are_excluded_from_the_claim_query(self):
+        state = self._state()
+        scheduler = crawler.HostScheduler()
+        scheduler.blocked_until["blocked.example"] = time.monotonic() + 900
+        self.assertEqual(
+            set(scheduler.busy_hosts()), {"blocked.example"}
+        )
+        item, _wait_hint = self._claim(state, scheduler)
+        self.assertEqual(item[1], "https://other.example/b")
+
+    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()

web/test_stats.py

@@ -0,0 +1,359 @@
+#!/usr/bin/env python3
+"""
+Tests für den Statserver (web/app.py), ohne Datenbank.
+
+Geprüft werden die DB-freien Teile: Parsen der psql-Zeile, der
+2-Stunden-Rhythmus für den exakten Bild-Count, die Wahl zwischen
+exakter Zählung und Stichprobe sowie die Einbindung der neuen
+Images-Anzeige in die HTML-Seite.
+"""
+
+import importlib.util
+import os
+import subprocess
+import unittest
+from datetime import datetime, timedelta
+from unittest import mock
+
+HERE = os.path.dirname(os.path.abspath(__file__))
+
+# web/app.py unter einem eindeutigen Modulnamen laden, damit es nicht mit
+# priorityweb/app.py (gleicher Dateiname) kollidiert.
+_spec = importlib.util.spec_from_file_location(
+    "jonassearch_stats_app", os.path.join(HERE, "app.py")
+)
+stats_app = importlib.util.module_from_spec(_spec)
+_spec.loader.exec_module(stats_app)
+
+
+class ParseStatsRowTest(unittest.TestCase):
+
+    def test_count_and_images(self):
+        self.assertEqual(
+            stats_app._parse_stats_row("981981|12345"), (981981, 12345)
+        )
+
+    def test_empty_sample_is_none(self):
+        self.assertEqual(stats_app._parse_stats_row("10|"), (10, None))
+        self.assertEqual(stats_app._parse_stats_row("10|\n"), (10, None))
+
+
+class ImagesExactDueTest(unittest.TestCase):
+
+    NOW = datetime(2026, 10, 17, 12, 0, 0)
+
+    def _entry(self, age, exact):
+        return {
+            "timestamp": (self.NOW - age).isoformat(),
+            "count": 1,
+            "images": 2,
+            "images_exact": exact,
+        }
+
+    def test_no_data_means_due(self):
+        self.assertTrue(stats_app.images_exact_due([], now=self.NOW))
+
+    def test_recent_exact_is_not_due(self):
+        data = [self._entry(timedelta(hours=1), True)]
+        self.assertFalse(stats_app.images_exact_due(data, now=self.NOW))
+
+    def test_two_hours_old_is_due(self):
+        data = [self._entry(timedelta(hours=2), True)]
+        self.assertTrue(stats_app.images_exact_due(data, now=self.NOW))
+
+    def test_estimates_in_between_do_not_reset_schedule(self):
+        data = [
+            self._entry(timedelta(hours=1, minutes=30), True),
+            self._entry(timedelta(minutes=30), False),
+        ]
+        self.assertFalse(stats_app.images_exact_due(data, now=self.NOW))
+
+    def test_old_entries_without_flag_are_ignored(self):
+        data = [{"timestamp": self.NOW.isoformat(), "count": 1}]
+        self.assertTrue(stats_app.images_exact_due(data, now=self.NOW))
+
+    def test_malformed_exact_timestamp_is_ignored(self):
+        data = [{"timestamp": "kaputt", "images_exact": True}]
+        self.assertTrue(stats_app.images_exact_due(data, now=self.NOW))
+
+
+class GetStatsTest(unittest.TestCase):
+
+    @mock.patch.object(stats_app, "_psql_scalar")
+    def test_exact_query(self, psql):
+        psql.return_value = "100|250"
+        self.assertEqual(stats_app.get_stats(True), (100, 250, True))
+        self.assertEqual(psql.call_args[0][0], stats_app.EXACT_IMAGE_SQL)
+        self.assertNotIn("TABLESAMPLE", stats_app.EXACT_IMAGE_SQL)
+
+    @mock.patch.object(stats_app, "_psql_scalar")
+    def test_estimate_query(self, psql):
+        psql.return_value = "100|250"
+        self.assertEqual(stats_app.get_stats(False), (100, 250, False))
+        self.assertEqual(psql.call_args[0][0], stats_app.ESTIMATE_IMAGE_SQL)
+        self.assertIn("TABLESAMPLE SYSTEM (1)", stats_app.ESTIMATE_IMAGE_SQL)
+
+    @mock.patch.object(stats_app, "_psql_scalar")
+    def test_empty_sample_on_small_table_falls_back_to_exact(self, psql):
+        psql.side_effect = ["10|", "10|20"]
+        self.assertEqual(stats_app.get_stats(False), (10, 20, True))
+        self.assertEqual(psql.call_count, 2)
+        self.assertEqual(psql.call_args_list[1][0][0],
+                         stats_app.EXACT_IMAGE_SQL)
+
+    @mock.patch.object(stats_app, "_psql_scalar")
+    def test_empty_sample_on_large_table_stays_unknown(self, psql):
+        psql.return_value = "200000|"
+        self.assertEqual(stats_app.get_stats(False), (200000, None, False))
+
+    @mock.patch.object(stats_app, "_psql_scalar")
+    def test_missing_pictures_column_falls_back_to_page_count(self, psql):
+        error = subprocess.CalledProcessError(
+            1, ["psql"], stderr='ERROR: column "pictures" does not exist'
+        )
+        psql.side_effect = [error, "10"]
+        self.assertEqual(stats_app.get_stats(False), (10, None, False))
+        self.assertEqual(psql.call_args_list[1][0][0],
+                         "SELECT COUNT(*) FROM pages;")
+
+
+class GetPendingTest(unittest.TestCase):
+
+    @mock.patch.object(stats_app, "_psql_scalar")
+    def test_counts_pending_queue_rows(self, psql):
+        psql.return_value = "1384711"
+        self.assertEqual(stats_app.get_pending(), 1384711)
+        self.assertIn("FROM crawl_queue", stats_app.PENDING_QUEUE_SQL)
+        self.assertIn("status = 'pending'", stats_app.PENDING_QUEUE_SQL)
+
+    @mock.patch.object(stats_app, "_psql_scalar")
+    def test_unavailable_queue_returns_none(self, psql):
+        psql.side_effect = subprocess.CalledProcessError(
+            1, ["psql"], stderr='ERROR: relation "crawl_queue" does not exist'
+        )
+        self.assertIsNone(stats_app.get_pending())
+
+    @mock.patch.object(stats_app, "_psql_scalar")
+    def test_unexpected_output_returns_none(self, psql):
+        psql.return_value = ""
+        self.assertIsNone(stats_app.get_pending())
+
+
+class CollectTest(unittest.TestCase):
+
+    def _run_collect(self, get_stats, get_pending):
+        with mock.patch.object(stats_app, "load_data", return_value=[]), \
+                mock.patch.object(stats_app, "save_data") as save_data, \
+                mock.patch.object(stats_app, "get_stats", get_stats), \
+                mock.patch.object(stats_app, "get_pending", get_pending), \
+                mock.patch.object(stats_app, "time") as time_mock:
+            time_mock.sleep.side_effect = KeyboardInterrupt
+            with self.assertRaises(KeyboardInterrupt):
+                stats_app.collect()
+        return save_data.call_args[0][0][0]
+
+    def test_pending_is_stored_with_every_sample(self):
+        entry = self._run_collect(
+            mock.Mock(return_value=(1000, 2000, True)),
+            mock.Mock(return_value=4711),
+        )
+        self.assertEqual(entry["pending"], 4711)
+        self.assertEqual(entry["count"], 1000)
+        self.assertEqual(entry["images"], 2000)
+
+    def test_missing_queue_does_not_break_the_sample(self):
+        entry = self._run_collect(
+            mock.Mock(return_value=(1000, 2000, False)),
+            mock.Mock(return_value=None),
+        )
+        self.assertIsNone(entry["pending"])
+        self.assertEqual(entry["count"], 1000)
+
+
+class StatsPageTest(unittest.TestCase):
+
+    def test_page_contains_image_counter(self):
+        client = stats_app.app.test_client()
+        response = client.get("/")
+        self.assertEqual(response.status_code, 200)
+        html = response.get_data(as_text=True)
+        self.assertIn('id="current-images"', html)
+        self.assertIn('id="current-images-meta"', html)
+        self.assertIn("Estimated · exact every 2 h", html)
+        self.assertIn("imagesExact", html)
+
+    def test_page_contains_6h_range_and_card(self):
+        client = stats_app.app.test_client()
+        response = client.get("/")
+        self.assertEqual(response.status_code, 200)
+        html = response.get_data(as_text=True)
+        self.assertIn('data-range="6h"', html)
+        self.assertIn('id="stat-delta-6h"', html)
+        self.assertIn('id="stat-delta-6h-hint"', html)
+        self.assertIn("'6h': 6 * 3600000", html)
+        self.assertIn("closestBefore(all, last.x - 6 * 3600000", html)
+
+    def test_page_contains_all_image_graphs(self):
+        client = stats_app.app.test_client()
+        response = client.get("/")
+        self.assertEqual(response.status_code, 200)
+        html = response.get_data(as_text=True)
+
+        self.assertIn("Images – all graphs", html)
+        for canvas in (
+            "imagesMainChart",
+            "imagesDeltaChart",
+            "imagesCumChart",
+            "imagesHourlyChart",
+            "imagesHodChart",
+        ):
+            self.assertIn('id="' + canvas + '"', html)
+        self.assertIn('id="records-images"', html)
+        self.assertIn('id="heatmap-images"', html)
+        self.assertIn('id="images-empty"', html)
+
+        # Image estimates are excluded by default and can be included with the persistent toggle.
+        self.assertIn('id="show-estimated-images"', html)
+        self.assertIn('type="checkbox" id="show-estimated-images"', html)
+        self.assertIn("Include estimated image counts in graphs", html)
+        self.assertIn("function toImagePoints(points, includeEstimated)", html)
+        self.assertIn("p.images !== null && p.images !== undefined", html)
+        self.assertIn("includeEstimated || p.imagesExact", html)
+        self.assertIn("localStorage.getItem('js-show-estimated-images')", html)
+        self.assertIn("localStorage.setItem('js-show-estimated-images'", html)
+        self.assertIn("toImagePoints(all, showEstimatedImages)", html)
+        self.assertIn("Only exact measurements (counted every 2 h) are plotted", html)
+        self.assertIn("Exact and estimated measurements are plotted", html)
+        self.assertIn("Only estimated image counts so far", html)
+        self.assertIn("No image counts in the selected range", html)
+        # The headline counter stays the latest collected value; only graphs toggle estimates.
+        self.assertIn("elImages.textContent = fmt(last.images)", html)
+        # Images estimates affect graphs only; other series keep their own points.
+        self.assertIn("renderCharts({\n        prefix: '',\n        points: fpoints", html)
+        self.assertIn("renderCharts({\n        prefix: 'pending',\n        points: pendingFpoints", html)
+        self.assertIn("renderHeatmap(imageAll, 'heatmap-images', 'Images')", html)
+        self.assertIn("renderRecords(imageDeltas, 'records-images', 'Images')", html)
+        # existing page chart ids must stay lowercase (mainChart, not MainChart)
+        self.assertIn("function chartId(prefix, suffix)", html)
+        self.assertIn("chartId(prefix, 'MainChart')", html)
+        # both series use the same time window (anchored to the latest page sample)
+        self.assertIn("filterRange(imageAll, last.x)", html)
+
+        # both metrics go through the same renderers, with the right units
+        self.assertIn("unit: 'Images'", html)
+        self.assertIn("unit: 'Pages'", html)
+        self.assertIn("renderRecords(deltas, 'records', 'Pages')", html)
+        self.assertIn("renderRecords(imageDeltas, 'records-images', 'Images')", html)
+        self.assertIn("renderHeatmap(all, 'heatmap', 'Pages')", html)
+        self.assertIn("renderHeatmap(imageAll, 'heatmap-images', 'Images')", html)
+
+    def test_page_has_metric_tabs_with_one_panel_each(self):
+        client = stats_app.app.test_client()
+        response = client.get("/")
+        self.assertEqual(response.status_code, 200)
+        html = response.get_data(as_text=True)
+
+        # three equally styled tab buttons (pages / images / queue) with icons
+        self.assertIn('class="metric-btn', html)
+        for metric in ("pages", "images", "queue"):
+            self.assertIn('data-metric="' + metric + '"', html)
+        self.assertIn("metric-icon", html)
+        self.assertIn("📄", html)
+        self.assertIn("🖼️", html)
+        self.assertIn("⏳", html)
+
+        # every metric has its own panel; only the initial one is active
+        self.assertEqual(html.count('class="metric-panel active" data-metric="pages"'), 1)
+        for metric in ("images", "queue"):
+            self.assertIn('class="metric-panel" data-metric="' + metric + '"', html)
+        self.assertIn('id="summary-images"', html)
+        self.assertIn('id="summary-queue"', html)
+
+        # switching tabs hides the other panels and is remembered
+        self.assertIn("function setMetric(next, persist)", html)
+        self.assertIn("panel.classList.toggle('active', panel.dataset.metric === activeMetric)", html)
+        self.assertIn("localStorage.setItem('js-metric', activeMetric)", html)
+        self.assertIn("localStorage.getItem('js-metric')", html)
+
+        # hidden panels must not create Chart.js instances (canvas has 0 px)
+        self.assertIn("function isVisible(el)", html)
+        self.assertIn("if (!target || !isVisible(target)) return;", html)
+
+        # summary cards are generated for the non-page tabs
+        self.assertIn("function renderSummary(containerId, all, points)", html)
+        self.assertIn("renderSummary('summary-images', imageAll, imageFpoints)", html)
+        self.assertIn("renderSummary('summary-queue', pendingAll, pendingFpoints)", html)
+
+        # exports follow the selected tab
+        self.assertIn("function exportSeries()", html)
+        self.assertIn("return { name: 'jonassearch_images'", html)
+        self.assertIn("return { name: 'jonassearch_queue'", html)
+
+    def test_page_has_target_eta_field_for_active_tab(self):
+        client = stats_app.app.test_client()
+        response = client.get("/")
+        self.assertEqual(response.status_code, 200)
+        html = response.get_data(as_text=True)
+
+        # a plain number field at the top, plus the computed duration
+        self.assertIn('id="goal-input"', html)
+        self.assertIn('type="number" inputmode="numeric"', html)
+        self.assertIn('id="goal-eta"', html)
+        self.assertIn('id="goal-hint"', html)
+        self.assertIn('id="goal-unit"', html)
+        self.assertIn("Reach target", html)
+
+        # the estimate uses the trend of the selected range (last 2 h first)
+        self.assertIn("function recentTrend(points)", html)
+        self.assertIn("const reg = recentTrend(fpoints);", html)
+        self.assertIn("function fmtEta(ms)", html)
+        self.assertIn("const speed = trend ? trend.slope : null;", html)
+        self.assertIn("const etaMs = remaining / speed * 3600000;", html)
+        self.assertIn("if (remaining === 0) {", html)
+        self.assertIn("Moving away from the target", html)
+
+        # one target per tab, restored on tab switch and persisted per browser
+        self.assertIn("function goalSeries()", html)
+        self.assertIn("goalTargets[activeMetric] || ''", html)
+        self.assertIn("localStorage.setItem('js-goal-' + activeMetric, raw)", html)
+        self.assertIn("localStorage.getItem('js-goal-' + metric)", html)
+        self.assertIn("toImagePoints(all, showEstimatedImages)", html)
+
+        # recomputed on refresh, tab switch and range switch
+        self.assertIn("renderSummary('summary-queue', pendingAll, pendingFpoints);\n    renderGoal();", html)
+        self.assertIn("goalInput.value = goalTargets[activeMetric] || '';\n        renderGoal();", html)
+
+    def test_page_contains_pending_queue_counter_and_graphs(self):
+        client = stats_app.app.test_client()
+        response = client.get("/")
+        self.assertEqual(response.status_code, 200)
+        html = response.get_data(as_text=True)
+
+        self.assertIn('id="current-pending"', html)
+        self.assertIn('id="current-pending-meta"', html)
+        self.assertIn("Queue – all graphs", html)
+        for canvas in (
+            "pendingMainChart",
+            "pendingDeltaChart",
+            "pendingCumChart",
+            "pendingHourlyChart",
+            "pendingHodChart",
+        ):
+            self.assertIn('id="' + canvas + '"', html)
+        self.assertIn('id="records-pending"', html)
+        self.assertIn('id="heatmap-pending"', html)
+        self.assertIn('id="pending-empty"', html)
+
+        # pending values are parsed and plotted without sampling/filtering
+        self.assertIn("function toPendingPoints(points)", html)
+        self.assertIn("p.pending !== null && p.pending !== undefined", html)
+        self.assertIn("filterRange(pendingAll, last.x)", html)
+        self.assertIn("unit: 'Pending'", html)
+        self.assertIn("renderRecords(pendingDeltas, 'records-pending', 'Pending')", html)
+        self.assertIn("renderHeatmap(pendingAll, 'heatmap-pending', 'Pending')", html)
+        self.assertIn("No queue counts yet", html)
+        self.assertIn("No queue counts in the selected range.", html)
+
+
+if __name__ == "__main__":
+    unittest.main(verbosity=2)

web/test_tui.py

@@ -0,0 +1,341 @@
+#!/usr/bin/env python3
+"""
+Tests für das Terminal-Dashboard (web/tui.py), ohne Terminal und ohne DB.
+
+Geprüft werden Datei-Lesen, Zeitfenster, Sparkline, die Textdarstellung
+und der Aufruf über die Kommandozeile (`--once`).
+"""
+
+import importlib.util
+import json
+import os
+import subprocess
+import sys
+import tempfile
+import unittest
+from datetime import datetime, timedelta
+
+HERE = os.path.dirname(os.path.abspath(__file__))
+ROOT = os.path.dirname(HERE)
+
+_spec = importlib.util.spec_from_file_location(
+    "jonassearch_stats_tui", os.path.join(HERE, "tui.py")
+)
+tui = importlib.util.module_from_spec(_spec)
+_spec.loader.exec_module(tui)
+
+try:
+    _stats_spec = importlib.util.spec_from_file_location(
+        "jonassearch_stats_app_for_tui", os.path.join(HERE, "app.py")
+    )
+    stats_app = importlib.util.module_from_spec(_stats_spec)
+    _stats_spec.loader.exec_module(stats_app)
+except Exception:  # z. B. Flask nicht installiert
+    stats_app = None
+
+NOW = datetime(2026, 10, 17, 12, 0, 0)
+
+
+def entry(timestamp, count, images=None, exact=None):
+    data = {"timestamp": timestamp.isoformat(), "count": count}
+    if images is not None:
+        data["images"] = images
+    if exact is not None:
+        data["images_exact"] = exact
+    return data
+
+
+def sample_entries():
+    return [
+        entry(NOW - timedelta(hours=2), 980_000, images=100, exact=True),
+        entry(NOW - timedelta(hours=1), 981_000),
+        entry(NOW - timedelta(minutes=30), 981_950, images=120, exact=False),
+    ]
+
+
+def normalized(entries):
+    """Rohdaten so aufbereiten, wie load_entries() sie liefert."""
+    result = []
+    for item in entries:
+        copy = dict(item)
+        copy["timestamp"] = tui.parse_timestamp(copy["timestamp"])
+        result.append(copy)
+    return result
+
+
+class FormatTest(unittest.TestCase):
+
+    def test_fmt_int_uses_dots(self):
+        self.assertEqual(tui.fmt_int(987832), "987.832")
+        self.assertEqual(tui.fmt_int(None), "-")
+
+    def test_fmt_delta(self):
+        self.assertEqual(tui.fmt_delta(950), "+950")
+        self.assertEqual(tui.fmt_delta(-12), "-12")
+        self.assertEqual(tui.fmt_delta(0), "0")
+        self.assertEqual(tui.fmt_delta(None), "-")
+
+    def test_fmt_duration(self):
+        self.assertEqual(tui.fmt_duration(30), "30 s")
+        self.assertEqual(tui.fmt_duration(90), "1 min")
+        self.assertEqual(tui.fmt_duration(5400), "1 h 30 min")
+        self.assertEqual(tui.fmt_duration(90000), "1 d 1 h")
+
+    def test_samples_label(self):
+        self.assertEqual(tui._samples_label(1), "1 sample")
+        self.assertEqual(tui._samples_label(2), "2 samples")
+
+    def test_relative_time(self):
+        self.assertEqual(tui.relative_time(NOW, NOW), "just now")
+        self.assertEqual(
+            tui.relative_time(NOW - timedelta(minutes=3), NOW), "3 min ago"
+        )
+
+
+class ParseTimestampTest(unittest.TestCase):
+
+    def test_iso_and_z(self):
+        self.assertEqual(
+            tui.parse_timestamp("2026-10-17T20:44:20.187761"),
+            datetime(2026, 10, 17, 20, 44, 20, 187761),
+        )
+        parsed = tui.parse_timestamp("2026-10-17T20:44:20Z")
+        self.assertIsNone(parsed.tzinfo)
+        self.assertEqual(
+            parsed,
+            datetime.fromisoformat("2026-10-17T20:44:20+00:00")
+            .astimezone().replace(tzinfo=None),
+        )
+
+    def test_invalid(self):
+        self.assertIsNone(tui.parse_timestamp("kaputt"))
+        self.assertIsNone(tui.parse_timestamp(""))
+        self.assertIsNone(tui.parse_timestamp(None))
+
+
+class LoadEntriesTest(unittest.TestCase):
+
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory()
+        self.path = os.path.join(self.tmp.name, "page_counts.json")
+
+    def tearDown(self):
+        self.tmp.cleanup()
+
+    def _write(self, data):
+        with open(self.path, "w", encoding="utf-8") as handle:
+            if isinstance(data, str):
+                handle.write(data)
+            else:
+                json.dump(data, handle)
+
+    def test_missing_file_is_empty(self):
+        self.assertEqual(tui.load_entries(self.path), [])
+
+    def test_sorts_and_skips_bad_rows(self):
+        self._write([
+            entry(NOW, 20),
+            "kaputt",
+            {"timestamp": "kaputt", "count": 1},
+            {"timestamp": NOW.isoformat(), "count": "x"},
+            entry(NOW - timedelta(minutes=30), 10),
+        ])
+        entries = tui.load_entries(self.path)
+        self.assertEqual([e["count"] for e in entries], [10, 20])
+        self.assertIsInstance(entries[0]["timestamp"], datetime)
+
+    def test_api_format_is_accepted(self):
+        self._write({"values": [entry(NOW, 5, images=7, exact=True)]})
+        entries = tui.load_entries(self.path)
+        self.assertEqual(len(entries), 1)
+        self.assertEqual(entries[0]["images"], 7)
+        self.assertTrue(entries[0]["images_exact"])
+
+    def test_invalid_images_becomes_none(self):
+        self._write([{"timestamp": NOW.isoformat(), "count": 1,
+                      "images": "viele"}])
+        self.assertIsNone(tui.load_entries(self.path)[0]["images"])
+
+    def test_corrupt_json_raises(self):
+        self._write("{not json")
+        with self.assertRaises(tui.StatsDataError):
+            tui.load_entries(self.path)
+
+    def test_wrong_toplevel_type_raises(self):
+        self._write(42)
+        with self.assertRaises(tui.StatsDataError):
+            tui.load_entries(self.path)
+
+
+class SharedScheduleTest(unittest.TestCase):
+
+    def test_exact_interval_matches_statserver(self):
+        if stats_app is None:
+            self.skipTest("web/app.py nicht importierbar (Flask fehlt)")
+        self.assertEqual(
+            tui.EXACT_IMAGE_INTERVAL, stats_app.EXACT_IMAGE_INTERVAL
+        )
+
+
+class WindowTest(unittest.TestCase):
+
+    def test_window_filters_old_entries(self):
+        entries = normalized([
+            entry(NOW - timedelta(days=3), 1),
+            entry(NOW - timedelta(hours=1), 2),
+        ])
+        self.assertEqual(
+            [e["count"] for e in tui.select_window(entries, "24h", NOW)], [2]
+        )
+        self.assertEqual(len(tui.select_window(entries, "all", NOW)), 2)
+
+    def test_latest_images_falls_back(self):
+        entries = normalized([
+            entry(NOW - timedelta(hours=1), 1, images=10, exact=True),
+            entry(NOW, 2),
+        ])
+        images, exact, timestamp = tui.latest_images(entries)
+        self.assertEqual(images, 10)
+        self.assertTrue(exact)
+        self.assertEqual(timestamp, entries[0]["timestamp"])
+
+    def test_next_exact_text(self):
+        recent = normalized([
+            entry(NOW - timedelta(minutes=30), 1, images=1, exact=True)
+        ])
+        self.assertEqual(
+            tui.next_exact_text(recent, NOW), "next exact in 1 h 30 min"
+        )
+        overdue = normalized([
+            entry(NOW - timedelta(hours=3), 1, images=1, exact=True)
+        ])
+        self.assertEqual(
+            tui.next_exact_text(overdue, NOW), "exact count due now"
+        )
+        self.assertEqual(
+            tui.next_exact_text(normalized([entry(NOW, 1)]), NOW),
+            "no exact count yet",
+        )
+
+
+class SparklineTest(unittest.TestCase):
+
+    def test_width_and_padding(self):
+        line = tui.sparkline([1, 2, 3], 10)
+        self.assertEqual(len(line), 10)
+        self.assertTrue(line.startswith("       "))
+
+    def test_single_sample(self):
+        self.assertEqual(tui.sparkline([5], 1), tui.SPARK_BLOCKS[4])
+
+    def test_increasing_values_use_increasing_blocks(self):
+        line = tui.sparkline([1, 2, 3, 4], 4)
+        indices = [tui.SPARK_BLOCKS.index(char) for char in line]
+        self.assertEqual(indices, sorted(indices))
+
+    def test_constant_values_do_not_crash(self):
+        self.assertEqual(tui.sparkline([7, 7, 7], 3), tui.SPARK_BLOCKS[4] * 3)
+
+    def test_empty(self):
+        self.assertEqual(tui.sparkline([], 10), "")
+
+
+class RenderTest(unittest.TestCase):
+
+    def test_shows_numbers_and_hints(self):
+        lines = tui.render_dashboard(
+            normalized(sample_entries()), now=NOW, width=100
+        )
+        text = "\n".join(lines)
+        self.assertIn("Pages      981.950", text)
+        self.assertIn("Images     120", text)
+        self.assertIn("estimated", text)
+        self.assertIn("exact count due now", text)
+        self.assertIn("+950 since previous", text)
+        self.assertIn("q quit", text)
+        self.assertIn("t window (24h)", text)
+        self.assertIn("Recent samples", text)
+
+    def test_lines_fit_width(self):
+        for width in (32, 48, 60, 80):
+            lines = tui.render_dashboard(
+                normalized(sample_entries()), now=NOW, width=width, rows=24
+            )
+            for line in lines:
+                self.assertLessEqual(len(line), width, repr(line))
+
+    def test_row_budget_keeps_footer(self):
+        lines = tui.render_dashboard(
+            normalized(sample_entries()), now=NOW, width=80, rows=12
+        )
+        self.assertLessEqual(len(lines), 12)
+        self.assertIn("q quit", lines[-1])
+
+    def test_unreadable_file_error(self):
+        temp = tempfile.TemporaryDirectory()
+        self.addCleanup(temp.cleanup)
+        path = os.path.join(temp.name, "page_counts.json")
+        with open(path, "w", encoding="utf-8") as handle:
+            handle.write("{kaputt")
+
+        try:
+            tui.load_entries(path)
+        except tui.StatsDataError as exc:
+            lines = tui.render_dashboard([], now=NOW, width=90, error=str(exc))
+        else:
+            self.fail("StatsDataError erwartet")
+
+        self.assertIn("invalid JSON", "\n".join(lines))
+        self.assertIn("q quit", "\n".join(lines))
+
+    def test_empty_data(self):
+        text = "\n".join(tui.render_dashboard([], now=NOW, width=80))
+        self.assertIn("No samples yet", text)
+
+
+class OnceModeTest(unittest.TestCase):
+
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory()
+        self.addCleanup(self.tmp.cleanup)
+        self.path = os.path.join(self.tmp.name, "page_counts.json")
+        with open(self.path, "w", encoding="utf-8") as handle:
+            json.dump(sample_entries(), handle)
+
+    def _run(self, *extra):
+        return subprocess.run(
+            [sys.executable, os.path.join(HERE, "tui.py"), "--once",
+             "--file", self.path, "--width", "100"] + list(extra),
+            capture_output=True,
+            text=True,
+            cwd=ROOT,
+        )
+
+    def test_once_prints_dashboard(self):
+        result = self._run()
+        self.assertEqual(result.returncode, 0, result.stderr)
+        self.assertIn("Pages", result.stdout)
+        self.assertIn("981.950", result.stdout)
+        self.assertIn("q quit", result.stdout)
+
+    def test_once_window_flag(self):
+        result = self._run("--window", "all", "--ascii")
+        self.assertEqual(result.returncode, 0, result.stderr)
+        self.assertIn("all data", result.stdout)
+
+    def test_once_corrupt_file_exits_nonzero(self):
+        with open(self.path, "w", encoding="utf-8") as handle:
+            handle.write("{kaputt")
+        result = self._run()
+        self.assertEqual(result.returncode, 1)
+        self.assertIn("error:", result.stderr)
+
+    def test_once_missing_file_is_empty_state(self):
+        os.unlink(self.path)
+        result = self._run()
+        self.assertEqual(result.returncode, 0, result.stderr)
+        self.assertIn("No samples yet", result.stdout)
+
+
+if __name__ == "__main__":
+    unittest.main()

web/tui.py

@@ -0,0 +1,640 @@
+#!/usr/bin/env python3
+"""
+Terminal-Dashboard für den Statserver.
+
+Liest dieselbe Datei wie `web/app.py` (`page_counts.json`) und zeigt
+Seitenzahl, Bildanzahl, Wachstum, einen Sparkline-Verlauf und die
+letzten Messwerte direkt im Terminal. Startet keinen Webserver und
+benötigt keine zusätzlichen Pakete (nur curses aus der Standardlib).
+
+Aufruf:
+    python3 web/tui.py                 # interaktiv (q beendet)
+    python3 web/tui.py --once          # Momentaufnahme für Skripte/Logs
+    python3 web/tui.py --window 7d     # 24h (Default) | 7d | 30d | all
+"""
+
+from __future__ import annotations
+
+import argparse
+import curses
+import json
+import locale
+import shutil
+import sys
+import time
+from datetime import datetime, timedelta
+from pathlib import Path
+
+DB_DATA = Path(__file__).parent / "page_counts.json"
+
+# Muss zu EXACT_IMAGE_INTERVAL in web/app.py passen (dort per Test
+# abgesichert): Bilder werden nur alle 2 Stunden exakt gezählt.
+EXACT_IMAGE_INTERVAL = 2 * 60 * 60  # Sekunden
+
+REFRESH_SECONDS = 2.0
+DEFAULT_WINDOW = "24h"
+WINDOWS = {
+    "24h": timedelta(hours=24),
+    "7d": timedelta(days=7),
+    "30d": timedelta(days=30),
+    "all": None,
+}
+WINDOW_ORDER = ("24h", "7d", "30d", "all")
+WINDOW_LABELS = {
+    "24h": "last 24 h",
+    "7d": "last 7 days",
+    "30d": "last 30 days",
+    "all": "all data",
+}
+
+SPARK_BLOCKS = "▁▂▃▄▅▆▇█"
+ASCII_BLOCKS = ".:-=+*#%"
+
+TABLE_ROWS_ONCE = 15
+
+
+class StatsDataError(Exception):
+    """Die Statistikdatei fehlt nicht, ist aber nicht lesbar/parsebar."""
+
+
+# --------------------------------------------------------------------------
+# Daten
+# --------------------------------------------------------------------------
+
+def parse_timestamp(value):
+    """ISO-Zeitstempel -> naive lokale datetime (oder None)."""
+    if not isinstance(value, str) or not value.strip():
+        return None
+
+    text = value.strip()
+    if text.endswith("Z"):
+        text = text[:-1] + "+00:00"
+
+    try:
+        timestamp = datetime.fromisoformat(text)
+    except ValueError:
+        return None
+
+    if timestamp.tzinfo is not None:
+        timestamp = timestamp.astimezone().replace(tzinfo=None)
+
+    return timestamp
+
+
+def load_entries(path=DB_DATA):
+    """
+    Messwerte laden, sortiert nach Zeitstempel.
+
+    Akzeptiert sowohl die Liste aus `page_counts.json` als auch das
+    Format der API (`{"values": [...]}`). Einträge ohne gültigen
+    Zeitstempel oder ohne Seitenzahl werden übersprungen.
+    """
+    path = Path(path)
+
+    try:
+        raw = json.loads(path.read_text(encoding="utf-8"))
+    except FileNotFoundError:
+        return []
+    except OSError as exc:
+        raise StatsDataError(f"{path}: {exc}") from exc
+    except json.JSONDecodeError as exc:
+        raise StatsDataError(
+            f"{path}: invalid JSON (line {exc.lineno}: {exc.msg})"
+        ) from exc
+
+    if isinstance(raw, dict):
+        raw = raw.get("values", [])
+
+    if not isinstance(raw, list):
+        raise StatsDataError(f"{path}: expected a JSON list of samples")
+
+    entries = []
+
+    for item in raw:
+        if not isinstance(item, dict):
+            continue
+
+        timestamp = parse_timestamp(item.get("timestamp"))
+        if timestamp is None:
+            continue
+
+        try:
+            count = int(item.get("count"))
+        except (TypeError, ValueError):
+            continue
+
+        entry = dict(item)
+        entry["timestamp"] = timestamp
+        entry["count"] = count
+
+        images = item.get("images")
+        if images is None:
+            entry["images"] = None
+        else:
+            try:
+                entry["images"] = int(images)
+            except (TypeError, ValueError):
+                entry["images"] = None
+
+        entries.append(entry)
+
+    entries.sort(key=lambda entry: entry["timestamp"])
+    return entries
+
+
+def select_window(entries, window, now=None):
+    """Nur Messwerte innerhalb des gewählten Zeitfensters."""
+    delta = WINDOWS.get(window, WINDOWS[DEFAULT_WINDOW])
+    if delta is None:
+        return list(entries)
+
+    now = now or datetime.now()
+    cutoff = now - delta
+    return [entry for entry in entries if entry["timestamp"] >= cutoff]
+
+
+def latest_images(entries):
+    """
+    (images, images_exact, timestamp) des neuesten Eintrags mit
+    Bildanzahl. Ältere Dateien haben kein `images`-Feld -> None.
+    """
+    for entry in reversed(entries):
+        if entry.get("images") is not None:
+            return (
+                entry["images"],
+                entry.get("images_exact") is True,
+                entry["timestamp"],
+            )
+    return None, False, None
+
+
+def next_exact_text(entries, now):
+    """Wann die nächste exakte Bildzählung fällig ist."""
+    last_exact = None
+
+    for entry in reversed(entries):
+        if entry.get("images_exact") is True:
+            last_exact = entry["timestamp"]
+            break
+
+    if last_exact is None:
+        return "no exact count yet"
+
+    remaining = EXACT_IMAGE_INTERVAL - (now - last_exact).total_seconds()
+    if remaining <= 0:
+        return "exact count due now"
+    return f"next exact in {fmt_duration(remaining)}"
+
+
+# --------------------------------------------------------------------------
+# Formatierung
+# --------------------------------------------------------------------------
+
+def fmt_int(value):
+    if value is None:
+        return "-"
+    return f"{int(value):,}".replace(",", ".")
+
+
+def fmt_delta(value):
+    if value is None:
+        return "-"
+    value = int(value)
+    if value == 0:
+        return "0"
+    if value > 0:
+        return "+" + fmt_int(value)
+    return "-" + fmt_int(abs(value))
+
+
+def fmt_duration(seconds):
+    seconds = max(0, int(round(seconds)))
+
+    if seconds < 60:
+        return f"{seconds} s"
+
+    minutes = seconds // 60
+    if minutes < 60:
+        return f"{minutes} min"
+
+    hours, minutes = divmod(minutes, 60)
+    if hours < 24:
+        return f"{hours} h {minutes} min"
+
+    days, hours = divmod(hours, 24)
+    return f"{days} d {hours} h"
+
+
+def relative_time(timestamp, now):
+    seconds = (now - timestamp).total_seconds()
+    if seconds < 5:
+        return "just now"
+    return f"{fmt_duration(seconds)} ago"
+
+
+def sparkline(values, width, blocks=SPARK_BLOCKS):
+    """Sparkline aus Zahlen; links mit Leerzeichen auf `width` aufgefüllt."""
+    if width <= 0 or not values:
+        return ""
+
+    values = list(values)[-width:]
+    low, high = min(values), max(values)
+
+    if high == low:
+        chars = [blocks[len(blocks) // 2]] * len(values)
+    else:
+        span = high - low
+        last = len(blocks) - 1
+        chars = [
+            blocks[min(last, int((value - low) / span * last + 0.5))]
+            for value in values
+        ]
+
+    return " " * (width - len(values)) + "".join(chars)
+
+
+def _fit(text, width):
+    return text[:max(0, width)]
+
+
+def _samples_label(count):
+    return f"{count} sample" if count == 1 else f"{count} samples"
+
+
+def _rate(window_entries):
+    """Wachstum in Seiten pro Stunde (None bei < 2 Messwerten)."""
+    if len(window_entries) < 2:
+        return None
+
+    first, last = window_entries[0], window_entries[-1]
+    hours = (last["timestamp"] - first["timestamp"]).total_seconds() / 3600
+
+    if hours <= 0:
+        return None
+
+    return (last["count"] - first["count"]) / hours
+
+
+def _deltas(entries):
+    """Änderung der Seitenzahl gegenüber dem vorherigen Messwert."""
+    result = {}
+    previous = None
+
+    for entry in entries:
+        result[id(entry)] = (
+            None if previous is None else entry["count"] - previous
+        )
+        previous = entry["count"]
+
+    return result
+
+
+def _time_label(timestamp, now, compact):
+    if compact:
+        return timestamp.strftime("%H:%M")
+    if timestamp.date() == now.date():
+        return timestamp.strftime("%H:%M")
+    return timestamp.strftime("%d.%m. %H:%M")
+
+
+def _table_line(time_text, pages, delta, images, flag, compact):
+    if compact:
+        return (
+            f" {time_text:<6} {pages:>10} {delta:>8}"
+            f" {images:>6} {flag}"
+        )
+    return (
+        f" {time_text:<13} {pages:>12} {delta:>9}"
+        f" {images:>10}  {flag}"
+    )
+
+
+def _build_table(window_entries, delta_by_id, now, width, limit):
+    """Neueste Messwerte zuerst; Kopfzeile als erste Zeile."""
+    compact = width < 62
+    header = _table_line(
+        "Time",
+        "Pages",
+        "Chg" if compact else "Change",
+        "Images",
+        "",
+        compact,
+    )
+
+    rows = []
+    newest_first = list(reversed(window_entries))
+
+    for entry in newest_first[:limit]:
+        flag = ""
+        if entry.get("images") is not None:
+            flag = "exact" if entry.get("images_exact") is True else "est."
+
+        rows.append(_table_line(
+            _time_label(entry["timestamp"], now, compact),
+            fmt_int(entry["count"]),
+            fmt_delta(delta_by_id.get(id(entry))),
+            fmt_int(entry.get("images")),
+            flag,
+            compact,
+        ))
+
+    return header, rows
+
+
+# --------------------------------------------------------------------------
+# Darstellung
+# --------------------------------------------------------------------------
+
+def render_dashboard(entries, window=DEFAULT_WINDOW, now=None, width=80,
+                     rows=None, source=None, error=None,
+                     blocks=SPARK_BLOCKS):
+    """
+    Dashboard als Liste von Textzeilen.
+
+    Reine Funktion ohne curses: dadurch testbar und auch für `--once`
+    bzw. Umleitungen nutzbar. `rows` begrenzt die Höhe (Terminal);
+    ohne Angabe werden bis zu 15 Messwerte angezeigt.
+    """
+    now = now or datetime.now()
+    width = max(24, int(width))
+    source = str(source or DB_DATA)
+
+    lines = [_fit(" JonasSearch stats", width)]
+
+    if error:
+        lines.append("")
+        lines.append(_fit(f" ! {error}", width))
+        lines.append("")
+        lines.append(_fit(" q quit | r retry", width))
+        return lines[:rows] if rows else lines
+
+    if not entries:
+        lines.append("")
+        lines.append(_fit(" No samples yet.", width))
+        lines.append(_fit(f" Waiting for the collector to write {source}", width))
+        lines.append("")
+        lines.append(_fit(" q quit | r reload", width))
+        return lines[:rows] if rows else lines
+
+    latest = entries[-1]
+    previous = entries[-2] if len(entries) > 1 else None
+    delta = None if previous is None else latest["count"] - previous["count"]
+
+    window_entries = select_window(entries, window, now)
+    window_note = ""
+    if not window_entries:
+        window_entries = [latest]
+        window_note = " (no recent samples)"
+
+    rate = _rate(window_entries)
+    images, images_exact, images_ts = latest_images(entries)
+
+    lines.append(_fit(
+        f" updated {latest['timestamp']:%d.%m.%Y %H:%M:%S}"
+        f" ({relative_time(latest['timestamp'], now)})",
+        width,
+    ))
+    lines.append("")
+
+    lines.append(_fit(f" Pages      {fmt_int(latest['count'])}", width))
+
+    growth = [fmt_delta(delta) + " since previous" if delta is not None
+              else "first sample"]
+    if rate is not None:
+        growth.append(
+            f"{fmt_delta(round(rate))}/h over {WINDOW_LABELS[window]}"
+            f" ({_samples_label(len(window_entries))})"
+        )
+    lines.append(_fit("            " + " | ".join(growth), width))
+    lines.append("")
+
+    lines.append(_fit(f" Images     {fmt_int(images)}", width))
+
+    if images is None:
+        images_detail = "not counted yet"
+    else:
+        kind = "exact" if images_exact else "estimated"
+        images_detail = (
+            f"{kind} | counted {images_ts:%H:%M} | exact every 2 h"
+            f" | {next_exact_text(entries, now)}"
+        )
+    lines.append(_fit("            " + images_detail, width))
+    lines.append("")
+
+    values = [entry["count"] for entry in window_entries]
+    low, high = min(values), max(values)
+    lines.append(_fit(
+        f" Pages | {WINDOW_LABELS[window]}{window_note}"
+        f" | {_samples_label(len(window_entries))}",
+        width,
+    ))
+
+    range_text = f"  ({fmt_int(low)} - {fmt_int(high)})"
+    spark_width = max(8, min(64, width - len(range_text) - 2))
+    spark = sparkline(values, spark_width, blocks)
+    lines.append(_fit(" " + spark + range_text, width))
+
+    delta_by_id = _deltas(entries)
+    footer = ["", _fit(
+        f" q quit | r reload | t window ({window}) | file: {source}",
+        width,
+    )]
+
+    if rows is None:
+        limit = TABLE_ROWS_ONCE
+    else:
+        # 3 Zeilen Overhead: Leerzeile, "Recent samples", Tabellenkopf
+        limit = max(0, rows - len(lines) - len(footer) - 3)
+
+    if limit > 0:
+        header, table = _build_table(
+            window_entries, delta_by_id, now, width, limit
+        )
+        lines += ["", _fit(" Recent samples", width), _fit(header, width)]
+        lines += [_fit(row, width) for row in table]
+
+    lines += footer
+
+    if rows is not None and len(lines) > rows:
+        # Sehr kleines Terminal: Kerninfo kürzen, Footer unten halten.
+        lines = lines[:max(0, rows - 1)] + [lines[-1]]
+
+    return lines
+
+
+# --------------------------------------------------------------------------
+# Interaktiv / einmalig
+# --------------------------------------------------------------------------
+
+def _blocks_for(args):
+    if args.ascii:
+        return ASCII_BLOCKS
+
+    encoding = (getattr(sys.stdout, "encoding", None) or "").lower()
+    return SPARK_BLOCKS if "utf" in encoding else ASCII_BLOCKS
+
+
+def _render(args, entries, now, width, rows=None):
+    return render_dashboard(
+        entries,
+        window=args.window,
+        now=now,
+        width=width,
+        rows=rows,
+        source=args.file,
+        blocks=_blocks_for(args),
+    )
+
+
+def emit_once(args, stream=None, errors=None):
+    """Momentaufnahme ausgeben (kein curses). Rückgabe: Exit-Code."""
+    stream = stream or sys.stdout
+    errors = errors or sys.stderr
+
+    try:
+        entries = load_entries(args.file)
+    except StatsDataError as exc:
+        errors.write(f"error: {exc}\n")
+        return 1
+
+    width = args.width or shutil.get_terminal_size((80, 24)).columns
+    lines = _render(args, entries, datetime.now(), width)
+    stream.write("\n".join(lines) + "\n")
+    return 0
+
+
+def next_window(window):
+    try:
+        index = WINDOW_ORDER.index(window)
+    except ValueError:
+        index = 0
+    return WINDOW_ORDER[(index + 1) % len(WINDOW_ORDER)]
+
+
+def run_interactive(args):
+    try:
+        curses.wrapper(_interactive_loop, args)
+    except KeyboardInterrupt:
+        pass
+    except curses.error as exc:
+        print(
+            f"Terminal-Dashboard nicht möglich ({exc}); "
+            f"gebe Momentaufnahme aus.",
+            file=sys.stderr,
+        )
+        return emit_once(args)
+
+    return 0
+
+
+def _interactive_loop(stdscr, args):
+    try:
+        curses.curs_set(0)
+    except curses.error:
+        pass
+
+    stdscr.timeout(max(50, int(args.interval * 1000)))
+
+    entries = []
+    error = None
+    loaded_at = 0.0
+    window = args.window
+
+    while True:
+        monotonic_now = time.monotonic()
+
+        if monotonic_now - loaded_at >= args.interval:
+            try:
+                entries = load_entries(args.file)
+                error = None
+            except StatsDataError as exc:
+                error = str(exc)
+            loaded_at = monotonic_now
+
+        height, width = stdscr.getmaxyx()
+        lines = render_dashboard(
+            entries,
+            window=window,
+            now=datetime.now(),
+            width=width,
+            rows=height,
+            source=args.file,
+            error=error,
+            blocks=_blocks_for(args),
+        )
+
+        stdscr.erase()
+        for index, line in enumerate(lines[:height]):
+            attributes = curses.A_BOLD if index == 0 else 0
+            try:
+                stdscr.addnstr(index, 0, line, max(0, width - 1), attributes)
+            except curses.error:
+                pass
+        stdscr.refresh()
+
+        key = stdscr.getch()
+
+        if key in (ord("q"), 27):
+            return
+        if key == ord("r"):
+            loaded_at = 0.0
+        elif key == ord("t"):
+            window = next_window(window)
+
+
+def main(argv=None):
+    parser = argparse.ArgumentParser(
+        description=(
+            "Terminal-Dashboard für den Statserver "
+            "(liest web/page_counts.json)."
+        ),
+    )
+    parser.add_argument(
+        "--file",
+        default=str(DB_DATA),
+        help=f"Statistikdatei (Default: {DB_DATA})",
+    )
+    parser.add_argument(
+        "--window",
+        choices=WINDOW_ORDER,
+        default=DEFAULT_WINDOW,
+        help="Zeitfenster für Verlauf und Tabelle (Default: 24h)",
+    )
+    parser.add_argument(
+        "--once",
+        action="store_true",
+        help="nur einmal ausgeben und beenden",
+    )
+    parser.add_argument(
+        "--width",
+        type=int,
+        default=None,
+        help="Breite für --once (Default: Terminalbreite)",
+    )
+    parser.add_argument(
+        "--interval",
+        type=float,
+        default=REFRESH_SECONDS,
+        help=f"Reload-Intervall in Sekunden (Default: {REFRESH_SECONDS})",
+    )
+    parser.add_argument(
+        "--ascii",
+        action="store_true",
+        help="ASCII-Sparkline statt Unicode-Blöcke",
+    )
+
+    args = parser.parse_args(argv)
+
+    try:
+        locale.setlocale(locale.LC_ALL, "")
+    except locale.Error:
+        pass
+
+    if args.once or not (sys.stdin.isatty() and sys.stdout.isatty()):
+        return emit_once(args)
+
+    return run_interactive(args)
+
+
+if __name__ == "__main__":
+    sys.exit(main())