From 682c0f4971662b5b6b6d7df43eb6a470d5223149 Mon Sep 17 00:00:00 2001 From: jhkaktus Date: Sun, 13 Sep 2026 01:57:46 +0200 Subject: [PATCH] add launcher --- .env.example | 28 ++ .gitignore | 5 + README.md | 45 ++- crawler.py | 743 ++++++++++++++++++++++++++++++++++++++-- launcher.py | 620 ++++++++++++++++++++++++++++++++++ web/app.py | 933 +++++++++++++++++++++++++++++++++++++++++---------- 6 files changed, 2165 insertions(+), 209 deletions(-) create mode 100644 launcher.py diff --git a/.env.example b/.env.example index 4b2c68e..ec437f2 100644 --- a/.env.example +++ b/.env.example @@ -26,6 +26,34 @@ JONASSEARCH_MAX_CONTENT_LENGTH=5242880 # User-Agent JONASSEARCH_USER_AGENT=JonasSearchBot/0.1 (+https://jonassearch.local/bot) +# ============================================================ +# Rolling Updates (Launcher + Crawler) +# ============================================================ + +# Version dieser Crawler-Codebasis (NICHT die Deployment-Generation). +# Muss bei jedem verteilten Crawler-Änderung erhöht werden. +JONASSEARCH_CRAWLER_VERSION=1 + +# Wie oft jeder Crawler außerhalb der Transaktionen nach einer neuen +# Deployment-Generation sieht (Sekunden). Der Check setzt nur ein +# internes Signal; beendet wird sich erst am nächsten sicheren Commit. +JONASSEARCH_UPDATE_CHECK_SECONDS=30 + +# Verzeichnis für gestagte Crawler-Releases (releases//crawler.py). +# Relative Pfade gelten relativ zur crawler.py. +#JONASSEARCH_RELEASES_DIR=releases + +# Launcher: Wohin mit neuen Versionen? Eine der beiden Quellen setzen, +# sonst kann der Launcher nur extern abgelegte Releases prüfen/aktivieren. +#JONASSEARCH_UPDATE_SOURCE_URL=https://example.com/jonassearch/releases +#JONASSEARCH_UPDATE_SOURCE_DIR=/srv/jonassearch/new-release + +# Launcher: Wartezeit nach Crash (Sekunden), bevor der Crawler neu startet +#JONASSEARCH_LAUNCHER_ERROR_RESTART_DELAY=10 + +# Launcher: Wartezeit vor dem Neustart nach einem Update-Zyklus +#JONASSEARCH_LAUNCHER_RESTART_DELAY=2 + # ============================================================ # Suchmaschine (search/) # ============================================================ diff --git a/.gitignore b/.gitignore index 08fb26e..605b1a2 100644 --- a/.gitignore +++ b/.gitignore @@ -178,3 +178,8 @@ cython_debug/ web/page_counts.json + +# Crawler rolling updates +releases/ +.active-crawler-version +crawler.py.bak-* diff --git a/README.md b/README.md index 8b88973..b7bdf3c 100644 --- a/README.md +++ b/README.md @@ -3,5 +3,48 @@ 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 \ No newline at end of file +- `web/` – Statistik-Webseite + +## Rolling Updates (Generation-Mechanismus) + +Zentrale Tabelle (eine Zeile, `id = 1`): + +```sql +CREATE TABLE IF NOT EXISTS deployment_state ( + id INTEGER PRIMARY KEY, + generation BIGINT NOT NULL, + required_version INTEGER NOT NULL +); +``` + +**Ablauf:** + +1. Einmalig initialisieren: `python3 crawler.py --init-deployment` +2. Launcher starten: `python3 launcher.py` (startet + überwacht den Crawler) +3. Neue Version verteilen: + - Neue `crawler.py` mit erhöhter `JONASSEARCH_CRAWLER_VERSION` bereitstellen + (Update-Quelle im `.env` setzen oder Datei nach `releases//crawler.py` legen) + - `python3 launcher.py --check-update` + → neue Version wird gestaget → geprüft (`--self-update`) → aktiviert + → **erst dann** wird die Generation in PostgreSQL erhöht (`--update-deployment`) +4. Alle laufenden alten Crawler merken das beim nächsten sicheren Punkt + (vor jedem relevanten Commit bzw. spätestens nach + `JONASSEARCH_UPDATE_CHECK_SECONDS` im Hintergrund-Check), + rollen ihre offene Transaktion sauber zurück und beenden sich mit + **Exit-Code 42** – nie mitten in einer Transaktion, kein hartes Killen. +5. Der jeweilige Launcher erkennt 42 und startet die bereits bereitgestellte + neue Version, die die neue Generation beim Start liest. + +Exit-Codes des Crawlers: `0` = normal beendet, `1` = Fehler/Crash, +`42` = ausschließlich „Update erforderlich". + +Nützliche Befehle: + +```bash +python3 crawler.py --version # installierte Version +python3 launcher.py --status # Generation + Versionen anzeigen +python3 crawler.py --self-update 2 # gestagte Release prüfen (ohne zu aktivieren) +python3 crawler.py --update-deployment 2 # Generation+1 und required_version=2 setzen +``` \ No newline at end of file diff --git a/crawler.py b/crawler.py index 1bcd4d2..337cd7c 100644 --- a/crawler.py +++ b/crawler.py @@ -18,6 +18,13 @@ Features: - robots.txt support with in-memory cache - Graceful shutdown - Basic crawler statistics +- Generation-based rolling updates: the crawler remembers the + deployment generation it started with, checks it before every + relevant commit (safe point) and exits with exit code 42 when a new + generation requires an update - the open transaction is rolled + back, never half-committed +- Launcher support: --version, --self-update N (verify a staged + release) and --update-deployment N (bump generation + required_version) Expected PostgreSQL schema: pages @@ -30,6 +37,8 @@ Connection defaults can be overridden with environment variables: JONASSEARCH_DB_NAME JONASSEARCH_DB_USER JONASSEARCH_DB_PASSWORD + JONASSEARCH_UPDATE_CHECK_SECONDS + JONASSEARCH_RELEASES_DIR Example: python3 crawler.py @@ -37,10 +46,14 @@ Example: import os import re +import sys import math import signal import time import logging +import argparse +import subprocess +import threading import multiprocessing as mp from collections import defaultdict from html import unescape @@ -59,11 +72,29 @@ try: except ImportError: BeautifulSoup = None -# .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 / ".env") +def _load_env_files(): + """ + .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). + + Eine nicht lesbare .env darf den Crawler nicht abstürzen lassen + (z. B. falscher Benutzer unter systemd): dann laufen die bereits + gesetzten Umgebungsvariablen unverändert weiter. + """ + for dotenv_path in (None, Path(__file__).resolve().parent / ".env"): + try: + load_dotenv(dotenv_path) + except OSError as exc: + logging.warning( + "[CONFIG] %s not readable (%s); continuing with the " + "current environment.", + dotenv_path or ".env (working directory)", + exc, + ) + + +_load_env_files() # ============================================================ @@ -99,6 +130,46 @@ USER_AGENT = os.getenv( "JonasSearchBot/0.1 (+https://jonassearch.local/bot)" ) +# ============================================================ +# DEPLOYMENT / UPDATE CONFIG +# ============================================================ + +# Version of THIS crawler code. Increased with every released crawler +# 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")) + +# Exit codes. 42 is reserved exclusively for "I must be updated" so +# the launcher can distinguish update, crash and normal exit. +EXIT_OK = 0 +EXIT_ERROR = 1 +EXIT_UPDATE_REQUIRED = 42 + +# How often the background checker re-reads the generation. It only +# sets an internal flag; the actual exit happens at the next safe +# commit point, never in the middle of a transaction. +UPDATE_CHECK_INTERVAL_SECONDS = float( + os.getenv("JONASSEARCH_UPDATE_CHECK_SECONDS", "30") +) + +# Staging directory for updates: the launcher stages a new version in +# releases//crawler.py, verifies it with --self-update and +# only then activates it. Relative paths resolve against the crawler +# file's directory, so they work from any working directory. +_RELEASES_RAW = os.getenv("JONASSEARCH_RELEASES_DIR", "releases") +CRAWLER_RELEASES_DIR = Path(_RELEASES_RAW) +if not CRAWLER_RELEASES_DIR.is_absolute(): + CRAWLER_RELEASES_DIR = ( + Path(__file__).resolve().parent / CRAWLER_RELEASES_DIR + ) + +# How long the launcher waits before restarting the crawler after it +# exited with EXIT_UPDATE_REQUIRED. +LAUNCHER_RESTART_DELAY = float( + os.getenv("JONASSEARCH_LAUNCHER_RESTART_DELAY", "2") +) + # Only crawl normal HTTP(S) pages. ALLOWED_SCHEMES = {"http", "https"} @@ -223,6 +294,252 @@ HOST_STATS_CACHE_SECONDS = 30 HOST_STATS_SAMPLE_LIMIT = 20000 +# ============================================================ +# DEPLOYMENT / UPDATE (generation-based rolling updates) +# ============================================================ +# +# The deployment_state table is the global switch for rolling +# updates (exactly one row, id = 1): +# +# CREATE TABLE IF NOT EXISTS deployment_state ( +# id INTEGER PRIMARY KEY, +# generation BIGINT NOT NULL, +# required_version INTEGER NOT NULL +# ) +# +# - generation: global deployment counter. The launcher bumps +# it when a new crawler version goes live. +# - required_version: the crawler version this generation requires. +# +# Every crawler instance reads the generation ONCE at startup ("my +# generation is N"). Before every relevant PostgreSQL commit +# (safe_commit) and additionally in a periodic background check it +# compares the current generation with its own: +# +# - identical -> everything fine, continue. +# - different -> the open transaction is rolled back cleanly and the +# process exits with EXIT_UPDATE_REQUIRED (42). The +# launcher recognizes that exit code, provisions the +# new version and starts it; the new process reads the +# new generation at startup and runs on. +# +# No process is ever killed mid-transaction: exits happen only at the +# crawler's own safe commit points. +# +# CRAWLER_VERSION (CONFIG section) is the version of THIS crawler +# code. It is NOT the generation and must be increased with every +# released crawler change. + + +class UpdateRequired(Exception): + """ + Raised at a safe point (before a commit) when a new deployment + generation is active. safe_commit() rolls the open transaction + back BEFORE raising, so no work is ever committed half-done. + """ + + +class UpdateSignal: + """ + Per-process update state (every crawler worker process owns one). + + generation: the deployment generation this process started with. + pending: True once a check noticed a new generation; the + process stops at the next safe commit point. + stop_checker: tells the background checker thread to finish. + """ + + def __init__(self): + self.generation = None + self.pending = False + self.stop_checker = False + + def check(self): + """Raise UpdateRequired if an update is pending.""" + if self.pending: + raise UpdateRequired( + "new deployment generation active; update required" + ) + + +update_signal = UpdateSignal() + + +def ensure_deployment_state_table(conn): + """ + Idempotent bootstrap for deployment_state (fresh databases): + creates the table if needed and makes sure exactly one row with + id = 1 exists. Safe to run from several crawler instances at once + (CREATE TABLE IF NOT EXISTS + ON CONFLICT DO NOTHING). + """ + with conn.cursor() as cur: + cur.execute(""" + CREATE TABLE IF NOT EXISTS deployment_state ( + id INTEGER PRIMARY KEY, + generation BIGINT NOT NULL, + required_version INTEGER NOT NULL + ) + """) + cur.execute(""" + INSERT INTO deployment_state (id, generation, required_version) + VALUES (1, 1, %s) + ON CONFLICT (id) DO NOTHING + """, (CRAWLER_VERSION,)) + conn.commit() + + +def _fetch_generation_row(conn): + """ + Read the deployment_state row (id = 1). If the table does not exist + yet, it is bootstrapped (see ensure_deployment_state_table) and the + row is read again. Returns the row or None (table without row). + """ + try: + with conn.cursor() as cur: + cur.execute( + "SELECT generation FROM deployment_state WHERE id = 1" + ) + row = cur.fetchone() + conn.rollback() # read-only; leave no open transaction behind + return row + except psycopg2.errors.UndefinedTable: + conn.rollback() + logging.warning( + "deployment_state table missing; creating it with " + "generation 1 and required_version %d.", + CRAWLER_VERSION, + ) + ensure_deployment_state_table(conn) + with conn.cursor() as cur: + cur.execute( + "SELECT generation FROM deployment_state WHERE id = 1" + ) + row = cur.fetchone() + conn.rollback() + return row + + +def get_current_generation(conn=None): + """ + Read the current deployment generation. Called once per process at + startup: this becomes "my generation" (update_signal.generation). + """ + own_conn = conn is None + if own_conn: + conn = db_connect() + try: + row = _fetch_generation_row(conn) + if row is None: + logging.warning( + "deployment_state has no row with id = 1; inserting " + "generation 1 with required_version %d.", + CRAWLER_VERSION, + ) + ensure_deployment_state_table(conn) + row = _fetch_generation_row(conn) + return int(row[0]) + finally: + if own_conn: + conn.close() + + +def periodic_update_check(): + """ + Regular update check OUTSIDE any transaction, so instances that + currently do not commit still notice a new generation. + + Runs as a daemon thread per worker and uses its own short-lived + connection, so it can never interfere with the worker's open + transaction. A new generation only sets update_signal.pending; + the actual exit happens at the next safe commit point. A failed + check (e.g. short DB hiccup) is simply retried next interval. + """ + while not ( + STOP.is_set() + or update_signal.pending + or update_signal.stop_checker + ): + time.sleep(UPDATE_CHECK_INTERVAL_SECONDS) + if STOP.is_set() or update_signal.pending or update_signal.stop_checker: + break + + try: + conn = db_connect() + except Exception: + continue + + try: + with conn.cursor() as cur: + cur.execute( + "SELECT generation FROM deployment_state WHERE id = 1" + ) + row = cur.fetchone() + conn.rollback() + except Exception: + try: + conn.rollback() + except Exception: + pass + continue + finally: + try: + conn.close() + except Exception: + pass + + if row is None: + continue + + try: + current = int(row[0]) + except (TypeError, ValueError): + continue + + if ( + not update_signal.pending + and update_signal.generation is not None + and current != update_signal.generation + ): + update_signal.pending = True + logging.warning( + "[UPDATE] Deployment generation changed: %d -> %d. " + "This worker (version %d) stops at the next safe " + "commit point.", + update_signal.generation, + current, + CRAWLER_VERSION, + ) + + +def safe_commit(conn, operation=None): + """ + The ONLY commit path for crawling work. + + 1. If an update is pending, roll the open transaction back and + raise UpdateRequired: the caller's writes are discarded + cleanly, nothing is committed half-done. + 2. Otherwise run the writes (operation) and commit them. + + Every crawler commit that does real work goes through here + (page saves, queue inserts, queue claim/finish, stats), so a + deployment switch can never cut a transaction in half. The + one-off startup commits (seeding, stats table) are routed + through this function as well. + """ + if update_signal.pending: + try: + conn.rollback() + except Exception: + logging.exception("Rollback before update exit failed.") + raise UpdateRequired( + "new deployment generation active; update required" + ) + + result = operation() if operation is not None else None + conn.commit() + return result + + # ============================================================ # LOGGING / SHUTDOWN # ============================================================ @@ -262,6 +579,8 @@ def db_connect(): def ensure_stats_table(conn): + """Create crawler_stats. Writes only - commit happens in + safe_commit().""" with conn.cursor() as cur: cur.execute(""" CREATE TABLE IF NOT EXISTS crawler_stats ( @@ -272,16 +591,16 @@ def ensure_stats_table(conn): failed INTEGER NOT NULL DEFAULT 0 ) """) - conn.commit() def write_stats(conn, crawled, queued, failed): + """Insert one stats row. Writes only - commit happens in + safe_commit().""" with conn.cursor() as cur: cur.execute(""" INSERT INTO crawler_stats (crawled, queued, failed) VALUES (%s, %s, %s) """, (crawled, queued, failed)) - conn.commit() def get_next_queue_item(conn): @@ -290,6 +609,8 @@ def get_next_queue_item(conn): SKIP LOCKED is the important part: several crawler processes can consume the same PostgreSQL queue without waiting for each other. + + Writes only - the commit happens in safe_commit(). """ with conn.cursor() as cur: cur.execute(""" @@ -310,23 +631,25 @@ def get_next_queue_item(conn): RETURNING q.id, q.url, q.source_url """) row = cur.fetchone() - conn.commit() return row def finish_queue_item(conn, queue_id, success=True): + """Mark a queue item done/failed. Writes only - commit happens in + safe_commit().""" with conn.cursor() as cur: cur.execute(""" UPDATE crawl_queue SET status = %s WHERE id = %s """, ("done" if success else "failed", queue_id)) - conn.commit() def requeue_stale_items(conn, older_than_seconds=3600): """ Recover URLs left in 'processing' after a crash. + + Writes only - commit happens in safe_commit(). """ with conn.cursor() as cur: cur.execute(""" @@ -336,7 +659,6 @@ def requeue_stale_items(conn, older_than_seconds=3600): AND last_attempt_at < now() - (%s * interval '1 second') """, (older_than_seconds,)) changed = cur.rowcount - conn.commit() if changed: logging.info("Requeued %d stale processing URLs.", changed) @@ -346,9 +668,10 @@ def _insert_queue_rows(conn, rows): Insert pre-ordered queue rows and return the number of newly inserted rows. - All pages run inside ONE transaction (commit only at the end), so - the batch stays atomic as before. rowcount is accumulated per page - because execute_values() does not aggregate rowcount across pages. + All pages run inside ONE transaction (committed by safe_commit() + after this function returns), so the batch stays atomic as before. + rowcount is accumulated per page because execute_values() does not + aggregate rowcount across pages. """ inserted = 0 with conn.cursor() as cur: @@ -367,7 +690,6 @@ def _insert_queue_rows(conn, rows): ) inserted += max(cur.rowcount, 0) - conn.commit() return inserted @@ -441,7 +763,9 @@ def queue_urls(conn, links, source_url): for attempt in range(1, QUEUE_INSERT_MAX_ATTEMPTS + 1): try: - inserted = _insert_queue_rows(conn, rows) + inserted = safe_commit( + conn, lambda: _insert_queue_rows(conn, rows) + ) if attempt > 1: logging.info( "[QUEUE] Insert succeeded on retry %d/%d.", @@ -1222,6 +1546,10 @@ def parse_html(html, base_url): # ============================================================ def save_page(conn, url, source_url, status, content_type, parsed, crawl_ms): + """ + Store/refresh one page. Writes only - commit happens in + safe_commit(). + """ # Sanitize everything that came from the web before it reaches # PostgreSQL. NUL bytes are legal in HTML but illegal in text columns. url = clean_text(url) @@ -1292,8 +1620,6 @@ def save_page(conn, url, source_url, status, content_type, parsed, crawl_ms): crawl_ms, )) - conn.commit() - # ============================================================ # WORKER @@ -1310,17 +1636,42 @@ def worker(worker_id): crawled = 0 failed = 0 last_stats = time.monotonic() + current_queue_id = None + update_requested = False try: conn = db_connect() + # Remember the generation this instance started with. Read ONCE + # per process: five crawler instances all remember their own + # generation (here: the same one, until a deployment bump). + update_signal.generation = get_current_generation(conn) + logging.info( - "Started. DB=%s:%s/%s", - DB_HOST, DB_PORT, DB_NAME + "Started. DB=%s:%s/%s version=%d generation=%d", + DB_HOST, DB_PORT, DB_NAME, + CRAWLER_VERSION, update_signal.generation, ) + # Regular update check OUTSIDE the transactions (own short-lived + # connection, daemon thread). It only sets update_signal.pending; + # the exit itself happens at the next safe commit point, never + # in the middle of a transaction. + checker_thread = threading.Thread( + target=periodic_update_check, + name=f"{process_name}-update-check", + daemon=True, + ) + checker_thread.start() + while not STOP.is_set(): - item = get_next_queue_item(conn) + # Safe point between page cycles: if the periodic check + # noticed a new generation, stop before starting new work. + if update_signal.pending: + update_requested = True + break + + item = safe_commit(conn, lambda: get_next_queue_item(conn)) if not item: logging.info("Queue empty. Worker waiting...") @@ -1328,6 +1679,7 @@ def worker(worker_id): continue queue_id, url, source_url = item + current_queue_id = queue_id logging.info("=" * 70) logging.info("[CRAWL] %s", url) @@ -1339,7 +1691,12 @@ def worker(worker_id): if error: logging.info("[HTTP ] %s -> %s", url, error) - finish_queue_item(conn, queue_id, success=False) + safe_commit( + conn, + lambda: finish_queue_item( + conn, queue_id, success=False + ), + ) failed += 1 continue @@ -1348,7 +1705,12 @@ def worker(worker_id): # A redirect normally arrives as the final URL in urllib. # The requested URL remains the canonical queue identity. if html is None: - finish_queue_item(conn, queue_id, success=False) + safe_commit( + conn, + lambda: finish_queue_item( + conn, queue_id, success=False + ), + ) failed += 1 continue @@ -1356,7 +1718,12 @@ def worker(worker_id): parsed = parse_html(html, url) except Exception as exc: logging.exception("[PARSE] %s: %s", url, exc) - finish_queue_item(conn, queue_id, success=False) + safe_commit( + conn, + lambda: finish_queue_item( + conn, queue_id, success=False + ), + ) failed += 1 continue @@ -1367,14 +1734,17 @@ def worker(worker_id): logging.info("[HEADS ] %d", len(parsed["headings"])) logging.info("[LINKS ] %d gefunden", len(parsed["links"])) - save_page( - conn=conn, - url=url, - source_url=source_url, - status=status, - content_type="text/html", - parsed=parsed, - crawl_ms=crawl_ms, + safe_commit( + conn, + lambda: save_page( + conn=conn, + url=url, + source_url=source_url, + status=status, + content_type="text/html", + parsed=parsed, + crawl_ms=crawl_ms, + ), ) inserted = queue_urls( @@ -1390,9 +1760,13 @@ def worker(worker_id): inserted, ) - finish_queue_item(conn, queue_id, success=True) + safe_commit( + conn, + lambda: finish_queue_item(conn, queue_id, success=True), + ) crawled += 1 + current_queue_id = None # Periodic stats; not on every page to avoid unnecessary DB load. if time.monotonic() - last_stats >= 30: @@ -1403,15 +1777,58 @@ def worker(worker_id): ) queued = cur.fetchone()[0] - write_stats(conn, crawled, queued, failed) + safe_commit( + conn, + lambda: write_stats(conn, crawled, queued, failed), + ) last_stats = time.monotonic() + except UpdateRequired: + raise except Exception: conn.rollback() + except UpdateRequired: + # A safe point (safe_commit) noticed the new generation and + # rolled back the open transaction BEFORE raising. Nothing was + # committed half-done. + update_requested = True + + # Best effort: give the claimed item back to the queue so the + # next version can crawl it immediately. (Should this fail, + # requeue_stale_items() recovers it later anyway.) This commit + # is wanted shutdown bookkeeping, not crawled content. + try: + if current_queue_id is not None: + with conn.cursor() as cur: + cur.execute( + "UPDATE crawl_queue " + "SET status = 'pending' " + "WHERE id = %s AND status = 'processing'", + (current_queue_id,), + ) + conn.commit() + logging.info( + "[UPDATE] Queue item %d requeued for the next version.", + current_queue_id, + ) + except Exception: + try: + conn.rollback() + except Exception: + pass + + logging.warning( + "[UPDATE] %s stops for update (started with generation %s, " + "version %d).", + process_name, + update_signal.generation, + CRAWLER_VERSION, + ) except Exception: logging.exception("Worker crashed.") raise finally: + update_signal.stop_checker = True if conn is not None: conn.close() @@ -1420,6 +1837,10 @@ def worker(worker_id): crawled, failed ) + if update_requested: + # Reserved exit code, exclusively for "update required". + sys.exit(EXIT_UPDATE_REQUIRED) + # ============================================================ # STARTUP @@ -1434,12 +1855,16 @@ def main(): print(f"Delay: {DEFAULT_DELAY:.1f} Sekunden") print(f"Robots-Cache: {ROBOTS_CACHE_SECONDS // 3600} Stunden") print(f"DB: {DB_HOST}:{DB_PORT}/{DB_NAME}") + print(f"Version: {CRAWLER_VERSION}") print() conn = db_connect() + ensure_deployment_state_table(conn) ensure_stats_table(conn) - requeue_stale_items(conn) + safe_commit(conn) + + safe_commit(conn, lambda: requeue_stale_items(conn)) # Seed URL. ON CONFLICT makes restarting harmless. queue_urls(conn, [normalize_url(START_URL)], None) @@ -1453,18 +1878,34 @@ def main(): p.start() workers.append(p) + update_requested = False + worker_crashed = False + try: while any(p.is_alive() for p in workers): time.sleep(1) - # If a worker dies unexpectedly, stop the remaining workers. for p in workers: - if p.exitcode not in (None, 0): + if p.exitcode == EXIT_UPDATE_REQUIRED: + # This worker reached its safe point and wants the + # update. Stop the remaining workers as well (they + # also exit at their own safe points, never + # mid-transaction) and propagate exit code 42. + if not update_requested: + logging.warning( + "[UPDATE] Worker %s requests update; the " + "other workers stop at their safe points.", + p.name, + ) + update_requested = True + STOP.set() + elif p.exitcode not in (None, 0): logging.error( "Worker %s exited with code %s.", p.name, p.exitcode, ) + worker_crashed = True STOP.set() except KeyboardInterrupt: @@ -1481,9 +1922,237 @@ def main(): p.terminate() p.join() + if update_requested: + # Reserved exit code: "update required". Only the launcher may + # interpret this code; normal errors use EXIT_ERROR (1). + print("Crawler stopped: update required.") + sys.exit(EXIT_UPDATE_REQUIRED) + + if worker_crashed: + # A normal error must NOT look like a normal end (exit 0) or an + # update (42): the launcher restarts on this code unchanged. + print("Crawler stopped: worker error.") + sys.exit(EXIT_ERROR) + print("Crawler stopped.") +# ============================================================ +# CLI (version / deployment / update checks) +# ============================================================ + +def build_arg_parser(): + parser = argparse.ArgumentParser( + prog="crawler.py", + description="JonasSearch PostgreSQL crawler with " + "generation-based rolling updates.", + ) + parser.add_argument( + "--version", + action="store_true", + help="print the crawler version (just the number) and exit", + ) + parser.add_argument( + "--self-update", + type=int, + metavar="VERSION", + help="verify a staged crawler release in " + "%s//crawler.py without activating it " + "(exit 0 = startable, 1 = broken/missing)" % CRAWLER_RELEASES_DIR, + ) + parser.add_argument( + "--update-deployment", + type=int, + metavar="VERSION", + help="bump the deployment generation and set required_version " + "to VERSION. Run ONLY after the new version has been " + "provisioned and verified - from this moment all running " + "crawlers stop at their next safe commit point.", + ) + parser.add_argument( + "--init-deployment", + action="store_true", + help="create deployment_state if needed and seed the initial " + "row (id=1, generation 1, required_version = this " + "crawler's version)", + ) + return parser + + +def cmd_version(): + # Deliberately machine-readable: the launcher parses this output. + print(CRAWLER_VERSION) + return EXIT_OK + + +def _basic_self_check(): + """ + Minimal smoke test used by --self-update when no releases directory + exists yet: this file must at least compile. + """ + try: + with open(__file__, "r", encoding="utf-8") as fh: + compile(fh.read(), __file__, "exec") + except Exception as exc: + logging.error("[UPDATE] Self-check failed: %s", exc) + return EXIT_ERROR + print("[UPDATE] Basic self-check passed.") + return EXIT_OK + + +def cmd_self_update(version): + """ + Verify that a staged release for `version` exists and is startable + - WITHOUT activating it. The launcher runs this with the CURRENT + (old) interpreter before swapping anything, so a broken release can + never become active (fallback guarantee). + + Checks: + 1. releases//crawler.py exists, + 2. `python3 releases//crawler.py --version` prints exactly + `version` and exits with EXIT_OK (imports, syntax and the + version constant of the new file are all checked). + """ + release_file = CRAWLER_RELEASES_DIR / str(version) / "crawler.py" + + if not CRAWLER_RELEASES_DIR.is_dir(): + logging.warning( + "[UPDATE] No releases directory (%s); running the basic " + "self-check only.", + CRAWLER_RELEASES_DIR, + ) + return _basic_self_check() + + if not release_file.is_file(): + logging.error( + "[UPDATE] Staged release missing: %s", release_file + ) + return EXIT_ERROR + + print(f"[UPDATE] Checking staged release: {release_file}") + try: + result = subprocess.run( + [sys.executable, str(release_file), "--version"], + capture_output=True, + text=True, + timeout=60, + ) + except Exception as exc: + logging.error("[UPDATE] Release check could not run: %s", exc) + return EXIT_ERROR + + printed = (result.stdout or "").strip() + if result.returncode != EXIT_OK or printed != str(version): + logging.error( + "[UPDATE] Staged release check FAILED (exit code %d, " + "version output %r, stderr: %s). The release is NOT " + "activated; the old version stays in place.", + result.returncode, + printed, + (result.stderr or "").strip()[-500:], + ) + return EXIT_ERROR + + print(f"[UPDATE] Release {version} verified and startable.") + return EXIT_OK + + +def cmd_update_deployment(required_version): + """ + The global switch point: atomically raise the deployment generation + and set the required crawler version in ONE statement. + + Run only AFTER the new version has been provisioned and verified on + the machines (see --self-update). From this moment every running + old crawler sees the new generation at its next safe commit point, + rolls its open transaction back and exits with EXIT_UPDATE_REQUIRED + - no process is ever killed mid-transaction. + """ + conn = None + try: + conn = db_connect() + ensure_deployment_state_table(conn) + with conn.cursor() as cur: + cur.execute( + """ + UPDATE deployment_state + SET generation = generation + 1, + required_version = %s + WHERE id = 1 + RETURNING generation, required_version + """, + (required_version,), + ) + row = cur.fetchone() + if row is None: + logging.error( + "deployment_state row (id = 1) missing; nothing updated." + ) + conn.rollback() + return EXIT_ERROR + conn.commit() + generation, activated_version = row + print( + f"[UPDATE] Deployment generation -> {generation}, " + f"required_version -> {activated_version}. Running crawlers " + "stop at their next safe commit point with exit code " + f"{EXIT_UPDATE_REQUIRED}." + ) + return EXIT_OK + except Exception: + logging.exception("Deployment update failed.") + if conn is not None: + try: + conn.rollback() + except Exception: + pass + return EXIT_ERROR + finally: + if conn is not None: + conn.close() + + +def cmd_init_deployment(): + """ + One-time initialization of deployment_state: create the table and + the single current-state row (id = 1) with generation 1 and the + currently running crawler version. Restarting is harmless. + """ + conn = None + try: + conn = db_connect() + ensure_deployment_state_table(conn) + row = _fetch_generation_row(conn) + generation = int(row[0]) if row else 1 + print( + f"[UPDATE] deployment_state ready: generation {generation}, " + f"required_version {CRAWLER_VERSION}." + ) + return EXIT_OK + except Exception: + logging.exception("Initializing deployment_state failed.") + return EXIT_ERROR + finally: + if conn is not None: + conn.close() + + +def main_cli(argv=None): + args = build_arg_parser().parse_args(argv) + + if args.version: + return cmd_version() + if args.self_update is not None: + return cmd_self_update(args.self_update) + if args.update_deployment is not None: + return cmd_update_deployment(args.update_deployment) + if args.init_deployment: + return cmd_init_deployment() + + main() + return EXIT_OK + + if __name__ == "__main__": mp.set_start_method("spawn", force=True) - main() + sys.exit(main_cli()) diff --git a/launcher.py b/launcher.py new file mode 100644 index 0000000..c0f86d9 --- /dev/null +++ b/launcher.py @@ -0,0 +1,620 @@ +#!/usr/bin/env python3 +""" +JonasSearch Crawler Launcher + +Starts and monitors the crawler and handles the generation-based +rolling update mechanism (see crawler.py, section +"DEPLOYMENT / UPDATE"). + +Division of labor: + + crawler.py (old version) + - remembers the deployment generation it started with + - checks it before every relevant commit (safe point) and + periodically in the background + - when a new generation appears it rolls the open transaction + back and exits with exit code 42 (EXIT_UPDATE_REQUIRED) + + launcher.py (this file) + - starts the crawler and waits for it to end + - interprets the exit code: + 0 -> normal end (e.g. Ctrl+C): nothing to do + 42 -> update required: provision the new version, + bump the generation, start the new version + else-> error/crash: restart the crawler unchanged + - update flow (in this order, so a broken update can never + take the crawler down): + 1. stage the new version completely on this machine + (releases//crawler.py) + 2. verify it startable: the OLD crawler verifies the + staged release via `--self-update VERSION` + 3. only then raise generation + required_version in + PostgreSQL (--update-deployment VERSION) + 4. restart the crawler (now the new version), which + reads the new generation at startup and runs on + +Exit codes of the launcher itself mirror the crawler's: 0 normal end, +42 update cycle completed, 1 configuration/provisioning error. + +Usage: + python3 launcher.py # start crawler (loop until Ctrl+C) + python3 launcher.py --once # run one crawler cycle, then exit + python3 launcher.py --check-update # only trigger the update flow + python3 launcher.py --status # show crawler + deployment state +""" + +import os +import sys +import time +import shutil +import signal +import logging +import argparse +import subprocess +from pathlib import Path +from urllib.request import Request, urlopen +from urllib.error import URLError + +from dotenv import load_dotenv + +# .env loading mirrors crawler.py: working directory first, then the +# launcher's own directory, so relative paths work from anywhere. An +# unreadable .env must not crash the launcher. +for _dotenv_path in (None, Path(__file__).resolve().parent / ".env"): + try: + load_dotenv(_dotenv_path) + except OSError as exc: + logging.warning( + "[CONFIG] %s not readable (%s); continuing with the " + "current environment.", + _dotenv_path or ".env (working directory)", + exc, + ) + +# Configure logging BEFORE importing crawler.py - importing the module +# would otherwise install its own logging.basicConfig first. +logging.basicConfig( + level=logging.INFO, + format="%(asctime)s [launcher] %(levelname)s %(message)s", + datefmt="%H:%M:%S", +) +log = logging.getLogger("launcher") + +# Make the crawler's CONFIG available (DB settings, CRAWLER_VERSION, +# exit codes, CRAWLER_RELEASES_DIR). crawler.py is import-safe: all +# side effects live under `if __name__ == "__main__":`. +sys.path.insert(0, str(Path(__file__).resolve().parent)) +import crawler as crawler_mod # noqa: E402 + +# crawler.py registers SIGINT/SIGTERM handlers at import time (for its +# spawned workers). The launcher needs the default behavior back so +# Ctrl+C raises KeyboardInterrupt here; the crawler child processes +# spawn fresh and get their own handlers. +signal.signal(signal.SIGINT, signal.default_int_handler) +signal.signal(signal.SIGTERM, signal.default_int_handler) + +# ============================================================ +# CONFIG +# ============================================================ + +# The crawler file this launcher starts (active/old version). +CRAWLER_ENTRY = Path( + os.getenv( + "JONASSEARCH_LAUNCHER_ENTRY", + str(Path(__file__).resolve().parent / "crawler.py"), + ) +) + +# Where new versions are staged before activation. +# crawler.py resolves relative JONASSEARCH_RELEASES_DIR paths against +# the crawler file's directory - mirror that here so both agree. +RELEASES_DIR = crawler_mod.CRAWLER_RELEASES_DIR + +# Index of the currently active version, kept next to the crawler. +ACTIVE_MARKER = CRAWLER_ENTRY.parent / ".active-crawler-version" + +# How new versions arrive on this machine (step 16 of the concept - +# deliberately kept simple and pluggable). Supported sources: +# - JONASSEARCH_UPDATE_SOURCE_URL: HTTP(S) download of crawler.py +# (the file is downloaded fresh for every update, so no local +# cache is needed) +# - JONASSEARCH_UPDATE_SOURCE_DIR: local directory containing +# crawler.py (e.g. a shared folder or a git checkout) +# Without a source the launcher can only verify and activate a release +# that was placed into releases//crawler.py externally. +UPDATE_SOURCE_URL = os.getenv("JONASSEARCH_UPDATE_SOURCE_URL", "").strip() +UPDATE_SOURCE_DIR = os.getenv("JONASSEARCH_UPDATE_SOURCE_DIR", "").strip() + +# Restart behavior after a crash (not an update!). +RESTART_DELAY_ON_ERROR = float( + os.getenv("JONASSEARCH_LAUNCHER_ERROR_RESTART_DELAY", "10") +) + + +# ============================================================ +# HELPERS +# ============================================================ + +def run_crawler_command(entry, *args, timeout=None): + """Run a crawler.py subcommand and return the CompletedProcess.""" + return subprocess.run( + [sys.executable, str(entry), *args], + capture_output=True, + text=True, + timeout=timeout, + ) + + +def read_active_version(): + """Version number of the currently active crawler (marker file).""" + try: + return int(ACTIVE_MARKER.read_text(encoding="utf-8").strip()) + except Exception: + return None + + +def write_active_version(version): + ACTIVE_MARKER.write_text(f"{version}\n", encoding="utf-8") + + +def read_installed_version(entry): + """Ask the crawler itself which version it is (source of truth).""" + try: + result = run_crawler_command(entry, "--version", timeout=30) + except Exception as exc: + log.error("Could not query crawler version (%s): %s", entry, exc) + return None + if result.returncode != crawler_mod.EXIT_OK: + log.error( + "crawler.py --version failed (exit %d): %s", + result.returncode, + (result.stderr or "").strip()[-300:], + ) + return None + try: + return int((result.stdout or "").strip()) + except ValueError: + log.error( + "crawler.py --version printed no version number: %r", + (result.stdout or "")[:100], + ) + return None + + +def fetch_remote_version(): + """Ask the update source which version it offers (may be None).""" + if UPDATE_SOURCE_URL: + try: + req = Request(UPDATE_SOURCE_URL, headers={"User-Agent": "launcher"}) + with urlopen(req, timeout=crawler_mod.REQUEST_TIMEOUT) as resp: + text = resp.read(1024 * 1024).decode("utf-8", errors="replace") + return int(text.strip()) + except (URLError, ValueError) as exc: + log.error( + "Could not query update source %s: %s", UPDATE_SOURCE_URL, exc + ) + return None + if UPDATE_SOURCE_DIR: + version_file = Path(UPDATE_SOURCE_DIR) / "VERSION" + try: + return int(version_file.read_text(encoding="utf-8").strip()) + except Exception as exc: + log.error( + "Could not read VERSION in update source %s: %s", + version_file, + exc, + ) + return None + return None + + +# ============================================================ +# UPDATE FLOW (steps 8, 9 and 15 of the concept) +# ============================================================ + +def stage_release(version): + """ + Step 1: make the new version fully available on this machine, + WITHOUT activating it: releases//crawler.py. + + Already staged releases are kept (idempotent). Without a configured + update source the release must already exist here. + """ + target_dir = RELEASES_DIR / str(version) + target_file = target_dir / "crawler.py" + + if target_file.is_file(): + log.info("Release %d already staged at %s.", version, target_file) + return target_file + + if UPDATE_SOURCE_URL: + base = UPDATE_SOURCE_URL.rsplit("/", 1)[0] + file_url = f"{base}/{version}/crawler.py" + log.info("Downloading release %d from %s ...", version, file_url) + req = Request(file_url, headers={"User-Agent": "launcher"}) + with urlopen(req, timeout=crawler_mod.REQUEST_TIMEOUT) as resp: + data = resp.read(64 * 1024 * 1024) + target_dir.mkdir(parents=True, exist_ok=True) + tmp_file = target_dir / "crawler.py.part" + tmp_file.write_bytes(data) + # Only complete files get their final name. + tmp_file.replace(target_file) + log.info("Release %d staged at %s.", version, target_file) + return target_file + + if UPDATE_SOURCE_DIR: + source_file = Path(UPDATE_SOURCE_DIR) / "crawler.py" + if not source_file.is_file(): + log.error("crawler.py missing in update source %s.", source_file) + return None + target_dir.mkdir(parents=True, exist_ok=True) + shutil.copy2(source_file, target_file) + log.info("Release %d staged at %s.", version, target_file) + return target_file + + log.error( + "Release %d is not staged (%s) and no update source is " + "configured (JONASSEARCH_UPDATE_SOURCE_URL or " + "JONASSEARCH_UPDATE_SOURCE_DIR). The generation was NOT " + "changed.", + version, + target_file, + ) + return None + + +def verify_release(version): + """ + Step 2: verify the staged release is complete and startable - + executed by the CURRENT (old) crawler, before anything is switched. + If this fails, the old version stays active (fallback guarantee). + """ + result = run_crawler_command( + CRAWLER_ENTRY, "--self-update", str(version), timeout=120 + ) + if result.returncode != crawler_mod.EXIT_OK: + log.error( + "Release %d FAILED the verification; nothing was switched.\n%s", + version, + (result.stdout or "") + (result.stderr or ""), + ) + return False + return True + + +def activate_release(version, staged_file): + """ + Step 3: switch the active crawler to the verified release. + + The old file is kept as crawler.py.bak- so there is always + a working fallback. The marker file records the active version; + the database switch itself happens in bump_deployment(). + """ + if staged_file.resolve() == CRAWLER_ENTRY.resolve(): + write_active_version(version) + return True + + backup = CRAWLER_ENTRY.with_name( + "crawler.py.bak-{}".format( + read_active_version() + or read_installed_version(CRAWLER_ENTRY) + or "unknown" + ) + ) + try: + shutil.copy2(CRAWLER_ENTRY, backup) + shutil.copy2(staged_file, CRAWLER_ENTRY) + except OSError as exc: + log.error( + "Switching to release %d failed (%s); the old version " + "stays active.", + version, + exc, + ) + return False + + # Sanity check: the switched file must identify itself correctly. + installed = read_installed_version(CRAWLER_ENTRY) + if installed != version: + log.error( + "Switched crawler reports version %s instead of %d; " + "restoring the backup.", + installed, + version, + ) + shutil.copy2(backup, CRAWLER_ENTRY) + return False + + write_active_version(version) + log.info( + "Release %d activated (backup of the old version: %s).", + version, + backup, + ) + return True + + +def bump_deployment(version): + """ + Step 4 - the global switch point: generation + 1 and + required_version = version in PostgreSQL. Only AFTER this moment do + running old crawlers notice the update (at their next safe commit + point) and exit with code 42. + """ + result = run_crawler_command( + CRAWLER_ENTRY, "--update-deployment", str(version), timeout=60 + ) + if result.returncode != crawler_mod.EXIT_OK: + log.error( + "Deployment update to version %d failed:\n%s", + version, + (result.stderr or "")[-500:], + ) + return False + log.info((result.stdout or "").strip()) + return True + + +def perform_update(required_version): + """ + Full update flow in the safe order (concept steps 8, 9 and 15): + + stage -> verify -> activate -> bump generation + + Returns True only if every step succeeded. + """ + log.info("Update to crawler version %d starting ...", required_version) + + staged = stage_release(required_version) + if staged is None: + return False + + if not verify_release(required_version): + return False + + if not activate_release(required_version, staged): + return False + + if not bump_deployment(required_version): + return False + + log.info("Update to crawler version %d completed.", required_version) + return True + + +# ============================================================ +# LAUNCH LOOP +# ============================================================ + +def start_crawler_process(): + log.info( + "Starting crawler: %s (version %s)", + CRAWLER_ENTRY, + read_installed_version(CRAWLER_ENTRY), + ) + # The crawler inherits stdout/stderr, so its logs stay visible. + return subprocess.Popen([sys.executable, str(CRAWLER_ENTRY)]) + + +def wait_for_process(process): + """Block until the process ends; return its exit code.""" + try: + return process.wait() + except KeyboardInterrupt: + log.info("Ctrl+C received; stopping crawler ...") + process.terminate() + try: + return process.wait(timeout=10) + except subprocess.TimeoutExpired: + process.kill() + return process.wait() + + +def read_required_version(): + """ + Read required_version from deployment_state (id = 1). Returns None + on database errors or a missing row. + """ + try: + conn = crawler_mod.db_connect() + try: + with conn.cursor() as cur: + cur.execute( + "SELECT required_version FROM deployment_state " + "WHERE id = 1" + ) + row = cur.fetchone() + conn.rollback() + finally: + conn.close() + except Exception as exc: + log.error("Deployment state could not be read: %s", exc) + return None + return int(row[0]) if row else None + + +def check_and_update(): + """ + Compare installed version with the required version from + PostgreSQL and run the update flow when necessary. + """ + try: + generation = crawler_mod.get_current_generation() + except Exception as exc: + log.error("Deployment state could not be read: %s", exc) + return False + + row_required = read_required_version() + if row_required is None: + log.error("deployment_state has no row with id = 1.") + return False + + required_version = row_required + installed = read_installed_version(CRAWLER_ENTRY) + remote = fetch_remote_version() + + log.info( + "Deployment: generation %d | required_version %d | " + "installed %s | update source offers %s", + generation, + required_version, + installed, + remote, + ) + + if installed is None: + return False + + if installed != required_version: + # The crawler would exit with 42 immediately anyway - provide + # the new version first so the restart is seamless. + return perform_update(required_version) + + if remote is not None and remote > installed: + # New version announced but not yet deployed: stage it now and + # only then flip the generation (concept step 16). + return perform_update(remote) + + log.info("Crawler is up to date.") + return True + + +def main(): + parser = argparse.ArgumentParser( + prog="launcher.py", + description="Starts and monitors the JonasSearch crawler and " + "handles generation-based rolling updates.", + ) + parser.add_argument( + "--once", + action="store_true", + help="run one crawler cycle and exit (no restart loop)", + ) + parser.add_argument( + "--check-update", + action="store_true", + help="check for a new version, run the update flow if needed " + "and exit", + ) + parser.add_argument( + "--status", + action="store_true", + help="show deployment state, installed version and exit", + ) + args = parser.parse_args() + + if args.status: + try: + generation = crawler_mod.get_current_generation() + except Exception as exc: + log.error("Deployment state could not be read: %s", exc) + return crawler_mod.EXIT_ERROR + installed = read_installed_version(CRAWLER_ENTRY) + required_version = read_required_version() + print( + f"generation: {generation}\n" + f"required_version: {required_version}\n" + f"installed: {installed}\n" + f"crawler file: {CRAWLER_ENTRY}\n" + f"releases dir: {RELEASES_DIR}" + ) + return ( + crawler_mod.EXIT_OK + if required_version is not None + else crawler_mod.EXIT_ERROR + ) + + if args.check_update: + return crawler_mod.EXIT_OK if check_and_update() else crawler_mod.EXIT_ERROR + + log.info( + "Launcher started. Crawler: %s | releases: %s", + CRAWLER_ENTRY, + RELEASES_DIR, + ) + + update_cycle_ran = False + + while True: + process = start_crawler_process() + exit_code = wait_for_process(process) + + if exit_code == crawler_mod.EXIT_UPDATE_REQUIRED: + # The crawler reached a safe point itself and wants the + # update. New version first, then the generation bump, then + # the restart (see perform_update()). + log.warning( + "Crawler exited with code %d: update required.", + crawler_mod.EXIT_UPDATE_REQUIRED, + ) + + required_version = read_required_version() + + if required_version is None: + log.error( + "Cannot determine the required version; starting the " + "existing crawler again." + ) + time.sleep(RESTART_DELAY_ON_ERROR) + continue + + installed = read_installed_version(CRAWLER_ENTRY) + if installed == required_version: + # Rare case: the version was already switched externally + # (e.g. by --check-update) - nothing to provision. + log.info("Version %d already active.", required_version) + elif not perform_update(required_version): + log.error( + "Update failed. Starting the existing version again " + "after %ds (it will exit with 42 at the next safe " + "point and the update will be retried).", + RESTART_DELAY_ON_ERROR, + ) + time.sleep(RESTART_DELAY_ON_ERROR) + continue + + update_cycle_ran = True + + if args.once: + # One cycle only: report the update to the caller + # (exit code 42) instead of restarting ourselves. + break + + try: + current_generation = crawler_mod.get_current_generation() + except Exception: + current_generation = "?" + log.info( + "Restarting the crawler (it reads generation %s at " + "startup) ...", + current_generation, + ) + time.sleep(crawler_mod.LAUNCHER_RESTART_DELAY) + continue + + if exit_code == crawler_mod.EXIT_OK: + # Normal end (e.g. Ctrl+C on the crawler): do not restart. + log.info("Crawler ended normally (exit 0).") + break + + # Anything else is an error/crash - NOT an update. + log.error( + "Crawler crashed (exit code %s). Restart in %ds ...", + exit_code, + RESTART_DELAY_ON_ERROR, + ) + time.sleep(RESTART_DELAY_ON_ERROR) + + if args.once: + break + + log.info("Launcher stopped.") + return ( + crawler_mod.EXIT_UPDATE_REQUIRED + if update_cycle_ran + else crawler_mod.EXIT_OK + ) + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/web/app.py b/web/app.py index f993a11..8fec781 100644 --- a/web/app.py +++ b/web/app.py @@ -7,14 +7,16 @@ import threading from datetime import datetime from pathlib import Path -from flask import Flask, jsonify, render_template_string +from flask import Flask, jsonify, render_template_string, Response DB_DATA = Path(__file__).parent / "page_counts.json" INTERVAL = 15 * 60 # 15 Minuten app = Flask(__name__) -HTML = """ +# r""" raw string: preserves \n etc. for the embedded JavaScript +# (a normal string would eat the backslashes and break the JS) +HTML = r""" @@ -38,6 +40,16 @@ HTML = """ --warn: #f5a623; } + html[data-theme="light"] { + --bg: #f4f6fa; + --card: #ffffff; + --border: #d8dde5; + --text: #1a1d23; + --muted: #6b7280; + --accent: #2f6fe4; + --accent2: #8a4fe0; + } + * { box-sizing: border-box; } body { @@ -49,12 +61,38 @@ HTML = """ .container { max-width: 1200px; margin: auto; padding: 30px 24px 60px; } - header { display: flex; flex-wrap: wrap; align-items: baseline; gap: 8px 16px; } + header { + display: flex; + flex-wrap: wrap; + align-items: baseline; + gap: 8px 16px; + position: sticky; + top: 0; + z-index: 10; + background: var(--bg); + padding: 10px 0 14px; + border-bottom: 1px solid var(--border); + } h1 { margin: 0; font-size: 1.9rem; letter-spacing: 0.3px; } .subtitle { color: var(--muted); } + .header-right { margin-left: auto; display: flex; gap: 8px; align-items: center; } + + .btn { + background: var(--card); + border: 1px solid var(--border); + color: var(--text); + font-size: 0.8rem; + padding: 6px 12px; + border-radius: 999px; + cursor: pointer; + text-decoration: none; + } + + .btn:hover { border-color: var(--accent); } + .current { font-size: 3.2rem; font-weight: 700; @@ -71,7 +109,7 @@ HTML = """ .current-meta { color: var(--muted); margin-bottom: 24px; } - .badges { display: flex; flex-wrap: wrap; gap: 8px; margin-bottom: 22px; } + .badges { display: flex; flex-wrap: wrap; gap: 8px; margin-bottom: 14px; } .badge { display: inline-flex; @@ -96,6 +134,24 @@ HTML = """ 50% { opacity: 0.35; } } + .ranges { display: flex; gap: 8px; margin-bottom: 18px; } + + .range-btn { + background: var(--card); + border: 1px solid var(--border); + color: var(--muted); + font-size: 0.82rem; + padding: 6px 14px; + border-radius: 999px; + cursor: pointer; + } + + .range-btn.active { + color: #fff; + background: var(--accent); + border-color: var(--accent); + } + .grid { display: grid; grid-template-columns: repeat(auto-fit, minmax(165px, 1fr)); @@ -134,6 +190,14 @@ HTML = """ margin-bottom: 22px; } + .chart-row { display: grid; grid-template-columns: repeat(auto-fit, minmax(420px, 1fr)); gap: 22px; } + .chart-row .chart-container { margin-bottom: 0; } + + @media (max-width: 950px) { + .chart-row { grid-template-columns: 1fr; } + .chart-row .chart-container { margin-bottom: 22px; } + } + .chart-box { position: relative; height: 340px; } .chart-box.small { height: 260px; } @@ -141,8 +205,42 @@ HTML = """ .chart-sub { color: var(--muted); font-size: 0.82rem; margin-bottom: 14px; } + .heatmap { overflow-x: auto; } + + .heatmap table { border-collapse: collapse; width: 100%; min-width: 720px; } + + .heatmap td, .heatmap th { + font-size: 0.68rem; + color: var(--muted); + padding: 2px; + text-align: center; + } + + .heatmap td.day { text-align: right; padding-right: 8px; white-space: nowrap; } + + .heatmap .cell { + width: 100%; + min-width: 22px; + height: 18px; + border-radius: 4px; + background: rgba(128,128,128,0.12); + } + .footer { color: var(--muted); font-size: 0.8rem; margin-top: 8px; } + table.records { width: 100%; border-collapse: collapse; font-size: 0.85rem; } + + table.records th, table.records td { + text-align: left; + padding: 7px 8px; + border-bottom: 1px solid var(--border); + } + + table.records th { color: var(--muted); font-weight: 500; font-size: 0.75rem; text-transform: uppercase; letter-spacing: 0.6px; } + table.records td.num { text-align: right; font-variant-numeric: tabular-nums; } + + .section-title { font-weight: 600; margin: 30px 0 12px; font-size: 1.05rem; } + @media (max-width: 600px) { .current { font-size: 2.4rem; } .container { padding: 20px 14px 40px; } @@ -158,6 +256,11 @@ HTML = """

JonasSearch

Indexed Pages – Live Statistics
+
+ Export CSV + Export JSON + +
@@ -171,6 +274,13 @@ HTML = """ First measurement: – Measurements: – Last update: – + Next refresh: – +
+ +
+ + +
@@ -185,20 +295,40 @@ HTML = """
Average over full history
-
Last 5 min
-
–
-
Latest interval change
-
-
-
Last hour
-
–
-
vs. measurement ~1 h ago
-
-
-
Last 24 h
+
Growth 24 h
–
vs. measurement ~24 h ago
+
+
Growth 7 days
+
–
+
vs. measurement ~7 d ago
+
+
+
Avg / h (24 h)
+
–
+
Mean of last 24 h changes
+
+
+
Best hour
+
–
+
Largest gain within one hour
+
+
+
Worst hour
+
–
+
Largest loss within one hour
+
+
+
Peak
+
–
+
Highest measurement ever
+
+
+
Lowest
+
–
+
Lowest measurement ever
+
Forecast next hour
–
@@ -208,16 +338,47 @@ HTML = """
Indexed pages over time
-
X-axis shows the real measurement timestamps · dashed line = linear-regression trend & 24 h forecast
+
Real measurement timestamps · dashed line = linear-regression trend & 24 h forecast
Change per 5 minutes
-
Computed from the raw measurements: each change is normalized to a 5-minute interval
+
Normalized to a 5-minute interval · grey = raw · colored = smoothed (moving average over 5 measurements)
+
+
Cumulative net growth
+
Total pages gained/lost since the start of the selected range · linear time axis
+
+
+ +
+
Growth per hour (last 48 h)
+
Sum of changes per hour bucket · green = growth, red = loss
+
+
+ +
+
+
Growth by hour of day
+
Average gain per 5 min, grouped by hour of day (whole history) · shows when the crawler is most active
+
+
+
+
Records – biggest changes
+
Largest gains and losses between two consecutive measurements (selected range)
+
+
+
+ +
Activity heatmap – gain per hour of day
+
+
Rows = days · columns = hours (0–23) · green = growth, red = loss, grey = no data
+
+
+
@@ -228,15 +389,26 @@ if (window.Chart && Chart.registerables) { Chart.register.apply(Chart, Chart.registerables); } -Chart.defaults.color = '#9aa0a6'; -Chart.defaults.borderColor = '#2c313a'; -Chart.defaults.font.family = "-apple-system, BlinkMacSystemFont, 'Segoe UI', Roboto, Arial, sans-serif"; - const ACCENT = '#4f8cff'; const ACCENT2 = '#b47aff'; const UP = '#2ecc71'; const DOWN = '#ff6b6b'; const FALLBACK_INTERVAL_MS = 15 * 60 * 1000; +const DELTA_SMOOTH_WINDOW = 5; +const REFRESH_S = 30; + +let rawValues = []; +let intervalMs = FALLBACK_INTERVAL_MS; +let range = 'all'; +let charts = {}; +let refreshLeft = REFRESH_S; + +function cssVar(name) { + return getComputedStyle(document.documentElement).getPropertyValue(name).trim(); +} + +function borderColor() { return cssVar('--border') || '#2c313a'; } +function mutedColor() { return cssVar('--muted') || '#9aa0a6'; } function fmt(n) { return Math.round(n).toLocaleString('de-DE'); } @@ -255,6 +427,10 @@ function fmtDateFull(ms) { return new Date(ms).toLocaleString('de-DE'); } +function fmtDay(ms) { + return new Date(ms).toLocaleDateString('de-DE', { day: '2-digit', month: '2-digit' }); +} + function fmtDur(ms) { const min = Math.round(ms / 60000); if (min < 1) return '<1 min'; @@ -295,6 +471,17 @@ function regress(points) { }; } +function movingAverage(values, window) { + const half = Math.floor(window / 2); + return values.map(function (_, i) { + let sum = 0, count = 0; + for (let j = i - half; j <= i + half; j++) { + if (j >= 0 && j < values.length) { sum += values[j]; count++; } + } + return sum / count; + }); +} + function median(arr) { if (!arr.length) return null; const s = arr.slice().sort(function (a, b) { return a - b; }); @@ -326,34 +513,174 @@ function setDeltaCard(id, idHint, delta, spanMs, per5) { hint.textContent = fmtSigned(spanMs > 0 ? delta * (3600000 / spanMs) : 0) + ' Pages/h · ' + fmtSigned(per5) + ' per 5 min over ' + fmtDur(spanMs); } -let mainChart = null; -let deltaChart = null; -let intervalMs = FALLBACK_INTERVAL_MS; - -function gradientFill(context) { - const chart = context.chart; - const ctx = chart.ctx; - const area = chart.chartArea; - if (!area) return 'rgba(79,140,255,0.15)'; - const g = ctx.createLinearGradient(0, area.top, 0, area.bottom); - g.addColorStop(0, 'rgba(79,140,255,0.28)'); - g.addColorStop(1, 'rgba(79,140,255,0.02)'); - return g; +function setSigned(id, value) { + const el = document.getElementById(id); + el.textContent = fmtSigned(value); + el.classList.remove('pos', 'neg'); + el.classList.add(value >= 0 ? 'pos' : 'neg'); } -const xAxisConfig = { - type: 'linear', - ticks: { - maxTicksLimit: 8, - maxRotation: 0, - autoSkip: true, - callback: function (v) { return fmtDate(v); } - }, - grid: { color: 'rgba(255,255,255,0.05)' } -}; +/* ---------- data helpers ---------- */ -function buildCharts(points, trendPoints, deltas) { - mainChart = new Chart(document.getElementById('mainChart'), { +function parsePoints(values) { + return (values || []) + .map(function (v) { return { x: new Date(v.timestamp).getTime(), y: v.count }; }) + .filter(function (p) { return !isNaN(p.x); }) + .sort(function (a, b) { return a.x - b.x; }); +} + +function filterRange(points) { + if (range === 'all' || !points.length) return points; + const last = points[points.length - 1].x; + const span = range === '24h' ? 24 * 3600000 : 7 * 24 * 3600000; + return points.filter(function (p) { return p.x >= last - span; }); +} + +function computeDeltas(points) { + const dts = []; + for (let i = 1; i < points.length; i++) dts.push(points[i].x - points[i - 1].x); + const iv = median(dts) || FALLBACK_INTERVAL_MS; + const deltas = []; + for (let i = 1; i < points.length; i++) { + const dt = points[i].x - points[i - 1].x; + const diff = points[i].y - points[i - 1].y; + if (dt <= 0 || dt > 4 * iv) continue; + const per5 = dt < 60000 ? diff : diff * (5 * 60000 / dt); + deltas.push({ x: points[i].x, y: per5, dt: dt, diff: diff }); + } + return { deltas: deltas, interval: iv }; +} + +/* ---------- status ---------- */ + +function setStatus(kind, text) { + const dot = document.getElementById('status-dot'); + dot.className = 'dot ' + kind + (kind === 'ok' ? ' live' : ''); + document.getElementById('status-text').textContent = text; +} + +/* ---------- theme ---------- */ + +function applyTheme(theme) { + document.documentElement.dataset.theme = theme; + document.getElementById('btn-theme').textContent = theme === 'light' ? '🌙 Dark' : '☀️ Light'; + try { localStorage.setItem('js-theme', theme); } catch (e) {} + if (rawValues.length) renderAll(); // re-render charts with new palette +} + +document.getElementById('btn-theme').addEventListener('click', function () { + applyTheme(document.documentElement.dataset.theme === 'light' ? 'dark' : 'light'); +}); + +(function () { + let saved = null; + try { saved = localStorage.getItem('js-theme'); } catch (e) {} + if (saved) document.documentElement.dataset.theme = saved; + document.getElementById('btn-theme').textContent = saved === 'light' ? '🌙 Dark' : '☀️ Light'; +})(); + +/* ---------- range buttons ---------- */ + +document.querySelectorAll('.range-btn').forEach(function (btn) { + btn.addEventListener('click', function () { + document.querySelectorAll('.range-btn').forEach(function (b) { b.classList.remove('active'); }); + btn.classList.add('active'); + range = btn.dataset.range; + update(); + }); +}); + +/* ---------- export ---------- */ + +function download(name, text, mime) { + const blob = new Blob([text], { type: mime }); + const a = document.createElement('a'); + a.href = URL.createObjectURL(blob); + a.download = name; + a.click(); + URL.revokeObjectURL(a.href); +} + +document.getElementById('btn-csv').addEventListener('click', function (e) { + e.preventDefault(); + const lines = ['timestamp;count']; + rawValues.forEach(function (v) { lines.push(v.timestamp + ';' + v.count); }); + download('jonassearch_pages.csv', lines.join('\n'), 'text/csv'); +}); + +document.getElementById('btn-json').addEventListener('click', function (e) { + e.preventDefault(); + download('jonassearch_pages.json', JSON.stringify({ values: rawValues }, null, 2), 'application/json'); +}); + +/* ---------- chart helpers ---------- */ + +function destroyCharts() { + Object.keys(charts).forEach(function (k) { + if (charts[k]) charts[k].destroy(); + }); + charts = {}; +} + +function baseScales(yFmt) { + return { + x: { + type: 'linear', + ticks: { + maxTicksLimit: 8, + maxRotation: 0, + autoSkip: true, + color: mutedColor(), + callback: function (v) { return fmtDate(v); } + }, + grid: { color: 'rgba(128,128,128,0.15)' } + }, + y: { + grace: '5%', + ticks: { color: mutedColor(), callback: yFmt }, + grid: { color: 'rgba(128,128,128,0.15)' } + } + }; +} + +function tooltipTitle(items) { return fmtDateFull(items[0].parsed.x); } + +function gradientFill(colorTop, colorBottom) { + return function (context) { + const chart = context.chart; + const area = chart.chartArea; + if (!area) return colorBottom; + const g = chart.ctx.createLinearGradient(0, area.top, 0, area.bottom); + g.addColorStop(0, colorTop); + g.addColorStop(1, colorBottom); + return g; + }; +} + +/* ---------- charts ---------- */ + +function renderCharts(points, deltas, smoothedDeltas) { + destroyCharts(); + const grid = 'rgba(128,128,128,0.15)'; + + /* 1. main chart + trend */ + const regWindow = []; + for (let i = points.length - 1; i >= 0; i--) { + regWindow.unshift(points[i]); + if (points.length - i >= 96) break; + if (points[points.length - 1].x - points[i].x >= 2 * 3600000) break; + } + const reg = regress(regWindow); + let trendPoints = []; + if (reg) { + const horizon = points[points.length - 1].x + 24 * 3600000; + trendPoints = [ + { x: regWindow[0].x, y: reg.at(regWindow[0].x) }, + { x: horizon, y: reg.at(horizon) } + ]; + } + + charts.main = new Chart(document.getElementById('mainChart'), { type: 'line', data: { datasets: [ @@ -367,7 +694,7 @@ function buildCharts(points, trendPoints, deltas) { pointHoverRadius: 5, tension: 0.25, fill: true, - backgroundColor: gradientFill, + backgroundColor: gradientFill('rgba(79,140,255,0.28)', 'rgba(79,140,255,0.02)'), spanGaps: false }, { @@ -388,19 +715,12 @@ function buildCharts(points, trendPoints, deltas) { maintainAspectRatio: false, normalized: true, interaction: { mode: 'nearest', axis: 'x', intersect: false }, - scales: { - x: xAxisConfig, - y: { - grace: '5%', - ticks: { callback: function (v) { return fmt(v); } }, - grid: { color: 'rgba(255,255,255,0.05)' } - } - }, + scales: baseScales(function (v) { return fmt(v); }), plugins: { - legend: { labels: { usePointStyle: true, boxWidth: 8 } }, + legend: { labels: { usePointStyle: true, boxWidth: 8, color: mutedColor() } }, tooltip: { callbacks: { - title: function (items) { return fmtDateFull(items[0].parsed.x); }, + title: tooltipTitle, label: function (item) { return item.datasetIndex === 0 ? (fmt(item.parsed.y) + ' Pages') @@ -412,21 +732,97 @@ function buildCharts(points, trendPoints, deltas) { } }); - deltaChart = new Chart(document.getElementById('deltaChart'), { - type: 'bar', + /* 2. delta chart */ + charts.delta = new Chart(document.getElementById('deltaChart'), { + type: 'line', + data: { + datasets: [ + { + label: 'Raw', + data: deltas, + parsing: false, + borderColor: 'rgba(128,128,128,0.30)', + borderWidth: 1, + pointRadius: 0, + tension: 0, + fill: false + }, + { + label: 'Change per 5 min', + data: smoothedDeltas, + parsing: false, + borderColor: ACCENT, + borderWidth: 2.5, + pointRadius: 0, + pointHoverRadius: 4, + tension: 0.35, + fill: false, + segment: { + borderColor: function (c) { return c.p1.parsed.y >= 0 ? UP : DOWN; } + } + } + ] + }, + options: { + responsive: true, + maintainAspectRatio: false, + normalized: true, + interaction: { mode: 'nearest', axis: 'x', intersect: false }, + scales: { + x: baseScales(function () { return ''; }).x, + y: { + ticks: { color: mutedColor(), callback: function (v) { return fmtSigned(v); } }, + grid: { + color: function (c) { return c.tick.value === 0 ? 'rgba(128,128,128,0.5)' : grid; } + } + } + }, + plugins: { + legend: { + display: true, + labels: { + usePointStyle: true, + boxWidth: 8, + color: mutedColor(), + filter: function (item) { return item.text !== 'Raw'; } + } + }, + tooltip: { + filter: function (item) { return item.datasetIndex === 1; }, + callbacks: { + title: tooltipTitle, + label: function (item) { + const raw = item.raw; + return [ + 'Smoothed: ' + fmtSigned(raw.y) + ' Pages per 5 min', + 'Raw: ' + fmtSigned(raw.raw) + ' per 5 min (change ' + fmtSigned(raw.diff) + ' in ' + fmtDur(raw.dt) + ')' + ]; + } + } + } + } + } + }); + + /* 3. cumulative net growth */ + const cum = points.map(function (p, i) { + return { x: p.x, y: p.y - points[0].y }; + }); + + charts.cum = new Chart(document.getElementById('cumChart'), { + type: 'line', data: { datasets: [{ - label: 'Change per 5 min', - data: deltas, + label: 'Net growth since range start', + data: cum, parsing: false, - backgroundColor: function (c) { - return c.raw && c.raw.y >= 0 ? 'rgba(46,204,113,0.75)' : 'rgba(255,107,107,0.75)'; - }, - borderColor: function (c) { - return c.raw && c.raw.y >= 0 ? UP : DOWN; - }, - borderWidth: 1, - maxBarThickness: 28 + borderColor: UP, + borderWidth: 2, + pointRadius: 0, + pointHoverRadius: 4, + tension: 0.25, + fill: true, + backgroundColor: gradientFill('rgba(46,204,113,0.22)', 'rgba(46,204,113,0.02)') }] }, options: { @@ -435,71 +831,224 @@ function buildCharts(points, trendPoints, deltas) { normalized: true, interaction: { mode: 'nearest', axis: 'x', intersect: false }, scales: { - x: xAxisConfig, + x: baseScales(function () { return ''; }).x, y: { - ticks: { callback: function (v) { return fmtSigned(v); } }, + ticks: { color: mutedColor(), callback: function (v) { return fmtSigned(v); } }, grid: { - color: function (c) { return c.tick.value === 0 ? 'rgba(255,255,255,0.35)' : 'rgba(255,255,255,0.05)'; } + color: function (c) { return c.tick.value === 0 ? 'rgba(128,128,128,0.5)' : grid; } } } }, plugins: { - legend: { display: false }, + legend: { labels: { usePointStyle: true, boxWidth: 8, color: mutedColor() } }, tooltip: { callbacks: { - title: function (items) { return fmtDateFull(items[0].parsed.x); }, - label: function (item) { - const raw = item.raw; - return [ - fmtSigned(raw.y) + ' Pages per 5 min', - 'Actual change: ' + fmtSigned(raw.diff) + ' in ' + fmtDur(raw.dt) - ]; - } + title: tooltipTitle, + label: function (item) { return fmtSigned(item.parsed.y) + ' Pages since range start'; } } } } } }); -} -async function update() { - let payload; - try { - const response = await fetch('/api/data'); - payload = await response.json(); - } catch (e) { - setStatus('offline', 'Fetch error – retrying…'); - return; - } + /* 4. hourly growth, last 48 h */ + const buckets = {}; + deltas.forEach(function (d) { + const h = Math.floor(d.x / 3600000) * 3600000; + buckets[h] = (buckets[h] || 0) + d.y; + }); + const hours = Object.keys(buckets).map(Number).sort(function (a, b) { return a - b; }).slice(-48); + const hourlyLabels = hours.map(function (h) { return fmtDate(h); }); + const hourlyValues = hours.map(function (h) { return Math.round(buckets[h]); }); - const measurements = (payload && payload.values) ? payload.values.slice() : []; - measurements.sort(function (a, b) { return new Date(a.timestamp) - new Date(b.timestamp); }); - - const points = measurements.map(function (v) { - return { x: new Date(v.timestamp).getTime(), y: v.count }; + charts.hourly = new Chart(document.getElementById('hourlyChart'), { + type: 'bar', + data: { + labels: hourlyLabels, + datasets: [{ + label: 'Gain per hour', + data: hourlyValues, + backgroundColor: hourlyValues.map(function (v) { return v >= 0 ? 'rgba(46,204,113,0.65)' : 'rgba(255,107,107,0.65)'; }), + borderRadius: 4 + }] + }, + options: { + responsive: true, + maintainAspectRatio: false, + interaction: { mode: 'nearest', axis: 'x', intersect: false }, + scales: { + x: { ticks: { color: mutedColor(), maxRotation: 45, autoSkip: true, maxTicksLimit: 12 }, grid: { display: false } }, + y: { + ticks: { color: mutedColor(), callback: function (v) { return fmtSigned(v); } }, + grid: { + color: function (c) { return c.tick.value === 0 ? 'rgba(128,128,128,0.5)' : grid; } + } + } + }, + plugins: { + legend: { labels: { usePointStyle: true, boxWidth: 8, color: mutedColor() } }, + tooltip: { callbacks: { label: function (item) { return fmtSigned(item.parsed.y) + ' Pages in this hour'; } } } + } + } }); + /* 5. growth by hour of day (whole history) */ + const hodSum = new Array(24).fill(0); + const hodCount = new Array(24).fill(0); + deltas.forEach(function (d) { + const h = new Date(d.x).getHours(); + hodSum[h] += d.y; + hodCount[h] += 1; + }); + const hodValues = hodSum.map(function (s, i) { + return hodCount[i] ? Math.round(s / hodCount[i] * 10) / 10 : 0; + }); + + charts.hod = new Chart(document.getElementById('hodChart'), { + type: 'bar', + data: { + labels: hodValues.map(function (_, i) { return (i < 10 ? '0' : '') + i + ':00'; }), + datasets: [{ + label: 'Avg gain per 5 min', + data: hodValues, + backgroundColor: hodValues.map(function (v) { return v >= 0 ? 'rgba(79,140,255,0.65)' : 'rgba(255,107,107,0.65)'; }), + borderRadius: 4 + }] + }, + options: { + responsive: true, + maintainAspectRatio: false, + scales: { + x: { ticks: { color: mutedColor(), maxRotation: 0, autoSkip: true }, grid: { display: false } }, + y: { + ticks: { color: mutedColor(), callback: function (v) { return fmtSigned(v); } }, + grid: { + color: function (c) { return c.tick.value === 0 ? 'rgba(128,128,128,0.5)' : grid; } + } + } + }, + plugins: { + legend: { labels: { usePointStyle: true, boxWidth: 8, color: mutedColor() } }, + tooltip: { callbacks: { label: function (item) { return fmtSigned(item.parsed.y) + ' Pages per 5 min (avg of ' + hodCount[item.dataIndex] + ' measurements)'; } } } + } + } + }); +} + +/* ---------- records table ---------- */ + +function renderRecords(deltas) { + const el = document.getElementById('records'); + if (deltas.length < 2) { + el.innerHTML = '
Not enough data yet.
'; + return; + } + const sorted = deltas.slice().sort(function (a, b) { return b.diff - a.diff; }); + const gains = sorted.slice(0, 5).filter(function (d) { return d.diff > 0; }); + const losses = sorted.slice().reverse().slice(0, 5).filter(function (d) { return d.diff < 0; }).reverse(); + + function rows(items, cls) { + return items.map(function (d) { + return '' + fmtDate(d.x) + '' + fmtSigned(d.diff) + '' + fmtDur(d.dt) + ''; + }).join(''); + } + + el.innerHTML = + '' + + '' + + rows(gains, 'pos') + + rows(losses, 'neg') + + '
TimeChangeSpan
'; +} + +/* ---------- heatmap ---------- */ + +function renderHeatmap(points) { + const el = document.getElementById('heatmap'); + if (!points.length) { el.innerHTML = ''; return; } + + function localDayKey(d) { + return d.getFullYear() + '-' + + ('0' + (d.getMonth() + 1)).slice(-2) + '-' + + ('0' + d.getDate()).slice(-2); + } + + const map = {}; + points.forEach(function (p) { + const d = new Date(p.x); + map[localDayKey(d) + '|' + d.getHours()] = p.y; + }); + + const days = {}; + Object.keys(map).forEach(function (k) { days[k.split('|')[0]] = true; }); + const dayList = Object.keys(days).sort(); + + // deltas per day+hour for coloring + const cellDeltas = {}; + let maxAbs = 0; + const sorted = points.slice().sort(function (a, b) { return a.x - b.x; }); + const byDayHour = {}; + sorted.forEach(function (p) { + const d = new Date(p.x); + const key = localDayKey(d) + '|' + d.getHours(); + if (byDayHour[key] !== undefined) { + cellDeltas[key] = p.y - byDayHour[key]; + maxAbs = Math.max(maxAbs, Math.abs(cellDeltas[key])); + } + byDayHour[key] = p.y; + }); + + let html = ''; + for (let h = 0; h < 24; h++) html += ''; + html += ''; + + dayList.forEach(function (day) { + html += ''; + for (let h = 0; h < 24; h++) { + const key = day + '|' + h; + const val = map[key]; + let style = ''; + let title = fmtDay(new Date(day + 'T12:00:00').getTime()) + ' ' + h + ':00 – no data'; + if (val !== undefined && cellDeltas[key] !== undefined) { + const dv = cellDeltas[key]; + const op = maxAbs ? Math.min(0.85, 0.15 + 0.7 * Math.abs(dv) / maxAbs) : 0.3; + const color = dv >= 0 ? '46,204,113' : '255,107,107'; + style = ' style="background:rgba(' + color + ',' + op.toFixed(2) + ')"'; + title = fmtDay(new Date(day + 'T12:00:00').getTime()) + ' ' + h + ':00 – ' + fmtSigned(dv) + ' Pages (' + fmt(val) + ' total)'; + } + html += ''; + } + html += ''; + }); + html += '
' + h + '
' + fmtDay(new Date(day + 'T12:00:00').getTime()) + '
'; + el.innerHTML = html; +} + +/* ---------- main update ---------- */ + +function renderAll() { + const all = parsePoints(rawValues); + const points = filterRange(all); const now = Date.now(); const elCurrent = document.getElementById('current'); const meta = document.getElementById('current-meta'); - document.getElementById('badge-count').textContent = 'Measurements: ' + points.length; + document.getElementById('badge-count').textContent = 'Measurements: ' + all.length; - if (!points.length) { + if (!all.length) { elCurrent.textContent = '–'; meta.textContent = 'No measurements yet – the collector stores one every 15 minutes.'; setStatus('offline', 'No data'); return; } - const last = points[points.length - 1]; - const first = points[0]; + const last = all[all.length - 1]; + const first = all[0]; const dts = []; - for (let i = 1; i < points.length; i++) dts.push(points[i].x - points[i - 1].x); + for (let i = 1; i < all.length; i++) dts.push(all[i].x - all[i - 1].x); intervalMs = median(dts) || FALLBACK_INTERVAL_MS; - document.getElementById('badge-interval').textContent = - 'Interval: ~' + fmtDur(intervalMs); + document.getElementById('badge-interval').textContent = 'Interval: ~' + fmtDur(intervalMs); elCurrent.textContent = fmt(last.y); meta.textContent = 'Last measurement: ' + fmtDateFull(last.x) + @@ -517,6 +1066,7 @@ async function update() { setStatus('offline', 'No data for ' + fmtDur(age)); } + /* fixed-window cards (full history) */ const totalHours = (last.x - first.x) / 3600000; const perHour = totalHours > 0 ? (last.y - first.y) / totalHours : null; const elPh = document.getElementById('stat-per-hour'); @@ -525,100 +1075,132 @@ async function update() { elPh.textContent = '–'; elPd.textContent = '–'; } else { - elPh.textContent = fmtSigned(perHour); - elPh.classList.remove('pos', 'neg'); - elPh.classList.add(perHour >= 0 ? 'pos' : 'neg'); - elPd.textContent = fmtSigned(perHour * 24); - elPd.classList.remove('pos', 'neg'); - elPd.classList.add(perHour >= 0 ? 'pos' : 'neg'); + setSigned('stat-per-hour', perHour); + setSigned('stat-per-day', perHour * 24); } - let deltas = []; - for (let i = 1; i < points.length; i++) { - const dt = points[i].x - points[i - 1].x; - const diff = points[i].y - points[i - 1].y; - if (dt <= 0 || dt > 4 * intervalMs) continue; - const per5 = dt < 60000 ? diff : diff * (5 * 60000 / dt); - deltas.push({ x: points[i].x, y: per5, dt: dt, diff: diff }); - } - - const lastDelta = deltas.length ? deltas[deltas.length - 1] : null; - if (lastDelta) { - const el = document.getElementById('stat-delta-5m'); - el.textContent = fmtSigned(lastDelta.diff); - el.classList.remove('pos', 'neg'); - el.classList.add(lastDelta.diff >= 0 ? 'pos' : 'neg'); - document.getElementById('stat-delta-5m-hint').textContent = - fmtSigned(lastDelta.y) + ' per 5 min · over ' + fmtDur(lastDelta.dt); - } else { - document.getElementById('stat-delta-5m').textContent = '–'; - document.getElementById('stat-delta-5m-hint').textContent = 'Needs at least 2 measurements'; - } - - const span1h = closestBefore(points, last.x - 3600000, Math.max(3 * intervalMs, 30 * 60000)); - setDeltaCard('stat-delta-1h', 'stat-delta-1h-hint', - span1h ? last.y - span1h.y : null, - span1h ? last.x - span1h.x : 0, - span1h ? (last.y - span1h.y) * (3600000 / Math.max(1, last.x - span1h.x)) * (5 / 60) : 0); - - const span24h = closestBefore(points, last.x - 24 * 3600000, Math.max(6 * intervalMs, 3 * 3600000)); + const span24h = closestBefore(all, last.x - 24 * 3600000, Math.max(6 * intervalMs, 3 * 3600000)); setDeltaCard('stat-delta-24h', 'stat-delta-24h-hint', span24h ? last.y - span24h.y : null, span24h ? last.x - span24h.x : 0, span24h ? (last.y - span24h.y) * (3600000 / Math.max(1, last.x - span24h.x)) * (5 / 60) : 0); + const span7d = closestBefore(all, last.x - 7 * 24 * 3600000, Math.max(12 * intervalMs, 6 * 3600000)); + setDeltaCard('stat-delta-7d', 'stat-delta-7d-hint', + span7d ? last.y - span7d.y : null, + span7d ? last.x - span7d.x : 0, + span7d ? (last.y - span7d.y) * (3600000 / Math.max(1, last.x - span7d.x)) * (5 / 60) : 0); + + /* range-dependent cards + charts */ + const fpoints = points.length >= 2 ? points : all; + const dd = computeDeltas(fpoints); + const deltas = dd.deltas; + + const smoothY = movingAverage(deltas.map(function (d) { return d.y; }), DELTA_SMOOTH_WINDOW); + const smoothedDeltas = deltas.map(function (d, i) { + return { x: d.x, y: smoothY[i], raw: d.y, dt: d.dt, diff: d.diff }; + }); + + /* avg/h over last 24 h: sum of all changes within 24 h ÷ hours actually covered */ + let sum24 = 0; + let windowStart = Math.max(fpoints[0].x, last.x - 24 * 3600000); + let coveredH = (last.x - windowStart) / 3600000; + deltas.forEach(function (d) { + if (d.x > last.x - 24 * 3600000) sum24 += d.diff; + }); + if (coveredH >= 0.5) { + setSigned('stat-avg-hour', sum24 / coveredH); + } else { + document.getElementById('stat-avg-hour').textContent = '–'; + } + + /* best / worst hour: best total gain within a sliding ~1 h window */ + let best = null, worst = null; + for (let i = 0; i < deltas.length; i++) { + let gain = 0, endMs = deltas[i].x; + for (let j = i; j < deltas.length; j++) { + if (deltas[j].x - deltas[i].x > 3600000) break; + gain += deltas[j].diff; + endMs = deltas[j].x; + } + if (!best || gain > best.gain) best = { gain: gain, x: endMs }; + if (!worst || gain < worst.gain) worst = { gain: gain, x: endMs }; + } + if (best && best.gain !== 0) { + setSigned('stat-best-hour', best.gain); + document.getElementById('stat-best-hour-hint').textContent = 'Around ' + fmtDate(best.x); + } + if (worst && worst.gain !== 0) { + setSigned('stat-worst-hour', worst.gain); + document.getElementById('stat-worst-hour-hint').textContent = 'Around ' + fmtDate(worst.x); + } + + /* peak / lowest (whole history) */ + let peak = all[0], low = all[0]; + all.forEach(function (p) { + if (p.y > peak.y) peak = p; + if (p.y < low.y) low = p; + }); + document.getElementById('stat-peak').textContent = fmt(peak.y); + document.getElementById('stat-peak-hint').textContent = 'On ' + fmtDateFull(peak.x); + document.getElementById('stat-min').textContent = fmt(low.y); + document.getElementById('stat-min-hint').textContent = 'On ' + fmtDateFull(low.x); + + /* forecast */ const regWindow = []; - for (let i = points.length - 1; i >= 0; i--) { - regWindow.unshift(points[i]); - if (points.length - i >= 96) break; - if (last.x - points[i].x >= 2 * 3600000) break; + for (let i = fpoints.length - 1; i >= 0; i--) { + regWindow.unshift(fpoints[i]); + if (fpoints.length - i >= 96) break; + if (fpoints[fpoints.length - 1].x - fpoints[i].x >= 2 * 3600000) break; } const reg = regress(regWindow); - - let trendPoints = []; if (reg) { - const horizon = last.x + 24 * 3600000; - trendPoints = [ - { x: regWindow[0].x, y: reg.at(regWindow[0].x) }, - { x: horizon, y: reg.at(horizon) } - ]; const elF = document.getElementById('stat-forecast'); - const fc = last.y + reg.slope; + const fc = fpoints[fpoints.length - 1].y + reg.slope; elF.textContent = fmt(fc); elF.classList.remove('pos', 'neg'); - elF.classList.add(fc >= last.y ? 'pos' : 'neg'); + elF.classList.add(fc >= fpoints[fpoints.length - 1].y ? 'pos' : 'neg'); document.getElementById('stat-forecast-hint').textContent = 'Trend ' + fmtSigned(reg.slope) + '/h · R² = ' + reg.r2.toFixed(3); document.getElementById('footer').textContent = 'Forecast: linear least-squares regression over the last ' + regWindow.length + - ' measurements (' + fmtDur(last.x - regWindow[0].x) + '), extrapolated +24 h · R² = ' + - reg.r2.toFixed(3) + ' · All statistics are computed in the browser.'; + ' measurements (' + fmtDur(fpoints[fpoints.length - 1].x - regWindow[0].x) + + '), extrapolated +24 h · R² = ' + reg.r2.toFixed(3) + + ' · Range: ' + (range === 'all' ? 'all' : range) + ' · All statistics are computed in the browser.'; } else { document.getElementById('stat-forecast').textContent = '–'; document.getElementById('stat-forecast-hint').textContent = 'Needs at least 2 measurements'; - document.getElementById('footer').textContent = - 'Forecast available once there are at least 2 measurements.'; + document.getElementById('footer').textContent = 'Forecast available once there are at least 2 measurements.'; } - if (!mainChart) { - buildCharts(points, trendPoints, deltas); + renderCharts(fpoints, deltas, smoothedDeltas); + renderRecords(deltas); + renderHeatmap(all); +} + +async function update() { + refreshLeft = REFRESH_S; + try { + const response = await fetch('/api/data'); + rawValues = (await response.json()).values || []; + } catch (e) { + setStatus('offline', 'Fetch error – retrying…'); + return; + } + renderAll(); +} + +setInterval(function () { + refreshLeft -= 1; + if (refreshLeft <= 0) { + document.getElementById('badge-refresh').textContent = 'Refreshing…'; } else { - mainChart.data.datasets[0].data = points; - mainChart.data.datasets[1].data = trendPoints; - mainChart.update('none'); - deltaChart.data.datasets[0].data = deltas; - deltaChart.update('none'); + document.getElementById('badge-refresh').textContent = 'Next refresh: ' + refreshLeft + ' s'; } -} - -function setStatus(kind, text) { - const dot = document.getElementById('status-dot'); - dot.className = 'dot ' + kind + (kind === 'ok' ? ' live' : ''); - document.getElementById('status-text').textContent = text; -} +}, 1000); update(); -setInterval(update, 30000); +setInterval(update, REFRESH_S * 1000); @@ -710,6 +1292,14 @@ def api_data(): }) +@app.route("/api/export") +def api_export(): + lines = ["timestamp;count"] + for entry in load_data(): + lines.append(f"{entry.get('timestamp', '')};{entry.get('count', '')}") + return Response("\n".join(lines), mimetype="text/csv") + + if __name__ == "__main__": # Collector im Hintergrund starten thread = threading.Thread(target=collect, daemon=True) @@ -719,5 +1309,6 @@ if __name__ == "__main__": app.run( host="0.0.0.0", port=8081, - threaded=True + threaded=True, + load_dotenv=False # sonst liest Flask beim Start .env im Projektordner -> PermissionError )