From d415f6952bd235ed954f95cda675a1ff281ee813 Mon Sep 17 00:00:00 2001 From: jonas H Date: Wed, 2 Sep 2026 19:54:49 +0200 Subject: [PATCH] add search --- .env.example | 15 ++ README.md | 6 +- search/README.md | 74 ++++++++ search/__init__.py | 22 +++ search/api.py | 371 +++++++++++++++++++++++++++++++++++++ search/cli.py | 153 +++++++++++++++ search/engine.py | 451 +++++++++++++++++++++++++++++++++++++++++++++ 7 files changed, 1091 insertions(+), 1 deletion(-) create mode 100644 search/README.md create mode 100644 search/__init__.py create mode 100644 search/api.py create mode 100644 search/cli.py create mode 100644 search/engine.py diff --git a/.env.example b/.env.example index e6fc6c0..4b2c68e 100644 --- a/.env.example +++ b/.env.example @@ -25,3 +25,18 @@ JONASSEARCH_MAX_CONTENT_LENGTH=5242880 # User-Agent JONASSEARCH_USER_AGENT=JonasSearchBot/0.1 (+https://jonassearch.local/bot) + +# ============================================================ +# Suchmaschine (search/) +# ============================================================ + +# PostgreSQL-Textsuch-Konfiguration (Stemming/Stopwörter) +JONASSEARCH_TS_CONFIG=german + +# tsvector-Spalte der Tabelle `pages` erzwingen (optional, +# wird automatisch erkannt) +#JONASSEARCH_VECTOR_COLUMN=search_vector + +# HTTP-API (search/api.py) +JONASSEARCH_API_HOST=0.0.0.0 +JONASSEARCH_API_PORT=8082 diff --git a/README.md b/README.md index 82013b6..8b88973 100644 --- a/README.md +++ b/README.md @@ -1,3 +1,7 @@ # search -das komplette serverjonas search \ No newline at end of file +das komplette serverjonas search + +- `crawler.py` – PostgreSQL-Crawler, baut den Suchindex auf +- `search/` – Suchmaschine darüber (CLI + HTTP-API), siehe [search/README.md](search/README.md) +- `web/` – Statistik-Webseite \ No newline at end of file diff --git a/search/README.md b/search/README.md new file mode 100644 index 0000000..d5e13eb --- /dev/null +++ b/search/README.md @@ -0,0 +1,74 @@ +# JonasSearch – Suchmaschine + +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`). + +## 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 | +|---|---|---| +| `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` | + +## Abhängigkeiten + +- `psycopg2` (DB-Zugriff) +- `python-dotenv` (`.env`-Datei) +- `flask` (nur für `search/api.py`) \ No newline at end of file diff --git a/search/__init__.py b/search/__init__.py new file mode 100644 index 0000000..c9fa74c --- /dev/null +++ b/search/__init__.py @@ -0,0 +1,22 @@ +""" +JonasSearch – Suchmaschine über den PostgreSQL-Index des Crawlers. + +Public API: + from search import SearchEngine, SearchHit, SearchResponse +""" + +from .engine import ( + SearchEngine, + SearchHit, + SearchResponse, + parse_terms, +) + +__all__ = [ + "SearchEngine", + "SearchHit", + "SearchResponse", + "parse_terms", +] + +__version__ = "1.0.0" \ No newline at end of file diff --git a/search/api.py b/search/api.py new file mode 100644 index 0000000..e4d80c3 --- /dev/null +++ b/search/api.py @@ -0,0 +1,371 @@ +#!/usr/bin/env python3 +""" +JonasSearch – Suchmaschine (HTTP-API). + +Einfacher Suchserver über den vom Crawler aufgebauten Index: + + python3 search/api.py + + 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) +""" + +import logging +import os + +from flask import Flask, jsonify, render_template_string, request + +try: + from .engine import SearchEngine +except ImportError: # python3 search/api.py (ohne -m) + from engine import SearchEngine + +API_HOST = os.getenv("JONASSEARCH_API_HOST", "0.0.0.0") +API_PORT = int(os.getenv("JONASSEARCH_API_PORT", "8082")) + +PAGE_HTML = """\ + + + + + + JonasSearch + + + + + +
+

JonasSearch

+
Suchmaschine über den Crawler-Index
+ + + +
+
+ +
+ + + + + +""" + + +def create_app() -> Flask: + app = Flask(__name__) + + # 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 + + +if __name__ == "__main__": + logging.basicConfig(level=logging.INFO) + + 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) + + create_app().run(host=API_HOST, port=API_PORT, threaded=True) \ No newline at end of file diff --git a/search/cli.py b/search/cli.py new file mode 100644 index 0000000..79ec803 --- /dev/null +++ b/search/cli.py @@ -0,0 +1,153 @@ +#!/usr/bin/env python3 +""" +JonasSearch – Suchmaschine (Kommandozeile). + +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 +""" + +import argparse +import html +import json +import re +import sys + +try: + from .engine import SearchEngine +except ImportError: # python3 search/cli.py (ohne -m) + from engine import SearchEngine + +_TAG_RE = re.compile(r"<[^>]+>") + + +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) + + +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 build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser( + prog="jonassearch", + description="JonasSearch – Suchmaschine (Kommandozeile).", + epilog=( + "Konfiguration über JONASSEARCH_* Umgebungsvariablen, " + "siehe .env.example." + ), + ) + + 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", + ) + + return parser + + +def main(argv=None) -> int: + args = build_parser().parse_args(argv) + + engine = SearchEngine(ts_config=args.config) + + try: + response = engine.search( + query=args.query, + page=args.page, + per_page=args.per_page, + domain=args.domain, + tld=args.tld, + ) + except Exception as exc: # z. B. DB nicht erreichbar, Tabelle fehlt + print(f"Fehler bei der Suche: {exc}", file=sys.stderr) + return 1 + + if args.json: + print(json.dumps(response.to_dict(), ensure_ascii=False, indent=2)) + else: + print(format_plain(response)) + + return 0 + + +if __name__ == "__main__": + sys.exit(main()) \ No newline at end of file diff --git a/search/engine.py b/search/engine.py new file mode 100644 index 0000000..f5504d2 --- /dev/null +++ b/search/engine.py @@ -0,0 +1,451 @@ +#!/usr/bin/env python3 +""" +JonasSearch – Suchmaschine (Engine). + +Volltextsuche über den Index, den crawler.py in PostgreSQL aufgebaut hat +(Tabelle `pages`). + +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. + +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 + +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) + +Beispiel: + from search.engine import SearchEngine + engine = SearchEngine() + result = engine.search("raspberry pi", page=1, per_page=10) +""" + +import logging +import os +import re +import time +from dataclasses import dataclass, field +from typing import List, Optional, Tuple + +from dotenv import load_dotenv +import psycopg2 + +load_dotenv() + +# ============================================================ +# KONFIGURATION +# ============================================================ + +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" + +# Optional: tsvector-Spalte der Tabelle `pages` erzwingen. +VECTOR_COLUMN = os.getenv("JONASSEARCH_VECTOR_COLUMN", "").strip() + +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= …" +) + +# 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'<', '<')" +) + + +# ============================================================ +# DATENTYPEN +# ============================================================ + +@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, + } + + +@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], + } + + +# ============================================================ +# HILFSFUNKTIONEN +# ============================================================ + +def parse_terms(raw_query: str) -> List[str]: + """ + Zerlegt eine Roh-Abfrage in einzelne Suchbegriffe. + + Es werden nur Wortzeichen behalten (auch Umlaute), alles andere + (Interpunktion, Sonderzeichen) wird als Trenner behandelt. + """ + terms = [t.lower() for t in _TERM_RE.findall(raw_query or "")] + return terms[:MAX_TERMS] + + +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() + " …" + + +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 + """ + + 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) + + total = cur.fetchone()[0] + + cur.execute(select_sql, params) + rows = cur.fetchall() + + hits = [] + for index, row in enumerate(rows): + ( + url, + title, + dom, + tld_value, + crawled_at, + headings, + snippet, + rank, + ) = row + + 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] + ], + ) + ) + + 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, + ) + + 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