From 19730e1711be1e381ec507ec75bf5a69f369c50e Mon Sep 17 00:00:00 2001 From: jhkaktus Date: Fri, 9 Oct 2026 16:42:15 +0200 Subject: [PATCH] add stuff --- .env.example | 54 +- README.md | 47 +- clear_domain_suggestions.sh | 142 +++ crawler.py | 548 ++++++++- priorityweb/jonassearch-priority.service | 4 +- search/README.md | 242 ++-- search/__init__.py | 24 +- search/api.py | 871 ++++++++------ search/cli.py | 352 ++++-- search/engine.py | 1318 +++++++++++++++------- search/jonassearch-engine.service | 42 + search/test_engine.py | 344 ++++++ test_crawler.py | 455 ++++++++ 13 files changed, 3448 insertions(+), 995 deletions(-) create mode 100755 clear_domain_suggestions.sh create mode 100644 search/jonassearch-engine.service create mode 100644 search/test_engine.py create mode 100644 test_crawler.py diff --git a/.env.example b/.env.example index ec437f2..498d41f 100644 --- a/.env.example +++ b/.env.example @@ -16,6 +16,21 @@ JONASSEARCH_WORKERS=4 # Mindestabstand zwischen Requests zur gleichen Domain JONASSEARCH_DELAY=3.0 +# Host-Scheduling: blockierte/rate-limitierte Domains werden automatisch +# übersprungen und später erneut versucht (mit exponentiellem Backoff). +# Kandidaten, die pro Claim-Runde in der Queue geprüft werden +#JONASSEARCH_CLAIM_BATCH=25 +# Maximale Anzahl geprüfter Kandidaten pro Claim +#JONASSEARCH_CLAIM_SCAN_LIMIT=200 +# Maximale Versuche pro URL (transiente Fehler werden erneut eingereiht), +# bevor der Eintrag endgültig failed wird +#JONASSEARCH_MAX_ATTEMPTS=5 +# Wartezeiten über dieser Grenze (Sekunden) werden nicht inline geschlafen: +# das Item wird zurückgestellt und eine andere Domain gecrawlt +#JONASSEARCH_MAX_INLINE_WAIT=10 +# Obergrenze für die Wartezeit, wenn gerade kein Host crawlbar ist +#JONASSEARCH_MAX_IDLE_SLEEP=30 + # robots.txt Cache JONASSEARCH_ROBOTS_CACHE_SECONDS=86400 @@ -32,7 +47,7 @@ JONASSEARCH_USER_AGENT=JonasSearchBot/0.1 (+https://jonassearch.local/bot) # Version dieser Crawler-Codebasis (NICHT die Deployment-Generation). # Muss bei jedem verteilten Crawler-Änderung erhöht werden. -JONASSEARCH_CRAWLER_VERSION=1 +JONASSEARCH_CRAWLER_VERSION=3 # Wie oft jeder Crawler außerhalb der Transaktionen nach einer neuen # Deployment-Generation sieht (Sekunden). Der Check setzt nur ein @@ -55,16 +70,43 @@ JONASSEARCH_UPDATE_CHECK_SECONDS=30 #JONASSEARCH_LAUNCHER_RESTART_DELAY=2 # ============================================================ -# Suchmaschine (search/) +# Suchmaschine (search/) – Port 8083 # ============================================================ -# PostgreSQL-Textsuch-Konfiguration (Stemming/Stopwörter) -JONASSEARCH_TS_CONFIG=german +# PostgreSQL-Textsuch-Konfiguration. MUSS zur Konfiguration passen, +# mit der der DB-Trigger den Suchvektor (pages.search_vector) baut – +# laut DB-Doku: 'simple'. Ansonsten finden die Queries nichts. +JONASSEARCH_TS_CONFIG=simple # tsvector-Spalte der Tabelle `pages` erzwingen (optional, # wird automatisch erkannt) #JONASSEARCH_VECTOR_COLUMN=search_vector -# HTTP-API (search/api.py) +# HTTP-API + Weboberfläche (search/api.py) JONASSEARCH_API_HOST=0.0.0.0 -JONASSEARCH_API_PORT=8082 +JONASSEARCH_API_PORT=8083 + +# Ranking: ts_rank_cd-Gewichte (Titel=A, Überschriften=B, Inhalt=C) +#JONASSEARCH_RANK_WEIGHTS=A=1.0,B=0.6,C=0.3,D=0.1 + +# Ranking: Aktualitätsbonus (Faktor auf den Text-Rang) +#JONASSEARCH_FRESHNESS_WEIGHT=0.25 +#JONASSEARCH_FRESHNESS_HALF_LIFE_DAYS=180 + +# Ranking: ts_rank_cd-Normalisierung (Bitmaske). 8 = Rang / +# (1 + log Dokumentlänge) – riesige Seiten dominieren sonst automatisch. +# 0 = keine Normalisierung. +#JONASSEARCH_RANK_NORMALIZATION=8 + +# Fallback bei sehr häufigen Begriffen: volles Ranking wird bei +# Zeitüberschreitung auf die ersten N Kandidaten (Indexreihenfolge) +# begrenzt, damit die Antwort immer kommt. +#JONASSEARCH_MAX_RANK_CANDIDATES=5000 + +# Schutz der großen DB: Statement-Timeouts (ms) +#JONASSEARCH_STATEMENT_TIMEOUT_MS=10000 +#JONASSEARCH_COUNT_TIMEOUT_MS=4000 + +# Paginierung +#JONASSEARCH_RESULTS_PER_PAGE=10 +#JONASSEARCH_MAX_RESULTS_PER_PAGE=50 diff --git a/README.md b/README.md index 7eeca71..398da5e 100644 --- a/README.md +++ b/README.md @@ -4,8 +4,12 @@ das komplette serverjonas search - `crawler.py` – PostgreSQL-Crawler, baut den Suchindex auf - `launcher.py` – startet/überwacht den Crawler und rollt Updates aus (siehe unten) -- `search/` – Suchmaschine darüber (CLI + HTTP-API), siehe [search/README.md](search/README.md) -- `web/` – Statistik-Webseite +- `search/` – Suchmaschine darüber (CLI + HTTP-API + Weboberfläche, Port 8083), + siehe [search/README.md](search/README.md) +- `web/` – Statistik-Webseite (Port 8081, `jonassearch-stats.service`) +- `priorityweb/` – Vorschlagsverfeinerer (Port 8082, `jonassearch-priority.service`) +- `search/jonassearch-engine.service` – systemd-Unit für die Suchmaschine + (Port 8083, `jonassearch-engine.service`) - `fix_pages_search_vector_trigger.sql` – einmalig laufen lassen: kappelt den Text, der im DB-Trigger `pages_search_vector_update()` in `to_tsvector()` einfließt (Postgres hartes Limit: 1.048.575 Bytes pro `tsvector`) @@ -38,6 +42,45 @@ Zusätzlich werden bekannt wertlose URLs (`pkg.go.dev/...?tab=importedby` – Megabyte-große reine Rückverweis-Listen) über `SKIP_URL_PATTERNS` schon vor dem Fetch übersprungen und als `done` markiert. +## Domain-Switching / Block-Handling + +Der Crawler wartet nie auf einer langsamen oder blockierten Domain: + +1. **Claim mit Host-Check:** Beim Holen des nächsten Queue-Eintrags prüft der + Worker pro Kandidat (Prioritätsreihenfolge, bis zu + `JONASSEARCH_CLAIM_SCAN_LIMIT` Kandidaten), ob der Host gerade crawlbar ist + (Per-Host-Delay aus robots.txt bzw. `JONASSEARCH_DELAY`, aktive + Backoff-Sperre). Nicht crawlbare Hosts werden übersprungen – der Worker + arbeitet automatisch auf einer anderen Domain weiter und kehrt zurück, + sobald der Host wieder darf. Ist gerade gar kein Host crawlbar, schläft + der Worker bis zum frühesten freigegebenen Host (gekappet bei + `JONASSEARCH_MAX_IDLE_SLEEP`, dann wird neu gescannt). +2. **Backoff bei Blocks:** HTTP 429/5xx/403/401 und Netzwerkfehler setzen den + Host auf exponentielles Backoff (z. B. 429: 60 s, sich verdoppelnd bis + 30 min; 403/401: 10 min bis 2 h). Ein `Retry-After`-Header wird zusätzlich + beachtet. Bei einem erfolgreichen Fetch zählt der Zähler wieder von vorn. +3. **Retry statt Verlust:** Transiente Fehler (429/5xx/Netzwerk) werden + wieder in die Queue gestellt (`pending`) statt endgültig `failed` – + begrenzt durch `JONASSEARCH_MAX_ATTEMPTS` (Standard 5), damit dauerhaft + kaputte Hosts nicht ewig rotieren. Wartezeiten größer als + `JONASSEARCH_MAX_INLINE_WAIT` (z. B. großer robots-crawl-delay) werden + nicht inline geschlafen: das Item wird ohne Versuchs-Zählung zurückgestellt + und solange eine andere Domain gecrawlt. + +## Nicht crawlbare Ziele (localhost, IPs, nicht-echte Domains) + +Der Crawler folgt nur echten, öffentlich auflösbaren Domains. Bereits in +`normalize_url()` werden abgelehnt und nie in die Queue gestellt: + +- `localhost`, `*.localhost`, `localhost.localdomain`, `*.local`, `*.internal` +- reine IP-Adressen als Host (IPv4 und IPv6, z. B. `127.0.0.1`, `http://[::1]/`) +- Hostnamen ohne Punkt (Einzel-Labels wie `http://myserver/` – LAN-Namen) + +Alte Queue-Einträge mit solchen Hosts werden beim Zugriff als `done` +markiert und nicht mehr geholt. Zeigt der Start-URL auf so einen Host, +beendet sich der Crawler mit einem klaren Fehler (Exit 1) statt endlos +zu rotieren. + ## Rolling Updates (Generation-Mechanismus) Zentrale Tabelle (eine Zeile, `id = 1`): diff --git a/clear_domain_suggestions.sh b/clear_domain_suggestions.sh new file mode 100755 index 0000000..b857e6c --- /dev/null +++ b/clear_domain_suggestions.sh @@ -0,0 +1,142 @@ +#!/usr/bin/env bash +# +# clear_domain_suggestions.sh +# +# Loescht ALLE Eintraege aus der Tabelle `domain_suggestions` — der +# Buchhaltungs-Tabelle des Vorschlagsverfeinerers (priorityweb/app.py, +# Port 8082). Die Liste "Letzte Vorschlaege" im Web-UI ist danach leer. +# +# WICHTIG: Die crawl_queue wird NICHT angefasst. Bereits priorisierte +# Domains bleiben mit Prioritaet 1000000 (SUGGEST_PRIORITY) vorne in +# der Queue und werden weiterhin als Naechstes gecrawlt. Falls auch +# diese Boosts aufgehoben werden sollen, zusaetzlich ausfuehren: +# +# DELETE FROM crawl_queue +# WHERE status = 'pending' AND priority >= 1000000; +# +# (Nur pending-Zeilen: laufende Verarbeitungen werden nicht abgeschossen.) +# +# Verwendung: +# ./clear_domain_suggestions.sh mit Rueckfrage +# ./clear_domain_suggestions.sh --yes ohne Rueckfrage (z. B. fuer Cron) +# ./clear_domain_suggestions.sh --help diesen Text anzeigen +# +# DB-Zugangsdaten kommen wie bei allen Komponenten aus der .env +# (JONASSEARCH_DB_*). Ist dort kein Passwort gesetzt oder die direkte +# Verbindung fehlgeschlagen, wird lokal per `sudo -u postgres psql` +# ausgefuehrt (gleicher Weg wie in web/app.py). + +set -euo pipefail +cd "$(dirname "$0")" + +# Muss zu SUGGEST_PRIORITY in priorityweb/app.py passen. +SUGGEST_PRIORITY=1000000 + +usage() { + sed -n '2,27p' "$0" | sed 's/^# \{0,1\}//' + exit 0 +} + +YES=0 +for arg in "$@"; do + case "$arg" in + -y|--yes) YES=1 ;; + -h|--help) usage ;; + *) echo "Unbekanntes Argument: $arg (--help fuer Nutzung)" >&2; exit 2 ;; + esac +done + +# --------------------------------------------------------------- +# Zugangsdaten aus .env laden (nur JONASSEARCH_DB_*-Zeilen) +# --------------------------------------------------------------- +if [[ -f .env ]]; then + set -a + # shellcheck disable=SC1091 + source <(grep -E '^JONASSEARCH_DB_[A-Z]+=' .env | tr -d '\r') + set +a +fi + +DB_HOST="${JONASSEARCH_DB_HOST:-127.0.0.1}" +DB_PORT="${JONASSEARCH_DB_PORT:-25432}" +DB_NAME="${JONASSEARCH_DB_NAME:-jonassearch}" +DB_USER="${JONASSEARCH_DB_USER:-jonassearch}" + +# --------------------------------------------------------------- +# psql-Wrapper: zuerst direkt mit Passwort, sonst als postgres-Superuser +# --------------------------------------------------------------- +psql_run() { + if [[ -n "${JONASSEARCH_DB_PASSWORD:-}" ]] && command -v psql >/dev/null 2>&1; then + if PGPASSWORD="${JONASSEARCH_DB_PASSWORD}" \ + psql -h "$DB_HOST" -p "$DB_PORT" -U "$DB_USER" -d "$DB_NAME" \ + -v ON_ERROR_STOP=1 "$@"; then + return 0 + fi + echo "Direkte DB-Verbindung fehlgeschlagen — Fallback: sudo -u postgres psql" >&2 + fi + sudo -u postgres psql -p "$DB_PORT" -d "$DB_NAME" -v ON_ERROR_STOP=1 "$@" +} + +die() { echo "FEHLER: $*" >&2; exit 3; } + +# --------------------------------------------------------------- +# 1) Existiert die Tabelle ueberhaupt? +# (Die App legt sie beim ersten Vorschlag selbst an.) +# --------------------------------------------------------------- +EXISTS="$(psql_run -t -A -c \ + "SELECT to_regclass('public.domain_suggestions') IS NOT NULL;")" \ + || die "keine Verbindung zur DB ${DB_HOST}:${DB_PORT}/${DB_NAME}" + +if [[ "$EXISTS" != "t" ]]; then + echo "Tabelle 'domain_suggestions' existiert nicht — es gibt nichts zu loeschen." + exit 0 +fi + +# --------------------------------------------------------------- +# 2) Zaehlen und bestaetigen lassen +# --------------------------------------------------------------- +BEFORE="$(psql_run -t -A -c "SELECT COUNT(*) FROM domain_suggestions;")" \ + || die "Konnte Anzahl der Eintraege nicht lesen." + +echo "DB : ${DB_USER}@${DB_HOST}:${DB_PORT}/${DB_NAME}" +echo "Eintraege : ${BEFORE}" + +if [[ "$BEFORE" == "0" ]]; then + echo "Tabelle ist bereits leer — nichts zu tun." + exit 0 +fi + +if (( ! YES )); then + read -r -p "Wirklich ALLE ${BEFORE} Eintraege aus 'domain_suggestions' loeschen? (yes/no) " answer + if [[ "$answer" != "yes" ]]; then + echo "Abgebrochen — nichts geloescht." + exit 1 + fi +fi + +# --------------------------------------------------------------- +# 3) Loeschen (DELETE statt TRUNCATE: kein ACCESS EXCLUSIVE Lock, +# vertraeglich mit gleichzeitig laufenden Vorschlaegen im UI) +# --------------------------------------------------------------- +DELETED="$(psql_run -t -A -c \ + "WITH d AS (DELETE FROM domain_suggestions RETURNING 1) \ + SELECT COUNT(*) FROM d;")" \ + || die "DELETE fehlgeschlagen." + +echo "Geloescht : ${DELETED} Eintraege (vorher: ${BEFORE})." +echo "Die Liste 'Letzte Vorschlaege' im Web-UI leert sich beim naechsten Refresh (max. 30 s)." + +# --------------------------------------------------------------- +# 4) Nur Hinweis: stecken in der crawl_queue noch priorisierte Domains? +# (gleicher schneller Index-Range wie /api/queue in der App) +# --------------------------------------------------------------- +BOOSTED="$(psql_run -t -A -c \ + "SELECT COUNT(*) FROM crawl_queue \ + WHERE status = 'pending' AND priority >= ${SUGGEST_PRIORITY};" 2>/dev/null)" \ + || BOOSTED="?" + +if [[ "$BOOSTED" != "0" ]]; then + echo "Hinweis: ${BOOSTED} priorisierte Domain(s) stecken noch in der crawl_queue" + echo " (dort bleiben sie auch nach diesem Skript vorne)." + echo " Aufheben mit:" + echo " DELETE FROM crawl_queue WHERE status = 'pending' AND priority >= ${SUGGEST_PRIORITY};" +fi diff --git a/crawler.py b/crawler.py index ad415b4..2a92a09 100644 --- a/crawler.py +++ b/crawler.py @@ -14,7 +14,17 @@ Features: bounded retries with rollback + backoff) - Smart priority queue: prefers never-crawled URLs, new domains and new hosts, with a spread bonus for little-explored territory -- Per-host crawl delay +- Host-aware scheduling: a worker never waits on a busy host - when + the current domain is rate-limited or blocked, the worker + automatically picks another domain and returns once the host is + crawlable again +- Per-host crawl delay (robots.txt) and per-host block backoff: + HTTP 429/5xx/403 and network errors pause the host with an + exponential backoff, Retry-After is honored +- Transient failures are requeued (bounded by JONASSEARCH_MAX_ATTEMPTS) + instead of being lost +- Non-crawlable targets (localhost, raw IPs, non-real domains) are + rejected before they reach the queue - robots.txt support with in-memory cache - Graceful shutdown - Basic crawler statistics @@ -52,10 +62,12 @@ import signal import time import logging import argparse +import ipaddress import subprocess import threading import multiprocessing as mp from collections import defaultdict +from email.utils import parsedate_to_datetime from html import unescape from pathlib import Path from urllib.parse import parse_qsl, urljoin, urlparse, urldefrag @@ -138,7 +150,7 @@ USER_AGENT = os.getenv( # change; completely independent from the deployment generation (the # generation is a global deployment counter in PostgreSQL, the version # identifies the crawler code). -CRAWLER_VERSION = int(os.getenv("JONASSEARCH_CRAWLER_VERSION", "1")) +CRAWLER_VERSION = int(os.getenv("JONASSEARCH_CRAWLER_VERSION", "3")) # Exit codes. 42 is reserved exclusively for "I must be updated" so # the launcher can distinguish update, crash and normal exit. @@ -173,6 +185,16 @@ LAUNCHER_RESTART_DELAY = float( # Only crawl normal HTTP(S) pages. ALLOWED_SCHEMES = {"http", "https"} +# Host suffixes reserved for loopback/private networks (never delegated +# in public DNS). Together with the localhost/IP checks in +# is_non_crawlable_host() they keep non-real domains out of the queue. +NON_CRAWLABLE_HOST_SUFFIXES = ( + ".internal", + ".local", + ".localdomain", + ".localhost", +) + # Ignore obvious non-HTML resources. BLOCKED_EXTENSIONS = { ".7z", ".avi", ".bin", ".bmp", ".bz2", ".class", ".csv", ".doc", @@ -244,6 +266,69 @@ QUEUE_INSERT_PAGE_SIZE = 500 QUEUE_INSERT_MAX_ATTEMPTS = 3 QUEUE_INSERT_RETRY_BASE_DELAY = 0.5 +# ============================================================ +# HOST SCHEDULING / BLOCK HANDLING CONFIG +# ============================================================ +# The queue claim examines candidates in priority order and picks the +# first URL whose host is currently crawlable (not rate-limited, not +# blocked). Hosts that would have to wait are skipped, so a worker +# automatically crawls another domain while the blocked one cools +# down - no worker sits idle in front of a 429. + +# Pending URLs examined per claim round (FOR UPDATE SKIP LOCKED, so +# several crawler processes stay safe on one shared queue). +QUEUE_CLAIM_BATCH = int(os.getenv("JONASSEARCH_CLAIM_BATCH", "25")) + +# Maximum number of candidates examined per claim (batch * rounds). +# Keeps the common case (a ready host in the first batch) cheap while +# still finding crawlable hosts behind a wall of blocked ones. +QUEUE_CLAIM_SCAN_LIMIT = int( + os.getenv("JONASSEARCH_CLAIM_SCAN_LIMIT", "200") +) + +# Cap for the "nothing is crawlable right now" sleep, so a host blocked +# for a long time cannot park the whole worker; after the cap the queue +# is simply re-scanned. +QUEUE_MAX_IDLE_SLEEP = float(os.getenv("JONASSEARCH_MAX_IDLE_SLEEP", "30")) + +# A URL may be attempted this many times (transient errors are +# requeued) before it is marked failed for good, so permanently broken +# hosts cannot spin forever. +QUEUE_MAX_ATTEMPTS = int(os.getenv("JONASSEARCH_MAX_ATTEMPTS", "5")) + +# Rate waits longer than this are NOT slept inline: the item is put +# back and the worker immediately looks for another domain instead of +# blocking on a slow host (e.g. a large robots.txt crawl-delay). +MAX_INLINE_RATE_WAIT = float(os.getenv("JONASSEARCH_MAX_INLINE_WAIT", "10")) + +# Backoff for blocked hosts, per HTTP status: (base seconds, cap +# seconds). The pause doubles with every consecutive block of the same +# host, so persistent blocks fade out of the schedule while a single +# hiccup costs only the base delay. +HOST_BLOCK_RULES = { + 401: (600.0, 7200.0), # auth required: probably permanent, long pause + 403: (600.0, 7200.0), # forbidden + 408: (30.0, 600.0), # request timeout + 425: (30.0, 600.0), # too early + 429: (60.0, 1800.0), # rate limited (Retry-After is honored on top) + 500: (30.0, 600.0), + 502: (30.0, 600.0), + 503: (30.0, 900.0), + 504: (30.0, 600.0), +} + +# Network-level failures (DNS, timeout, connection reset) get a short +# backoff as well - a dead host must not be retried in a tight loop. +HOST_BLOCK_NETWORK_BASE = 30.0 +HOST_BLOCK_NETWORK_MAX = 300.0 + +# Absolute upper bound for one host pause (also caps Retry-After). +HOST_BLOCK_HARD_CAP = 7200.0 + +# HTTP statuses whose queue items are requeued (retried later) instead +# of being failed terminally. +TRANSIENT_HTTP_STATUSES = {408, 425, 429, 500, 502, 503, 504} + # ============================================================ # URL PRIORITY CONFIG # ============================================================ @@ -646,35 +731,135 @@ def write_stats(conn, crawled, queued, failed): """, (crawled, queued, failed)) -def get_next_queue_item(conn): +def claim_next_queue_item(conn, scheduler, robots): """ - Atomically claim one pending URL. + Atomically claim the best currently-crawlable pending URL. - SKIP LOCKED is the important part: several crawler processes can - consume the same PostgreSQL queue without waiting for each other. + Candidates are read in priority order (same ORDER BY as before) in + small batches with FOR UPDATE SKIP LOCKED, so several crawler + processes can still share one queue without waiting for each + other. For every candidate the HostScheduler decides whether its + host may be contacted right now (per-host crawl delay, active + block backoff). The first ready candidate is claimed; hosts that + would have to wait are skipped, so the worker automatically works + on another domain while a rate-limited or blocked host cools down. + + Returns (item, wait_hint): + item (id, url, source_url) of the claimed URL, or None + wait_hint seconds until the soonest examined host becomes + crawlable again (None when the queue is empty) + + Writes only - the commit happens in safe_commit(). + """ + now = time.monotonic() + seen_ids = [] + scanned = 0 + wait_hint = None + item = None + + while item is None and scanned < QUEUE_CLAIM_SCAN_LIMIT: + with conn.cursor() as cur: + if seen_ids: + cur.execute(""" + SELECT id, url, source_url + FROM crawl_queue + WHERE status = 'pending' + AND NOT (id = ANY(%s)) + ORDER BY priority DESC, discovered_at ASC, id ASC + FOR UPDATE SKIP LOCKED + LIMIT %s + """, (seen_ids, QUEUE_CLAIM_BATCH)) + else: + cur.execute(""" + SELECT id, url, source_url + FROM crawl_queue + WHERE status = 'pending' + ORDER BY priority DESC, discovered_at ASC, id ASC + FOR UPDATE SKIP LOCKED + LIMIT %s + """, (QUEUE_CLAIM_BATCH,)) + candidates = cur.fetchall() + + if not candidates: + break + + for queue_id, url, source_url in candidates: + seen_ids.append(queue_id) + scanned += 1 + + wait_for = scheduler.wait_time(url, robots.delay(url), now) + if wait_for > 0.0: + if wait_hint is None or wait_for < wait_hint: + wait_hint = wait_for + continue + + # Host is crawlable: claim it. The row is already locked by + # this transaction (FOR UPDATE SKIP LOCKED), so the + # status='pending' guard is just belt and braces against + # races with other claim paths. + with conn.cursor() as cur: + cur.execute(""" + UPDATE crawl_queue + SET status = 'processing', + attempts = attempts + 1, + last_attempt_at = now() + WHERE id = %s AND status = 'pending' + RETURNING id, url, source_url + """, (queue_id,)) + row = cur.fetchone() + + if row: + item = row + break + + if len(candidates) < QUEUE_CLAIM_BATCH: + break + + return item, wait_hint + + +def defer_queue_item(conn, queue_id): + """ + Put a claimed item back WITHOUT burning an attempt: the host turned + out not to be crawlable at fetch time (e.g. a robots.txt crawl-delay + larger than MAX_INLINE_RATE_WAIT was discovered only after + claiming). The item stays pending and is retried when the host's + schedule allows it - the worker crawls other domains until then. Writes only - the commit happens in safe_commit(). """ with conn.cursor() as cur: cur.execute(""" - WITH next_item AS ( - SELECT id - FROM crawl_queue - WHERE status = 'pending' - ORDER BY priority DESC, discovered_at ASC, id ASC - FOR UPDATE SKIP LOCKED - LIMIT 1 - ) - UPDATE crawl_queue q - SET status = 'processing', - attempts = attempts + 1, - last_attempt_at = now() - FROM next_item - WHERE q.id = next_item.id - RETURNING q.id, q.url, q.source_url - """) + UPDATE crawl_queue + SET status = 'pending', + attempts = GREATEST(attempts - 1, 0) + WHERE id = %s + """, (queue_id,)) + + +def requeue_queue_item(conn, queue_id): + """ + Requeue an item after a transient failure (HTTP 429/5xx, network + errors): the host got a backoff, the URL is retried afterwards. + After QUEUE_MAX_ATTEMPTS attempts the item is marked failed for + good, so a permanently broken host cannot spin forever. + + Returns the new status ('pending' or 'failed'), or None. + + Writes only - the commit happens in safe_commit(). + """ + with conn.cursor() as cur: + cur.execute(""" + UPDATE crawl_queue + SET status = CASE + WHEN attempts >= %s THEN 'failed' + ELSE 'pending' + END + WHERE id = %s + RETURNING status + """, (QUEUE_MAX_ATTEMPTS, queue_id)) row = cur.fetchone() - return row + return row[0] if row else None def finish_queue_item(conn, queue_id, success=True): @@ -1290,6 +1475,54 @@ def clean_text_list(values): # URL NORMALIZATION # ============================================================ +def is_non_crawlable_host(hostname): + """ + True for hosts the crawler must never touch: loopback names, + private-use suffixes, raw IP literals (IPv4/IPv6) and other + non-resolvable (non-real) host names. A public web crawler can only + follow real, publicly resolving domains - everything else is a LAN + target, a loopback service or a literal address. + + Applied centrally in normalize_url(), so neither the seed URL nor + any discovered link ever reaches the queue. + """ + if not hostname: + return True + host = hostname.strip(".").lower() + if not host: + return True + if host == "localhost" or host == "localhost.localdomain": + return True + if host.endswith(NON_CRAWLABLE_HOST_SUFFIXES): + return True + # Raw IP literals (IPv4 and IPv6) are not real domains. Note that + # urlparse already strips the brackets of IPv6 URLs, so "::1" is + # compared directly. + try: + ipaddress.ip_address(host) + return True + except ValueError: + pass + # Single-label names (no dot) can never be public domains. + if "." not in host: + return True + return False + + +def is_non_crawlable_url(url): + """ + Convenience wrapper for queue entries: True when the URL points at + a non-crawlable host (localhost, raw IP, non-real domain) or cannot + be parsed at all. Unparseable URLs are treated as non-crawlable so + one malformed entry can never crash the worker. + """ + try: + hostname = (urlparse(url).hostname or "").lower() + except ValueError: + return True + return is_non_crawlable_host(hostname) + + def normalize_url(url, base_url=None): if not url: return None @@ -1319,6 +1552,11 @@ def normalize_url(url, base_url=None): if not hostname: return None + # Never queue loopback, LAN or raw-IP targets (localhost, IPs, + # non-real domains). + if is_non_crawlable_host(hostname): + return None + # Drop default ports. port = parsed.port if (scheme == "http" and port == 80) or (scheme == "https" and port == 443): @@ -1423,20 +1661,159 @@ class RobotsCache: # HTTP # ============================================================ -class RateLimiter: +class HostScheduler: + """ + Per-process host availability tracker - the heart of the automatic + domain switching. + + The scheduler keeps, per host: + last_request: monotonic time of the last request start, so + per-host delays (DEFAULT_DELAY / robots.txt + crawl-delay) are honored. + blocked_until: monotonic time until which the host is considered + blocked (HTTP 429/5xx/403, network errors, + Retry-After). + block_count: consecutive blocks of the host; the pause doubles + with every consecutive block and resets on a + successful fetch. + + claim_next_queue_item() asks wait_time() for every queue candidate: + a host that would have to wait is skipped, so the worker crawls + another domain instead and returns to the blocked host when its + backoff has expired. No worker ever sleeps in front of a + rate-limited host while other domains are crawlable. + """ + def __init__(self): - self.last_request = defaultdict(float) + self.last_request = {} + self.blocked_until = {} + self.block_count = {} + + @staticmethod + def _host(url): + try: + return (urlparse(url).netloc or "").lower() + except ValueError: + return "" + + def wait_time(self, url, delay, now=None): + """Seconds until url's host may be contacted again (0.0 = now).""" + if now is None: + now = time.monotonic() + host = self._host(url) + if not host: + return 0.0 + next_allowed = max( + self.last_request.get(host, 0.0) + delay, + self.blocked_until.get(host, 0.0), + ) + return max(0.0, next_allowed - now) + + def is_ready(self, url, delay, now=None): + return self.wait_time(url, delay, now) <= 0.0 def wait(self, url, delay): - host = urlparse(url).netloc.lower() - now = time.monotonic() - wait_for = delay - (now - self.last_request[host]) + """ + Enforce the per-host delay directly before a request. + Returns True when the request may start now (the wait was slept + inline). Returns False when the wait would exceed + MAX_INLINE_RATE_WAIT (e.g. a large robots.txt crawl-delay that + claim_next_queue_item could not know yet, because robots.txt is + only loaded during the first fetch): the caller then defers the + item instead of parking the worker, and the host's schedule + stays untouched, so the item is retried exactly when the delay + has passed. + """ + wait_for = self.wait_time(url, delay) + if wait_for > MAX_INLINE_RATE_WAIT: + return False if wait_for > 0: - logging.info("[RATE] %s: %.1fs", host, wait_for) + if wait_for >= 1.0: + logging.info("[RATE ] %s: %.1fs", self._host(url), wait_for) time.sleep(wait_for) + self.last_request[self._host(url)] = time.monotonic() + return True - self.last_request[host] = time.monotonic() + def _register_block(self, host, seconds): + count = self.block_count.get(host, 0) + 1 + self.block_count[host] = count + blocked = min(seconds, HOST_BLOCK_HARD_CAP) + self.blocked_until[host] = time.monotonic() + blocked + return count, blocked + + def note_success(self, url): + """Successful fetch: the host's consecutive-block counter resets.""" + self.block_count.pop(self._host(url), None) + + def note_http_error(self, url, status, retry_after=None): + """ + Pause the host after a block-ish HTTP status (429, 5xx, 403, + ...). An exponential backoff applies; a Retry-After header is + honored on top of it. + """ + rule = HOST_BLOCK_RULES.get(status) + if rule is None: + return + host = self._host(url) + base, cap = rule + seconds = min(cap, base * (2 ** self.block_count.get(host, 0))) + if retry_after is not None: + seconds = max(seconds, retry_after) + count, blocked = self._register_block(host, seconds) + logging.info( + "[BLOCK] %s: HTTP %d -> host paused for %.0fs (block #%d)", + host, status, blocked, count, + ) + + def note_network_error(self, url): + """Short backoff after DNS/timeout/connection failures.""" + host = self._host(url) + seconds = min( + HOST_BLOCK_NETWORK_MAX, + HOST_BLOCK_NETWORK_BASE * (2 ** self.block_count.get(host, 0)), + ) + count, blocked = self._register_block(host, seconds) + logging.info( + "[BLOCK] %s: network error -> host paused for %.0fs (block #%d)", + host, blocked, count, + ) + + +def _retry_after_seconds(headers): + """ + Parse a Retry-After header (delta-seconds or HTTP date). Returns + the number of seconds, or None when absent/unparseable. + """ + try: + value = headers.get("Retry-After") if headers is not None else None + except Exception: + return None + if not value: + return None + value = value.strip() + if value.isdigit(): + return float(value) + try: + retry_at = parsedate_to_datetime(value) + return max(0.0, retry_at.timestamp() - time.time()) + except Exception: + return None + + +def is_transient_error(error): + """ + True for fetch errors worth retrying later (the item is requeued + and the host gets a backoff): network-level failures and the HTTP + statuses in TRANSIENT_HTTP_STATUSES. Everything else (404, robots, + too large, not HTML, ...) stays a terminal failure. + """ + if not error: + return False + if error.startswith("network:"): + return True + match = re.match(r"^http_(\d{3})$", error) + return bool(match) and int(match.group(1)) in TRANSIENT_HTTP_STATUSES def is_skipped_url(url): @@ -1448,14 +1825,20 @@ def is_skipped_url(url): return any(pattern.match(url) for pattern in SKIP_URL_PATTERNS) -def fetch(url, limiter, robots): +def fetch(url, scheduler, robots): parsed = urlparse(url) if not robots.allowed(url): return None, None, "robots" delay = robots.delay(url) - limiter.wait(url, delay) + if not scheduler.wait(url, delay): + # The host needs a longer break than we are willing to sleep + # for inline (e.g. a large robots.txt crawl-delay discovered + # only now, because robots.txt is loaded on the first fetch): + # report it so the worker defers the item and crawls another + # domain in the meantime. + return None, None, "host_busy" req = Request( url, @@ -1501,8 +1884,18 @@ def fetch(url, limiter, robots): return status, text, None except HTTPError as e: + # Blocked or rate-limited? Put the host on backoff so the next + # claims automatically prefer other domains. + scheduler.note_http_error( + url, + e.code, + retry_after=_retry_after_seconds( + getattr(e, "headers", None) + ), + ) return e.code, None, f"http_{e.code}" except (URLError, TimeoutError, OSError) as e: + scheduler.note_network_error(url) return None, None, f"network:{e}" except Exception as e: return None, None, f"error:{e}" @@ -1702,10 +2095,11 @@ def worker(worker_id): conn = None robots = RobotsCache() - limiter = RateLimiter() + scheduler = HostScheduler() crawled = 0 failed = 0 + retried = 0 last_stats = time.monotonic() current_queue_id = None update_requested = False @@ -1742,11 +2136,26 @@ def worker(worker_id): update_requested = True break - item = safe_commit(conn, lambda: get_next_queue_item(conn)) + item, wait_hint = safe_commit( + conn, + lambda: claim_next_queue_item(conn, scheduler, robots), + ) if not item: - logging.info("Queue empty. Worker waiting...") - time.sleep(5) + if wait_hint: + # Every candidate host is rate-limited or blocked + # right now; sleep until the soonest one becomes + # crawlable (capped, then the queue is re-scanned). + sleep_for = min(wait_hint, QUEUE_MAX_IDLE_SLEEP) + logging.info( + "[WAIT ] All candidate hosts busy or blocked; " + "worker sleeps %.1fs", + sleep_for, + ) + else: + sleep_for = 5 + logging.info("Queue empty. Worker waiting...") + time.sleep(sleep_for) continue queue_id, url, source_url = item @@ -1770,12 +2179,53 @@ def worker(worker_id): ) continue + # Hosts that may never be crawled (localhost, raw IPs, + # non-real domains) can still sit in the queue from before + # this rule existed: drop them here so they are not + # fetched and not retried endlessly. + if is_non_crawlable_url(url): + logging.info("[SKIP ] %s (non-crawlable host)", url) + safe_commit( + conn, + lambda: finish_queue_item( + conn, queue_id, success=True + ), + ) + continue + started = time.monotonic() - status, html, error = fetch(url, limiter, robots) + status, html, error = fetch(url, scheduler, robots) if error: logging.info("[HTTP ] %s -> %s", url, error) + + if error == "host_busy": + # Host needs a longer break than we sleep for + # inline: put the item back (without burning an + # attempt) and crawl another domain instead. + safe_commit( + conn, + lambda: defer_queue_item(conn, queue_id), + ) + continue + + if is_transient_error(error): + # The host is already on backoff (set in fetch); + # requeue the URL for a later attempt. + new_status = safe_commit( + conn, + lambda: requeue_queue_item(conn, queue_id), + ) + retried += 1 + logging.info( + "[RETRY] %s -> %s (max %d attempts)", + url, + new_status or "requeued", + QUEUE_MAX_ATTEMPTS, + ) + continue + safe_commit( conn, lambda: finish_queue_item( @@ -1786,6 +2236,7 @@ def worker(worker_id): continue logging.info("[HTTP ] %s -> HTTP %s", url, status) + scheduler.note_success(url) # A redirect normally arrives as the final URL in urllib. # The requested URL remains the canonical queue identity. @@ -1941,8 +2392,8 @@ def worker(worker_id): conn.close() logging.info( - "Stopped. crawled=%d failed=%d", - crawled, failed + "Stopped. crawled=%d failed=%d retried=%d", + crawled, failed, retried ) if update_requested: @@ -1961,6 +2412,7 @@ def main(): print(f"Start: {START_URL}") print(f"Workers: {WORKERS}") print(f"Delay: {DEFAULT_DELAY:.1f} Sekunden") + print("Auto-Switch: aktiv (blockierte Domains werden übersprungen)") print(f"Robots-Cache: {ROBOTS_CACHE_SECONDS // 3600} Stunden") print(f"DB: {DB_HOST}:{DB_PORT}/{DB_NAME}") print(f"Version: {CRAWLER_VERSION}") @@ -1975,7 +2427,19 @@ def main(): safe_commit(conn, lambda: requeue_stale_items(conn)) # Seed URL. ON CONFLICT makes restarting harmless. - queue_urls(conn, [normalize_url(START_URL)], None) + seed_url = normalize_url(START_URL) + if seed_url is None: + # Either an invalid URL or a non-crawlable host (localhost, + # raw IP, non-real domain): crawling can never work, so fail + # clearly instead of spinning on an empty queue. + conn.close() + logging.error( + "Start URL %s is not crawlable (invalid or non-public " + "host such as localhost or an IP address).", + START_URL, + ) + return EXIT_ERROR + queue_urls(conn, [seed_url], None) conn.close() @@ -2043,6 +2507,7 @@ def main(): sys.exit(EXIT_ERROR) print("Crawler stopped.") + return EXIT_OK # ============================================================ @@ -2257,8 +2722,7 @@ def main_cli(argv=None): if args.init_deployment: return cmd_init_deployment() - main() - return EXIT_OK + return main() if __name__ == "__main__": diff --git a/priorityweb/jonassearch-priority.service b/priorityweb/jonassearch-priority.service index e81f734..5ae0b4b 100644 --- a/priorityweb/jonassearch-priority.service +++ b/priorityweb/jonassearch-priority.service @@ -19,11 +19,11 @@ Type=simple User=jonassearch Group=jonassearch -WorkingDirectory=/srv/serverjonas/jonassearch +WorkingDirectory=/srv/serverjonas/search EnvironmentFile=/srv/serverjonas/search/.env -ExecStart=/usr/bin/python3 /srv/serverjonas/jonassearch/priorityweb/app.py +ExecStart=/usr/bin/python3 /srv/serverjonas/search/priorityweb/app.py Restart=on-failure RestartSec=10 diff --git a/search/README.md b/search/README.md index d5e13eb..5e09ab5 100644 --- a/search/README.md +++ b/search/README.md @@ -1,74 +1,178 @@ -# JonasSearch – Suchmaschine +# JonasSearch Suchmaschine (`search/`) -Volltextsuche über den Index, den `crawler.py` in PostgreSQL aufgebaut hat. -Die Engine sucht in der Tabelle `pages` und nutzt dabei PostgreSQLs -Volltextsuche (`tsvector` / `tsquery`). +Die eigentliche Websuchmaschine über der PostgreSQL-Datenbank `jonassearch`: +Volltextsuche auf dem vorhandenen `tsvector`-Index (`pages.search_vector`, +GIN-Index `idx_pages_search`) mit mehrstufigem Ranking, Filtern, Snippets, +Weboberfläche, HTTP-API und Kommandozeile. -## Starten - -CLI: - -```bash -python3 search/cli.py "raspberry pi" -python3 search/cli.py "postgres" --domain example.com --page 2 -python3 search/cli.py "amplifier" --json -``` - -HTTP-API + Suchseite: - -```bash -python3 search/api.py -# Suchseite: http://localhost:8082/ -# JSON-API: http://localhost:8082/search?q=raspberry%20pi -``` - -Als Bibliothek: - -```python -from search import SearchEngine - -engine = SearchEngine() -result = engine.search("postgres", page=1, per_page=10) - -for hit in result.hits: - print(hit.position, hit.title, hit.url) -``` - -## Funktionsweise - -- **Index-Nutzung:** Die Engine sucht automatisch die tsvector-Spalte der - Tabelle `pages` (z. B. `search_vector`, die ein Trigger pflegt) und - durchsucht sie per `@@`-Operator. Fehlt eine solche Spalte, wird der - Suchvektor zur Laufzeit aus `title` + `headings` + `content` berechnet – - die Suche funktioniert dann genauso, nur langsamer. -- **Ranking:** Sortierung nach `ts_rank` absteigend, danach nach - Crawl-Zeitpunkt. -- **Snippets:** `ts_headline` erzeugt Ausschnitte, in denen die Suchbegriffe - mit `` hervorgehoben sind. -- **Präfix-Suche:** Jeder Begriff wird mit `:*` gesucht, „raspberr" findet - also auch „raspberry". -- **UND-Verknüpfung:** Mehrere Begriffe müssen alle vorkommen. -- **Paginierung:** `--page` / `--per-page` (max. 50). -- **Filter:** `--domain` (exakte Domain) und `--tld` (z. B. `.de`). - -## Umgebungsvariablen - -Wie der Crawler (siehe `.env.example`): - -| Variable | Bedeutung | Standard | +| Baustein | Zweck | Port | |---|---|---| -| `JONASSEARCH_DB_HOST` | DB-Host | `127.0.0.1` | -| `JONASSEARCH_DB_PORT` | DB-Port | `25432` | -| `JONASSEARCH_DB_NAME` | DB-Name | `jonassearch` | -| `JONASSEARCH_DB_USER` | DB-User | `jonassearch` | -| `JONASSEARCH_DB_PASSWORD` | DB-Passwort | leer | -| `JONASSEARCH_TS_CONFIG` | PostgreSQL-Textsuch-Konfiguration | `german` | -| `JONASSEARCH_VECTOR_COLUMN` | tsvector-Spalte erzwingen (optional) | automatisch | -| `JONASSEARCH_API_HOST` | Bind-Adresse der API | `0.0.0.0` | -| `JONASSEARCH_API_PORT` | Port der API | `8082` | +| `search/api.py` | HTTP-API + Weboberfläche | **8083** | +| `search/cli.py` | Kommandozeile (ohne Webserver) | – | +| `search/engine.py` | Suchlogik (von api/cli genutzt) | – | +| `search/jonassearch-engine.service` | systemd-Dienst | 8083 | -## Abhängigkeiten +Bestehende Dienste: Statserver 8081 (`web/app.py`), Vorschlagsverfeinerer +8082 (`priorityweb/app.py`), Suchmaschine **8083**. -- `psycopg2` (DB-Zugriff) -- `python-dotenv` (`.env`-Datei) -- `flask` (nur für `search/api.py`) \ No newline at end of file +## Wie gesucht wird + +1. **Retrieval:** Die Anfrage wird per `websearch_to_tsquery` geparst und + per `@@` gegen `pages.search_vector` über den GIN-Index ausgewertet. + Nutzersyntax: `wort1 wort2` (UND), `"exakte phrase"`, `OR`, + `-ausschluss`. Da die Suchvektoren mit der Textsuch-Konfiguration + `simple` gebaut werden (DB-Trigger), matched die Suche ohne + Stemming – exakte Lexeme, Groß-/Kleinschreibung egal. +2. **Ranking (in SQL):** + + ```text + score = ts_rank_cd(Gewichte, search_vector, tsquery, Norm=8) + – Titel=A, Überschriften=B, Inhalt=C + * (1 + 0.25 · exp(−Alter_in_Tagen / 180)) – Aktualität (crawled_at/discovered_at) + ``` + + Die Gewichte sind über `JONASSEARCH_RANK_WEIGHTS` einstellbar + (`A=1.0,B=0.6,C=0.3,D=0.1`). Die ts_rank_cd-Normalisierung (Bitmaske 8 + = Rang / (1 + log Dokumentlänge), `JONASSEARCH_RANK_NORMALIZATION`) + verhindert, dass riesige Seiten kurze, präzise Treffer dominieren. + Nicht in die Formel eingeflossen (noch): Link-Autorität aus + `pages.links`/`source_url` – dafür liegt mit dem gespeicherten + Webgraph die Grundlage bereit. + + **Fallback bei sehr häufigen Begriffen:** Ist das volle Ranking über + alle Kandidaten nach `JONASSEARCH_STATEMENT_TIMEOUT_MS` nicht fertig + (z. B. Einzelwort mit zigtausend Treffern), wird automatisch ein + zweiter Versuch über die ersten `JONASSEARCH_MAX_RANK_CANDIDATES` + (Default 5000) Kandidaten gefahren – die Antwort kommt dann garantiert, + die Antwortmarke `ranking_capped: true` zeigt die Einschränkung an. + +3. **Filter** (kombinierbar): `site=` (Domain inkl. Subdomains), `tld=`, + `after=`/`before=` (ISO-Datum oder `24h`/`7d`/`30d`/`365d`), nur + `http_status = 200` und nur HTML-Inhalt (jeweils abschaltbar). +4. **Snippets:** `ts_headline` wird nur für die Treffer der angeforderten + Seite berechnet – `content` (TOAST-lastig) wird nie für die ganze + Kandidatenmenge gelesen. Markierungen im Snippet: `⟦…⟧`. + +## Schutz der Produktivdatenbank (~10 GB) + +- Jede Operation nutzt eine eigene Verbindung mit `statement_timeout` + (Default 10 s); Zeitüberschreitung → HTTP 504 bzw. CLI-Hinweis zum + Eingrenzen, niemals eine blockierte Datenbank. +- Die Gesamt-Trefferzahl läuft mit eigenem, kürzerem Timeout (4 s) und + ist **best effort**: bei Überschreitung liefert die API `total: null` + („viele Treffer“), die Ergebnisse bleiben unbeeinflusst. +- `/api/stats` nutzt nur Schätzwerte aus `pg_class`/`pg_stats`, keine + Full-Scans. + +## HTTP-API (Port 8083) + +| Endpunkt | Beschreibung | +|---|---| +| `GET /` | Weboberfläche (Suchseite, Deutsch, Dark/Light) | +| `GET /api/search` | Suche – Parameter siehe unten | +| `GET /api/suggest?q=begr` | Autocomplete (Präfix auf letztem Wort, `to_tsquery 'begr:*'`) | +| `GET /api/page?id=42` oder `?url=…` | gespeicherte Seite (Inhalt gekürzt) | +| `GET /api/stats` | Index-Statistik (Seiten, Domains, Queue, DB-Größe) | +| `GET /health` | Healthcheck inkl. DB-Status | + +Parameter von `/api/search`: + +| Parameter | Bedeutung | Default | +|---|---|---| +| `q` | Suchbegriffe (Pflicht) | – | +| `page`, `per_page` | Paginierung (max 50/Seite) | 1, 10 | +| `site` | Domain-Filter, mehrfach möglich (`site=example.com`) | – | +| `tld` | TLD-Filter, mehrfach möglich (`tld=de`) | – | +| `after` / `before` | Zeitfilter: ISO oder `24h`/`7d`/`2w` | – | +| `sort` | `relevance` (Default) oder `freshness` | relevance | +| `include_errors=1` | auch Seiten ohne HTTP 200 | aus | +| `include_non_html=1` | auch Nicht-HTML-Dokumente | aus | +| `no_snippets=1` | Snippets weglassen (schneller) | aus | + +Beispiel: + +```bash +curl "http://localhost:8083/api/search?q=volltextsuche&site=postgresql.org&after=30d" +``` + +Antwort (Auszug): + +```json +{ + "query": "volltextsuche", + "parsed_query": "'volltextsuche'", + "page": 1, "per_page": 10, + "total": 1234, + "has_more": true, + "results": [ + { + "id": 42, "url": "https://…", "title": "…", + "domain": "…", "tld": ".de", "http_status": 200, + "crawled_at": "2026-09-30T18:12:03+00:00", + "score": 0.081203, "text_rank": 0.078341, "freshness": 1.074, + "snippet": "… PostgreSQL ⟦Volltextsuche⟧ mit tsvector …" + } + ], + "took_ms": 31.4 +} +``` + +Fehler: `400` ungültige Parameter · `404` Seite unbekannt · +`503` Datenbank nicht erreichbar · `504` Zeitlimit (bitte eingrenzen). + +## Kommandozeile + +```bash +python3 search/cli.py "postgresql volltextsuche" +python3 search/cli.py "crawler" --site example.com --sort freshness --limit 20 +python3 search/cli.py "python" --after 7d +python3 search/cli.py --suggest "suchmasch" +python3 search/cli.py --stats +python3 search/cli.py --page-id 42 +python3 search/cli.py "test" --json +``` + +## Installation als Dienst + +```bash +sudo cp /srv/serverjonas/search/search/jonassearch-engine.service \ + /etc/systemd/system/ +sudo systemctl daemon-reload +sudo systemctl enable --now jonassearch-engine.service +journalctl -u jonassearch-engine -f +``` + +Port-Auflösung: `--port` (manueller Start) > `JONASSEARCH_API_PORT` +(`.env`) > **8083**. Der systemd-Dienst pinnt `JONASSEARCH_API_PORT=8083` +*nach* dem EnvironmentFile und überschreibt damit einen eventuell +abweichenden Wert aus `.env` – der Dienst läuft garantiert auf 8083. + +## Konfiguration (.env) + +| Variable | Default | Bedeutung | +|---|---|---| +| `JONASSEARCH_DB_HOST/PORT/NAME/USER/PASSWORD` | wie crawler.py | Datenbankzugriff | +| `JONASSEARCH_API_HOST` | `0.0.0.0` | Bind-Adresse | +| `JONASSEARCH_API_PORT` | `8083` | Port (Dienst pinnt 8083) | +| `JONASSEARCH_TS_CONFIG` | `simple` | Muss zur Trigger-Konfiguration des Suchvektors passen! | +| `JONASSEARCH_VECTOR_COLUMN` | auto | tsvector-Spalte erzwingen (sonst automatische Erkennung) | +| `JONASSEARCH_RANK_WEIGHTS` | `A=1.0,B=0.6,C=0.3,D=0.1` | ts_rank_cd-Gewichte | +| `JONASSEARCH_FRESHNESS_WEIGHT` | `0.25` | Stärke des Aktualitätsbonus | +| `JONASSEARCH_FRESHNESS_HALF_LIFE_DAYS` | `180` | Halbwertszeit der Aktualität | +| `JONASSEARCH_RANK_NORMALIZATION` | `8` | ts_rank_cd-Normalisierung (8 = 1+log Länge, 0 = aus) | +| `JONASSEARCH_MAX_RANK_CANDIDATES` | `5000` | Kandidaten-Fenster des Ranking-Fallbacks | +| `JONASSEARCH_STATEMENT_TIMEOUT_MS` | `10000` | Timeout pro Suchstatement | +| `JONASSEARCH_COUNT_TIMEOUT_MS` | `4000` | Timeout der Trefferzählung | +| `JONASSEARCH_RESULTS_PER_PAGE` | `10` | Treffer pro Seite | +| `JONASSEARCH_MAX_RESULTS_PER_PAGE` | `50` | Maximum pro Seite | +| `JONASSEARCH_HEADLINE_SOURCE_CHARS` | `100000` | `content`-Anteil für ts_headline | + +## Tests + +```bash +python3 search/test_engine.py +``` + +Getestet werden die DB-freien Teile: Parameter-Normalisierung, Gewichte, +Zeitfilter, SQL-Zusammensetzung, Query-Parsing. Alles ohne +Datenbankzugriff. diff --git a/search/__init__.py b/search/__init__.py index c9fa74c..18ccdc6 100644 --- a/search/__init__.py +++ b/search/__init__.py @@ -1,22 +1,10 @@ """ -JonasSearch – Suchmaschine über den PostgreSQL-Index des Crawlers. +JonasSearch Suchmaschine (Package `search`). -Public API: - from search import SearchEngine, SearchHit, SearchResponse +Module: + engine – Suchlogik über PostgreSQL-Volltextindex (pages.search_vector) + api – HTTP-API + Weboberfläche (Flask, Port 8083) + cli – Kommandozeilen-Werkzeug (python3 search/cli.py) """ -from .engine import ( - SearchEngine, - SearchHit, - SearchResponse, - parse_terms, -) - -__all__ = [ - "SearchEngine", - "SearchHit", - "SearchResponse", - "parse_terms", -] - -__version__ = "1.0.0" \ No newline at end of file +__version__ = "1.0.0" diff --git a/search/api.py b/search/api.py index 14ac90f..41ef634 100644 --- a/search/api.py +++ b/search/api.py @@ -1,395 +1,588 @@ #!/usr/bin/env python3 """ -JonasSearch – Suchmaschine (HTTP-API). +JonasSearch Suchmaschine – HTTP-API + Weboberfläche (api.py) -Einfacher Suchserver über den vom Crawler aufgebauten Index: +Läuft auf PORT = Vorschlagsverfeinerer (8082) + 1 = 8083. - python3 search/api.py +Endpunkte: + GET / Weboberfläche (Suchseite) + GET /api/search Volltextsuche (q, page, per_page, site, tld, + after, before, sort, include_errors, + include_non_html, snippets) + GET /api/suggest Autocomplete (q, limit) + GET /api/page einzelne Seite (id oder url) + GET /api/stats Index-Statistiken + GET /health Healthcheck (DB-Status) - curl 'http://localhost:8082/search?q=raspberry%20pi' - -Unter / gibt es eine kleine Suchseite, unter /search die JSON-API. - -Konfiguration über Umgebungsvariablen: - JONASSEARCH_DB_HOST / PORT / NAME / USER / PASSWORD - JONASSEARCH_TS_CONFIG - JONASSEARCH_API_HOST (Standard: 0.0.0.0) - JONASSEARCH_API_PORT (Standard: 8082) +Port-Auflösung: --port > JONASSEARCH_API_PORT (.env) > 8083. +Der systemd-Dienst (search/jonassearch-engine.service) pinnt +JONASSEARCH_API_PORT=8083 NACH dem EnvironmentFile und überschreibt +damit einen eventuell abweichenden Wert aus .env. """ import logging import os +import sys +import traceback +from pathlib import Path -from flask import Flask, jsonify, render_template_string, request +from dotenv import load_dotenv -try: - from .engine import SearchEngine -except ImportError: # python3 search/api.py (ohne -m) - from engine import SearchEngine +# .env laden BEVOR search.engine importiert wird (liest die +# JONASSEARCH_*-Variablen bei Importzeit): erst Working Directory, +# dann .env neben dem Projektstamm – wie launcher.py. +for _dotenv_path in (None, Path(__file__).resolve().parent.parent / ".env"): + try: + load_dotenv(_dotenv_path) + except OSError: + pass -API_HOST = os.getenv("JONASSEARCH_API_HOST", "0.0.0.0") -API_PORT = int(os.getenv("JONASSEARCH_API_PORT", "8082")) +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) -PAGE_HTML = """\ +import psycopg2 # noqa: E402 +from flask import Flask, jsonify, render_template_string, request # noqa: E402 + +try: # Paketkontext (python3 -m search.api) + from . import engine # noqa: E402 +except ImportError: # Direktstart: python3 search/api.py + from search import engine # noqa: E402 + +logging.basicConfig( + level=logging.INFO, + format="%(asctime)s [search-api] %(levelname)s %(message)s", + datefmt="%H:%M:%S", +) +log = logging.getLogger("jonassearch.search-api") + +DEFAULT_PORT = 8083 + +app = Flask(__name__) + + +# ============================================================ +# HELFER +# ============================================================ + +def _truthy(value): + return str(value or "").strip().lower() in ("1", "true", "yes", "on", "ja") + + +def _page_param(name, default): + raw = request.args.get(name) + if raw is None or str(raw).strip() == "": + return default + try: + return int(raw) + except ValueError: + raise engine.QueryError(f"{name} muss eine Zahl sein.") + + +def _search_params(): + """Query-Parameter der Request -> engine.search()-Argumente.""" + q = request.args.get("q", "").strip() + if not q: + raise engine.QueryError( + "Parameter 'q' (Suchbegriffe) fehlt oder ist leer." + ) + return dict( + query=q, + page=_page_param("page", 1), + per_page=_page_param("per_page", None), + sites=request.args.getlist("site") + request.args.getlist("sites"), + tlds=request.args.getlist("tld") + request.args.getlist("tlds"), + after=request.args.get("after"), + before=request.args.get("before"), + sort=request.args.get("sort", "relevance"), + include_errors=_truthy(request.args.get("include_errors")), + include_non_html=_truthy(request.args.get("include_non_html")), + with_snippets=not _truthy(request.args.get("no_snippets")), + ) + + +# ============================================================ +# API-ENDPUNKTE +# ============================================================ + +@app.route("/api/search") +def api_search(): + try: + result = engine.search(**_search_params()) + return jsonify(result) + except engine.QueryError as exc: + return jsonify({"error": str(exc)}), 400 + except engine.SearchTimeout as exc: + return jsonify({"error": str(exc)}), 504 + except psycopg2.OperationalError as exc: + log.error("Datenbank nicht erreichbar: %s", exc) + return jsonify({"error": "Datenbank nicht erreichbar."}), 503 + except engine.SearchEngineError as exc: + log.error("Suchfehler: %s", exc) + return jsonify({"error": str(exc)}), 500 + except Exception: + log.error("Unerwarteter Fehler in /api/search:\n%s", + traceback.format_exc()) + return jsonify({"error": "Interner Fehler."}), 500 + + +@app.route("/api/suggest") +def api_suggest(): + try: + limit = _page_param("limit", engine.MAX_SUGGEST) + return jsonify({ + "query": request.args.get("q", ""), + "suggestions": engine.suggest(request.args.get("q", ""), limit), + }) + except engine.QueryError as exc: + return jsonify({"error": str(exc)}), 400 + except psycopg2.OperationalError: + # Vorschläge sind optional: still leere Liste liefern. + return jsonify({"query": request.args.get("q", ""), + "suggestions": []}) + + +@app.route("/api/page") +def api_page(): + try: + page_id = request.args.get("id") + url = request.args.get("url") + page = engine.get_page( + page_id=page_id, + url=url, + max_content_chars=_page_param( + "max_content_chars", engine.PAGE_CONTENT_CHARS + ), + ) + if page is None: + return jsonify({"error": "Seite nicht gefunden."}), 404 + return jsonify(page) + except engine.QueryError as exc: + return jsonify({"error": str(exc)}), 400 + except psycopg2.OperationalError as exc: + log.error("Datenbank nicht erreichbar: %s", exc) + return jsonify({"error": "Datenbank nicht erreichbar."}), 503 + + +@app.route("/api/stats") +def api_stats(): + try: + return jsonify(engine.stats()) + except psycopg2.OperationalError as exc: + log.error("Datenbank nicht erreichbar: %s", exc) + return jsonify({"error": "Datenbank nicht erreichbar."}), 503 + + +@app.route("/health") +def health(): + status = engine.health() + return jsonify(status), (200 if status.get("ok") else 503) + + +# ============================================================ +# WEBOBERFLÄCHE +# ============================================================ + +HTML = r""" - - - JonasSearch - - + + +JonasSearch – Websuche + - -
-

JonasSearch

-
Suchmaschine über den Crawler-Index
+
+
+

JonasSearch

+ +
- +
+
+ + + +
+
+ + + +
+
-
-
- +
+
+ + + +
+ Beispiele: postgreSQL volltextsuche · + "exacte phrase" · + crawler -python · + postgres OR sqlite +
+ +
JonasSearch Suchmaschine · PostgreSQL-Volltextsuche + über gecrawlte Webseiten · Port 8083
- """ -def create_app() -> Flask: - app = Flask(__name__) +@app.route("/") +def index(): + return render_template_string(HTML) - # Umlaute nicht als \uXXXX escapen. - try: - app.json.ensure_ascii = False - except AttributeError: - pass - - engine = SearchEngine() - - @app.route("/") - def index(): - return render_template_string(PAGE_HTML) - - @app.route("/search") - def search(): - query = request.args.get("q", "").strip() - - if not query: - return jsonify({"error": "Parameter 'q' fehlt."}), 400 - - page_arg = request.args.get("page", "1") - per_page_arg = request.args.get("per_page", "10") - - try: - page = max(1, int(page_arg)) - except ValueError: - return jsonify({"error": "Parameter 'page' muss eine Zahl sein."}), 400 - - try: - per_page = max(1, min(int(per_page_arg), 50)) - except ValueError: - return jsonify({"error": "Parameter 'per_page' muss eine Zahl sein."}), 400 - - domain = request.args.get("domain") or None - tld = request.args.get("tld") or None - - try: - response = engine.search( - query=query, - page=page, - per_page=per_page, - domain=domain, - tld=tld, - ) - except Exception as exc: - logging.exception("Suche fehlgeschlagen") - return jsonify({"error": f"Suche fehlgeschlagen: {exc}"}), 500 - - return jsonify(response.to_dict()) - - return app +# ============================================================ +# START +# ============================================================ if __name__ == "__main__": - logging.basicConfig(level=logging.INFO) + import argparse - print("=" * 60) - print(" JonasSearch – Suchmaschine (HTTP-API)") - print(f" URL: http://{API_HOST}:{API_PORT}/") - print(f" JSON: http://{API_HOST}:{API_PORT}/search?q=...") - print("=" * 60) + parser = argparse.ArgumentParser( + prog="search/api.py", + description="JonasSearch Suchmaschine (HTTP-API + Weboberfläche)", + ) + parser.add_argument("--host", default=None, + help="Bind-Adresse (Default: JONASSEARCH_API_HOST " + "oder 0.0.0.0)") + parser.add_argument("--port", type=int, default=None, + help="Port (Default: JONASSEARCH_API_PORT oder 8083)") + args = parser.parse_args() - create_app().run(host=API_HOST, port=API_PORT, threaded=True) \ No newline at end of file + host = args.host or os.getenv("JONASSEARCH_API_HOST", + os.getenv("HOST", "0.0.0.0")) + port = args.port or int(os.getenv("JONASSEARCH_API_PORT", + str(DEFAULT_PORT))) + + print("=" * 70) + print(" JonasSearch Suchmaschine") + print("=" * 70) + print(f"Statserver : Port 8081 (jonassearch-stats.service)") + print(f"Priorität : Port 8082 (jonassearch-priority.service)") + print(f"Suchmaschine: http://{host}:{port}") + print(f"DB : {engine.DB_HOST}:{engine.DB_PORT}/{engine.DB_NAME}") + print(f"TS-Config : {engine.TS_CONFIG}") + print("=" * 70) + + app.run( + host=host, + port=port, + debug=False, + threaded=True, + load_dotenv=False, # .env wurde oben schon geladen + ) diff --git a/search/cli.py b/search/cli.py index 79ec803..1fd3810 100644 --- a/search/cli.py +++ b/search/cli.py @@ -1,153 +1,259 @@ #!/usr/bin/env python3 """ -JonasSearch – Suchmaschine (Kommandozeile). +JonasSearch Suchmaschine – Kommandozeile (cli.py) -Suche im vom Crawler aufgebauten Index direkt vom Terminal: - - python3 search/cli.py "raspberry pi" - python3 search/cli.py "postgres" --domain example.com --page 2 - python3 search/cli.py "amplifier" --json - -Konfiguration über Umgebungsvariablen (siehe .env.example): - JONASSEARCH_DB_HOST / PORT / NAME / USER / PASSWORD - JONASSEARCH_TS_CONFIG +Beispiele: + python3 search/cli.py "postgresql volltextsuche" + python3 search/cli.py "crawler" --site example.com --sort freshness + python3 search/cli.py "python" --after 7d --limit 20 + python3 search/cli.py --suggest "suchmasch" + python3 search/cli.py --stats + python3 search/cli.py --page-id 42 + python3 search/cli.py --url https://example.com/seite + python3 search/cli.py "test" --json """ import argparse -import html import json -import re import sys +import textwrap +from pathlib import Path -try: - from .engine import SearchEngine -except ImportError: # python3 search/cli.py (ohne -m) - from engine import SearchEngine +from dotenv import load_dotenv -_TAG_RE = re.compile(r"<[^>]+>") +# .env laden, bevor engine importiert wird (wie api.py). +for _dotenv_path in (None, Path(__file__).resolve().parent.parent / ".env"): + try: + load_dotenv(_dotenv_path) + except OSError: + pass + +try: # Paketkontext (python3 -m search.cli) + from . import engine +except ImportError: # Direktstart: python3 search/cli.py + sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) + from search import engine -def strip_markup(text: str) -> str: - """Entfernt -Tags und dekodiert HTML-Entities fürs Terminal.""" - text = _TAG_RE.sub("", text or "") - return html.unescape(text) +# ------------------------------------------------------------ +# Ausgabe-Helfer +# ------------------------------------------------------------ + +class Style: + def __init__(self, enabled): + self.bold = "\033[1m" if enabled else "" + self.green = "\033[32m" if enabled else "" + self.blue = "\033[36m" if enabled else "" + self.dim = "\033[2m" if enabled else "" + self.reset = "\033[0m" if enabled else "" -def format_plain(response, show_rank=True) -> str: - lines = [] - total = response.total - unit = "Ergebnis" if total == 1 else "Ergebnisse" - - header = f"{total} {unit} für „{response.query}“ ({response.took_ms} ms)" - if response.pages > 0: - header += f" – Seite {response.page}/{response.pages}" - lines.append(header) - lines.append("") - - for hit in response.hits: - if show_rank: - lines.append(f"{hit.position:4d}. {hit.title}") - else: - lines.append(f" {hit.title}") - - lines.append(f" {hit.url}") - - if hit.snippet: - lines.append(f" {strip_markup(hit.snippet)}") - - meta = [] - if hit.domain: - meta.append(hit.domain) - if hit.crawled_at: - meta.append(hit.crawled_at[:10]) - if hit.rank: - meta.append(f"Relevanz {hit.rank:.4f}") - if meta: - lines.append(f" {' · '.join(meta)}") - - lines.append("") - - if not response.hits: - lines.append("Keine Treffer. Versuche weniger oder andere Begriffe.") - - return "\n".join(lines) +def _highlight(snippet, style): + """⟦…⟧-Markierungen -> ANSI-Hervorhebung (oder Plaintext).""" + return (snippet + .replace(engine.SNIPPET_START, + style.bold + style.green) + .replace(engine.SNIPPET_STOP, style.reset)) -def build_parser() -> argparse.ArgumentParser: +def _fmt_date(iso): + if not iso: + return "–" + return str(iso)[:16].replace("T", " ") + + +def print_results(data, style): + print( + f"{style.dim}" + f"{'Etwa ' if data['total'] is not None else ''}" + f"{data['total'] if data['total'] is not None else 'Viele'} " + f"Treffer · Seite {data['page']} · " + f"{data['took_ms'] / 1000:.2f} s" + f"{style.reset}" + ) + if not data["parsed_query"]: + print(f"{style.dim}Hinweis: Anfrage enthält keine durchsuchbaren " + f"Begriffe.{style.reset}") + if not data["results"]: + print("Keine Treffer.") + return + + for i, r in enumerate(data["results"], start=1): + print() + print(f"{style.bold}{i}. {r['title']}{style.reset}") + print(f"{style.blue}{r['url']}{style.reset}") + if r["snippet"]: + wrapped = textwrap.fill( + _highlight(r["snippet"], style), width=100, + initial_indent=" ", subsequent_indent=" ", + ) + print(wrapped) + meta = (f" {style.dim}{r['domain'] or '–'} · " + f"{_fmt_date(r['crawled_at'])} · " + f"Score {r['score']:.4f}") + if r["http_status"] not in (200, None): + meta += f" · HTTP {r['http_status']}" + print(meta + style.reset) + + print() + nav = [] + if data["page"] > 1: + nav.append(f"--page {data['page'] - 1}") + if data["has_more"]: + nav.append(f"--page {data['page'] + 1}") + if nav: + print(f"{style.dim}Weitere Seite: " + " | ".join(nav) + style.reset) + + +def print_page(page, style, max_chars): + print(f"{style.bold}{page.get('title') or '(ohne Titel)'}{style.reset}") + print(f"{style.blue}{page['url']}{style.reset}") + print( + f"{style.dim}ID {page['id']} · {page['domain'] or '–'} · " + f"HTTP {page['http_status']} · {page['content_type'] or '–'} · " + f"{page['content_length']} Zeichen" + f"{' (gekürzt)' if page['content_truncated'] else ''}\n" + f"Gefunden am {_fmt_date(page.get('discovered_at'))} · " + f"Gecrawlt am {_fmt_date(page.get('crawled_at'))} · " + f"{page['crawl_time_ms'] or '–'} ms · " + f"{page['links_total']} Links{style.reset}" + ) + if page.get("headings"): + print(f"\n{style.bold}Überschriften:{style.reset}") + for h in page["headings"][:15]: + print(f" – {h}") + content = (page.get("content") or "")[:max_chars] + if content: + print(f"\n{style.bold}Inhalt (Auszug):{style.reset}") + print(textwrap.fill(content, width=100)) + + +def print_stats(s, style): + def fmt(value): + return "–" if value is None else f"{value:,}".replace(",", ".") + + print(f"{style.bold}JonasSearch Index-Statistik{style.reset}") + print(f" Seiten (geschätzt) : {fmt(s.get('pages'))}") + print(f" Domains (geschätzt): {fmt(s.get('domains'))}") + print(f" Queue gesamt : {fmt(s.get('queue'))}") + print(f" Queue ausstehend : {fmt(s.get('queue_pending'))}") + print(f" Letzter Crawl : {_fmt_date(s.get('last_crawl'))}") + print(f" Datenbankgröße : {s.get('db_size') or '–'}") + + +# ------------------------------------------------------------ +# Main +# ------------------------------------------------------------ + +def build_parser(): parser = argparse.ArgumentParser( - prog="jonassearch", - description="JonasSearch – Suchmaschine (Kommandozeile).", - epilog=( - "Konfiguration über JONASSEARCH_* Umgebungsvariablen, " - "siehe .env.example." - ), + prog="search/cli.py", + description="JonasSearch Volltextsuche über der PostgreSQL-" + "Datenbank (pages.search_vector).", + epilog='Query-Syntax: "exakte Phrase", OR, -ausschluss', ) - - parser.add_argument( - "query", - help="Suchabfrage, z. B. \"raspberry pi\"", - ) - parser.add_argument( - "--page", - type=int, - default=1, - help="Seite der Ergebnisse (Standard: 1)", - ) - parser.add_argument( - "--per-page", - type=int, - default=10, - help="Ergebnisse pro Seite, max. 50 (Standard: 10)", - ) - parser.add_argument( - "--domain", - default=None, - help="Nur Seiten dieser exakten Domain (z. B. example.com)", - ) - parser.add_argument( - "--tld", - default=None, - help="Nur Seiten dieser TLD (z. B. .de oder .com)", - ) - parser.add_argument( - "--config", - default=None, - help="PostgreSQL-Textsuch-Konfiguration " - "(Standard: german, oder JONASSEARCH_TS_CONFIG)", - ) - parser.add_argument( - "--json", - action="store_true", - help="Ergebnisse als JSON ausgeben", - ) - + parser.add_argument("query", nargs="?", + help="Suchbegriffe (Websearch-Syntax)") + parser.add_argument("--site", action="append", default=[], + help="Auf Domain einschränken (mehrfach möglich)") + parser.add_argument("--tld", action="append", default=[], + help="Auf Top-Level-Domain einschränken (z. B. de)") + parser.add_argument("--after", help="frühestes Datum: 2026-01-31 oder " + "24h/7d/2w relativ") + parser.add_argument("--before", help="spätestes Datum (exklusiv) oder " + "24h/7d/2w relativ") + parser.add_argument("--sort", default="relevance", + choices=["relevance", "freshness"], + help="Relevanz (Default) oder Aktualität") + parser.add_argument("--page", type=int, default=1) + parser.add_argument("--limit", type=int, default=None, + help="Treffer pro Seite (Default 10, max 50)") + parser.add_argument("--no-snippets", action="store_true", + help="keine ts_headline-Snippets laden") + parser.add_argument("--include-errors", action="store_true", + help="auch Seiten ohne HTTP 200 liefern") + parser.add_argument("--include-non-html", action="store_true", + help="auch Nicht-HTML-Inhalte liefern") + parser.add_argument("--json", action="store_true", + help="rohe JSON-Ausgabe") + parser.add_argument("--no-color", action="store_true") + parser.add_argument("--suggest", metavar="BEGRIFF", + help="Autocomplete-Vorschläge anzeigen") + parser.add_argument("--stats", action="store_true", + help="Index-Statistik anzeigen") + parser.add_argument("--page-id", type=int, + help="gespeicherte Seite per ID anzeigen") + parser.add_argument("--url", help="gespeicherte Seite per URL anzeigen") return parser -def main(argv=None) -> int: +def main(argv=None): args = build_parser().parse_args(argv) - - engine = SearchEngine(ts_config=args.config) + style = Style(sys.stdout.isatty() and not args.no_color) try: - response = engine.search( - query=args.query, + if args.stats: + print_stats(engine.stats(), style) + return 0 + + if args.suggest is not None: + suggestions = engine.suggest(args.suggest) + if not suggestions: + print("Keine Vorschläge.") + return 0 + for s in suggestions: + print(f" {s['title']} {style.dim}({s['domain']})" + f"{style.reset}") + return 0 + + if args.page_id is not None or args.url is not None: + page = engine.get_page(page_id=args.page_id, url=args.url) + if page is None: + print("Seite nicht gefunden.", file=sys.stderr) + return 1 + if args.json: + print(json.dumps(page, ensure_ascii=False, indent=2)) + else: + print_page(page, style, max_chars=3000) + return 0 + + if not args.query: + build_parser().error("Suchbegriffe fehlen (oder --stats / " + "--suggest / --page-id / --url nutzen).") + + data = engine.search( + args.query, page=args.page, - per_page=args.per_page, - domain=args.domain, - tld=args.tld, + per_page=args.limit, + sites=args.site, + tlds=args.tld, + after=args.after, + before=args.before, + sort=args.sort, + include_errors=args.include_errors, + include_non_html=args.include_non_html, + with_snippets=not args.no_snippets, ) - except Exception as exc: # z. B. DB nicht erreichbar, Tabelle fehlt - print(f"Fehler bei der Suche: {exc}", file=sys.stderr) + if args.json: + print(json.dumps(data, ensure_ascii=False, indent=2)) + else: + print_results(data, style) + return 0 + + except engine.QueryError as exc: + print(f"Fehlerhafte Anfrage: {exc}", file=sys.stderr) + return 2 + except engine.SearchTimeout: + print("Zeitlimit überschritten – bitte eingrenzen (site=, nach=, " + "weitere Begriffe).", file=sys.stderr) + return 4 + except engine.SearchEngineError as exc: + print(f"Suchfehler: {exc}", file=sys.stderr) + return 3 + except Exception as exc: + print(f"Fehler: {exc}", file=sys.stderr) return 1 - if args.json: - print(json.dumps(response.to_dict(), ensure_ascii=False, indent=2)) - else: - print(format_plain(response)) - - return 0 - if __name__ == "__main__": - sys.exit(main()) \ No newline at end of file + sys.exit(main()) diff --git a/search/engine.py b/search/engine.py index 2970a1d..22e4472 100644 --- a/search/engine.py +++ b/search/engine.py @@ -1,456 +1,986 @@ #!/usr/bin/env python3 """ -JonasSearch – Suchmaschine (Engine). +JonasSearch Suchmaschine – Kern (engine.py) -Volltextsuche über den Index, den crawler.py in PostgreSQL aufgebaut hat -(Tabelle `pages`). +Die eigentliche Suchlogik über der PostgreSQL-Datenbank `jonassearch`. -Die Engine nutzt, wenn vorhanden, die vom Trigger gepflegte tsvector-Spalte -der Tabelle `pages` (Suchvektor) und sucht damit mit PostgreSQL's -Volltextsuche. Fehlt eine solche Spalte (z. B. weil die Tabelle manuell -angelegt wurde), wird der Suchvektor zur Laufzeit aus title/headings/content -berechnet – die Suche funktioniert dann genauso, nur langsamer. +Retrieval: + Die Tabelle `public.pages` hält pro Webseite u. a. url, title, + headings (text[]), content, 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 + `idx_pages_search`. Kandidaten werden per `@@` über diesen Index + ermittelt. -Features: -- Rangfolge über ts_rank (Titel/Inhalt gewichtet nach Index-Spalte) -- Snippets mit hervorgehobenen Suchbegriffen (ts_headline, ) -- Präfix-Suche ("raspberr" findet "raspberry") -- UND-Verknüpfung der Suchbegriffe, Phrasen über Anführungszeichen nicht nötig -- Paginierung -- Filter nach Domain und TLD +Query-Syntax: + Die Nutzersuche läuft über `websearch_to_tsquery` – damit sind + Anführungszeichen für exakte Phrasen, `OR` und `-ausschluss` + möglich, ohne dass eine fehlerhafte Syntax den Request abschießt. -Konfiguration über Umgebungsvariablen (siehe .env.example): - JONASSEARCH_DB_HOST / PORT / NAME / USER / PASSWORD - JONASSEARCH_TS_CONFIG Textsuch-Konfiguration (Standard: german) - JONASSEARCH_VECTOR_COLUMN tsvector-Spalte erzwingen (optional) +Ranking (mehrstufig, in SQL): + score = ts_rank_cd(Gewichte, search_vector, tsquery, Norm) [A/B/C] + * (1 + Frischegewicht * exp(-Alter_in_Tagen / Halbwertszeit)) -Beispiel: - from search.engine import SearchEngine - engine = SearchEngine() - result = engine.search("raspberry pi", page=1, per_page=10) + Die ts_rank_cd-Normalisierung (Bitmaske 8 = Rang / 1 + log + Dokumentlänge) verhindert, dass riesige Seiten kurze, präzise + Treffer dominieren. Aktualität basiert auf COALESCE(crawled_at, + discovered_at); Dokumentqualität/HTTP-Metadaten fließen über die + Standardfilter ein (nur http_status = 200, nur HTML-Inhalt). + Link-basierte Autoritätssignale sind bewusst noch nicht Teil der + Formel. + + Bei sehr häufigen Begriffen (Zeitlimit des vollen Rankings) wird + automatisch auf ein begrenztes Kandidatenfenster ausgewichen + (MAX_RANK_CANDIDATES, Antwortmarke ranking_capped). + +Schutz der Produktivdatenbank: + - Jede Operation baut eine eigene Verbindung mit + `statement_timeout` (Default 10 s) auf – keine Request kann die + Datenbank blockieren. + - Der Gesamt-Trefferzähler läuft mit kürzerem Timeout (4 s) und ist + "best effort": bei Überschreitung wird `total = None` geliefert, + die Ergebnisse selbst bleiben davon unberührt. + - `content` (TOAST-lastig, bis 200 KB indexiert) wird nur für die + Treffer der angeforderten Seite aus der DB gelesen (Snippets via + `ts_headline`), nicht für die komplette Kandidatenmenge. """ import logging import os import re import time -from dataclasses import dataclass, field -from pathlib import Path -from typing import List, Optional, Tuple +from datetime import datetime, timedelta, timezone -from dotenv import load_dotenv import psycopg2 +from psycopg2.extras import RealDictCursor + +log = logging.getLogger("jonassearch.search") +log.addHandler(logging.NullHandler()) -# .env zuerst aus dem Arbeitsverzeichnis laden (Standard), zusätzlich aus -# dem Projektordner – so findet der Prozess die Konfiguration auch dann, -# wenn er von woanders gestartet wird (z. B. via systemd). -load_dotenv() -load_dotenv(Path(__file__).resolve().parent.parent / ".env") # ============================================================ -# KONFIGURATION +# FEHLERKLASSEN # ============================================================ +class SearchEngineError(Exception): + """Basisfehler der Suchmaschine.""" + + +class QueryError(SearchEngineError, ValueError): + """Ungültige Suchanfrage / ungültige Parameter (-> HTTP 400).""" + + +class SearchTimeout(SearchEngineError, TimeoutError): + """Suchanfrage hat das Statement-Timeout überschritten (-> HTTP 504).""" + + +# ============================================================ +# KONFIGURATION (Environment, wie crawler.py / priorityweb) +# ============================================================ + +# Wichtig: die Query-Konfiguration muss zur Konfiguration passen, mit der +# der DB-Trigger den Suchvektor gebaut hat (laut DB-Doku: 'simple'). +TS_CONFIG = os.getenv("JONASSEARCH_TS_CONFIG", "simple").strip() or "simple" + +# tsvector-Spalte erzwingen (optional, sonst automatische Erkennung). +VECTOR_COLUMN_ENV = os.getenv("JONASSEARCH_VECTOR_COLUMN", "").strip() + +# Statement-Timeouts (Schutz der ~10-GB-Produktivdatenbank). +STATEMENT_TIMEOUT_MS = int(os.getenv("JONASSEARCH_STATEMENT_TIMEOUT_MS", "10000")) +COUNT_TIMEOUT_MS = int(os.getenv("JONASSEARCH_COUNT_TIMEOUT_MS", "4000")) + +# Paginierung +DEFAULT_PER_PAGE = int(os.getenv("JONASSEARCH_RESULTS_PER_PAGE", "10")) +MAX_PER_PAGE = int(os.getenv("JONASSEARCH_MAX_RESULTS_PER_PAGE", "50")) + +# Ranking: Frische-Signal (Aktualitätsbonus auf den ts_rank_cd). +FRESHNESS_WEIGHT = float(os.getenv("JONASSEARCH_FRESHNESS_WEIGHT", "0.25")) +FRESHNESS_HALF_LIFE_DAYS = float( + os.getenv("JONASSEARCH_FRESHNESS_HALF_LIFE_DAYS", "180") +) + +# ts_rank_cd-Normalisierung (Bitmaske): 8 = Rang / (1 + log(Dokumentlänge)) +# – verhindert, dass riesige Seiten automatisch jeden kurzen, präzisen +# Treffer dominieren. 0 = keine Normalisierung. +RANK_NORMALIZATION = int(os.getenv("JONASSEARCH_RANK_NORMALIZATION", "8")) + +# Bei sehr häufigen Begriffen (z. B. Einzelwort mit zigtausend Treffern) +# wäre ts_rank_cd über alle Kandidaten zu langsam. Passiert das, wird +# automatisch ein zweiter Versuch über die ersten MAX_RANK_CANDIDATES +# Kandidaten (Indexreihenfolge) gefahren – garantiert schnelle Antwort, +# Ranking dann innerhalb dieses Fensters (Antwortflag ranking_capped). +MAX_RANK_CANDIDATES = int(os.getenv("JONASSEARCH_MAX_RANK_CANDIDATES", "5000")) + +# ts_rank_cd-Gewichte. Reihenfolge im DB-Trigger: Titel = A, +# Überschriften = B, Inhalt = C, (D = ungewichteter Rest). +# Format: "A=1.0,B=0.6,C=0.3,D=0.1" +DEFAULT_RANK_WEIGHTS = {"A": 1.0, "B": 0.6, "C": 0.3, "D": 0.1} + +# Snippets: ts_headline liest nur die ersten N Zeichen von `content` +# (TOAST-Schonung; die indexierten 200 KB kommen praktisch nie in ein +# Snippet-Fenster). +HEADLINE_SOURCE_MAX_CHARS = int( + os.getenv("JONASSEARCH_HEADLINE_SOURCE_CHARS", "100000") +) +HEADLINE_OPTIONS = ( + "MaxWords=40, MinWords=15, ShortWord=2, MaxFragments=2, " + "StartSel=⟦, StopSel=⟧" +) + +# Einzelabruf einer Seite +PAGE_CONTENT_CHARS = int(os.getenv("JONASSEARCH_PAGE_CONTENT_CHARS", "50000")) +PAGE_MAX_LINKS = 100 + +# Vorschläge (Autocomplete) +MAX_SUGGEST = 10 + DB_HOST = os.getenv("JONASSEARCH_DB_HOST", "127.0.0.1") DB_PORT = int(os.getenv("JONASSEARCH_DB_PORT", "25432")) DB_NAME = os.getenv("JONASSEARCH_DB_NAME", "jonassearch") DB_USER = os.getenv("JONASSEARCH_DB_USER", "jonassearch") DB_PASSWORD = os.getenv("JONASSEARCH_DB_PASSWORD", "") -# PostgreSQL-Textsuch-Konfiguration (Stemming / Stopwörter). -TS_CONFIG = os.getenv("JONASSEARCH_TS_CONFIG", "german").strip() or "german" +# Markierungen im ts_headline-Snippet (werden im Frontend/CLI zu +# Hervorhebungen umgesetzt; bewusst KEIN HTML, damit der Inhalt nie +# ungefiltert als Markup landet). +SNIPPET_START = "⟦" +SNIPPET_STOP = "⟧" -# Optional: tsvector-Spalte der Tabelle `pages` erzwingen. -VECTOR_COLUMN = os.getenv("JONASSEARCH_VECTOR_COLUMN", "").strip() +# ------------------------------------------------------------ +# Validierung +# ------------------------------------------------------------ -MAX_PER_PAGE = 50 -MAX_TERMS = 32 - -# Suchbegriffe: nur Wortzeichen (auch Umlaute) zulassen. -_TERM_RE = re.compile(r"\w+", re.UNICODE) - -# ts_headline-Optionen: kurze Snippets mit -Hervorhebung. -HIGHLIGHT_OPTIONS = ( - "StartSel=, StopSel=, " - "MaxWords=32, MinWords=12, MaxFragments=2, " - "FragmentDelimiter= …" +# Sehr defensive Domain-Prüfung (wie priorityweb/app.py). +DOMAIN_RE = re.compile( + r"^(?=.{1,253}\.)([a-z0-9]([a-z0-9-]{0,61}[a-z0-9])?\.)+" + r"[a-z]{2,63}$" ) -# eslint-ähnlicher Guard, damit ts_headline-Snippets im Browser sicher sind. -# Der Dokumenttext wird VOR dem Hervorheben escaped (& und <), damit nur die -# von ts_headline eingefügten -Tags im Snippet übrig bleiben. -_ESCAPE_FOR_HEADLINE = ( - "replace(replace(p.content, E'&', '&'), E'<', '<')" -) +# SQL-Identifier für die auto-erkannte tsvector-Spalte. +IDENT_RE = re.compile(r"^[a-z_][a-z0-9_]{0,62}$") + +# Relative Zeitspannen wie 24h / 7d / 2w. +_RELATIVE_RE = re.compile(r"^(\d+)\s*([hdw])$", re.IGNORECASE) + +_SORT_ALIASES = { + "relevance": "relevance", + "score": "relevance", + "rank": "relevance", + "relevanz": "relevance", + "freshness": "freshness", + "fresh": "freshness", + "date": "freshness", + "time": "freshness", + "aktualitaet": "freshness", +} # ============================================================ -# DATENTYPEN +# DATENBANK # ============================================================ -@dataclass -class SearchHit: - """Ein einzelnes Suchergebnis (eine indexierte Seite).""" - - url: str - title: str - snippet: str - domain: str = "" - tld: str = "" - crawled_at: Optional[str] = None - rank: float = 0.0 - position: int = 0 - headings: List[str] = field(default_factory=list) - - def to_dict(self) -> dict: - return { - "url": self.url, - "title": self.title, - "snippet": self.snippet, - "domain": self.domain, - "tld": self.tld, - "crawled_at": self.crawled_at, - "rank": round(self.rank, 6), - "position": self.position, - "headings": self.headings, - } +def db_connect(statement_timeout_ms=None): + """Neue Verbindung (autocommit, read-only Nutzung, Timeout gesetzt).""" + timeout = ( + STATEMENT_TIMEOUT_MS + if statement_timeout_ms is None + else int(statement_timeout_ms) + ) + conn = psycopg2.connect( + host=DB_HOST, + port=DB_PORT, + dbname=DB_NAME, + user=DB_USER, + password=DB_PASSWORD, + connect_timeout=5, + application_name="jonassearch-search", + options=f"-c statement_timeout={timeout}", + ) + conn.autocommit = True + return conn -@dataclass -class SearchResponse: - """Antwort einer Suche inkl. Metadaten.""" - - query: str - total: int - page: int - per_page: int - pages: int - took_ms: int - hits: List[SearchHit] = field(default_factory=list) - - def to_dict(self) -> dict: - return { - "query": self.query, - "total": self.total, - "page": self.page, - "per_page": self.per_page, - "pages": self.pages, - "took_ms": self.took_ms, - "hits": [h.to_dict() for h in self.hits], - } +def _set_statement_timeout(cur, ms): + """Timeout zur Laufzeit umschalten (int()-gecastet -> keine Injection).""" + cur.execute(f"SET statement_timeout = {int(ms)}") -# ============================================================ -# HILFSFUNKTIONEN -# ============================================================ +# tsvector-Spalte von `pages` (einmalig ermittelt, dann gecacht). +_vector_column_cache = None -def parse_terms(raw_query: str) -> List[str]: + +def resolve_vector_column(conn): """ - Zerlegt eine Roh-Abfrage in einzelne Suchbegriffe. + Liefert den Namen der tsvector-Spalte von public.pages. - Es werden nur Wortzeichen behalten (auch Umlaute), alles andere - (Interpunktion, Sonderzeichen) wird als Trenner behandelt. + Priorität: JONASSEARCH_VECTOR_COLUMN > 'search_vector' > erste + gefundene tsvector-Spalte. Der Name wird gegen IDENT_RE geprüft, + bevor er in SQL eingebettet wird. """ - terms = [t.lower() for t in _TERM_RE.findall(raw_query or "")] - return terms[:MAX_TERMS] + global _vector_column_cache + if VECTOR_COLUMN_ENV: + if not IDENT_RE.match(VECTOR_COLUMN_ENV): + raise SearchEngineError( + f"Ungültiger Spaltenname in JONASSEARCH_VECTOR_COLUMN: " + f"{VECTOR_COLUMN_ENV!r}" + ) + return VECTOR_COLUMN_ENV -def _shorten(text: Optional[str], max_len: int) -> str: - """Kürzt Text an einer Wortgrenze und hängt '…' an.""" - if not text: - return "" - if len(text) <= max_len: - return text - cut = text.rfind(" ", 0, max_len) - if cut < max_len // 2: - cut = max_len - return text[:cut].rstrip() + " …" + if _vector_column_cache: + return _vector_column_cache - -def _find_vector(conn) -> Tuple[Optional[str], Optional[str]]: - """ - Findet die Suchvektor-Spalte der Tabelle `pages`. - - Rückgabe: (column_name, None) wenn eine tsvector-Spalte existiert, - sonst (None, sql_expression) für eine Laufzeit-Berechnung aus - title/headings/content. - """ with conn.cursor() as cur: cur.execute( - "SELECT column_name, data_type " - "FROM information_schema.columns " - "WHERE table_name = 'pages' " - "AND table_schema = ANY (current_schemas(false))" - ) - columns = dict(cur.fetchall()) - - if not columns: - raise RuntimeError( - "Index-Tabelle 'pages' existiert nicht in der Datenbank." - ) - - # Explizit konfigurierte Spalte hat Vorrang. - if VECTOR_COLUMN: - if columns.get(VECTOR_COLUMN) == "tsvector": - logging.info("Nutze Index-Spalte '%s'.", VECTOR_COLUMN) - return VECTOR_COLUMN, None - logging.warning( - "Konfigurierte Spalte '%s' ist kein tsvector – " - "Fallback wird verwendet.", - VECTOR_COLUMN, - ) - - # Bevorzugt Spalten, die wie ein Suchvektor heißen, sonst irgendein tsvector. - preferred = [ - name for name in columns - if columns[name] == "tsvector" - and ("vector" in name.lower() or "search" in name.lower()) - ] - if preferred: - logging.info("Nutze Index-Spalte '%s'.", preferred[0]) - return preferred[0], None - - for name, data_type in columns.items(): - if data_type == "tsvector": - logging.info("Nutze Index-Spalte '%s'.", name) - return name, None - - # Kein tsvector-Index: Vektor zur Laufzeit aufbauen. - parts = [] - if "title" in columns: - parts.append("title") - if "headings" in columns: - headings_sql = ( - "array_to_string(headings, ' ')" - if columns["headings"] == "ARRAY" - else "headings" - ) - parts.append(headings_sql) - if "content" in columns: - parts.append("content") - - expression = " || ' ' || ".join(parts) if parts else "''" - logging.warning( - "Keine tsvector-Spalte in 'pages' gefunden – " - "Suchvektor wird zur Laufzeit berechnet (%s).", - expression, - ) - return None, expression - - -# ============================================================ -# SUCHMASCHINE -# ============================================================ - -class SearchEngine: - """ - Volltextsuche über die vom Crawler aufgebaute Tabelle `pages`. - """ - - def __init__( - self, - *, - host: str = DB_HOST, - port: int = DB_PORT, - dbname: str = DB_NAME, - user: str = DB_USER, - password: str = DB_PASSWORD, - ts_config: Optional[str] = None, - ): - self.host = host - self.port = port - self.dbname = dbname - self.user = user - self.password = password - self.ts_config = (ts_config or TS_CONFIG).strip() or "german" - # Cache für die erkannte Suchvektor-Definition. - self._vector_cache: Optional[Tuple[Optional[str], Optional[str]]] = None - - def _connect(self): - return psycopg2.connect( - host=self.host, - port=self.port, - dbname=self.dbname, - user=self.user, - password=self.password, - connect_timeout=10, - application_name="jonassearch-search", - ) - - def _vector(self, conn) -> Tuple[Optional[str], Optional[str]]: - if self._vector_cache is None: - self._vector_cache = _find_vector(conn) - return self._vector_cache - - def search( - self, - query: str, - page: int = 1, - per_page: int = 10, - domain: Optional[str] = None, - tld: Optional[str] = None, - ) -> SearchResponse: - """ - Führt eine Suche aus. - - `query` kann beliebig formuliert werden; die Begriffe werden - UND-verknüpft und mit Präfix-Suche ausgeführt. `domain`/`tld` - schränken auf eine exakte Domain bzw. TLD ein. - """ - started = time.monotonic() - - try: - page = max(1, int(page)) - except (TypeError, ValueError): - page = 1 - try: - per_page = max(1, min(int(per_page), MAX_PER_PAGE)) - except (TypeError, ValueError): - per_page = 10 - - terms = parse_terms(query) - - if not terms: - return self._empty_response(query, page, per_page, started) - - query_text = " & ".join(f"{term}:*" for term in terms) - - conn = self._connect() - try: - vector_sql, _ = self._prepare_vector(conn) - query_sql = "to_tsquery(%(config)s, %(querytext)s)" - - params = { - "config": self.ts_config, - "querytext": query_text, - "hlopts": HIGHLIGHT_OPTIONS, - "limit": per_page, - "offset": (page - 1) * per_page, - } - - filters = "" - if domain: - filters += " AND p.domain = %(domain)s" - params["domain"] = domain - if tld: - filters += " AND p.tld = %(tld)s" - params["tld"] = tld - - count_sql = ( - "SELECT count(*) FROM pages p " - f"WHERE {vector_sql} @@ {query_sql}{filters}" - ) - - select_sql = f""" - SELECT - p.url, - p.title, - p.domain, - p.tld, - p.crawled_at, - p.headings, - ts_headline( - %(config)s, - {_ESCAPE_FOR_HEADLINE}, - {query_sql}, - %(hlopts)s - ) AS snippet, - ts_rank({vector_sql}, {query_sql}) AS rank - FROM pages p - WHERE {vector_sql} @@ {query_sql}{filters} - ORDER BY rank DESC, p.crawled_at DESC, p.url ASC - LIMIT %(limit)s OFFSET %(offset)s """ + SELECT column_name + FROM information_schema.columns + WHERE table_schema = 'public' + AND table_name = 'pages' + AND data_type = 'tsvector' + ORDER BY (column_name = 'search_vector') DESC, column_name + LIMIT 1 + """ + ) + row = cur.fetchone() - with conn.cursor() as cur: - try: - cur.execute(count_sql, params) - except psycopg2.Error as exc: - # Z. B. wenn die Abfrage nur aus Stopwörtern besteht - # (to_tsquery liefert dann einen leeren Suchvektor). - logging.info("Abfrage ohne Treffer-Möglichkeit: %s", exc) - return self._empty_response(query, page, per_page, started) + if not row: + raise SearchEngineError( + "Keine tsvector-Spalte in public.pages gefunden. Wurde der " + "Suchindex (Trigger pages_search_vector_update) angelegt?" + ) + column = row[0] + if not IDENT_RE.match(column): + raise SearchEngineError( + f"Unerwarteter Spaltenname für den Suchvektor: {column!r}" + ) + _vector_column_cache = column + log.debug("tsvector-Spalte erkannt: %s", column) + return column - total = cur.fetchone()[0] - cur.execute(select_sql, params) - rows = cur.fetchall() +# ============================================================ +# PARAMETER-NORMALISIERUNG (rein, ohne DB – gut testbar) +# ============================================================ - hits = [] - for index, row in enumerate(rows): - ( - url, - title, - dom, - tld_value, - crawled_at, - headings, - snippet, - rank, - ) = row +def _as_list(value): + """None/str/list -> Liste von Strings (str wird gesplittet an ',').""" + if value is None: + return [] + if isinstance(value, str): + parts = value.split(",") + else: + parts = [] + for item in value: + if isinstance(item, str): + parts.extend(item.split(",")) + else: + parts.append(item) + return [str(p).strip() for p in parts if str(p).strip()] - title = title or url - snippet = snippet or title - hits.append( - SearchHit( - url=url, - title=_shorten(title, 300), - snippet=_shorten(snippet, 600), - domain=dom or "", - tld=tld_value or "", - crawled_at=crawled_at.isoformat() - if crawled_at is not None - else None, - rank=float(rank or 0.0), - position=(page - 1) * per_page + index + 1, - headings=[ - _shorten(h, 200) for h in (headings or [])[:5] - ], - ) +def normalize_sites(values): + """ + Benutzereingaben -> saubere Domainliste (kleingeschrieben, ohne + Schema/Pfad/Port/www). Akzeptiert auch URLs und '*.example.com'. + """ + out = [] + for raw in _as_list(values): + v = raw.strip().lower() + if "://" in v: + v = v.split("://", 1)[1] + v = v.split("/", 1)[0] + v = v.split("?", 1)[0] + v = v.split("#", 1)[0] + v = v.split("@", 1)[-1] + v = v.rsplit(":", 1)[0] + if v.startswith("*."): + v = v[2:] + if v.startswith("www."): + v = v[4:] + v = v.rstrip(".") + if not v: + continue + if not DOMAIN_RE.match(v): + raise QueryError(f"Ungültige Domain-Filterangabe: {raw!r}") + if v not in out: + out.append(v) + return out + + +def normalize_tlds(values): + """ + 'de', '.DE', ['de', 'at'] -> ['.de', '.at']. + + Der Crawler speichert die TLD MIT führendem Punkt (z. B. '.de'). + """ + out = [] + for raw in _as_list(values): + v = raw.strip().lower().lstrip(".") + if not v: + continue + if not re.match(r"^[a-z0-9-]{2,63}$", v): + raise QueryError(f"Ungültige TLD-Filterangabe: {raw!r}") + dotted = "." + v + if dotted not in out: + out.append(dotted) + return out + + +def normalize_sort(value): + v = str(value or "relevance").strip().lower() + if v not in _SORT_ALIASES: + raise QueryError( + f"Ungültige Sortierung: {value!r} " + f"(erlaubt: relevance | freshness)" + ) + return _SORT_ALIASES[v] + + +def parse_time_bound(value, which="after"): + """ + Zeilfilter parsen. + + Erlaubt: ISO-Datum/Zeit ('2026-01-31', '2026-01-31T12:00:00', mit + oder ohne 'Z'/Offset) sowie relative Angaben '24h', '7d', '2w' + (Ergebnis ist immer now() - Zeitspanne; 'after=7d' heißt also + "letzte 7 Tage", 'before=7d' "älter als 7 Tage"). + """ + if value is None or str(value).strip() == "": + return None + + v = str(value).strip() + + m = _RELATIVE_RE.match(v) + if m: + amount, unit = int(m.group(1)), m.group(2).lower() + if amount <= 0: + raise QueryError(f"Ungültige Zeitspanne: {value!r}") + delta = { + "h": timedelta(hours=amount), + "d": timedelta(days=amount), + "w": timedelta(weeks=amount), + }[unit] + return datetime.now(timezone.utc) - delta + + try: + dt = datetime.fromisoformat(v.replace("Z", "+00:00")) + except ValueError: + raise QueryError(f"Ungültiges Datumsfilter ({which}): {value!r}") + + if dt.tzinfo is None: + dt = dt.replace(tzinfo=timezone.utc) + return dt + + +def parse_rank_weights(spec=None): + """ + Parst JONASSEARCH_RANK_WEIGHTS ("A=1.0,B=0.6,C=0.3,D=0.1") und + liefert das Array in der von ts_rank_cd erwarteten Reihenfolge + [D, C, B, A]. + """ + weights = dict(DEFAULT_RANK_WEIGHTS) + if spec: + for part in str(spec).split(","): + part = part.strip() + if not part: + continue + if "=" not in part: + raise QueryError( + f"Ungültiges Gewicht {part!r} – Format: A=1.0,B=0.6,C=0.3,D=0.1" ) + key, _, val = part.partition("=") + key = key.strip().upper() + if key not in ("A", "B", "C", "D"): + raise QueryError( + f"Ungültiges Gewichtssignal {key!r} – erlaubt: A, B, C, D" + ) + try: + weights[key] = float(val) + except ValueError: + raise QueryError(f"Ungültiger Zahlenwert für {key}: {val!r}") + if not (0.0 <= weights[key] <= 1e6): + raise QueryError(f"Gewicht {key} außerhalb des zulässigen Bereichs") + return [weights["D"], weights["C"], weights["B"], weights["A"]] - total = int(total or 0) - return SearchResponse( - query=query, - total=total, - page=page, - per_page=per_page, - pages=(total + per_page - 1) // per_page if total else 0, - took_ms=int((time.monotonic() - started) * 1000), - hits=hits, +def prefix_lexeme(query): + """ + Letztes 'normales' Wort der Eingabe für die Autocomplete-Präfixsuche + ('hallo welt' -> 'welt:*'). Nicht nutzbare Eingaben -> None. + """ + if not query: + return None + tokens = re.findall(r"\w+", str(query), re.UNICODE) + for token in reversed(tokens): + if len(token) >= 2: + return token.lower() + ":*" + return None + + +RANK_WEIGHTS = parse_rank_weights(os.getenv("JONASSEARCH_RANK_WEIGHTS")) + + +# ============================================================ +# SQL-BAUSTEINE +# ============================================================ + +def _filter_clauses(sites, tlds, after, before, include_errors, + include_non_html): + """WHERE-Bausteine + Parameter für die Standardfilter.""" + clauses = [] + params = {} + + if sites: + # domain speichert den vollen Hostnamen (z. B. www.example.com): + # exakte Domain ODER Subdomain. + clauses.append( + "(p.domain = ANY(%(sites)s) OR p.domain LIKE ANY(%(site_pats)s))" + ) + params["sites"] = sites + params["site_pats"] = ["%." + s for s in sites] + + if tlds: + clauses.append("(p.tld = ANY(%(tlds)s))") + params["tlds"] = tlds + + if not include_errors: + # Nur erfolgreich gecrawlte Seiten. + clauses.append("(p.http_status = 200)") + + if not include_non_html: + # Nur HTML-Dokumente (leerer content_type zulassen). + # Achtung: %% statt % – der Filter-SQL-String wird immer mit + # Parameter-Dict an psycopg2 übergeben, nackte Prozentzeichen + # würden als Platzhalter geparst. + clauses.append( + "(p.content_type LIKE 'text/html%%' OR p.content_type IS NULL)" + ) + + if after is not None: + clauses.append("(COALESCE(p.crawled_at, p.discovered_at) >= %(after)s)") + params["after"] = after + + if before is not None: + clauses.append("(COALESCE(p.crawled_at, p.discovered_at) < %(before)s)") + params["before"] = before + + sql = (" AND " + " AND ".join(clauses)) if clauses else "" + return sql, params + + +def _search_sql(vector_col, sort, filter_sql, capped=False): + """ + Ranking-Query: Kandidaten via GIN-Index, bewertet + sortiert. + + capped=False: volles Ranking über alle Kandidaten (exakt, aber bei + sehr häufigen Begriffen langsam). + + capped=True: Ranking nur über die ersten MAX_RANK_CANDIDATES + Kandidaten (Subquery-LIMIT in Indexreihenfolge) – bounded latency + als Fallback nach einem Timeout des vollen Rankings. + """ + if sort == "relevance": + order = "(r.text_rank * f.fresh) DESC, r.text_rank DESC, p.id ASC" + else: + order = ( + "COALESCE(p.crawled_at, p.discovered_at) DESC NULLS LAST, p.id ASC" + ) + + # 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 + """ + + if capped: + source = """ + FROM ( + SELECT p.id, p.url, p.title, p.domain, p.tld, p.http_status, + p.content_type, 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{filter_sql} + LIMIT %(rank_cap)s + ) p + CROSS JOIN LATERAL ( + SELECT ts_rank_cd(%(weights)s::real[], p.vector, + websearch_to_tsquery(%(ts_config)s, %(query)s), + %(norm)s::integer) AS text_rank + ) r + """.format(vector_col=vector_col, filter_sql=filter_sql) + else: + source = f""" + FROM pages p + CROSS JOIN LATERAL ( + SELECT websearch_to_tsquery(%(ts_config)s, %(query)s) AS tsq + ) q + CROSS JOIN LATERAL ( + SELECT ts_rank_cd(%(weights)s::real[], p.{vector_col}, q.tsq, + %(norm)s::integer) AS text_rank + ) r + """ + fresh_lateral + f""" + WHERE p.{vector_col} @@ q.tsq{filter_sql} + """ + + if capped: + source += fresh_lateral + + return f""" + SELECT + p.id, + p.url, + p.title, + p.domain, + p.tld, + p.http_status, + p.content_type, + p.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 _headline_sql(): + """Snippets nur für die Treffer der angeforderten Seite.""" + return """ + SELECT p.id, + ts_headline(%(ts_config)s, + left(p.content, %(hl_chars)s), + q.tsq, + %(hl_opts)s) AS snippet + FROM pages p + CROSS JOIN LATERAL ( + SELECT websearch_to_tsquery(%(ts_config)s, %(query)s) AS tsq + ) q + WHERE p.id = ANY(%(ids)s) + """ + + +def _count_sql(vector_col, filter_sql): + return f""" + SELECT count(*) 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{filter_sql} + """ + + +def _iso(dt): + if isinstance(dt, datetime): + return dt.isoformat() + return None + + +def _round(value, digits=6): + try: + return round(float(value), digits) + except (TypeError, ValueError): + return 0.0 + + +# ============================================================ +# ÖFFENTLICHE API +# ============================================================ + +def search(query, *, page=1, per_page=None, sites=None, tlds=None, + after=None, before=None, sort="relevance", + include_errors=False, include_non_html=False, + with_snippets=True): + """ + Volltextsuche über `pages`. + + Rückgabe (dict): query, parsed_query, page, per_page, sort, total + (None = nicht gezählt/Zeitlimit), has_more, results[], took_ms. + Jedes Result: id, url, title, domain, tld, http_status, content_type, + crawled_at, score, text_rank, freshness, snippet (⟦…⟧-Markierungen). + """ + q = (query or "").strip() + if not q: + raise QueryError("Leere Suchanfrage.") + + try: + page = max(1, int(page)) + per_page = DEFAULT_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_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, + include_errors, include_non_html, + ) + params = { + "ts_config": TS_CONFIG, + "query": q, + "weights": RANK_WEIGHTS, + "norm": RANK_NORMALIZATION, + "fresh_weight": FRESHNESS_WEIGHT, + "fresh_half_life": FRESHNESS_HALF_LIFE_DAYS, + "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: + # Wie hat PostgreSQL die Anfrage interpretiert? (Transparenz + # + Erkennung "keine durchsuchbaren Begriffe") + 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 = [] + snippets = {} + total = None + ranking_capped = False + + if not empty_tsq: + _set_statement_timeout(cur, STATEMENT_TIMEOUT_MS) + try: + cur.execute(_search_sql(vector_col, sort, filter_sql), + params) + rows = cur.fetchall() + except psycopg2.errors.QueryCanceled: + # Sehr häufiger Begriff: volles Ranking über alle + # Kandidaten ist zu langsam. Zweiter Versuch über + # ein begrenztes Kandidatenfenster (bounded latency). + log.info( + "Volles Ranking hat das Zeitlimit überschritten; " + "Fallback auf %d Kandidaten.", MAX_RANK_CANDIDATES + ) + ranking_capped = True + try: + cur.execute( + _search_sql(vector_col, sort, filter_sql, + capped=True), + params, + ) + rows = cur.fetchall() + except psycopg2.errors.QueryCanceled: + raise SearchTimeout( + "Suchanfrage hat das Zeitlimit überschritten – " + "bitte weiter eingrenzen (z. B. site=, after= " + "oder zusätzliche Begriffe)." + ) + + if with_snippets and rows: + ids = [r["id"] for r in rows[:per_page]] + cur.execute(_headline_sql(), { + "ts_config": TS_CONFIG, + "query": q, + "hl_chars": HEADLINE_SOURCE_MAX_CHARS, + "hl_opts": HEADLINE_OPTIONS, + "ids": ids, + }) + for row in cur.fetchall(): + snippets[row["id"]] = row["snippet"] or "" + + # Gesamtzahl nur "best effort": eigenes (kürzeres) Timeout, + # bei Überschreitung bleibt total=None. + _set_statement_timeout(cur, COUNT_TIMEOUT_MS) + try: + cur.execute(_count_sql(vector_col, filter_sql), params) + total = int(cur.fetchone()["n"]) + except psycopg2.Error as exc: + log.warning("Trefferzählung abgebrochen: %s", exc) + total = None + + has_more = len(rows) > per_page + rows = rows[:per_page] + + results = [] + for r in rows: + results.append({ + "id": r["id"], + "url": r["url"], + "title": (r["title"] or "").strip() or r["url"], + "domain": r["domain"] or "", + "tld": r["tld"] or "", + "http_status": r["http_status"], + "content_type": r["content_type"], + "crawled_at": _iso(r["crawled_at"]), + "score": _round(r["score"]), + "text_rank": _round(r["text_rank"]), + "freshness": _round(r["freshness"], 3), + "snippet": (snippets.get(r["id"]) or "").strip(), + }) + + 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), + } + finally: + conn.close() + + +def get_page(page_id=None, url=None, *, max_content_chars=None): + """ + Einzelne Seite per id oder url. content ist auf + JONASSEARCH_PAGE_CONTENT_CHARS (Default 50000) gekürzt; links auf + PAGE_MAX_LINKS. content_truncated zeigt die Kürzung an. + """ + if (page_id is None) == (url is None): + raise QueryError("Genau eins von 'id' oder 'url' angeben.") + if page_id is not None: + try: + page_id = int(page_id) + except (TypeError, ValueError): + raise QueryError("id muss eine Zahl sein.") + if page_id <= 0: + raise QueryError("id muss positiv sein.") + if url is not None: + url = str(url).strip() + if not url or len(url) > 2048: + raise QueryError("Ungültige url.") + + limit_chars = ( + PAGE_CONTENT_CHARS + if max_content_chars is None + else max(200, min(int(max_content_chars), 1000000)) + ) + + conn = db_connect() + try: + with conn.cursor(cursor_factory=RealDictCursor) as cur: + if page_id is not None: + cur.execute( + """ + SELECT id, url, title, headings, domain, tld, source_url, + http_status, content_type, discovered_at, + crawled_at, crawl_time_ms, + length(content) AS content_length, + links, + left(content, %(max_chars)s) AS content + FROM pages + WHERE id = %(id)s + """, + {"id": page_id, "max_chars": limit_chars}, + ) + else: + cur.execute( + """ + SELECT id, url, title, headings, domain, tld, source_url, + http_status, content_type, discovered_at, + crawled_at, crawl_time_ms, + length(content) AS content_length, + links, + left(content, %(max_chars)s) AS content + FROM pages + WHERE url = %(url)s + """, + {"url": url, "max_chars": limit_chars}, + ) + row = cur.fetchone() + finally: + conn.close() + + if not row: + return None + + content = row["content"] or "" + content_length = row["content_length"] or 0 + links = list(row["links"] or [])[:PAGE_MAX_LINKS] + + return { + "id": row["id"], + "url": row["url"], + "title": (row["title"] or "").strip(), + "headings": list(row["headings"] or []), + "domain": row["domain"] or "", + "tld": row["tld"] or "", + "source_url": row["source_url"], + "http_status": row["http_status"], + "content_type": row["content_type"], + "discovered_at": _iso(row["discovered_at"]), + "crawled_at": _iso(row["crawled_at"]), + "crawl_time_ms": row["crawl_time_ms"], + "content_length": content_length, + "content_truncated": content_length > len(content), + "content": content, + "links": links, + "links_total": len(row["links"] or []), + } + + +def suggest(query, limit=None): + """ + Autocomplete-Vorschläge: Präfixsuche auf dem letzten Wort + (to_tsquery 'begr:*') gegen den Suchvektor, gerankt nach Relevanz. + """ + lexeme = prefix_lexeme(query) + if not lexeme: + return [] + + try: + limit = MAX_SUGGEST if limit is None else int(limit) + except (TypeError, ValueError): + raise QueryError("limit muss eine Zahl sein.") + limit = min(max(1, limit), MAX_SUGGEST) + + conn = db_connect() + try: + vector_col = resolve_vector_column(conn) + with conn.cursor(cursor_factory=RealDictCursor) as cur: + cur.execute( + f""" + SELECT p.title, p.url, p.domain + FROM pages p + CROSS JOIN LATERAL ( + SELECT to_tsquery(%(ts_config)s, %(lexeme)s) AS tsq + ) q + CROSS JOIN LATERAL ( + SELECT ts_rank_cd(p.{vector_col}, q.tsq) AS tr + ) r + WHERE p.{vector_col} @@ q.tsq + AND COALESCE(p.title, '') <> '' + ORDER BY r.tr DESC, p.id ASC + LIMIT %(limit)s + """, + {"ts_config": TS_CONFIG, "lexeme": lexeme, "limit": limit}, + ) + rows = cur.fetchall() + except psycopg2.errors.QueryCanceled: + return [] # Vorschläge sind optional – nie zum Fehlerfall machen + finally: + conn.close() + + out, seen = [], set() + for row in rows: + title = (row["title"] or "").strip() + if not title or title.lower() in seen: + continue + seen.add(title.lower()) + out.append({ + "title": title, + "url": row["url"], + "domain": row["domain"] or "", + }) + return out + + +def stats(): + """ + Günstige Betriebsstatistiken (Schätzungen aus pg_class/pg_stats, + keine teuren Full-Scans über die ~10-GB-Tabelle). + """ + conn = db_connect() + try: + with conn.cursor(cursor_factory=RealDictCursor) as cur: + _set_statement_timeout(cur, STATEMENT_TIMEOUT_MS) + cur.execute( + """ + SELECT + (SELECT reltuples::bigint + FROM pg_class + WHERE oid = 'public.pages'::regclass) AS pages, + (SELECT reltuples::bigint + FROM pg_class + WHERE oid = 'public.crawl_queue'::regclass) AS queue, + (SELECT max(crawled_at) FROM pages) AS last_crawl, + pg_size_pretty(pg_database_size(current_database())) + AS db_size + """ + ) + row = dict(cur.fetchone() or {}) + + # Offene Queue-Einträge: best effort mit kurzem Timeout + # (Index auf (status, priority, discovered_at) vorhanden). + pending = None + _set_statement_timeout(cur, COUNT_TIMEOUT_MS) + try: + cur.execute( + "SELECT count(*) AS n FROM crawl_queue " + "WHERE status = 'pending'" + ) + pending = int(cur.fetchone()["n"]) + except psycopg2.Error as exc: + log.warning("Queue-Zählung abgebrochen: %s", exc) + + # Distinct-Domain-Schätzung aus den Planner-Statistiken. + domains = None + try: + cur.execute( + """ + SELECT n_distinct + FROM pg_stats + WHERE schemaname = 'public' + AND tablename = 'pages' + AND attname = 'domain' + """ + ) + nd = cur.fetchone() + if nd and nd["n_distinct"] is not None: + nd_val = float(nd["n_distinct"]) + pages = row.get("pages") or 0 + if nd_val >= 0: + domains = int(nd_val) + elif pages > 0: + domains = int(abs(nd_val) * pages) + except psycopg2.Error as exc: + log.debug("Domain-Schätzung fehlgeschlagen: %s", exc) + finally: + conn.close() + + return { + "pages": row.get("pages"), + "queue": row.get("queue"), + "queue_pending": pending, + "domains": domains, + "last_crawl": _iso(row.get("last_crawl")), + "db_size": row.get("db_size"), + } + + +def health(): + """Healthcheck für Monitoring/systemd – wirft nie.""" + try: + conn = db_connect(statement_timeout_ms=3000) + try: + with conn.cursor() as cur: + cur.execute("SELECT 1") + cur.fetchone() finally: conn.close() - - def _prepare_vector(self, conn) -> Tuple[str, None]: - """ - Liefert das SQL für den Suchvektor: - - vorhandene tsvector-Spalte, z. B. `search_vector` - - oder Laufzeit-Ausdruck `to_tsvector(%(config)s, ...)` - """ - column, expression = self._vector(conn) - - if column is not None: - return column, None - - return f"to_tsvector(%(config)s, {expression})", None - - def _empty_response( - self, - query: str, - page: int, - per_page: int, - started: float, - ) -> SearchResponse: - return SearchResponse( - query=query, - total=0, - page=page, - per_page=per_page, - pages=0, - took_ms=int((time.monotonic() - started) * 1000), - hits=[], - ) \ No newline at end of file + return {"ok": True, "database": "up", "ts_config": TS_CONFIG} + except Exception as exc: + return {"ok": False, "database": "down", "error": str(exc)[:200]} diff --git a/search/jonassearch-engine.service b/search/jonassearch-engine.service new file mode 100644 index 0000000..e4cb58d --- /dev/null +++ b/search/jonassearch-engine.service @@ -0,0 +1,42 @@ +# JonasSearch Suchmaschine (Volltextsuch-Weboberfläche + HTTP-API) +# +# Installieren (als root): +# cp /srv/serverjonas/search/search/jonassearch-engine.service \ +# /etc/systemd/system/ +# systemctl daemon-reload +# systemctl enable --now jonassearch-engine.service +# +# Ports: Statserver 8081 · Vorschlagsverfeinerer 8082 · Suchmaschine 8083 +# +# Der Port ist unten über Environment= fest auf 8083 gepinnt; diese +# Zuweisung gilt NACH dem EnvironmentFile und überschreibt damit einen +# eventuell abweichenden JONASSEARCH_API_PORT-Wert aus .env. + +[Unit] +Description=JonasSearch Suchmaschine (HTTP-API + Weboberfläche) +After=network-online.target postgresql@17-jonassearch.service +Wants=network-online.target + +[Service] +Type=simple + +User=jonassearch +Group=jonassearch + +WorkingDirectory=/srv/serverjonas/search + +EnvironmentFile=/srv/serverjonas/search/.env + +Environment=JONASSEARCH_API_PORT=8083 + +ExecStart=/usr/bin/python3 /srv/serverjonas/search/search/api.py + +Restart=on-failure +RestartSec=10 + +# Sauberer Shutdown +KillSignal=SIGTERM +TimeoutStopSec=30 + +[Install] +WantedBy=multi-user.target diff --git a/search/test_engine.py b/search/test_engine.py new file mode 100644 index 0000000..58efc7e --- /dev/null +++ b/search/test_engine.py @@ -0,0 +1,344 @@ +#!/usr/bin/env python3 +""" +Local test suite for the search engine (no database required). + +Only the pure, DB-free parts are tested: parameter normalization, +SQL assembly and the query-parsing helpers. Database-backed behaviour +is covered by the statement timeouts in production. +""" + +import os +import re +import sys +import unittest +from datetime import timedelta, timezone +from unittest import mock + +sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) + + +def _load_env_file(): + """ + Load ../.env like the systemd EnvironmentFile does, so the tests + also run outside systemd. Must happen before engine.py is imported, + which reads the JONASSEARCH_* variables at import time. + """ + env_path = os.path.join( + os.path.dirname(os.path.abspath(__file__)), + os.pardir, + ".env", + ) + + try: + with open(env_path) as handle: + for line in handle: + line = line.strip() + if not line or line.startswith("#") or "=" not in line: + continue + key, _, value = line.partition("=") + os.environ.setdefault(key.strip(), value.strip()) + except OSError: + pass + + +_load_env_file() + +import engine # noqa: E402 + + +class ParseRankWeightsTest(unittest.TestCase): + + def test_defaults_map_to_pg_order(self): + # ts_rank_cd erwartet [D, C, B, A] + self.assertEqual( + engine.parse_rank_weights(None), + [0.1, 0.3, 0.6, 1.0], + ) + + def test_custom_spec(self): + self.assertEqual( + engine.parse_rank_weights("A=1.0,B=0.5,C=0.25,D=0.05"), + [0.05, 0.25, 0.5, 1.0], + ) + + def test_partial_spec_keeps_defaults(self): + self.assertEqual( + engine.parse_rank_weights("B=0.9"), + [0.1, 0.3, 0.9, 1.0], + ) + + def test_invalid_key(self): + with self.assertRaises(engine.QueryError): + engine.parse_rank_weights("E=1.0") + + def test_invalid_value(self): + with self.assertRaises(engine.QueryError): + engine.parse_rank_weights("A=xyz") + + def test_missing_equals(self): + with self.assertRaises(engine.QueryError): + engine.parse_rank_weights("A 1.0") + + +class NormalizeSitesTest(unittest.TestCase): + + def test_plain(self): + self.assertEqual(engine.normalize_sites(["example.com"]), + ["example.com"]) + + def test_url_and_www(self): + self.assertEqual( + engine.normalize_sites(["https://www.Example.com/pfad?x=1"]), + ["example.com"], + ) + + def test_wildcard(self): + self.assertEqual(engine.normalize_sites("*.example.com"), + ["example.com"]) + + def test_dedup_and_multiple(self): + self.assertEqual( + engine.normalize_sites(["example.com", "EXAMPLE.com", "de.example.org"]), + ["example.com", "de.example.org"], + ) + + def test_empty(self): + self.assertEqual(engine.normalize_sites(None), []) + self.assertEqual(engine.normalize_sites(["", " "]), []) + + def test_invalid_raises(self): + with self.assertRaises(engine.QueryError): + engine.normalize_sites(["not a domain!"]) + + +class NormalizeTldsTest(unittest.TestCase): + + def test_plain_gets_leading_dot(self): + # Der Crawler speichert die TLD mit führendem Punkt ('.de'). + self.assertEqual(engine.normalize_tlds(["de"]), [".de"]) + + def test_dot_and_case(self): + self.assertEqual(engine.normalize_tlds([".DE", "at"]), [".de", ".at"]) + + def test_invalid_raises(self): + with self.assertRaises(engine.QueryError): + engine.normalize_tlds(["de/)"]) + + def test_empty(self): + self.assertEqual(engine.normalize_tlds(None), []) + + +class NormalizeSortTest(unittest.TestCase): + + def test_defaults(self): + self.assertEqual(engine.normalize_sort(None), "relevance") + self.assertEqual(engine.normalize_sort("relevance"), "relevance") + self.assertEqual(engine.normalize_sort("freshness"), "freshness") + + def test_aliases(self): + self.assertEqual(engine.normalize_sort("date"), "freshness") + self.assertEqual(engine.normalize_sort("score"), "relevance") + + def test_invalid(self): + with self.assertRaises(engine.QueryError): + engine.normalize_sort("banana") + + +class ParseTimeBoundTest(unittest.TestCase): + + def test_empty_is_none(self): + self.assertIsNone(engine.parse_time_bound(None)) + self.assertIsNone(engine.parse_time_bound("")) + + def test_relative_days(self): + before = engine.parse_time_bound("7d", "after") + now = datetime_now() + delta = now - before + self.assertGreater(delta.total_seconds(), 7 * 86400 - 5) + self.assertLess(delta.total_seconds(), 7 * 86400 + 5) + + def test_relative_hours_and_weeks(self): + before = engine.parse_time_bound("24h", "after") + self.assertAlmostEqual( + (datetime_now() - before).total_seconds(), 86400, delta=5 + ) + before = engine.parse_time_bound("2w", "after") + self.assertAlmostEqual( + (datetime_now() - before).total_seconds(), 14 * 86400, delta=5 + ) + + def test_relative_zero_raises(self): + with self.assertRaises(engine.QueryError): + engine.parse_time_bound("0d") + + def test_iso_date_becomes_utc(self): + dt = engine.parse_time_bound("2026-01-31", "after") + self.assertEqual(dt.tzinfo, timezone.utc) + self.assertEqual((dt.year, dt.month, dt.day), (2026, 1, 31)) + + def test_iso_datetime_with_z(self): + dt = engine.parse_time_bound("2026-01-31T12:00:00Z", "before") + self.assertEqual(dt.utcoffset(), timedelta(0)) + self.assertEqual(dt.hour, 12) + + def test_naive_datetime_becomes_utc(self): + dt = engine.parse_time_bound("2026-01-31 12:00:00", "after") + self.assertEqual(dt.tzinfo, timezone.utc) + + def test_invalid_raises(self): + with self.assertRaises(engine.QueryError): + engine.parse_time_bound("gestern") + + +def datetime_now(): + return engine.datetime.now(timezone.utc) + + +class PrefixLexemeTest(unittest.TestCase): + + def test_last_word(self): + self.assertEqual(engine.prefix_lexeme("hallo welt"), "welt:*") + + def test_lowercased(self): + self.assertEqual(engine.prefix_lexeme("SuchMASCHINE"), "suchmaschine:*") + + def test_skips_short_tokens(self): + self.assertEqual(engine.prefix_lexeme("a b suchbegriff"), + "suchbegriff:*") + + def test_punctuation_only(self): + self.assertIsNone(engine.prefix_lexeme("!!! ...")) + + def test_umlauts(self): + self.assertEqual(engine.prefix_lexeme("größen"), "größen:*") + + def test_empty(self): + self.assertIsNone(engine.prefix_lexeme("")) + self.assertIsNone(engine.prefix_lexeme(None)) + + +class FilterClausesTest(unittest.TestCase): + + def test_defaults_filter_errors_and_non_html(self): + sql, params = engine._filter_clauses([], [], None, None, + False, False) + self.assertIn("p.http_status = 200", sql) + self.assertIn("text/html%", sql) + self.assertEqual(params, {}) + + def test_include_errors(self): + sql, _ = engine._filter_clauses([], [], None, None, True, False) + self.assertNotIn("http_status", sql) + + def test_sites_clause(self): + sql, params = engine._filter_clauses( + engine.normalize_sites(["example.com"]), [], None, None, + False, False, + ) + self.assertIn("p.domain = ANY(%(sites)s)", sql) + self.assertIn("p.domain LIKE ANY(%(site_pats)s)", sql) + self.assertEqual(params["sites"], ["example.com"]) + self.assertEqual(params["site_pats"], ["%.example.com"]) + + def test_time_bounds(self): + after = engine.datetime(2026, 1, 1, tzinfo=timezone.utc) + before = engine.datetime(2026, 2, 1, tzinfo=timezone.utc) + sql, params = engine._filter_clauses( + [], [], after, before, False, False, + ) + self.assertIn("COALESCE(p.crawled_at, p.discovered_at) >= %(after)s", + sql) + self.assertIn("COALESCE(p.crawled_at, p.discovered_at) < %(before)s", + sql) + self.assertEqual(params["after"], after) + self.assertEqual(params["before"], before) + + +class SearchSqlTest(unittest.TestCase): + + def test_relevance_ordering(self): + sql = engine._search_sql("search_vector", "relevance", "") + self.assertIn( + "ts_rank_cd(%(weights)s::real[], p.search_vector, q.tsq,", + sql, + ) + self.assertIn("%(norm)s::integer", sql) + self.assertIn("(r.text_rank * f.fresh) DESC", sql) + self.assertIn("websearch_to_tsquery(%(ts_config)s, %(query)s)", sql) + self.assertIn("LIMIT %(limit)s OFFSET %(offset)s", sql) + # content darf im Ranking-Query nicht gelesen werden (TOAST-Schonung) + self.assertIsNone(re.search(r"\bp\.content\b", sql), + "content-Spalte im Ranking-Query gelesen") + + def test_capped_fallback_sql(self): + sql = engine._search_sql("search_vector", "relevance", "", + capped=True) + self.assertIn("LIMIT %(rank_cap)s", sql) + self.assertIn("p.search_vector AS vector", sql) + self.assertIn("ts_rank_cd(%(weights)s::real[], p.vector,", sql) + self.assertIn("LIMIT %(limit)s OFFSET %(offset)s", sql) + # auch der gekappte Query darf kein content lesen + self.assertIsNone(re.search(r"\bp\.content\b", sql)) + + def test_freshness_ordering(self): + sql = engine._search_sql("search_vector", "freshness", " AND x") + self.assertIn( + "COALESCE(p.crawled_at, p.discovered_at) DESC NULLS LAST", sql + ) + self.assertIn(" AND x", sql) + + def test_vector_column_is_interpolated(self): + sql = engine._search_sql("my_vector", "relevance", "") + self.assertIn("p.my_vector", sql) + self.assertNotIn("p.search_vector", sql) + + +class CountSqlTest(unittest.TestCase): + + def test_count_uses_filters(self): + sql = engine._count_sql("search_vector", " AND p.http_status = 200") + self.assertIn("count(*) AS n", sql) + self.assertIn("p.search_vector @@ q.tsq AND p.http_status = 200", sql) + + +class SetStatementTimeoutTest(unittest.TestCase): + + def test_casts_to_int(self): + cur = mock.Mock() + engine._set_statement_timeout(cur, "4000") + cur.execute.assert_called_once_with("SET statement_timeout = 4000") + + +class SearchValidationTest(unittest.TestCase): + + def test_empty_query_raises(self): + with self.assertRaises(engine.QueryError): + engine.search(" ") + + def test_bad_page_raises(self): + with self.assertRaises(engine.QueryError): + engine.search("test", page="x") + + def test_per_page_capped(self): + with mock.patch.object(engine, "db_connect") as db: + conn = mock.MagicMock() + db.return_value = conn + # 1. fetchone: geparste tsquery, 2. fetchone: Trefferzahl + conn.cursor.return_value.__enter__.return_value\ + .fetchone.side_effect = [{"tsq": "test"}, {"n": 0}] + conn.cursor.return_value.__enter__.return_value\ + .fetchall.return_value = [] + with mock.patch.object(engine, "resolve_vector_column", + return_value="search_vector"): + data = engine.search("test", per_page=10000) + self.assertEqual(data["per_page"], engine.MAX_PER_PAGE) + + def test_query_error_for_bad_filters(self): + with mock.patch.object(engine, "db_connect") as db: + db.return_value = mock.MagicMock() + with self.assertRaises(engine.QueryError): + engine.search("test", sites=["ungültig!"]) + + +if __name__ == "__main__": + unittest.main(verbosity=2) diff --git a/test_crawler.py b/test_crawler.py new file mode 100644 index 0000000..199ae00 --- /dev/null +++ b/test_crawler.py @@ -0,0 +1,455 @@ +#!/usr/bin/env python3 +""" +Local test suite for the crawler (no database required). + +Only the pure, DB-free parts are tested: the HostScheduler (per-host +delays, block backoff, Retry-After), the transient-error classification +and the queue-claim decision logic (blocked hosts are skipped, another +domain is picked) against an in-memory queue substitute. +""" + +import os +import sys +import time +import unittest +from email import utils as email_utils +from unittest import mock + +sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) + + +def _load_env_file(): + """ + Load .env like the systemd EnvironmentFile does, so the tests also + run outside systemd. Must happen before crawler.py is imported, + which reads the JONASSEARCH_* variables at import time. + """ + env_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), ".env") + + try: + with open(env_path) as handle: + for line in handle: + line = line.strip() + if not line or line.startswith("#") or "=" not in line: + continue + key, _, value = line.partition("=") + os.environ.setdefault(key.strip(), value.strip()) + except OSError: + pass + + +_load_env_file() + +import crawler # noqa: E402 + + +class StubRobots: + """robots.txt stub: constant delay, no network.""" + + def __init__(self, delay=3.0): + self.delay_value = delay + + def delay(self, url): + return self.delay_value + + +class FakeState: + """In-memory crawl_queue substitute for claim/defer/requeue.""" + + def __init__(self, rows=None): + self.rows = rows if rows is not None else [] + + def add(self, **kwargs): + row = { + "id": len(self.rows) + 1, + "url": "", + "source": None, + "status": "pending", + "priority": 0, + "attempts": 0, + } + row.update(kwargs) + self.rows.append(row) + return row["id"] + + +class FakeCursor: + """Just enough SQL surface for the crawler's queue functions.""" + + def __init__(self, state): + self._state = state + self._rows = [] + + def execute(self, query, params=None): + query = " ".join(query.split()) + params = params or () + + if "SELECT id, url, source_url" in query: + batch = params[-1] + exclude = set(params[0]) if len(params) == 2 else set() + candidates = [ + row for row in self._state.rows + if row["status"] == "pending" and row["id"] not in exclude + ] + candidates.sort(key=lambda row: (-row["priority"], row["id"])) + self._rows = [ + (row["id"], row["url"], row["source"]) + for row in candidates[:batch] + ] + elif "SET status = 'processing'" in query: + (queue_id,) = params + self._rows = [] + for row in self._state.rows: + if row["id"] == queue_id and row["status"] == "pending": + row["status"] = "processing" + row["attempts"] += 1 + self._rows = [(row["id"], row["url"], row["source"])] + break + elif "GREATEST(attempts - 1, 0)" in query: + (queue_id,) = params + for row in self._state.rows: + if row["id"] == queue_id: + row["status"] = "pending" + row["attempts"] = max(row["attempts"] - 1, 0) + elif "WHEN attempts >=" in query: + max_attempts, queue_id = params + self._rows = [] + for row in self._state.rows: + if row["id"] == queue_id: + row["status"] = ( + "failed" if row["attempts"] >= max_attempts + else "pending" + ) + self._rows = [(row["status"],)] + else: + raise AssertionError("unexpected query: %s" % query[:120]) + + def fetchone(self): + return self._rows[0] if self._rows else None + + def fetchall(self): + return self._rows + + def __enter__(self): + return self + + def __exit__(self, *_exc): + return False + + +class FakeConnection: + def __init__(self, state): + self.state = state + + def cursor(self): + return FakeCursor(self.state) + + def commit(self): + pass + + def rollback(self): + pass + + +class HostSchedulerTest(unittest.TestCase): + + def test_first_request_is_ready(self): + scheduler = crawler.HostScheduler() + self.assertEqual( + scheduler.wait_time("https://a.example/x", 3.0), 0.0 + ) + + def test_delay_is_honored_after_request(self): + scheduler = crawler.HostScheduler() + scheduler.wait("https://a.example/x", 0.0) + remaining = scheduler.wait_time("https://a.example/y", 3.0) + self.assertGreater(remaining, 2.5) + self.assertLessEqual(remaining, 3.0) + + def test_short_wait_is_slept_inline(self): + scheduler = crawler.HostScheduler() + scheduler.wait("https://a.example/x", 0.0) + with mock.patch("time.sleep") as sleep: + self.assertTrue(scheduler.wait("https://a.example/y", 3.0)) + sleep.assert_called_once() + + def test_long_wait_is_deferred_without_touching_schedule(self): + scheduler = crawler.HostScheduler() + scheduler.wait("https://a.example/x", 0.0) + recorded = scheduler.last_request["a.example"] + with mock.patch("time.sleep") as sleep: + self.assertFalse(scheduler.wait("https://a.example/y", 60.0)) + sleep.assert_not_called() + self.assertEqual(scheduler.last_request["a.example"], recorded) + + def test_block_backoff_doubles(self): + scheduler = crawler.HostScheduler() + url = "https://a.example/x" + scheduler.note_http_error(url, 429) + first = scheduler.wait_time("https://a.example/y", 3.0) + scheduler.note_http_error(url, 429) + second = scheduler.wait_time("https://a.example/y", 3.0) + self.assertGreater(first, crawler.HOST_BLOCK_RULES[429][0] * 0.9) + self.assertGreater(second, first * 1.5) + + def test_retry_after_is_honored(self): + scheduler = crawler.HostScheduler() + scheduler.note_http_error( + "https://a.example/x", 429, retry_after=1200 + ) + remaining = scheduler.wait_time("https://a.example/y", 3.0) + self.assertGreater(remaining, 1000) + self.assertLessEqual(remaining, crawler.HOST_BLOCK_HARD_CAP) + + def test_block_never_exceeds_hard_cap(self): + scheduler = crawler.HostScheduler() + url = "https://a.example/x" + for _ in range(40): + scheduler.note_http_error(url, 429) + remaining = scheduler.wait_time(url, 3.0) + self.assertLessEqual(remaining, crawler.HOST_BLOCK_HARD_CAP + 1) + + def test_success_resets_block_counter(self): + scheduler = crawler.HostScheduler() + url = "https://a.example/x" + scheduler.note_http_error(url, 429) + scheduler.note_http_error(url, 429) + scheduler.note_success(url) + scheduler.note_http_error(url, 429) + remaining = scheduler.wait_time("https://a.example/y", 3.0) + self.assertLessEqual( + remaining, crawler.HOST_BLOCK_RULES[429][0] + 1 + ) + + def test_non_block_status_is_ignored(self): + scheduler = crawler.HostScheduler() + scheduler.note_http_error("https://a.example/x", 404) + self.assertEqual( + scheduler.wait_time("https://a.example/y", 3.0), 0.0 + ) + + def test_network_error_backs_off(self): + scheduler = crawler.HostScheduler() + scheduler.note_network_error("https://a.example/x") + remaining = scheduler.wait_time("https://a.example/y", 3.0) + self.assertGreater(remaining, 20) + + def test_unrelated_host_is_not_blocked(self): + scheduler = crawler.HostScheduler() + scheduler.note_http_error("https://a.example/x", 429) + self.assertEqual( + scheduler.wait_time("https://b.example/y", 3.0), 0.0 + ) + + +class ErrorClassificationTest(unittest.TestCase): + + def test_network_errors_are_transient(self): + self.assertTrue(crawler.is_transient_error("network:timeout")) + self.assertTrue(crawler.is_transient_error("network:[Errno -2] x")) + + def test_rate_limit_and_server_errors_are_transient(self): + for status in (408, 425, 429, 500, 502, 503, 504): + self.assertTrue(crawler.is_transient_error("http_%d" % status)) + + def test_terminal_errors_are_not_transient(self): + for error in ( + "http_404", "http_410", "robots", "too_large", + "not_html", "error:boom", "host_busy", None, "", + ): + self.assertFalse(crawler.is_transient_error(error)) + + def test_retry_after_seconds_digit(self): + self.assertEqual( + crawler._retry_after_seconds({"Retry-After": "120"}), 120.0 + ) + + def test_retry_after_http_date(self): + headers = { + "Retry-After": email_utils.formatdate( + time.time() + 120, usegmt=True + ) + } + value = crawler._retry_after_seconds(headers) + self.assertGreater(value, 100) + self.assertLess(value, 140) + + def test_retry_after_missing_or_garbage(self): + self.assertIsNone(crawler._retry_after_seconds({})) + self.assertIsNone( + crawler._retry_after_seconds({"Retry-After": "soon"}) + ) + self.assertIsNone(crawler._retry_after_seconds(None)) + + +class QueueClaimTest(unittest.TestCase): + + def _state(self): + state = FakeState() + state.add(url="https://blocked.example/a", priority=500) + state.add(url="https://other.example/b", priority=100) + state.add(url="https://blocked.example/c", priority=400) + return state + + def _claim(self, state, scheduler=None): + scheduler = scheduler or crawler.HostScheduler() + return crawler.claim_next_queue_item( + FakeConnection(state), scheduler, StubRobots() + ) + + def test_highest_priority_ready_host_wins(self): + item, wait_hint = self._claim(self._state()) + self.assertEqual(item[1], "https://blocked.example/a") + self.assertIsNone(wait_hint) + + def test_blocked_host_is_skipped_for_other_domain(self): + scheduler = crawler.HostScheduler() + scheduler.blocked_until["blocked.example"] = time.monotonic() + 900 + item, wait_hint = self._claim(self._state(), scheduler) + self.assertEqual(item[1], "https://other.example/b") + self.assertGreater(wait_hint, 800) + + def test_rate_limited_host_is_skipped(self): + scheduler = crawler.HostScheduler() + scheduler.wait("https://blocked.example/a", 0.0) + item, _wait_hint = self._claim(self._state(), scheduler) + self.assertEqual(item[1], "https://other.example/b") + + def test_only_blocked_hosts_left_reports_wait_hint(self): + state = FakeState() + state.add(url="https://blocked.example/a", priority=500) + scheduler = crawler.HostScheduler() + scheduler.blocked_until["blocked.example"] = time.monotonic() + 900 + item, wait_hint = self._claim(state, scheduler) + self.assertIsNone(item) + self.assertGreater(wait_hint, 800) + + def test_empty_queue(self): + item, wait_hint = self._claim(FakeState()) + self.assertIsNone(item) + self.assertIsNone(wait_hint) + + def test_claim_marks_processing_and_counts_attempt(self): + state = self._state() + self._claim(state) + self.assertEqual(state.rows[0]["status"], "processing") + self.assertEqual(state.rows[0]["attempts"], 1) + self.assertEqual(state.rows[1]["status"], "pending") + + def test_claim_finds_ready_host_behind_blocked_wall(self): + state = FakeState() + for index in range(30): + state.add( + url="https://blocked.example/%d" % index, + priority=300 - index, + ) + state.add(url="https://ready.example/x", priority=10) + scheduler = crawler.HostScheduler() + scheduler.blocked_until["blocked.example"] = time.monotonic() + 900 + item, _wait_hint = self._claim(state, scheduler) + self.assertEqual(item[1], "https://ready.example/x") + + def test_defer_does_not_burn_an_attempt(self): + state = FakeState() + queue_id = state.add( + url="https://slow.example/a", + status="processing", + attempts=1, + ) + crawler.defer_queue_item(FakeConnection(state), queue_id) + row = state.rows[0] + self.assertEqual(row["status"], "pending") + self.assertEqual(row["attempts"], 0) + + def test_requeue_retries_below_attempt_cap(self): + state = FakeState() + queue_id = state.add( + url="https://x.example/a", + status="processing", + attempts=1, + ) + status = crawler.requeue_queue_item(FakeConnection(state), queue_id) + self.assertEqual(status, "pending") + self.assertEqual(state.rows[0]["status"], "pending") + + def test_requeue_fails_after_attempt_cap(self): + state = FakeState() + queue_id = state.add( + url="https://x.example/a", + status="processing", + attempts=crawler.QUEUE_MAX_ATTEMPTS, + ) + status = crawler.requeue_queue_item(FakeConnection(state), queue_id) + self.assertEqual(status, "failed") + self.assertEqual(state.rows[0]["status"], "failed") + + +class HostFilterTest(unittest.TestCase): + + def test_localhost_and_private_suffixes_are_blocked(self): + for host in ( + "localhost", "sub.localhost", "localhost.localdomain", + "myserver.local", "app.internal", + ): + self.assertTrue( + crawler.is_non_crawlable_host(host), host + ) + + def test_ip_literals_are_blocked(self): + for host in ( + "127.0.0.1", "10.0.0.5", "192.168.1.10", "172.16.0.1", + "8.8.8.8", "::1", "fe80::1", "2001:db8::1", + "::ffff:127.0.0.1", + ): + self.assertTrue( + crawler.is_non_crawlable_host(host), host + ) + + def test_single_label_and_empty_names_are_blocked(self): + for host in ("mymachine", "", " ", "."): + self.assertTrue( + crawler.is_non_crawlable_host(host), repr(host) + ) + + def test_real_domains_are_allowed(self): + for host in ( + "example.com", "www.example.com", + "sub.domain.example.co.uk", "xn--bcher-kva.example", + "example.com.", + ): + self.assertFalse( + crawler.is_non_crawlable_host(host), host + ) + + def test_normalize_url_rejects_non_public_targets(self): + for url in ( + "http://localhost/x", + "http://127.0.0.1/x", + "http://[::1]:8080/x", + "http://10.1.2.3/a?b=c", + "http://intranet.local/x", + "http://host/x", + ): + self.assertIsNone(crawler.normalize_url(url), url) + + def test_normalize_url_keeps_real_domains(self): + self.assertEqual( + crawler.normalize_url("http://www.example.com/x?y=1"), + "http://www.example.com/x?y=1", + ) + + def test_is_non_crawlable_url_handles_malformed_urls(self): + self.assertTrue(crawler.is_non_crawlable_url("http://localhost/x")) + self.assertTrue(crawler.is_non_crawlable_url("http://[::1]/")) + self.assertFalse(crawler.is_non_crawlable_url("https://example.com/")) + + def test_malformed_url_is_treated_as_non_crawlable(self): + # urlparse raises ValueError on malformed IPv6 brackets; the + # filter must not crash on it. + self.assertTrue( + crawler.is_non_crawlable_url("http://[::1") + ) + + +if __name__ == "__main__": + unittest.main()