#!/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.

QUEUE PENDING —
IN BEARBEITUNG —
PRIORISIERTE DOMAINS —

Letzte Vorschläge

Die zuletzt priorisierten Domains und ihr Queue-Status

Lade…
""" if __name__ == "__main__": host = os.getenv("HOST", "0.0.0.0") # Bookkeeping table for first run. try: ensure_suggestions_table_standalone() except Exception as exc: print(f"WARN: could not ensure domain_suggestions table: {exc}") start_queue_worker() print("=" * 70) print(" JonasSearch Vorschlagsverfeinerer Gamma 0.1") print("=" * 70) print(f"Statserver : Port {STATSERVER_PORT}") print(f"HTTP : {host}:{PORT} (Statserver + 1)") print(f"DB : {DB_HOST}:{DB_PORT}/{DB_NAME}") print(f"Priorität : {SUGGEST_PRIORITY}") print("=" * 70) app.run( host=host, port=PORT, debug=False, threaded=True, )