add pictures

.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

@@ -13,6 +13,8 @@ das komplette serverjonas search
 - `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)
 
 ## Sehr große Seiten / tsvector-Limit
 
@@ -47,14 +49,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
+   Worker, ob der Host gerade crawlbar ist (Per-Host-Delay aus robots.txt bzw.
-   `JONASSEARCH_CLAIM_SCAN_LIMIT` Kandidaten), ob der Host gerade crawlbar ist
+   `JONASSEARCH_DELAY`, aktive Backoff-Sperre). Nicht crawlbare Hosts werden
-   (Per-Host-Delay aus robots.txt bzw. `JONASSEARCH_DELAY`, aktive
+   übersprungen – der Worker arbeitet automatisch auf einer anderen Domain
-   Backoff-Sperre). Nicht crawlbare Hosts werden übersprungen – der Worker
+   weiter und kehrt zurück, sobald der Host wieder darf. Ist gerade gar kein
-   arbeitet automatisch auf einer anderen Domain weiter und kehrt zurück,
+   Host crawlbar, schläft der Worker bis zum frühesten freigegebenen Host
-   sobald der Host wieder darf. Ist gerade gar kein Host crawlbar, schläft
+   (gekappet bei `JONASSEARCH_MAX_IDLE_SLEEP`, dann wird neu gescannt).
-   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 +77,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
+    other. Hosts that the HostScheduler already knows to be busy
-    host may be contacted right now (per-host crawl delay, active
+    (per-host crawl delay, active block backoff) are excluded by the
-    block backoff). The first ready candidate is claimed; hosts that
+    query itself. For the remaining candidates the scheduler decides
-    would have to wait are skipped, so the worker automatically works
+    whether a host may be contacted right now. The first ready
-    on another domain while a rate-limited or blocked host cools down.
+    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("""
-                cur.execute("""
+                SELECT id, url, source_url
-                    SELECT id, url, source_url
+                FROM crawl_queue
-                    FROM crawl_queue
+                WHERE status = 'pending'
-                    WHERE status = 'pending'
+                  AND NOT (id = ANY(%s::bigint[]))
-                      AND NOT (id = ANY(%s))
+                  AND lower(split_part(split_part(url, '://', 2), '/', 1))
-                    ORDER BY priority DESC, discovered_at ASC, id ASC
+                      <> ALL(%s::text[])
-                    FOR UPDATE SKIP LOCKED
+                ORDER BY priority DESC, discovered_at ASC, id ASC
-                    LIMIT %s
+                FOR UPDATE SKIP LOCKED
-                """, (seen_ids, QUEUE_CLAIM_BATCH))
+                LIMIT %s
-            else:
+            """, (seen_ids, excluded_hosts, QUEUE_CLAIM_BATCH))
-                cur.execute("""
-                    SELECT id, url, source_url
-                    FROM crawl_queue
-                    WHERE status = 'pending'
-                    ORDER BY priority DESC, discovered_at ASC, id ASC
-                    FOR UPDATE SKIP LOCKED
-                    LIMIT %s
-                """, (QUEUE_CLAIM_BATCH,))
             candidates = cur.fetchall()
 
         if not candidates:
@@ -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
+Zeitfilter, SQL-Zusammensetzung (Web- und Bildersuche), Query-Parsing
-Datenbankzugriff.
+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 = """
+    fresh_lateral = _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
-    """
 
     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(