add stuff

.env.example

@@ -22,6 +22,11 @@ JONASSEARCH_DELAY=3.0
 #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
@@ -47,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=3
+JONASSEARCH_CRAWLER_VERSION=5
 
 # Wie oft jeder Crawler außerhalb der Transaktionen nach einer neuen
 # Deployment-Generation sieht (Sekunden). Der Check setzt nur ein
@@ -107,6 +112,14 @@ JONASSEARCH_API_PORT=8083
 #JONASSEARCH_STATEMENT_TIMEOUT_MS=10000
 #JONASSEARCH_COUNT_TIMEOUT_MS=4000
 
-# Paginierung
+# 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

@@ -6,13 +6,52 @@ das komplette serverjonas search
 - `launcher.py` – startet/überwacht den Crawler und rollt Updates aus (siehe unten)
 - `search/` – Suchmaschine darüber (CLI + HTTP-API + Weboberfläche, Port 8083),
   siehe [search/README.md](search/README.md)
-- `web/` – Statistik-Webseite (Port 8081, `jonassearch-stats.service`)
+- `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
 
@@ -47,14 +86,22 @@ Fetch übersprungen und als `done` markiert.
 Der Crawler wartet nie auf einer langsamen oder blockierten Domain:
 
 1. **Claim mit Host-Check:** Beim Holen des nächsten Queue-Eintrags prüft der
-   Worker pro Kandidat (Prioritätsreihenfolge, bis zu
-   `JONASSEARCH_CLAIM_SCAN_LIMIT` Kandidaten), ob der Host gerade crawlbar ist
-   (Per-Host-Delay aus robots.txt bzw. `JONASSEARCH_DELAY`, aktive
-   Backoff-Sperre). Nicht crawlbare Hosts werden übersprungen – der Worker
-   arbeitet automatisch auf einer anderen Domain weiter und kehrt zurück,
-   sobald der Host wieder darf. Ist gerade gar kein Host crawlbar, schläft
-   der Worker bis zum frühesten freigegebenen Host (gekappet bei
-   `JONASSEARCH_MAX_IDLE_SLEEP`, dann wird neu gescannt).
+   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
@@ -67,6 +114,36 @@ Der Crawler wartet nie auf einer langsamen oder blockierten Domain:
    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

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 '{}';

crawler.py

@@ -7,7 +7,11 @@ PostgreSQL-based parallel crawler.
 Features:
 - PostgreSQL queue with row-level locking (FOR UPDATE SKIP LOCKED)
 - Multiple crawler workers can safely share one queue
-- Stores title, headings, cleaned text and links
+- Stores title, headings, cleaned text, links and image URLs
+- Image URLs from <img src/srcset>, <picture>/<source>, og:image,
+  twitter:image and link rel="image_src" are resolved to absolute
+  http(s) URLs, deduplicated and stored in pages.pictures
+  (migration: add_pages_pictures.sql)
 - Search vector is maintained by the PostgreSQL trigger
 - URL deduplication via UNIQUE(url)
 - Deadlock-safe parallel queue inserts (deterministic ordering,
@@ -37,7 +41,7 @@ Features:
   release) and --update-deployment N (bump generation + required_version)
 
 Expected PostgreSQL schema:
-    pages
+    pages             (incl. pictures text[], see add_pages_pictures.sql)
     crawl_queue
     crawler_stats
 
@@ -150,7 +154,7 @@ USER_AGENT = os.getenv(
 # change; completely independent from the deployment generation (the
 # generation is a global deployment counter in PostgreSQL, the version
 # identifies the crawler code).
-CRAWLER_VERSION = int(os.getenv("JONASSEARCH_CRAWLER_VERSION", "3"))
+CRAWLER_VERSION = int(os.getenv("JONASSEARCH_CRAWLER_VERSION", "5"))
 
 # Exit codes. 42 is reserved exclusively for "I must be updated" so
 # the launcher can distinguish update, crash and normal exit.
@@ -235,6 +239,15 @@ MAX_HEADING_LENGTH = 1000
 MAX_TEXT_LENGTH = 2_000_000
 MAX_LINKS = 10_000
 
+# Image URLs per page. A page with thousands of <img>/srcset entries
+# would otherwise bloat pages.pictures for no search value; the first
+# images in document order are kept.
+MAX_PICTURES = 1_000
+
+# Hard cap for one image URL (text[] has no B-tree limit, but absurd
+# data: URLs / tracking parameters are not worth storing).
+MAX_IMAGE_URL_LENGTH = 2_000
+
 # Cap for the text that feeds the full-text index. PostgreSQL
 # hard-limits a single tsvector value to 1,048,575 bytes (not
 # configurable), and the trigger pages_search_vector_update() builds
@@ -281,11 +294,23 @@ QUEUE_CLAIM_BATCH = int(os.getenv("JONASSEARCH_CLAIM_BATCH", "25"))
 
 # Maximum number of candidates examined per claim (batch * rounds).
 # Keeps the common case (a ready host in the first batch) cheap while
-# still finding crawlable hosts behind a wall of blocked ones.
+# still finding crawlable hosts behind a small wall of blocked ones.
+# Larger walls are handled in SQL: hosts that are known to be busy are
+# excluded from the candidate query itself (see busy_hosts()), so a
+# blocked host with thousands of queued URLs cannot hide the rest of
+# the queue behind it.
 QUEUE_CLAIM_SCAN_LIMIT = int(
     os.getenv("JONASSEARCH_CLAIM_SCAN_LIMIT", "200")
 )
 
+# Upper bound for the host list that is passed into the claim query.
+# Longest-running blocks are kept first, because those form the wall at
+# the head of the priority order; hosts beyond the cap are still
+# skipped by the per-candidate check, they are just not pre-filtered.
+QUEUE_CLAIM_BUSY_HOST_LIMIT = int(
+    os.getenv("JONASSEARCH_CLAIM_BUSY_HOST_LIMIT", "1000")
+)
+
 # Cap for the "nothing is crawlable right now" sleep, so a host blocked
 # for a long time cannot park the whole worker; after the cap the queue
 # is simply re-scanned.
@@ -738,11 +763,18 @@ def claim_next_queue_item(conn, scheduler, robots):
     Candidates are read in priority order (same ORDER BY as before) in
     small batches with FOR UPDATE SKIP LOCKED, so several crawler
     processes can still share one queue without waiting for each
-    other. For every candidate the HostScheduler decides whether its
-    host may be contacted right now (per-host crawl delay, active
-    block backoff). The first ready candidate is claimed; hosts that
-    would have to wait are skipped, so the worker automatically works
-    on another domain while a rate-limited or blocked host cools down.
+    other. Hosts that the HostScheduler already knows to be busy
+    (per-host crawl delay, active block backoff) are excluded by the
+    query itself. For the remaining candidates the scheduler decides
+    whether a host may be contacted right now. The first ready
+    candidate is claimed, so the worker automatically works on another
+    domain while a rate-limited or blocked host cools down.
+
+    The SQL-level host filter is what keeps a blocked host from
+    starving the whole queue: blocked rows stay 'pending' at the top
+    of the priority order, and without the filter a wall of them could
+    fill every scan window forever while tens of thousands of other
+    hosts wait behind it.
 
     Returns (item, wait_hint):
       item      (id, url, source_url) of the claimed URL, or None
@@ -757,27 +789,28 @@ def claim_next_queue_item(conn, scheduler, robots):
     wait_hint = None
     item = None
 
+    busy_hosts = scheduler.busy_hosts(now)
+    # Longest waits first: those hosts form the wall at the head of the
+    # priority order and are the most valuable ones to exclude.
+    excluded_hosts = [
+        host for host, _wait in sorted(
+            busy_hosts.items(), key=lambda pair: pair[1], reverse=True
+        )
+    ][:QUEUE_CLAIM_BUSY_HOST_LIMIT]
+
     while item is None and scanned < QUEUE_CLAIM_SCAN_LIMIT:
         with conn.cursor() as cur:
-            if seen_ids:
-                cur.execute("""
-                    SELECT id, url, source_url
-                    FROM crawl_queue
-                    WHERE status = 'pending'
-                      AND NOT (id = ANY(%s))
-                    ORDER BY priority DESC, discovered_at ASC, id ASC
-                    FOR UPDATE SKIP LOCKED
-                    LIMIT %s
-                """, (seen_ids, QUEUE_CLAIM_BATCH))
-            else:
-                cur.execute("""
-                    SELECT id, url, source_url
-                    FROM crawl_queue
-                    WHERE status = 'pending'
-                    ORDER BY priority DESC, discovered_at ASC, id ASC
-                    FOR UPDATE SKIP LOCKED
-                    LIMIT %s
-                """, (QUEUE_CLAIM_BATCH,))
+            cur.execute("""
+                SELECT id, url, source_url
+                FROM crawl_queue
+                WHERE status = 'pending'
+                  AND NOT (id = ANY(%s::bigint[]))
+                  AND lower(split_part(split_part(url, '://', 2), '/', 1))
+                      <> ALL(%s::text[])
+                ORDER BY priority DESC, discovered_at ASC, id ASC
+                FOR UPDATE SKIP LOCKED
+                LIMIT %s
+            """, (seen_ids, excluded_hosts, QUEUE_CLAIM_BATCH))
             candidates = cur.fetchall()
 
         if not candidates:
@@ -815,6 +848,18 @@ def claim_next_queue_item(conn, scheduler, robots):
         if len(candidates) < QUEUE_CLAIM_BATCH:
             break
 
+    if item is None and scanned == 0 and excluded_hosts:
+        # The busy-host filter removed every pending URL from the query
+        # before it could return a single candidate. Report when the
+        # soonest host is crawlable again - but only if there really are
+        # pending URLs; otherwise the queue is simply empty.
+        with conn.cursor() as cur:
+            cur.execute(
+                "SELECT 1 FROM crawl_queue WHERE status = 'pending' LIMIT 1"
+            )
+            if cur.fetchone() is not None:
+                wait_hint = min(busy_hosts.values())
+
     return item, wait_hint
 
 
@@ -1584,6 +1629,48 @@ def normalize_url(url, base_url=None):
     return normalized
 
 
+def normalize_image_url(raw, base_url):
+    """
+    Turn an image reference from HTML into an absolute http(s) URL.
+
+    Unlike normalize_url() this does NOT reject image file extensions
+    (that list exists to keep binary resources out of the page queue).
+    Relative paths and protocol-relative URLs are resolved against the
+    page URL; fragments are dropped because they do not identify a
+    different image resource. data:, javascript:, ftp: and other
+    non-http(s) sources are rejected, as are empty/whitespace values,
+    non-public hosts and URLs that are unreasonably long.
+    """
+    if not isinstance(raw, str):
+        return None
+
+    raw = unescape(raw).strip()
+    if not raw:
+        return None
+
+    try:
+        absolute = urljoin(base_url, raw)
+        parsed = urlparse(absolute)
+    except ValueError:
+        return None
+
+    if parsed.scheme.lower() not in ALLOWED_SCHEMES:
+        return None
+    if not parsed.hostname:
+        return None
+    if is_non_crawlable_host(parsed.hostname):
+        return None
+
+    # Fragments never select a different image file.
+    clean, _fragment = urldefrag(absolute)
+    if any(char.isspace() for char in clean):
+        return None
+    if len(clean) > MAX_IMAGE_URL_LENGTH:
+        return None
+
+    return clean
+
+
 # ============================================================
 # ROBOTS
 # ============================================================
@@ -1688,6 +1775,10 @@ class HostScheduler:
         self.last_request = {}
         self.blocked_until = {}
         self.block_count = {}
+        # Delay last used for a host (robots.txt crawl-delay or
+        # DEFAULT_DELAY), so busy_hosts() can report the host's schedule
+        # without looking at any URL again.
+        self.host_delay = {}
 
     @staticmethod
     def _host(url):
@@ -1703,12 +1794,54 @@ class HostScheduler:
         host = self._host(url)
         if not host:
             return 0.0
+        self.host_delay[host] = delay
+        last = self.last_request.get(host)
+        # A host that was never contacted has no pending crawl delay:
+        # monotonic() has an arbitrary origin, so treating "never" as 0
+        # and adding a large robots delay would wrongly block the host
+        # for a long time on a freshly booted machine.
+        rate_until = last + delay if last is not None else 0.0
         next_allowed = max(
-            self.last_request.get(host, 0.0) + delay,
+            rate_until,
             self.blocked_until.get(host, 0.0),
         )
         return max(0.0, next_allowed - now)
 
+    def busy_hosts(self, now=None):
+        """
+        Map of hosts that cannot be contacted right now to the number
+        of seconds they still have to wait.
+
+        The queue claim passes these hosts into SQL: skipping busy hosts
+        only within the small candidate window would let a wall of
+        blocked URLs at the head of the priority order hide every other
+        crawlable host in the queue (the wall stays pending at the top,
+        so each rescan looks at the very same rows forever).
+
+        Expired entries are pruned here because this runs once per
+        claim - the per-process state then cannot grow without bound on
+        a queue with millions of distinct hosts.
+        """
+        if now is None:
+            now = time.monotonic()
+
+        busy = {}
+
+        for host, last in list(self.last_request.items()):
+            next_allowed = last + self.host_delay.get(host, 0.0)
+            if next_allowed > now:
+                busy[host] = next_allowed - now
+            else:
+                del self.last_request[host]
+
+        for host, until in list(self.blocked_until.items()):
+            if until > now:
+                busy[host] = max(busy.get(host, 0.0), until - now)
+            else:
+                del self.blocked_until[host]
+
+        return busy
+
     def is_ready(self, url, delay, now=None):
         return self.wait_time(url, delay, now) <= 0.0
 
@@ -1725,14 +1858,16 @@ class HostScheduler:
         stays untouched, so the item is retried exactly when the delay
         has passed.
         """
+        host = self._host(url)
+        self.host_delay[host] = delay
         wait_for = self.wait_time(url, delay)
         if wait_for > MAX_INLINE_RATE_WAIT:
             return False
         if wait_for > 0:
             if wait_for >= 1.0:
-                logging.info("[RATE ] %s: %.1fs", self._host(url), wait_for)
+                logging.info("[RATE ] %s: %.1fs", host, wait_for)
             time.sleep(wait_for)
-        self.last_request[self._host(url)] = time.monotonic()
+        self.last_request[host] = time.monotonic()
         return True
 
     def _register_block(self, host, seconds):
@@ -1926,6 +2061,104 @@ def clean_text_bs4(soup):
     return text[:MAX_TEXT_LENGTH]
 
 
+# Common attributes used for lazy-loaded images. They are not
+# standardized, but these three are reliable enough in practice.
+IMAGE_LAZY_SRC_ATTRS = ("data-src", "data-lazy-src", "data-original")
+
+# srcset appears on <img> and on <source> inside <picture>.
+IMAGE_SRCSET_ATTRS = ("srcset", "data-srcset")
+
+# Social/social-preview image metadata.
+IMAGE_META_KEYS = frozenset({
+    "og:image",
+    "og:image:url",
+    "og:image:secure_url",
+    "twitter:image",
+    "twitter:image:src",
+})
+
+
+def _srcset_candidates(value):
+    """
+    Yield the URL part of every srcset candidate.
+
+    srcset format: "img.jpg 1x, img@2x.jpg 2x" (or a width descriptor).
+    Parsed like the HTML spec: whitespace separates URL and descriptor,
+    a trailing comma ends a candidate. This keeps commas inside URLs
+    (including data: URLs, which are rejected later) from being split
+    into bogus relative candidates.
+    """
+    if not value:
+        return
+
+    tokens = str(value).split()
+    index = 0
+    while index < len(tokens):
+        candidate = tokens[index]
+        index += 1
+        if not candidate:
+            continue
+
+        ends_candidate = candidate.endswith(",")
+        candidate = candidate.rstrip(",").strip()
+        if not candidate:
+            continue
+
+        if not ends_candidate:
+            # The next token is the descriptor ("2x", "640w"), if any.
+            index += 1
+
+        yield candidate
+
+
+def extract_pictures(soup, base_url):
+    """
+    All absolute, deduplicated image URLs of a parsed document.
+
+    Sources: <img src>, lazy-loading data-* attributes, <img srcset>,
+    <picture> via <source srcset>, Open-Graph/Twitter preview images and
+    <link rel="image_src">. Order of first appearance is preserved.
+    """
+    pictures = []
+    seen = set()
+
+    def add(raw):
+        if len(pictures) >= MAX_PICTURES:
+            return
+        normalized = normalize_image_url(raw, base_url)
+        if not normalized or normalized in seen:
+            return
+        seen.add(normalized)
+        pictures.append(normalized)
+
+    for tag in soup.find_all("img"):
+        add(tag.get("src"))
+        for attr in IMAGE_LAZY_SRC_ATTRS:
+            add(tag.get(attr))
+        for attr in IMAGE_SRCSET_ATTRS:
+            for candidate in _srcset_candidates(tag.get(attr)):
+                add(candidate)
+
+    for tag in soup.find_all("source"):
+        for attr in IMAGE_SRCSET_ATTRS:
+            for candidate in _srcset_candidates(tag.get(attr)):
+                add(candidate)
+
+    for tag in soup.find_all("meta"):
+        key = (tag.get("property") or tag.get("name") or "").strip().lower()
+        if key in IMAGE_META_KEYS:
+            add(tag.get("content"))
+
+    for tag in soup.find_all("link"):
+        rel = tag.get("rel") or []
+        if isinstance(rel, str):
+            rel = rel.lower().split()
+        if any(str(value).lower() == "image_src" for value in rel):
+            add(tag.get("href"))
+
+    return pictures
+
+
 def parse_html(html, base_url):
     if BeautifulSoup is None:
         raise RuntimeError(
@@ -1965,6 +2198,10 @@ def parse_html(html, base_url):
         if len(links) >= MAX_LINKS:
             break
 
+    # Extract images BEFORE clean_text_bs4() mutates the tree (it
+    # removes navigation, noscript, svg, ... elements).
+    pictures = extract_pictures(soup, base_url)
+
     content = clean_text_bs4(soup)
 
     domain = urlparse(base_url).hostname or ""
@@ -1981,6 +2218,7 @@ def parse_html(html, base_url):
         "headings": headings,
         "content": content,
         "links": links,
+        "pictures": pictures,
         "domain": domain,
         "tld": tld,
     }
@@ -2006,6 +2244,11 @@ def save_page(conn, url, source_url, status, content_type, parsed, crawl_ms):
         "headings": clean_text_list(parsed["headings"]),
         "content": clean_text(parsed["content"]),
         "links": clean_text_list(parsed["links"]),
+        # parse_html() already deduplicates; repeat it here so a caller
+        # with a hand-built parsed dict can never write duplicates.
+        "pictures": list(dict.fromkeys(
+            clean_text_list(parsed.get("pictures") or [])
+        ))[:MAX_PICTURES],
         "domain": clean_text(parsed["domain"]),
         "tld": clean_text(parsed["tld"]),
     }
@@ -2037,6 +2280,7 @@ def save_page(conn, url, source_url, status, content_type, parsed, crawl_ms):
                 headings,
                 content,
                 links,
+                pictures,
                 domain,
                 tld,
                 source_url,
@@ -2047,7 +2291,7 @@ def save_page(conn, url, source_url, status, content_type, parsed, crawl_ms):
                 crawl_time_ms
             )
             VALUES (
-                %s, %s, %s, %s, %s, %s, %s, %s, %s, %s,
+                %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s,
                 COALESCE(
                     (SELECT discovered_at
                      FROM crawl_queue
@@ -2062,6 +2306,7 @@ def save_page(conn, url, source_url, status, content_type, parsed, crawl_ms):
                 headings = EXCLUDED.headings,
                 content = EXCLUDED.content,
                 links = EXCLUDED.links,
+                pictures = EXCLUDED.pictures,
                 domain = EXCLUDED.domain,
                 tld = EXCLUDED.tld,
                 source_url = EXCLUDED.source_url,
@@ -2075,6 +2320,7 @@ def save_page(conn, url, source_url, status, content_type, parsed, crawl_ms):
             safe_parsed["headings"],
             index_content,
             safe_parsed["links"],
+            safe_parsed["pictures"],
             safe_parsed["domain"],
             safe_parsed["tld"],
             source_url,
@@ -2149,7 +2395,8 @@ def worker(worker_id):
                     sleep_for = min(wait_hint, QUEUE_MAX_IDLE_SLEEP)
                     logging.info(
                         "[WAIT ] All candidate hosts busy or blocked; "
-                        "worker sleeps %.1fs",
+                        "soonest host free in %.1fs, worker sleeps %.1fs",
+                        wait_hint,
                         sleep_for,
                     )
                 else:
@@ -2269,6 +2516,7 @@ def worker(worker_id):
             logging.info("[DOMAIN] %s", parsed["domain"])
             logging.info("[HEADS ] %d", len(parsed["headings"]))
             logging.info("[LINKS ] %d gefunden", len(parsed["links"]))
+            logging.info("[IMGS  ] %d gefunden", len(parsed["pictures"]))
 
             try:
                 safe_commit(

search/README.md

@@ -3,7 +3,7 @@
 Die eigentliche Websuchmaschine über der PostgreSQL-Datenbank `jonassearch`:
 Volltextsuche auf dem vorhandenen `tsvector`-Index (`pages.search_vector`,
 GIN-Index `idx_pages_search`) mit mehrstufigem Ranking, Filtern, Snippets,
-Weboberfläche, HTTP-API und Kommandozeile.
+Bildersuche (Tab „Bilder“), Weboberfläche, HTTP-API und Kommandozeile.
 
 | Baustein | Zweck | Port |
 |---|---|---|
@@ -53,6 +53,26 @@ Bestehende Dienste: Statserver 8081 (`web/app.py`), Vorschlagsverfeinerer
    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`
@@ -70,6 +90,7 @@ Bestehende Dienste: Statserver 8081 (`web/app.py`), Vorschlagsverfeinerer
 |---|---|
 | `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) |
@@ -117,6 +138,27 @@ Antwort (Auszug):
 }
 ```
 
+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).
 
@@ -165,14 +207,19 @@ abweichenden Wert aus `.env` – der Dienst läuft garantiert auf 8083.
 | `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, Query-Parsing. Alles ohne
-Datenbankzugriff.
+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/api.py

@@ -9,6 +9,8 @@ Endpunkte:
     GET /api/search     Volltextsuche (q, page, per_page, site, tld,
                         after, before, sort, include_errors,
                         include_non_html, snippets)
+    GET /api/images     Bildersuche (q, page, per_page, site, tld,
+                        after, before, sort)
     GET /api/suggest    Autocomplete (q, limit)
     GET /api/page       einzelne Seite (id oder url)
     GET /api/stats      Index-Statistiken
@@ -124,6 +126,41 @@ def api_search():
         return jsonify({"error": "Interner Fehler."}), 500
 
 
+@app.route("/api/images")
+def api_images():
+    try:
+        q = request.args.get("q", "").strip()
+        if not q:
+            raise engine.QueryError(
+                "Parameter 'q' (Suchbegriffe) fehlt oder ist leer."
+            )
+        result = engine.search_images(
+            query=q,
+            page=_page_param("page", 1),
+            per_page=_page_param("per_page", None),
+            sites=request.args.getlist("site") + request.args.getlist("sites"),
+            tlds=request.args.getlist("tld") + request.args.getlist("tlds"),
+            after=request.args.get("after"),
+            before=request.args.get("before"),
+            sort=request.args.get("sort", "relevance"),
+        )
+        return jsonify(result)
+    except engine.QueryError as exc:
+        return jsonify({"error": str(exc)}), 400
+    except engine.SearchTimeout as exc:
+        return jsonify({"error": str(exc)}), 504
+    except psycopg2.OperationalError as exc:
+        log.error("Datenbank nicht erreichbar: %s", exc)
+        return jsonify({"error": "Datenbank nicht erreichbar."}), 503
+    except engine.SearchEngineError as exc:
+        log.error("Bilder-Suchfehler: %s", exc)
+        return jsonify({"error": str(exc)}), 500
+    except Exception:
+        log.error("Unerwarteter Fehler in /api/images:\n%s",
+                  traceback.format_exc())
+        return jsonify({"error": "Interner Fehler."}), 500
+
+
 @app.route("/api/suggest")
 def api_suggest():
     try:
@@ -250,6 +287,28 @@ button:disabled { opacity: .45; cursor: default; filter: none; }
     font-size: .82rem; padding: 7px 14px; outline: none;
 }
 .filters input:focus, .filters select:focus { border-color: var(--accent); }
+.tabs { display: flex; gap: 4px; margin: 18px 0 14px;
+        border-bottom: 1px solid var(--border); }
+.tabs button { background: none; color: var(--muted); border: none;
+              border-bottom: 2px solid transparent; border-radius: 0;
+              padding: 9px 16px; font-size: .95rem; font-weight: 600; }
+.tabs button:hover { color: var(--text); filter: none; }
+.tabs button.active { color: var(--accent);
+                      border-bottom-color: var(--accent); }
+.img-grid { display: grid; gap: 12px;
+            grid-template-columns: repeat(auto-fill, minmax(150px, 1fr)); }
+.img-card { position: relative; display: block; aspect-ratio: 1 / 1;
+            overflow: hidden; border-radius: 12px; background: var(--card);
+            border: 1px solid var(--border); text-decoration: none;
+            color: var(--text); }
+.img-card img { width: 100%; height: 100%; object-fit: cover;
+               display: block; background: var(--card); }
+.img-card .caption { position: absolute; left: 0; right: 0; bottom: 0;
+                     padding: 22px 10px 8px; font-size: .74rem;
+                     line-height: 1.3; color: #fff; white-space: nowrap;
+                     overflow: hidden; text-overflow: ellipsis;
+                     background: linear-gradient(transparent,
+                                 rgba(0, 0, 0, .8)); }
 .suggest {
     position: absolute; top: 54px; left: 0; right: 90px; z-index: 20;
     background: var(--card); border: 1px solid var(--border);
@@ -294,6 +353,8 @@ footer { color: var(--muted); font-size: .76rem; margin-top: 60px;
     .wrap { padding: 24px 12px 40px; }
     .searchbar { flex-direction: column; }
     .suggest { right: 0; top: 106px; }
+    .img-grid { grid-template-columns: repeat(auto-fill, minmax(110px, 1fr));
+                gap: 8px; }
 }
 </style>
 </head>
@@ -326,6 +387,11 @@ footer { color: var(--muted); font-size: .76rem; margin-top: 60px;
     </div>
   </form>
 
+  <div class="tabs" role="tablist">
+    <button type="button" id="tab-web" class="active" role="tab">Web</button>
+    <button type="button" id="tab-images" role="tab">Bilder</button>
+  </div>
+
   <div id="info" class="info"></div>
   <div id="results"></div>
 
@@ -349,7 +415,8 @@ footer { color: var(--muted); font-size: .76rem; margin-top: 60px;
 <script>
 'use strict';
 const $ = id => document.getElementById(id);
-const state = { q: "", site: "", range: "", sort: "relevance", page: 1 };
+const state = { q: "", site: "", range: "", sort: "relevance",
+                page: 1, view: "web" };
 let suggestTimer = null;
 
 function esc(s) {
@@ -378,10 +445,16 @@ function fromHash() {
     state.range = p.get("after") || "";
     state.sort = p.get("sort") || "relevance";
     state.page = Math.max(1, parseInt(p.get("page") || "1", 10) || 1);
+    state.view = p.get("view") === "images" ? "images" : "web";
     $("q").value = state.q;
     $("site").value = state.site;
     $("range").value = state.range;
     $("sort").value = state.sort;
+    updateTabs();
+}
+function updateTabs() {
+    $("tab-web").classList.toggle("active", state.view !== "images");
+    $("tab-images").classList.toggle("active", state.view === "images");
 }
 function toHash() {
     const p = new URLSearchParams();
@@ -389,6 +462,7 @@ function toHash() {
     if (state.site) p.set("site", state.site);
     if (state.range) p.set("after", state.range);
     if (state.sort && state.sort !== "relevance") p.set("sort", state.sort);
+    if (state.view === "images") p.set("view", "images");
     if (state.page > 1) p.set("page", String(state.page));
     history.replaceState(null, "", p.toString() ? "#" + p.toString() : "#");
 }
@@ -412,9 +486,12 @@ async function run() {
     }
     $("hint").style.display = "none";
     $("info").className = "info";
-    $("info").textContent = "Suche …";
+    $("info").textContent = state.view === "images"
+        ? "Suche Bilder …" : "Suche …";
     $("results").innerHTML = "";
 
+    if (state.view === "images") { await runImages(); return; }
+
     const p = new URLSearchParams({ q: state.q, page: state.page });
     if (state.site) p.set("site", state.site);
     if (state.range) p.set("after", state.range);
@@ -476,6 +553,69 @@ async function run() {
     $("next").disabled = !data.has_more;
 }
 
+async function runImages() {
+    const p = new URLSearchParams({ q: state.q, page: state.page });
+    if (state.site) p.set("site", state.site);
+    if (state.range) p.set("after", state.range);
+    if (state.sort) p.set("sort", state.sort);
+
+    let data;
+    try {
+        const res = await fetch("/api/images?" + p.toString());
+        data = await res.json();
+        if (!res.ok) {
+            $("info").className = "info error";
+            $("info").textContent = data.error || ("Fehler " + res.status);
+            return;
+        }
+    } catch (e) {
+        $("info").className = "info error";
+        $("info").textContent = "Suchdienst nicht erreichbar.";
+        return;
+    }
+
+    const totalTxt = data.total != null
+        ? "Etwa " + fmtInt(data.total) + " Bilder"
+        : "Bilder";
+    const capHint = data.ranking_capped
+        ? " · Ranking auf Top-Kandidaten eingegrenzt" : "";
+    $("info").textContent = totalTxt + " · Seite " + data.page +
+        " · " + (data.took_ms / 1000).toFixed(2) + " s" + capHint;
+
+    if (!data.results.length) {
+        $("results").innerHTML =
+            '<div class="info">Keine Bilder zu diesen Begriffen gefunden. ' +
+            'Andere Begriffe oder weniger Filter probieren.</div>';
+        $("pager").style.display = "none";
+        return;
+    }
+
+    $("results").innerHTML = '<div class="img-grid">' +
+        data.results.map(r => {
+            const label = r.title || r.domain || prettyUrl(r.page_url);
+            return '<a class="img-card" href="' + esc(r.page_url) + '" ' +
+                'target="_blank" rel="noopener" title="' + esc(label) +
+                '"><img src="' + esc(r.image_url) + '" alt="' + esc(label) +
+                '" loading="lazy" decoding="async" ' +
+                'referrerpolicy="no-referrer"><span class="caption">' +
+                esc(label) + '</span></a>';
+        }).join("") + '</div>';
+
+    // Fehlende/blockierte Bilder dürfen die Ergebnisseite nicht
+    // beeinträchtigen: defekte Kacheln werden entfernt.
+    $("results").querySelectorAll(".img-card img").forEach(img => {
+        img.addEventListener("error", () => {
+            const card = img.closest(".img-card");
+            if (card) card.remove();
+        });
+    });
+
+    $("pager").style.display = "";
+    $("pageinfo").textContent = "Seite " + data.page;
+    $("prev").disabled = data.page <= 1;
+    $("next").disabled = !data.has_more;
+}
+
 /* ---------- Vorschläge (Autocomplete) ---------- */
 function hideSuggest() { $("suggest").classList.add("hidden"); }
 $("q").addEventListener("input", () => {
@@ -512,6 +652,16 @@ $("f").addEventListener("submit", e => {
     e.preventDefault();
     readInputs(); toHash(); run();
 });
+function setView(view) {
+    if (state.view === view) return;
+    state.view = view;
+    state.page = 1;
+    updateTabs();
+    toHash();
+    if (state.q) run();
+}
+$("tab-web").addEventListener("click", () => setView("web"));
+$("tab-images").addEventListener("click", () => setView("images"));
 $("prev").addEventListener("click", () => {
     if (state.page > 1) { state.page--; toHash(); run(); }
 });

search/engine.py

@@ -6,7 +6,8 @@ Die eigentliche Suchlogik über der PostgreSQL-Datenbank `jonassearch`.
 
 Retrieval:
     Die Tabelle `public.pages` hält pro Webseite u. a. url, title,
-    headings (text[]), content, domain, tld, http_status, content_type,
+    headings (text[]), content, links (text[]), pictures (text[],
+    absolute Bild-URLs), domain, tld, http_status, content_type,
     discovered_at/crawled_at sowie die per DB-Trigger gepflegte Spalte
     `search_vector` (tsvector, Titel = Gewicht A, Überschriften = B,
     Inhalt = C, Textsuch-Konfiguration 'simple') und den GIN-Index
@@ -144,6 +145,21 @@ HEADLINE_OPTIONS = (
 # Einzelabruf einer Seite
 PAGE_CONTENT_CHARS = int(os.getenv("JONASSEARCH_PAGE_CONTENT_CHARS", "50000"))
 PAGE_MAX_LINKS = 100
+PAGE_MAX_PICTURES = 200
+
+# Bildersuche (Tab "Bilder")
+# Standard-/Maximalzahl der Bildkacheln pro Ergebnisseite; pro
+# Quellseite werden höchstens IMAGES_PER_SOURCE_PAGE Bilder gezeigt,
+# damit eine Seite mit vielen Bildern die Ergebnisse nicht dominiert.
+IMAGE_RESULTS_PER_PAGE = int(
+    os.getenv("JONASSEARCH_IMAGE_RESULTS_PER_PAGE", "40")
+)
+MAX_IMAGE_RESULTS_PER_PAGE = int(
+    os.getenv("JONASSEARCH_MAX_IMAGE_RESULTS_PER_PAGE", "100")
+)
+IMAGES_PER_SOURCE_PAGE = int(
+    os.getenv("JONASSEARCH_IMAGES_PER_SOURCE_PAGE", "4")
+)
 
 # Vorschläge (Autocomplete)
 MAX_SUGGEST = 10
@@ -483,6 +499,20 @@ def _filter_clauses(sites, tlds, after, before, include_errors,
     return sql, params
 
 
+def _fresh_lateral():
+    """Frische-Signal (Aktualitätsbonus) als LATERAL-Join auf pages p."""
+    return """
+        CROSS JOIN LATERAL (
+            SELECT 1.0 + %(fresh_weight)s * exp(
+                       - GREATEST(0, EXTRACT(EPOCH FROM (
+                             now() - COALESCE(p.crawled_at, p.discovered_at)
+                         )))::double precision
+                       / 86400.0 / GREATEST(%(fresh_half_life)s, 0.01)
+                   ) AS fresh
+        ) f
+    """
+
+
 def _search_sql(vector_col, sort, filter_sql, capped=False):
     """
     Ranking-Query: Kandidaten via GIN-Index, bewertet + sortiert.
@@ -502,16 +532,7 @@ def _search_sql(vector_col, sort, filter_sql, capped=False):
         )
 
     # Frische-Signal: braucht p.crawled_at + p.discovered_at.
-    fresh_lateral = """
-        CROSS JOIN LATERAL (
-            SELECT 1.0 + %(fresh_weight)s * exp(
-                       - GREATEST(0, EXTRACT(EPOCH FROM (
-                             now() - COALESCE(p.crawled_at, p.discovered_at)
-                         )))::double precision
-                       / 86400.0 / GREATEST(%(fresh_half_life)s, 0.01)
-                   ) AS fresh
-        ) f
-    """
+    fresh_lateral = _fresh_lateral()
 
     if capped:
         source = """
@@ -620,6 +641,113 @@ def _count_sql(vector_col, filter_sql):
     """
 
 
+def _image_search_sql(vector_col, sort, filter_sql, capped=False):
+    """
+    Bildersuche: dieselben Kandidaten, Filter und dasselbe Ranking wie
+    die Websuche, aber pro Quellseite höchstens IMAGES_PER_SOURCE_PAGE
+    Bilder aus pages.pictures.
+
+    capped=True begrenzt die Kandidatenmenge vor dem Ranking (wie der
+    Fallback der Websuche), damit auch sehr häufige Begriffe antworten.
+    """
+    if sort == "relevance":
+        order = (
+            "(r.text_rank * f.fresh) DESC, r.text_rank DESC, "
+            "p.id ASC, i.ord ASC"
+        )
+    else:
+        order = (
+            "COALESCE(p.crawled_at, p.discovered_at) DESC NULLS LAST, "
+            "p.id ASC, i.ord ASC"
+        )
+
+    # Nur echte externe Bildquellen (Altbestände könnten data:-URLs
+    # enthalten); die Reihenfolge im Array bleibt erhalten.
+    images = f"""
+        INNER JOIN LATERAL (
+            SELECT picture AS image_url, ord
+            FROM unnest(p.pictures) WITH ORDINALITY AS pic(picture, ord)
+            WHERE picture ~* '^https?://'
+            ORDER BY ord
+            LIMIT %(images_per_source)s
+        ) i ON TRUE
+        CROSS JOIN LATERAL (
+            SELECT ts_rank_cd(%(weights)s::real[], p.{vector_col}, q.tsq,
+                              %(norm)s::integer) AS text_rank
+        ) r
+    """
+
+    if capped:
+        source = f"""
+        FROM (
+            SELECT p.id, p.url, p.title, p.domain, p.tld, p.pictures,
+                   p.discovered_at,
+                   COALESCE(p.crawled_at, p.discovered_at) AS crawled_at,
+                   p.{vector_col} AS vector
+            FROM pages p
+            CROSS JOIN LATERAL (
+                SELECT websearch_to_tsquery(%(ts_config)s, %(query)s) AS tsq
+            ) q
+            WHERE p.{vector_col} @@ q.tsq
+              AND p.pictures IS NOT NULL{filter_sql}
+            LIMIT %(rank_cap)s
+        ) p
+        CROSS JOIN LATERAL (
+            SELECT websearch_to_tsquery(%(ts_config)s, %(query)s) AS tsq
+        ) q
+        {images}
+        {_fresh_lateral()}
+        """
+    else:
+        source = f"""
+        FROM pages p
+        CROSS JOIN LATERAL (
+            SELECT websearch_to_tsquery(%(ts_config)s, %(query)s) AS tsq
+        ) q
+        {images}
+        {_fresh_lateral()}
+        WHERE p.{vector_col} @@ q.tsq
+          AND p.pictures IS NOT NULL{filter_sql}
+        """
+
+    return f"""
+        SELECT
+            i.image_url,
+            p.url AS page_url,
+            p.title,
+            p.domain,
+            p.tld,
+            COALESCE(p.crawled_at, p.discovered_at) AS crawled_at,
+            r.text_rank,
+            f.fresh AS freshness,
+            (r.text_rank * f.fresh) AS score
+        {source}
+        ORDER BY {order}
+        LIMIT %(limit)s OFFSET %(offset)s
+    """
+
+
+def _image_count_sql(vector_col, filter_sql):
+    """
+    Geschätzte Bildanzahl: pro Seite wird höchstens die angezeigte
+    Anzahl gezählt (die Abfrage muss die Bilder nicht einzeln zählen).
+    """
+    return f"""
+        SELECT COALESCE(SUM(
+                   LEAST(
+                       COALESCE(cardinality(p.pictures), 0),
+                       %(images_per_source)s
+                   )
+               ), 0)::bigint AS n
+        FROM pages p
+        CROSS JOIN LATERAL (
+            SELECT websearch_to_tsquery(%(ts_config)s, %(query)s) AS tsq
+        ) q
+        WHERE p.{vector_col} @@ q.tsq
+          AND p.pictures IS NOT NULL{filter_sql}
+    """
+
+
 def _iso(dt):
     if isinstance(dt, datetime):
         return dt.isoformat()
@@ -824,6 +952,157 @@ def search(query, *, page=1, per_page=None, sites=None, tlds=None,
         conn.close()
 
 
+def search_images(query, *, page=1, per_page=None, sites=None, tlds=None,
+                  after=None, before=None, sort="relevance"):
+    """
+    Bildersuche über die auf `pages.pictures` gespeicherten URLs.
+
+    Kandidatenseiten und Ranking entsprechen der Websuche (derselbe
+    Suchvektor, dieselben Filter und dieselbe Relevanz-/Frischeformel);
+    pro Quellseite werden höchstens IMAGES_PER_SOURCE_PAGE Bilder
+    geliefert.
+
+    Rückgabe (dict): query, parsed_query, page, per_page, sort, total
+    (geschätzt; None = nicht gezählt/Zeitlimit), has_more, results[],
+    ranking_capped, took_ms. Jedes Result: image_url, page_url, title,
+    domain, tld, crawled_at, score, text_rank, freshness.
+    """
+    q = (query or "").strip()
+    if not q:
+        raise QueryError("Leere Suchanfrage.")
+
+    try:
+        page = max(1, int(page))
+        per_page = (
+            IMAGE_RESULTS_PER_PAGE if per_page is None else int(per_page)
+        )
+    except (TypeError, ValueError):
+        raise QueryError("page/per_page müssen Zahlen sein.")
+    per_page = min(max(1, per_page), MAX_IMAGE_RESULTS_PER_PAGE)
+    offset = (page - 1) * per_page
+
+    sort = normalize_sort(sort)
+    site_list = normalize_sites(sites)
+    tld_list = normalize_tlds(tlds)
+    after_dt = parse_time_bound(after, "after")
+    before_dt = parse_time_bound(before, "before")
+
+    started = time.monotonic()
+    conn = db_connect()
+    try:
+        vector_col = resolve_vector_column(conn)
+        filter_sql, filter_params = _filter_clauses(
+            site_list, tld_list, after_dt, before_dt,
+            False, False,
+        )
+        params = {
+            "ts_config": TS_CONFIG,
+            "query": q,
+            "weights": RANK_WEIGHTS,
+            "norm": RANK_NORMALIZATION,
+            "fresh_weight": FRESHNESS_WEIGHT,
+            "fresh_half_life": FRESHNESS_HALF_LIFE_DAYS,
+            "images_per_source": IMAGES_PER_SOURCE_PAGE,
+            "limit": per_page + 1,  # +1 -> exaktes has_more
+            "offset": offset,
+            "rank_cap": MAX_RANK_CANDIDATES,
+        }
+        params.update(filter_params)
+
+        with conn.cursor(cursor_factory=RealDictCursor) as cur:
+            cur.execute(
+                "SELECT websearch_to_tsquery(%s, %s)::text AS tsq",
+                (TS_CONFIG, q),
+            )
+            parsed_tsq = ((cur.fetchone() or {}).get("tsq") or "").strip()
+            empty_tsq = not parsed_tsq
+
+            rows = []
+            total = None
+            ranking_capped = False
+
+            if not empty_tsq:
+                _set_statement_timeout(cur, STATEMENT_TIMEOUT_MS)
+                try:
+                    cur.execute(
+                        _image_search_sql(vector_col, sort, filter_sql),
+                        params,
+                    )
+                    rows = cur.fetchall()
+                except psycopg2.errors.QueryCanceled:
+                    # Wie bei der Websuche: zuerst ein begrenztes
+                    # Kandidatenfenster ranken, statt den Request zu
+                    # verlieren.
+                    ranking_capped = True
+                    log.info(
+                        "Bilder-Ranking übersprungen (Timeout); "
+                        "Fallback auf %d Kandidaten.",
+                        params["rank_cap"],
+                    )
+                    try:
+                        cur.execute(
+                            _image_search_sql(
+                                vector_col, sort, filter_sql, capped=True,
+                            ),
+                            params,
+                        )
+                        rows = cur.fetchall()
+                    except psycopg2.errors.QueryCanceled:
+                        raise SearchTimeout(
+                            "Die Bildersuche dauert ungewöhnlich lange – "
+                            "probiere es nochmal oder nutze einen Filter "
+                            "wie site=<domain>."
+                        )
+
+                _set_statement_timeout(cur, COUNT_TIMEOUT_MS)
+                try:
+                    cur.execute(
+                        _image_count_sql(vector_col, filter_sql), params,
+                    )
+                    total = int(cur.fetchone()["n"])
+                except psycopg2.Error as exc:
+                    log.warning("Bilder-Zählung abgebrochen: %s", exc)
+                    total = None
+
+        has_more = len(rows) > per_page
+        rows = rows[:per_page]
+
+        results = []
+        for r in rows:
+            page_url = r["page_url"]
+            results.append({
+                "image_url": r["image_url"],
+                "page_url": page_url,
+                "title": (r["title"] or "").strip() or page_url,
+                "domain": r["domain"] or "",
+                "tld": r["tld"] or "",
+                "crawled_at": _iso(r["crawled_at"]),
+                "score": _round(r["score"]),
+                "text_rank": _round(r["text_rank"]),
+                "freshness": _round(r["freshness"], 3),
+            })
+
+        return {
+            "query": q,
+            "parsed_query": parsed_tsq,
+            "page": page,
+            "per_page": per_page,
+            "sort": sort,
+            "ranking_capped": ranking_capped,
+            "total": total,
+            "has_more": has_more,
+            "results": results,
+            "took_ms": round((time.monotonic() - started) * 1000, 1),
+        }
+    except psycopg2.errors.UndefinedColumn as exc:
+        raise SearchEngineError(
+            "Die Spalte pages.pictures fehlt. Bitte zuerst "
+            "add_pages_pictures.sql einspielen."
+        ) from exc
+    finally:
+        conn.close()
+
+
 def get_page(page_id=None, url=None, *, max_content_chars=None):
     """
     Einzelne Seite per id oder url. content ist auf
@@ -861,6 +1140,7 @@ def get_page(page_id=None, url=None, *, max_content_chars=None):
                            crawled_at, crawl_time_ms,
                            length(content) AS content_length,
                            links,
+                           pictures,
                            left(content, %(max_chars)s) AS content
                     FROM pages
                     WHERE id = %(id)s
@@ -875,6 +1155,7 @@ def get_page(page_id=None, url=None, *, max_content_chars=None):
                            crawled_at, crawl_time_ms,
                            length(content) AS content_length,
                            links,
+                           pictures,
                            left(content, %(max_chars)s) AS content
                     FROM pages
                     WHERE url = %(url)s
@@ -891,6 +1172,7 @@ def get_page(page_id=None, url=None, *, max_content_chars=None):
     content = row["content"] or ""
     content_length = row["content_length"] or 0
     links = list(row["links"] or [])[:PAGE_MAX_LINKS]
+    pictures = list(row["pictures"] or [])[:PAGE_MAX_PICTURES]
 
     return {
         "id": row["id"],
@@ -910,6 +1192,8 @@ def get_page(page_id=None, url=None, *, max_content_chars=None):
         "content": content,
         "links": links,
         "links_total": len(row["links"] or []),
+        "pictures": pictures,
+        "pictures_total": len(row["pictures"] or []),
     }
 
 

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

@@ -301,6 +301,47 @@ class CountSqlTest(unittest.TestCase):
         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):
@@ -340,5 +381,38 @@ class SearchValidationTest(unittest.TestCase):
                 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

@@ -73,6 +73,12 @@ class FakeState:
         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."""
 
@@ -84,12 +90,20 @@ class FakeCursor:
         query = " ".join(query.split())
         params = params or ()
 
-        if "SELECT id, url, source_url" in query:
+        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) == 2 else set()
+            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
+                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 = [
@@ -241,6 +255,35 @@ class HostSchedulerTest(unittest.TestCase):
             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):
 
@@ -282,6 +325,151 @@ class ErrorClassificationTest(unittest.TestCase):
         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):
@@ -307,11 +495,14 @@ class QueueClaimTest(unittest.TestCase):
         scheduler.blocked_until["blocked.example"] = time.monotonic() + 900
         item, wait_hint = self._claim(self._state(), scheduler)
         self.assertEqual(item[1], "https://other.example/b")
-        self.assertGreater(wait_hint, 800)
+        # 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()
-        scheduler.wait("https://blocked.example/a", 0.0)
+        # 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")
 
@@ -349,6 +540,39 @@ class QueueClaimTest(unittest.TestCase):
         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(

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())