#!/usr/bin/env python3
"""
JonasSearch Vorschlagsverfeinerer Gamma 0.1
Lets users submit domains that are put at a very high priority in the
crawler's PostgreSQL queue (crawl_queue), so they get crawled next.
Runs on PORT = statserver port + 1 (default 8082).
How the queue jump works
------------------------
The crawler claims work with:
ORDER BY priority DESC, discovered_at ASC, id ASC
Normal crawler-discovered URLs sit at priority 0..500. A suggestion is
upserted at SUGGEST_PRIORITY (1,000,000), which sorts it above the whole
existing queue WITHOUT touching the other ~13M pending rows. The table
is far too large for mass UPDATEs: such a query blocks the crawler for
minutes. Multiple suggestions all share SUGGEST_PRIORITY and are
crawled in submission order (discovered_at, id), so no demotion is
needed there either.
"""
import os
import re
import threading
import time
from datetime import datetime, timezone
import psycopg2
from flask import Flask, jsonify, render_template_string, request
from psycopg2.extras import RealDictCursor
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", "")
# Port of the statserver (jonassearch-stats.service, web/app.py).
STATSERVER_PORT = int(os.getenv("JONASSEARCH_STATSERVER_PORT", "8081"))
PORT = int(os.getenv("PORT", str(STATSERVER_PORT + 1)))
# Very high priority for user suggestions. The current pending maximum
# in the queue is 500 (crawler-assigned), so this always sorts on top.
SUGGEST_PRIORITY = 1_000_000
# Hard safety net: this UI must never run away on the huge queue table.
STATEMENT_TIMEOUT_MS = int(os.getenv("JONASSEARCH_STATEMENT_TIMEOUT_MS", "15000"))
# Max suggestions shown in the recent list.
MAX_RECENT = 50
# Very defensive domain validation (ASCII only).
DOMAIN_RE = re.compile(
r"^(?=.{1,253}\.)([a-z0-9]([a-z0-9-]{0,61}[a-z0-9])?\.)+"
r"[a-z]{2,63}$"
)
app = Flask(__name__)
def db():
return psycopg2.connect(
host=DB_HOST,
port=DB_PORT,
dbname=DB_NAME,
user=DB_USER,
password=DB_PASSWORD,
connect_timeout=5,
application_name="jonassearch-priority",
options=f"-c statement_timeout={STATEMENT_TIMEOUT_MS}",
)
def query(sql, params=None):
conn = db()
try:
with conn.cursor(cursor_factory=RealDictCursor) as cur:
cur.execute(sql, params)
if cur.description:
return cur.fetchall()
return []
finally:
conn.close()
def clean_domain(raw):
"""
Turn arbitrary user input into a registrable domain.
Accepts "example.com", "https://www.example.com/pfad",
"www.example.com", strips scheme, path, port, "www." prefix.
Returns lowercase domain or None if invalid.
"""
if not raw:
return None
value = raw.strip().lower()
# Accept full URLs too.
if "://" in value:
value = value.split("://", 1)[1]
# Cut off path, query, port and userinfo.
value = value.split("/", 1)[0]
value = value.split("?", 1)[0]
value = value.split("#", 1)[0]
value = value.split("@", 1)[-1]
value = value.rsplit(":", 1)[0]
# Strip a leading www. so we prioritize the whole domain.
if value.startswith("www."):
value = value[4:]
value = value.rstrip(".")
if not DOMAIN_RE.match(value):
return None
return value
def ensure_suggestions_table(conn):
"""Bookkeeping table for the UI; the queue itself stays untouched."""
with conn.cursor() as cur:
cur.execute("""
CREATE TABLE IF NOT EXISTS domain_suggestions (
id BIGSERIAL PRIMARY KEY,
domain TEXT NOT NULL UNIQUE,
requested_at TIMESTAMPTZ NOT NULL DEFAULT now(),
request_count INTEGER NOT NULL DEFAULT 1,
last_status TEXT
)
""")
conn.commit()
def ensure_suggestions_table_standalone():
"""Create the bookkeeping table with its own short-lived connection."""
conn = db()
try:
ensure_suggestions_table(conn)
finally:
conn.close()
def suggest_domain(conn, domain):
"""
Put the domain at the very front of the crawl queue.
Uses the passed connection (so tests can run against temp tables)
and only touches rows by their unique url index — never the whole
table.
"""
url = f"https://{domain}/"
with conn.cursor() as cur:
# The domain itself: insert at top priority. If it already
# exists, pull it back to pending at top priority — unless a
# worker is processing it right now, in which case we leave it
# alone and let the worker finish.
cur.execute("""
INSERT INTO crawl_queue (url, source_url, priority, status)
VALUES (%s, %s, %s, 'pending')
ON CONFLICT (url) DO UPDATE SET
priority = EXCLUDED.priority,
status = 'pending',
last_attempt_at = NULL
WHERE crawl_queue.status IS DISTINCT FROM 'processing'
""", (url, "vorschlagsverfeinerer", SUGGEST_PRIORITY))
# UI bookkeeping.
cur.execute("""
INSERT INTO domain_suggestions (domain, last_status)
VALUES (%s, 'queued')
ON CONFLICT (domain) DO UPDATE SET
requested_at = now(),
request_count = domain_suggestions.request_count + 1,
last_status = 'queued'
""", (domain,))
conn.commit()
def urls_ahead_of(domain):
"""
How many pending URLs are claimed before this suggestion.
The crawler picks: priority DESC, then discovered_at ASC, then id
ASC. All suggestions share SUGGEST_PRIORITY, so only other
suggestions submitted earlier can be ahead.
"""
row = query("""
SELECT COUNT(*) AS ahead
FROM crawl_queue
WHERE status = 'pending'
AND (
priority > %(P)s
OR (
priority = %(P)s
AND (discovered_at, id) < (
SELECT discovered_at, id
FROM crawl_queue
WHERE url = %(url)s
)
)
)
""", {"P": SUGGEST_PRIORITY, "url": f"https://{domain}/"})
return int(row[0]["ahead"] or 0)
# ============================================================
# ROUTES
# ============================================================
@app.get("/")
def index():
return render_template_string(HTML, port=PORT)
@app.get("/api/suggest")
def api_suggest():
"""Convenience: /api/suggest?domain=example.com also works."""
return _handle_suggest(request.args.get("domain", ""))
@app.post("/api/suggest")
def api_suggest_post():
body = request.get_json(silent=True) or {}
domain_input = body.get("domain") or request.form.get("domain", "")
return _handle_suggest(domain_input)
def _handle_suggest(raw_domain):
domain = clean_domain(raw_domain)
if not domain:
return jsonify({
"ok": False,
"error": (
"Bitte eine gültige Domain eingeben, z. B. "
"example.com oder https://www.example.com/seite"
),
}), 400
conn = None
try:
conn = db()
suggest_domain(conn, domain)
except Exception as exc:
if conn is not None:
conn.rollback()
conn.close()
conn = None
# First run without the bookkeeping table: create it and retry
# once (42P01 = undefined_table).
if getattr(exc, "pgcode", None) == "42P01":
try:
ensure_suggestions_table_standalone()
conn = db()
suggest_domain(conn, domain)
except Exception as retry_exc:
if conn is not None:
conn.rollback()
return jsonify({
"ok": False,
"error": f"Datenbankfehler: {retry_exc}",
}), 503
else:
return jsonify({
"ok": False,
"error": f"Datenbankfehler: {exc}",
}), 503
finally:
if conn is not None:
conn.close()
try:
ahead = urls_ahead_of(domain)
except Exception:
ahead = None
return jsonify({
"ok": True,
"domain": domain,
"priority": SUGGEST_PRIORITY,
"urls_ahead": ahead,
})
@app.get("/api/suggestions")
def api_suggestions():
"""Recent suggestions for the UI list."""
try:
rows = query("""
SELECT domain,
requested_at,
request_count,
last_status,
EXISTS (
SELECT 1 FROM crawl_queue q
WHERE q.url = 'https://' || domain_suggestions.domain || '/'
AND q.status = 'pending'
) AS still_pending
FROM domain_suggestions
ORDER BY requested_at DESC
LIMIT %s
""", (MAX_RECENT,))
except Exception:
rows = []
return jsonify([
{
"domain": r["domain"],
"requested_at": r["requested_at"].isoformat(),
"request_count": int(r["request_count"]),
"last_status": r["last_status"],
"still_pending": bool(r["still_pending"]),
}
for r in rows
])
# ============================================================
# QUEUE SNAPSHOT (background refresh)
# ============================================================
# The pending/processing counts scan a 13M-row queue (several seconds
# of index-only scan). They must never run inside a request: a
# background thread refreshes them every 60 s and /api/queue serves
# the cached snapshot. The "boosted" count only reads the tiny
# priority >= SUGGEST_PRIORITY index range and is always fast.
QUEUE_REFRESH_SECONDS = int(os.getenv("JONASSEARCH_QUEUE_REFRESH_SECONDS", "60"))
_queue_lock = threading.Lock()
_queue_snapshot = None
def compute_queue_snapshot():
"""Full queue stats; may take several seconds on the big table."""
row = query("""
SELECT
(SELECT COUNT(*) FROM crawl_queue
WHERE status = 'pending') AS pending,
(SELECT COUNT(*) FROM crawl_queue
WHERE status = 'processing') AS processing,
(SELECT COUNT(*) FROM crawl_queue
WHERE status = 'pending'
AND priority >= %s) AS boosted
""", (SUGGEST_PRIORITY,))[0]
return {
"pending": int(row["pending"] or 0),
"processing": int(row["processing"] or 0),
"boosted": int(row["boosted"] or 0),
}
def _queue_worker():
global _queue_snapshot
while True:
try:
snapshot = compute_queue_snapshot()
with _queue_lock:
_queue_snapshot = snapshot
time.sleep(QUEUE_REFRESH_SECONDS)
except Exception:
# DB hiccup: retry shortly; the UI keeps the last snapshot.
time.sleep(15)
def start_queue_worker():
thread = threading.Thread(
target=_queue_worker,
name="queue-stats",
daemon=True,
)
thread.start()
@app.get("/api/queue")
def api_queue():
"""Small live status for the header widgets (cached, see above)."""
with _queue_lock:
snapshot = _queue_snapshot
if snapshot is not None:
return jsonify({
**snapshot,
"online": True,
"timestamp": datetime.now(timezone.utc).isoformat(),
})
# No snapshot yet (service just started): fall back to the fast
# boosted-only count instead of blocking the request for seconds.
try:
row = query("""
SELECT COUNT(*) AS boosted
FROM crawl_queue
WHERE status = 'pending'
AND priority >= %s
""", (SUGGEST_PRIORITY,))[0]
except Exception:
return jsonify({"online": False}), 503
return jsonify({
"online": True,
"pending": None,
"processing": None,
"boosted": int(row["boosted"] or 0),
"timestamp": datetime.now(timezone.utc).isoformat(),
})
@app.get("/api/health")
def api_health():
try:
query("SELECT 1")
return jsonify({"status": "ok"})
except Exception as exc:
return jsonify({"status": "error", "error": str(exc)}), 503
# ============================================================
# HTML (inline, no extra static files needed)
# ============================================================
HTML = """
JonasSearch · Vorschlagsverfeinerer
SERVERJONASSEARCH
Vorschlagsverfeinerer
Gamma 0.1 · Domain vorschlagen und die Warteschlange überholen
Verbinde…
Domain priorisieren
Eingabe als Domain oder komplette URL — z. B.
example.com oder
https://www.example.com/seite.
Der Vorschlag landet mit sehr hoher Priorität direkt vorne in der
Crawl-Queue.