Serverjonas/search
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

add search

Browse Source
This commit is contained in:
jhkaktus
2026-09-02 19:54:49 +02:00
parent f1c7ecc14d
commit d415f6952b
7 changed files with 1091 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.env.example
+15
View File
@@ -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
README.md
+5 -1
View File
@@ -1,3 +1,7 @@
# search
das komplette serverjonas search
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
search/README.md
+74
View File
@@ -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 `<mark>` hervorgehoben sind.
- **Präfix-Suche:** Jeder Begriff wird mit `:*` gesucht, „raspberr" findet
also auch „raspberry".
- **UND-Verknüpfung:** Mehrere Begriffe müssen alle vorkommen.
- **Paginierung:** `--page` / `--per-page` (max. 50).
- **Filter:** `--domain` (exakte Domain) und `--tld` (z. B. `.de`).
## Umgebungsvariablen
Wie der Crawler (siehe `.env.example`):
| Variable | Bedeutung | Standard |
|---|---|---|
| `JONASSEARCH_DB_HOST` | DB-Host | `127.0.0.1` |
| `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`)
search/__init__.py
+22
View File
@@ -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"
search/api.py
+371
View File
@@ -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 = """\
<!DOCTYPE html>
<html lang="de">
<head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<title>JonasSearch</title>
<style>
body {
margin: 0;
background: #111;
color: #eee;
font-family: Arial, sans-serif;
}
.container {
max-width: 800px;
margin: auto;
padding: 30px;
}
h1 {
margin: 0 0 4px;
}
.subtitle {
color: #888;
margin-bottom: 24px;
}
form.search {
display: flex;
gap: 8px;
margin-bottom: 16px;
}
input[type="text"] {
flex: 1;
padding: 12px 14px;
font-size: 1rem;
border-radius: 10px;
border: 1px solid #333;
background: #1b1b1b;
color: #eee;
}
button {
padding: 12px 20px;
font-size: 1rem;
border: none;
border-radius: 10px;
background: #2563eb;
color: #fff;
cursor: pointer;
}
button:hover {
background: #1d4ed8;
}
.meta {
color: #aaa;
margin-bottom: 20px;
}
.result {
background: #1b1b1b;
border-radius: 12px;
padding: 16px 18px;
margin-bottom: 12px;
}
.result h3 {
margin: 0 0 4px;
font-size: 1.05rem;
}
.result h3 a {
color: #60a5fa;
text-decoration: none;
}
.result h3 a:hover {
text-decoration: underline;
}
.result .url {
color: #4ade80;
font-size: 0.85rem;
margin-bottom: 6px;
overflow-wrap: anywhere;
}
.result .snippet {
color: #ccc;
line-height: 1.5;
}
.result .snippet mark {
background: #ca8a04;
color: #111;
border-radius: 3px;
padding: 0 2px;
}
.result .meta {
color: #777;
font-size: 0.8rem;
margin-top: 8px;
margin-bottom: 0;
}
.pagination {
display: flex;
gap: 10px;
margin-top: 20px;
}
.pagination a {
color: #60a5fa;
text-decoration: none;
}
.empty {
color: #888;
text-align: center;
margin-top: 40px;
}
</style>
</head>
<body>
<div class="container">
<h1>JonasSearch</h1>
<div class="subtitle">Suchmaschine über den Crawler-Index</div>
<form class="search" id="searchform">
<input
type="text"
id="q"
placeholder="Suchen…"
autofocus
autocomplete="off"
>
<button type="submit">Suchen</button>
</form>
<div class="meta" id="meta"></div>
<div id="results"></div>
<div class="pagination" id="pagination"></div>
</div>
<script>
const form = document.getElementById("searchform");
const input = document.getElementById("q");
const meta = document.getElementById("meta");
const results = document.getElementById("results");
const pagination = document.getElementById("pagination");
let currentQuery = "";
let currentPage = 1;
function urlFor(page) {
return "/search?q=" + encodeURIComponent(currentQuery) + "&page=" + page;
}
function render(data) {
meta.textContent = data.total
? data.total + " Ergebnisse in " + data.took_ms + " ms"
+ (data.pages > 1 ? " – Seite " + data.page + "/" + data.pages : "")
: "Keine Ergebnisse für „" + data.query + "“";
results.innerHTML = "";
for (const hit of data.hits) {
const box = document.createElement("div");
box.className = "result";
const title = document.createElement("h3");
const link = document.createElement("a");
link.href = hit.url;
link.target = "_blank";
link.rel = "noopener";
link.textContent = hit.title;
title.appendChild(link);
box.appendChild(title);
const url = document.createElement("div");
url.className = "url";
url.textContent = hit.url;
box.appendChild(url);
// Das Snippet enthält vom Server bereits <mark>-Tags;
// der Text selbst ist serverseitig escaped (XSS-sicher).
const snippet = document.createElement("div");
snippet.className = "snippet";
snippet.innerHTML = hit.snippet;
box.appendChild(snippet);
const metaLine = document.createElement("div");
metaLine.className = "meta";
const parts = [];
if (hit.domain) parts.push(hit.domain);
if (hit.crawled_at) parts.push(hit.crawled_at.slice(0, 10));
metaLine.textContent = parts.join(" · ");
if (metaLine.textContent) box.appendChild(metaLine);
results.appendChild(box);
}
pagination.innerHTML = "";
if (data.pages > 1) {
if (data.page > 1) {
const prev = document.createElement("a");
prev.href = "#";
prev.textContent = "← Zurück";
prev.addEventListener("click", e => {
e.preventDefault();
loadPage(data.page - 1);
});
pagination.appendChild(prev);
}
const label = document.createElement("span");
label.textContent = "Seite " + data.page + " / " + data.pages;
pagination.appendChild(label);
if (data.page < data.pages) {
const next = document.createElement("a");
next.href = "#";
next.textContent = "Weiter →";
next.addEventListener("click", e => {
e.preventDefault();
loadPage(data.page + 1);
});
pagination.appendChild(next);
}
}
}
async function loadPage(page) {
const response = await fetch(urlFor(page));
const data = await response.json();
currentPage = data.page;
render(data);
window.scrollTo({ top: 0, behavior: "smooth" });
}
form.addEventListener("submit", async event => {
event.preventDefault();
const query = input.value.trim();
if (!query) return;
currentQuery = query;
await loadPage(1);
history.replaceState(null, "", "/?q=" + encodeURIComponent(query));
});
// Beim Laden vorbefüllen (z. B. nach Reload mit ?q=)
const initial = new URLSearchParams(location.search).get("q");
if (initial) {
input.value = initial;
currentQuery = initial;
loadPage(1);
}
</script>
</body>
</html>
"""
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)
search/cli.py
+153
View File
@@ -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 <mark>-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())
search/engine.py
+451
View File
@@ -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, <mark>)
- 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 <mark>-Hervorhebung.
HIGHLIGHT_OPTIONS = (
"StartSel=<mark>, StopSel=</mark>, "
"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 <mark>-Tags im Snippet übrig bleiben.
_ESCAPE_FOR_HEADLINE = (
"replace(replace(p.content, E'&', '&amp;'), E'<', '&lt;')"
)
# ============================================================
# 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=[],
)