diff --git a/README.md b/README.md
index 2492dd9..bdf9452 100644
--- a/README.md
+++ b/README.md
@@ -6,7 +6,10 @@ das komplette serverjonas search
- `launcher.py` – startet/überwacht den Crawler und rollt Updates aus (siehe unten)
- `search/` – Suchmaschine darüber (CLI + HTTP-API + Weboberfläche, Port 8083),
siehe [search/README.md](search/README.md)
-- `web/` – Statistik-Webseite (Port 8081, `jonassearch-stats.service`)
+- `web/` – Statistik-Webseite (Port 8081, `jonassearch-stats.service`).
+ Seiten werden alle 30 Min. gezählt; Bilder alle 2 h exakt, dazwischen nur
+ geschätzt. Alle Diagramme zeigen ausschließlich die exakten Messpunkte.
+- `web/tui.py` – Terminal-Dashboard für dieselben Statistiken (siehe unten)
- `priorityweb/` – Vorschlagsverfeinerer (Port 8082, `jonassearch-priority.service`)
- `search/jonassearch-engine.service` – systemd-Unit für die Suchmaschine
(Port 8083, `jonassearch-engine.service`)
@@ -16,6 +19,27 @@ das komplette serverjonas search
- `add_pages_pictures.sql` – einmalig laufen lassen: legt `pages.pictures`
(`text[]`) für die Bild-URLs der gecrawlten Seiten an (idempotent)
+## Terminal-Dashboard (`web/tui.py`)
+
+Zeigt Seiten- und Bildanzahl, Wachstum, einen Sparkline-Verlauf und die
+letzten Messwerte direkt im Terminal. Liest dieselbe
+`web/page_counts.json` wie der Statserver – kein Webserver, keine
+zusätzlichen Pakete (nur `curses` aus der Standardlib).
+
+```bash
+python3 web/tui.py # interaktiv, Aktualisierung alle 2 s
+python3 web/tui.py --window 7d # 24h (Default) | 7d | 30d | all
+python3 web/tui.py --once # Momentaufnahme, z. B. für Logs/Skripte
+python3 web/tui.py --once --width 100 --window all
+```
+
+Tasten im interaktiven Modus: `q` beendet, `r` lädt die Datei sofort neu,
+`t` wechselt das Zeitfenster. `--ascii` ersetzt die Unicode-Blöcke der
+Sparkline. Läuft die Ausgabe nicht in einem Terminal (Pipe, Cron), wird
+automatisch die `--once`-Ausgabe verwendet. Fehlt die Datei, zeigt das
+Dashboard „No samples yet“; ist sie defekt, erscheint die Fehlermeldung
+im Dashboard (bei `--once` auf stderr, Exit-Code 1).
+
## Sehr große Seiten / tsvector-Limit
PostgreSQL begrenzt einen einzelnen `tsvector`-Wert hart auf 1.048.575 Bytes
diff --git a/web/app.py b/web/app.py
index 473fc43..9b05a38 100644
--- a/web/app.py
+++ b/web/app.py
@@ -145,7 +145,7 @@ HTML = r"""
50% { opacity: 0.35; }
}
- .ranges { display: flex; gap: 8px; margin-bottom: 18px; }
+ .ranges { display: flex; flex-wrap: wrap; gap: 8px; margin-bottom: 18px; }
.range-btn {
background: var(--card);
@@ -302,6 +302,7 @@ HTML = r"""
+
@@ -315,6 +316,11 @@ HTML = r"""
–
Average over full history
+
Growth 6 h
+
–
+
vs. measurement ~6 h ago
+
Growth 24 h
–
@@ -400,6 +406,55 @@ HTML = r"""
+ Images – all graphs
+ Only exact measurements (counted every 2 h) are plotted – the estimates in between are excluded.
+
+ No image counts yet – they appear after the next collector run with the image migration in place.
+
+
+
+
Images over time
+
Image URLs found while crawling · exact count every 2 h · only exact measurements are plotted (no estimates) · dashed line = trend & 24 h forecast
+
+
+
Change per 5 minutes
+
Normalized to a 5-minute interval · spans between two exact counts (2 h) · grey = raw · colored = smoothed (moving average over 5 measurements)
+
+
+
Cumulative net growth
+
Total images gained/lost since the start of the selected range · linear time axis
+
+
+
Images per hour (last 48 h)
+
Sum of changes per hour bucket · green = growth, red = loss
+
+
+
+
Image growth by hour of day
+
Average gain per 5 min, grouped by hour of day (whole history)
+
+
+
+
Records – biggest image changes
+
Largest gains and losses between two consecutive measurements (selected range)
+
+
+
Activity heatmap – image gain per hour of day
+
+
Rows = days · columns = hours (0–23) · green = growth, red = loss, grey = no data
+
+
Not enough data yet.
';
return;
@@ -985,7 +1068,7 @@ function renderRecords(deltas) {
el.innerHTML =
'' +
- '| Time | Change | Span |
|---|
' +
+ '| Time | Change (' + unit + ') | Span |
|---|
' +
rows(gains, 'pos') +
rows(losses, 'neg') +
'
';
@@ -993,8 +1076,8 @@ function renderRecords(deltas) {
/* ---------- heatmap ---------- */
-function renderHeatmap(points) {
- const el = document.getElementById('heatmap');
+function renderHeatmap(points, id, unit) {
+ const el = document.getElementById(id);
if (!points.length) { el.innerHTML = ''; return; }
function localDayKey(d) {
@@ -1044,7 +1127,7 @@ function renderHeatmap(points) {
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)';
+ title = fmtDay(new Date(day + 'T12:00:00').getTime()) + ' ' + h + ':00 – ' + fmtSigned(dv) + ' ' + unit + ' (' + fmt(val) + ' total)';
}
html += ' | ';
}
@@ -1123,6 +1206,12 @@ function renderAll() {
setSigned('stat-per-day', perHour * 24);
}
+ const span6h = closestBefore(all, last.x - 6 * 3600000, Math.max(3 * intervalMs, 3600000));
+ setDeltaCard('stat-delta-6h', 'stat-delta-6h-hint',
+ span6h ? last.y - span6h.y : null,
+ span6h ? last.x - span6h.x : 0,
+ span6h ? (last.y - span6h.y) * (3600000 / Math.max(1, last.x - span6h.x)) * (5 / 60) : 0);
+
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,
@@ -1217,9 +1306,55 @@ function renderAll() {
document.getElementById('footer').textContent = 'Forecast available once there are at least 2 measurements.';
}
- renderCharts(fpoints, deltas, smoothedDeltas);
- renderRecords(deltas);
- renderHeatmap(all);
+ /* images: same graphs, but only exact measurements (every 2 h) */
+ const imageAll = toImagePoints(all);
+ const imageFpoints = filterRange(imageAll, last.x);
+ const imageDeltas = computeDeltas(imageFpoints).deltas;
+ const imageSmoothY = movingAverage(imageDeltas.map(function (d) { return d.y; }), DELTA_SMOOTH_WINDOW);
+ const imageSmoothedDeltas = imageDeltas.map(function (d, i) {
+ return { x: d.x, y: imageSmoothY[i], raw: d.y, dt: d.dt, diff: d.diff };
+ });
+ const imagesEmpty = document.getElementById('images-empty');
+ const hasImageValues = all.some(function (p) { return p.images !== null; });
+ if (!imageAll.length) {
+ imagesEmpty.textContent = hasImageValues
+ ? 'Only estimated image counts so far – the graphs show exact counts only (every 2 h).'
+ : 'No image counts yet – they appear after the next collector run with the image migration in place.';
+ imagesEmpty.style.display = '';
+ } else if (!imageFpoints.length) {
+ imagesEmpty.textContent = 'No exact image counts in the selected range (counted every 2 h).';
+ imagesEmpty.style.display = '';
+ } else {
+ imagesEmpty.style.display = 'none';
+ }
+
+ destroyCharts();
+ renderCharts({
+ prefix: '',
+ points: fpoints,
+ deltas: deltas,
+ smoothedDeltas: smoothedDeltas,
+ unit: 'Pages',
+ accent: ACCENT,
+ accentBar: 'rgba(79,140,255,0.65)',
+ gradTop: 'rgba(79,140,255,0.28)',
+ gradBottom: 'rgba(79,140,255,0.02)'
+ });
+ renderCharts({
+ prefix: 'images',
+ points: imageFpoints,
+ deltas: imageDeltas,
+ smoothedDeltas: imageSmoothedDeltas,
+ unit: 'Images',
+ accent: ACCENT2,
+ accentBar: 'rgba(180,122,255,0.65)',
+ gradTop: 'rgba(180,122,255,0.28)',
+ gradBottom: 'rgba(180,122,255,0.02)'
+ });
+ renderRecords(deltas, 'records', 'Pages');
+ renderRecords(imageDeltas, 'records-images', 'Images');
+ renderHeatmap(all, 'heatmap', 'Pages');
+ renderHeatmap(imageAll, 'heatmap-images', 'Images');
}
async function update() {
diff --git a/web/test_stats.py b/web/test_stats.py
index 813987a..5f72473 100644
--- a/web/test_stats.py
+++ b/web/test_stats.py
@@ -129,6 +129,59 @@ class StatsPageTest(unittest.TestCase):
self.assertIn("Estimated · exact every 2 h", html)
self.assertIn("imagesExact", html)
+ def test_page_contains_6h_range_and_card(self):
+ client = stats_app.app.test_client()
+ response = client.get("/")
+ self.assertEqual(response.status_code, 200)
+ html = response.get_data(as_text=True)
+ self.assertIn('data-range="6h"', html)
+ self.assertIn('id="stat-delta-6h"', html)
+ self.assertIn('id="stat-delta-6h-hint"', html)
+ self.assertIn("'6h': 6 * 3600000", html)
+ self.assertIn("closestBefore(all, last.x - 6 * 3600000", html)
+
+ def test_page_contains_all_image_graphs(self):
+ client = stats_app.app.test_client()
+ response = client.get("/")
+ self.assertEqual(response.status_code, 200)
+ html = response.get_data(as_text=True)
+
+ self.assertIn("Images – all graphs", html)
+ for canvas in (
+ "imagesMainChart",
+ "imagesDeltaChart",
+ "imagesCumChart",
+ "imagesHourlyChart",
+ "imagesHodChart",
+ ):
+ self.assertIn('id="' + canvas + '"', html)
+ self.assertIn('id="records-images"', html)
+ self.assertIn('id="heatmap-images"', html)
+ self.assertIn('id="images-empty"', html)
+
+ # the image series skips entries without an image count (old data)
+ # and only plots exact measurements - no estimates between the 2 h counts
+ self.assertIn("function toImagePoints(points)", html)
+ self.assertIn("p.images !== null && p.images !== undefined", html)
+ self.assertIn(".filter(function (p) { return p.imagesExact; })", html)
+ self.assertIn("only exact measurements are plotted (no estimates)", html)
+ self.assertIn("Only exact measurements (counted every 2 h) are plotted", html)
+ self.assertIn("Only estimated image counts so far", html)
+ self.assertIn("No exact image counts in the selected range", html)
+ # existing page chart ids must stay lowercase (mainChart, not MainChart)
+ self.assertIn("function chartId(prefix, suffix)", html)
+ self.assertIn("chartId(prefix, 'MainChart')", html)
+ # both series use the same time window (anchored to the latest page sample)
+ self.assertIn("filterRange(imageAll, last.x)", html)
+
+ # both metrics go through the same renderers, with the right units
+ self.assertIn("unit: 'Images'", html)
+ self.assertIn("unit: 'Pages'", html)
+ self.assertIn("renderRecords(deltas, 'records', 'Pages')", html)
+ self.assertIn("renderRecords(imageDeltas, 'records-images', 'Images')", html)
+ self.assertIn("renderHeatmap(all, 'heatmap', 'Pages')", html)
+ self.assertIn("renderHeatmap(imageAll, 'heatmap-images', 'Images')", html)
+
if __name__ == "__main__":
unittest.main(verbosity=2)
diff --git a/web/test_tui.py b/web/test_tui.py
new file mode 100644
index 0000000..ec042cd
--- /dev/null
+++ b/web/test_tui.py
@@ -0,0 +1,341 @@
+#!/usr/bin/env python3
+"""
+Tests für das Terminal-Dashboard (web/tui.py), ohne Terminal und ohne DB.
+
+Geprüft werden Datei-Lesen, Zeitfenster, Sparkline, die Textdarstellung
+und der Aufruf über die Kommandozeile (`--once`).
+"""
+
+import importlib.util
+import json
+import os
+import subprocess
+import sys
+import tempfile
+import unittest
+from datetime import datetime, timedelta
+
+HERE = os.path.dirname(os.path.abspath(__file__))
+ROOT = os.path.dirname(HERE)
+
+_spec = importlib.util.spec_from_file_location(
+ "jonassearch_stats_tui", os.path.join(HERE, "tui.py")
+)
+tui = importlib.util.module_from_spec(_spec)
+_spec.loader.exec_module(tui)
+
+try:
+ _stats_spec = importlib.util.spec_from_file_location(
+ "jonassearch_stats_app_for_tui", os.path.join(HERE, "app.py")
+ )
+ stats_app = importlib.util.module_from_spec(_stats_spec)
+ _stats_spec.loader.exec_module(stats_app)
+except Exception: # z. B. Flask nicht installiert
+ stats_app = None
+
+NOW = datetime(2026, 10, 17, 12, 0, 0)
+
+
+def entry(timestamp, count, images=None, exact=None):
+ data = {"timestamp": timestamp.isoformat(), "count": count}
+ if images is not None:
+ data["images"] = images
+ if exact is not None:
+ data["images_exact"] = exact
+ return data
+
+
+def sample_entries():
+ return [
+ entry(NOW - timedelta(hours=2), 980_000, images=100, exact=True),
+ entry(NOW - timedelta(hours=1), 981_000),
+ entry(NOW - timedelta(minutes=30), 981_950, images=120, exact=False),
+ ]
+
+
+def normalized(entries):
+ """Rohdaten so aufbereiten, wie load_entries() sie liefert."""
+ result = []
+ for item in entries:
+ copy = dict(item)
+ copy["timestamp"] = tui.parse_timestamp(copy["timestamp"])
+ result.append(copy)
+ return result
+
+
+class FormatTest(unittest.TestCase):
+
+ def test_fmt_int_uses_dots(self):
+ self.assertEqual(tui.fmt_int(987832), "987.832")
+ self.assertEqual(tui.fmt_int(None), "-")
+
+ def test_fmt_delta(self):
+ self.assertEqual(tui.fmt_delta(950), "+950")
+ self.assertEqual(tui.fmt_delta(-12), "-12")
+ self.assertEqual(tui.fmt_delta(0), "0")
+ self.assertEqual(tui.fmt_delta(None), "-")
+
+ def test_fmt_duration(self):
+ self.assertEqual(tui.fmt_duration(30), "30 s")
+ self.assertEqual(tui.fmt_duration(90), "1 min")
+ self.assertEqual(tui.fmt_duration(5400), "1 h 30 min")
+ self.assertEqual(tui.fmt_duration(90000), "1 d 1 h")
+
+ def test_samples_label(self):
+ self.assertEqual(tui._samples_label(1), "1 sample")
+ self.assertEqual(tui._samples_label(2), "2 samples")
+
+ def test_relative_time(self):
+ self.assertEqual(tui.relative_time(NOW, NOW), "just now")
+ self.assertEqual(
+ tui.relative_time(NOW - timedelta(minutes=3), NOW), "3 min ago"
+ )
+
+
+class ParseTimestampTest(unittest.TestCase):
+
+ def test_iso_and_z(self):
+ self.assertEqual(
+ tui.parse_timestamp("2026-10-17T20:44:20.187761"),
+ datetime(2026, 10, 17, 20, 44, 20, 187761),
+ )
+ parsed = tui.parse_timestamp("2026-10-17T20:44:20Z")
+ self.assertIsNone(parsed.tzinfo)
+ self.assertEqual(
+ parsed,
+ datetime.fromisoformat("2026-10-17T20:44:20+00:00")
+ .astimezone().replace(tzinfo=None),
+ )
+
+ def test_invalid(self):
+ self.assertIsNone(tui.parse_timestamp("kaputt"))
+ self.assertIsNone(tui.parse_timestamp(""))
+ self.assertIsNone(tui.parse_timestamp(None))
+
+
+class LoadEntriesTest(unittest.TestCase):
+
+ def setUp(self):
+ self.tmp = tempfile.TemporaryDirectory()
+ self.path = os.path.join(self.tmp.name, "page_counts.json")
+
+ def tearDown(self):
+ self.tmp.cleanup()
+
+ def _write(self, data):
+ with open(self.path, "w", encoding="utf-8") as handle:
+ if isinstance(data, str):
+ handle.write(data)
+ else:
+ json.dump(data, handle)
+
+ def test_missing_file_is_empty(self):
+ self.assertEqual(tui.load_entries(self.path), [])
+
+ def test_sorts_and_skips_bad_rows(self):
+ self._write([
+ entry(NOW, 20),
+ "kaputt",
+ {"timestamp": "kaputt", "count": 1},
+ {"timestamp": NOW.isoformat(), "count": "x"},
+ entry(NOW - timedelta(minutes=30), 10),
+ ])
+ entries = tui.load_entries(self.path)
+ self.assertEqual([e["count"] for e in entries], [10, 20])
+ self.assertIsInstance(entries[0]["timestamp"], datetime)
+
+ def test_api_format_is_accepted(self):
+ self._write({"values": [entry(NOW, 5, images=7, exact=True)]})
+ entries = tui.load_entries(self.path)
+ self.assertEqual(len(entries), 1)
+ self.assertEqual(entries[0]["images"], 7)
+ self.assertTrue(entries[0]["images_exact"])
+
+ def test_invalid_images_becomes_none(self):
+ self._write([{"timestamp": NOW.isoformat(), "count": 1,
+ "images": "viele"}])
+ self.assertIsNone(tui.load_entries(self.path)[0]["images"])
+
+ def test_corrupt_json_raises(self):
+ self._write("{not json")
+ with self.assertRaises(tui.StatsDataError):
+ tui.load_entries(self.path)
+
+ def test_wrong_toplevel_type_raises(self):
+ self._write(42)
+ with self.assertRaises(tui.StatsDataError):
+ tui.load_entries(self.path)
+
+
+class SharedScheduleTest(unittest.TestCase):
+
+ def test_exact_interval_matches_statserver(self):
+ if stats_app is None:
+ self.skipTest("web/app.py nicht importierbar (Flask fehlt)")
+ self.assertEqual(
+ tui.EXACT_IMAGE_INTERVAL, stats_app.EXACT_IMAGE_INTERVAL
+ )
+
+
+class WindowTest(unittest.TestCase):
+
+ def test_window_filters_old_entries(self):
+ entries = normalized([
+ entry(NOW - timedelta(days=3), 1),
+ entry(NOW - timedelta(hours=1), 2),
+ ])
+ self.assertEqual(
+ [e["count"] for e in tui.select_window(entries, "24h", NOW)], [2]
+ )
+ self.assertEqual(len(tui.select_window(entries, "all", NOW)), 2)
+
+ def test_latest_images_falls_back(self):
+ entries = normalized([
+ entry(NOW - timedelta(hours=1), 1, images=10, exact=True),
+ entry(NOW, 2),
+ ])
+ images, exact, timestamp = tui.latest_images(entries)
+ self.assertEqual(images, 10)
+ self.assertTrue(exact)
+ self.assertEqual(timestamp, entries[0]["timestamp"])
+
+ def test_next_exact_text(self):
+ recent = normalized([
+ entry(NOW - timedelta(minutes=30), 1, images=1, exact=True)
+ ])
+ self.assertEqual(
+ tui.next_exact_text(recent, NOW), "next exact in 1 h 30 min"
+ )
+ overdue = normalized([
+ entry(NOW - timedelta(hours=3), 1, images=1, exact=True)
+ ])
+ self.assertEqual(
+ tui.next_exact_text(overdue, NOW), "exact count due now"
+ )
+ self.assertEqual(
+ tui.next_exact_text(normalized([entry(NOW, 1)]), NOW),
+ "no exact count yet",
+ )
+
+
+class SparklineTest(unittest.TestCase):
+
+ def test_width_and_padding(self):
+ line = tui.sparkline([1, 2, 3], 10)
+ self.assertEqual(len(line), 10)
+ self.assertTrue(line.startswith(" "))
+
+ def test_single_sample(self):
+ self.assertEqual(tui.sparkline([5], 1), tui.SPARK_BLOCKS[4])
+
+ def test_increasing_values_use_increasing_blocks(self):
+ line = tui.sparkline([1, 2, 3, 4], 4)
+ indices = [tui.SPARK_BLOCKS.index(char) for char in line]
+ self.assertEqual(indices, sorted(indices))
+
+ def test_constant_values_do_not_crash(self):
+ self.assertEqual(tui.sparkline([7, 7, 7], 3), tui.SPARK_BLOCKS[4] * 3)
+
+ def test_empty(self):
+ self.assertEqual(tui.sparkline([], 10), "")
+
+
+class RenderTest(unittest.TestCase):
+
+ def test_shows_numbers_and_hints(self):
+ lines = tui.render_dashboard(
+ normalized(sample_entries()), now=NOW, width=100
+ )
+ text = "\n".join(lines)
+ self.assertIn("Pages 981.950", text)
+ self.assertIn("Images 120", text)
+ self.assertIn("estimated", text)
+ self.assertIn("exact count due now", text)
+ self.assertIn("+950 since previous", text)
+ self.assertIn("q quit", text)
+ self.assertIn("t window (24h)", text)
+ self.assertIn("Recent samples", text)
+
+ def test_lines_fit_width(self):
+ for width in (32, 48, 60, 80):
+ lines = tui.render_dashboard(
+ normalized(sample_entries()), now=NOW, width=width, rows=24
+ )
+ for line in lines:
+ self.assertLessEqual(len(line), width, repr(line))
+
+ def test_row_budget_keeps_footer(self):
+ lines = tui.render_dashboard(
+ normalized(sample_entries()), now=NOW, width=80, rows=12
+ )
+ self.assertLessEqual(len(lines), 12)
+ self.assertIn("q quit", lines[-1])
+
+ def test_unreadable_file_error(self):
+ temp = tempfile.TemporaryDirectory()
+ self.addCleanup(temp.cleanup)
+ path = os.path.join(temp.name, "page_counts.json")
+ with open(path, "w", encoding="utf-8") as handle:
+ handle.write("{kaputt")
+
+ try:
+ tui.load_entries(path)
+ except tui.StatsDataError as exc:
+ lines = tui.render_dashboard([], now=NOW, width=90, error=str(exc))
+ else:
+ self.fail("StatsDataError erwartet")
+
+ self.assertIn("invalid JSON", "\n".join(lines))
+ self.assertIn("q quit", "\n".join(lines))
+
+ def test_empty_data(self):
+ text = "\n".join(tui.render_dashboard([], now=NOW, width=80))
+ self.assertIn("No samples yet", text)
+
+
+class OnceModeTest(unittest.TestCase):
+
+ def setUp(self):
+ self.tmp = tempfile.TemporaryDirectory()
+ self.addCleanup(self.tmp.cleanup)
+ self.path = os.path.join(self.tmp.name, "page_counts.json")
+ with open(self.path, "w", encoding="utf-8") as handle:
+ json.dump(sample_entries(), handle)
+
+ def _run(self, *extra):
+ return subprocess.run(
+ [sys.executable, os.path.join(HERE, "tui.py"), "--once",
+ "--file", self.path, "--width", "100"] + list(extra),
+ capture_output=True,
+ text=True,
+ cwd=ROOT,
+ )
+
+ def test_once_prints_dashboard(self):
+ result = self._run()
+ self.assertEqual(result.returncode, 0, result.stderr)
+ self.assertIn("Pages", result.stdout)
+ self.assertIn("981.950", result.stdout)
+ self.assertIn("q quit", result.stdout)
+
+ def test_once_window_flag(self):
+ result = self._run("--window", "all", "--ascii")
+ self.assertEqual(result.returncode, 0, result.stderr)
+ self.assertIn("all data", result.stdout)
+
+ def test_once_corrupt_file_exits_nonzero(self):
+ with open(self.path, "w", encoding="utf-8") as handle:
+ handle.write("{kaputt")
+ result = self._run()
+ self.assertEqual(result.returncode, 1)
+ self.assertIn("error:", result.stderr)
+
+ def test_once_missing_file_is_empty_state(self):
+ os.unlink(self.path)
+ result = self._run()
+ self.assertEqual(result.returncode, 0, result.stderr)
+ self.assertIn("No samples yet", result.stdout)
+
+
+if __name__ == "__main__":
+ unittest.main()
diff --git a/web/tui.py b/web/tui.py
new file mode 100644
index 0000000..7f31919
--- /dev/null
+++ b/web/tui.py
@@ -0,0 +1,640 @@
+#!/usr/bin/env python3
+"""
+Terminal-Dashboard für den Statserver.
+
+Liest dieselbe Datei wie `web/app.py` (`page_counts.json`) und zeigt
+Seitenzahl, Bildanzahl, Wachstum, einen Sparkline-Verlauf und die
+letzten Messwerte direkt im Terminal. Startet keinen Webserver und
+benötigt keine zusätzlichen Pakete (nur curses aus der Standardlib).
+
+Aufruf:
+ python3 web/tui.py # interaktiv (q beendet)
+ python3 web/tui.py --once # Momentaufnahme für Skripte/Logs
+ python3 web/tui.py --window 7d # 24h (Default) | 7d | 30d | all
+"""
+
+from __future__ import annotations
+
+import argparse
+import curses
+import json
+import locale
+import shutil
+import sys
+import time
+from datetime import datetime, timedelta
+from pathlib import Path
+
+DB_DATA = Path(__file__).parent / "page_counts.json"
+
+# Muss zu EXACT_IMAGE_INTERVAL in web/app.py passen (dort per Test
+# abgesichert): Bilder werden nur alle 2 Stunden exakt gezählt.
+EXACT_IMAGE_INTERVAL = 2 * 60 * 60 # Sekunden
+
+REFRESH_SECONDS = 2.0
+DEFAULT_WINDOW = "24h"
+WINDOWS = {
+ "24h": timedelta(hours=24),
+ "7d": timedelta(days=7),
+ "30d": timedelta(days=30),
+ "all": None,
+}
+WINDOW_ORDER = ("24h", "7d", "30d", "all")
+WINDOW_LABELS = {
+ "24h": "last 24 h",
+ "7d": "last 7 days",
+ "30d": "last 30 days",
+ "all": "all data",
+}
+
+SPARK_BLOCKS = "▁▂▃▄▅▆▇█"
+ASCII_BLOCKS = ".:-=+*#%"
+
+TABLE_ROWS_ONCE = 15
+
+
+class StatsDataError(Exception):
+ """Die Statistikdatei fehlt nicht, ist aber nicht lesbar/parsebar."""
+
+
+# --------------------------------------------------------------------------
+# Daten
+# --------------------------------------------------------------------------
+
+def parse_timestamp(value):
+ """ISO-Zeitstempel -> naive lokale datetime (oder None)."""
+ if not isinstance(value, str) or not value.strip():
+ return None
+
+ text = value.strip()
+ if text.endswith("Z"):
+ text = text[:-1] + "+00:00"
+
+ try:
+ timestamp = datetime.fromisoformat(text)
+ except ValueError:
+ return None
+
+ if timestamp.tzinfo is not None:
+ timestamp = timestamp.astimezone().replace(tzinfo=None)
+
+ return timestamp
+
+
+def load_entries(path=DB_DATA):
+ """
+ Messwerte laden, sortiert nach Zeitstempel.
+
+ Akzeptiert sowohl die Liste aus `page_counts.json` als auch das
+ Format der API (`{"values": [...]}`). Einträge ohne gültigen
+ Zeitstempel oder ohne Seitenzahl werden übersprungen.
+ """
+ path = Path(path)
+
+ try:
+ raw = json.loads(path.read_text(encoding="utf-8"))
+ except FileNotFoundError:
+ return []
+ except OSError as exc:
+ raise StatsDataError(f"{path}: {exc}") from exc
+ except json.JSONDecodeError as exc:
+ raise StatsDataError(
+ f"{path}: invalid JSON (line {exc.lineno}: {exc.msg})"
+ ) from exc
+
+ if isinstance(raw, dict):
+ raw = raw.get("values", [])
+
+ if not isinstance(raw, list):
+ raise StatsDataError(f"{path}: expected a JSON list of samples")
+
+ entries = []
+
+ for item in raw:
+ if not isinstance(item, dict):
+ continue
+
+ timestamp = parse_timestamp(item.get("timestamp"))
+ if timestamp is None:
+ continue
+
+ try:
+ count = int(item.get("count"))
+ except (TypeError, ValueError):
+ continue
+
+ entry = dict(item)
+ entry["timestamp"] = timestamp
+ entry["count"] = count
+
+ images = item.get("images")
+ if images is None:
+ entry["images"] = None
+ else:
+ try:
+ entry["images"] = int(images)
+ except (TypeError, ValueError):
+ entry["images"] = None
+
+ entries.append(entry)
+
+ entries.sort(key=lambda entry: entry["timestamp"])
+ return entries
+
+
+def select_window(entries, window, now=None):
+ """Nur Messwerte innerhalb des gewählten Zeitfensters."""
+ delta = WINDOWS.get(window, WINDOWS[DEFAULT_WINDOW])
+ if delta is None:
+ return list(entries)
+
+ now = now or datetime.now()
+ cutoff = now - delta
+ return [entry for entry in entries if entry["timestamp"] >= cutoff]
+
+
+def latest_images(entries):
+ """
+ (images, images_exact, timestamp) des neuesten Eintrags mit
+ Bildanzahl. Ältere Dateien haben kein `images`-Feld -> None.
+ """
+ for entry in reversed(entries):
+ if entry.get("images") is not None:
+ return (
+ entry["images"],
+ entry.get("images_exact") is True,
+ entry["timestamp"],
+ )
+ return None, False, None
+
+
+def next_exact_text(entries, now):
+ """Wann die nächste exakte Bildzählung fällig ist."""
+ last_exact = None
+
+ for entry in reversed(entries):
+ if entry.get("images_exact") is True:
+ last_exact = entry["timestamp"]
+ break
+
+ if last_exact is None:
+ return "no exact count yet"
+
+ remaining = EXACT_IMAGE_INTERVAL - (now - last_exact).total_seconds()
+ if remaining <= 0:
+ return "exact count due now"
+ return f"next exact in {fmt_duration(remaining)}"
+
+
+# --------------------------------------------------------------------------
+# Formatierung
+# --------------------------------------------------------------------------
+
+def fmt_int(value):
+ if value is None:
+ return "-"
+ return f"{int(value):,}".replace(",", ".")
+
+
+def fmt_delta(value):
+ if value is None:
+ return "-"
+ value = int(value)
+ if value == 0:
+ return "0"
+ if value > 0:
+ return "+" + fmt_int(value)
+ return "-" + fmt_int(abs(value))
+
+
+def fmt_duration(seconds):
+ seconds = max(0, int(round(seconds)))
+
+ if seconds < 60:
+ return f"{seconds} s"
+
+ minutes = seconds // 60
+ if minutes < 60:
+ return f"{minutes} min"
+
+ hours, minutes = divmod(minutes, 60)
+ if hours < 24:
+ return f"{hours} h {minutes} min"
+
+ days, hours = divmod(hours, 24)
+ return f"{days} d {hours} h"
+
+
+def relative_time(timestamp, now):
+ seconds = (now - timestamp).total_seconds()
+ if seconds < 5:
+ return "just now"
+ return f"{fmt_duration(seconds)} ago"
+
+
+def sparkline(values, width, blocks=SPARK_BLOCKS):
+ """Sparkline aus Zahlen; links mit Leerzeichen auf `width` aufgefüllt."""
+ if width <= 0 or not values:
+ return ""
+
+ values = list(values)[-width:]
+ low, high = min(values), max(values)
+
+ if high == low:
+ chars = [blocks[len(blocks) // 2]] * len(values)
+ else:
+ span = high - low
+ last = len(blocks) - 1
+ chars = [
+ blocks[min(last, int((value - low) / span * last + 0.5))]
+ for value in values
+ ]
+
+ return " " * (width - len(values)) + "".join(chars)
+
+
+def _fit(text, width):
+ return text[:max(0, width)]
+
+
+def _samples_label(count):
+ return f"{count} sample" if count == 1 else f"{count} samples"
+
+
+def _rate(window_entries):
+ """Wachstum in Seiten pro Stunde (None bei < 2 Messwerten)."""
+ if len(window_entries) < 2:
+ return None
+
+ first, last = window_entries[0], window_entries[-1]
+ hours = (last["timestamp"] - first["timestamp"]).total_seconds() / 3600
+
+ if hours <= 0:
+ return None
+
+ return (last["count"] - first["count"]) / hours
+
+
+def _deltas(entries):
+ """Änderung der Seitenzahl gegenüber dem vorherigen Messwert."""
+ result = {}
+ previous = None
+
+ for entry in entries:
+ result[id(entry)] = (
+ None if previous is None else entry["count"] - previous
+ )
+ previous = entry["count"]
+
+ return result
+
+
+def _time_label(timestamp, now, compact):
+ if compact:
+ return timestamp.strftime("%H:%M")
+ if timestamp.date() == now.date():
+ return timestamp.strftime("%H:%M")
+ return timestamp.strftime("%d.%m. %H:%M")
+
+
+def _table_line(time_text, pages, delta, images, flag, compact):
+ if compact:
+ return (
+ f" {time_text:<6} {pages:>10} {delta:>8}"
+ f" {images:>6} {flag}"
+ )
+ return (
+ f" {time_text:<13} {pages:>12} {delta:>9}"
+ f" {images:>10} {flag}"
+ )
+
+
+def _build_table(window_entries, delta_by_id, now, width, limit):
+ """Neueste Messwerte zuerst; Kopfzeile als erste Zeile."""
+ compact = width < 62
+ header = _table_line(
+ "Time",
+ "Pages",
+ "Chg" if compact else "Change",
+ "Images",
+ "",
+ compact,
+ )
+
+ rows = []
+ newest_first = list(reversed(window_entries))
+
+ for entry in newest_first[:limit]:
+ flag = ""
+ if entry.get("images") is not None:
+ flag = "exact" if entry.get("images_exact") is True else "est."
+
+ rows.append(_table_line(
+ _time_label(entry["timestamp"], now, compact),
+ fmt_int(entry["count"]),
+ fmt_delta(delta_by_id.get(id(entry))),
+ fmt_int(entry.get("images")),
+ flag,
+ compact,
+ ))
+
+ return header, rows
+
+
+# --------------------------------------------------------------------------
+# Darstellung
+# --------------------------------------------------------------------------
+
+def render_dashboard(entries, window=DEFAULT_WINDOW, now=None, width=80,
+ rows=None, source=None, error=None,
+ blocks=SPARK_BLOCKS):
+ """
+ Dashboard als Liste von Textzeilen.
+
+ Reine Funktion ohne curses: dadurch testbar und auch für `--once`
+ bzw. Umleitungen nutzbar. `rows` begrenzt die Höhe (Terminal);
+ ohne Angabe werden bis zu 15 Messwerte angezeigt.
+ """
+ now = now or datetime.now()
+ width = max(24, int(width))
+ source = str(source or DB_DATA)
+
+ lines = [_fit(" JonasSearch stats", width)]
+
+ if error:
+ lines.append("")
+ lines.append(_fit(f" ! {error}", width))
+ lines.append("")
+ lines.append(_fit(" q quit | r retry", width))
+ return lines[:rows] if rows else lines
+
+ if not entries:
+ lines.append("")
+ lines.append(_fit(" No samples yet.", width))
+ lines.append(_fit(f" Waiting for the collector to write {source}", width))
+ lines.append("")
+ lines.append(_fit(" q quit | r reload", width))
+ return lines[:rows] if rows else lines
+
+ latest = entries[-1]
+ previous = entries[-2] if len(entries) > 1 else None
+ delta = None if previous is None else latest["count"] - previous["count"]
+
+ window_entries = select_window(entries, window, now)
+ window_note = ""
+ if not window_entries:
+ window_entries = [latest]
+ window_note = " (no recent samples)"
+
+ rate = _rate(window_entries)
+ images, images_exact, images_ts = latest_images(entries)
+
+ lines.append(_fit(
+ f" updated {latest['timestamp']:%d.%m.%Y %H:%M:%S}"
+ f" ({relative_time(latest['timestamp'], now)})",
+ width,
+ ))
+ lines.append("")
+
+ lines.append(_fit(f" Pages {fmt_int(latest['count'])}", width))
+
+ growth = [fmt_delta(delta) + " since previous" if delta is not None
+ else "first sample"]
+ if rate is not None:
+ growth.append(
+ f"{fmt_delta(round(rate))}/h over {WINDOW_LABELS[window]}"
+ f" ({_samples_label(len(window_entries))})"
+ )
+ lines.append(_fit(" " + " | ".join(growth), width))
+ lines.append("")
+
+ lines.append(_fit(f" Images {fmt_int(images)}", width))
+
+ if images is None:
+ images_detail = "not counted yet"
+ else:
+ kind = "exact" if images_exact else "estimated"
+ images_detail = (
+ f"{kind} | counted {images_ts:%H:%M} | exact every 2 h"
+ f" | {next_exact_text(entries, now)}"
+ )
+ lines.append(_fit(" " + images_detail, width))
+ lines.append("")
+
+ values = [entry["count"] for entry in window_entries]
+ low, high = min(values), max(values)
+ lines.append(_fit(
+ f" Pages | {WINDOW_LABELS[window]}{window_note}"
+ f" | {_samples_label(len(window_entries))}",
+ width,
+ ))
+
+ range_text = f" ({fmt_int(low)} - {fmt_int(high)})"
+ spark_width = max(8, min(64, width - len(range_text) - 2))
+ spark = sparkline(values, spark_width, blocks)
+ lines.append(_fit(" " + spark + range_text, width))
+
+ delta_by_id = _deltas(entries)
+ footer = ["", _fit(
+ f" q quit | r reload | t window ({window}) | file: {source}",
+ width,
+ )]
+
+ if rows is None:
+ limit = TABLE_ROWS_ONCE
+ else:
+ # 3 Zeilen Overhead: Leerzeile, "Recent samples", Tabellenkopf
+ limit = max(0, rows - len(lines) - len(footer) - 3)
+
+ if limit > 0:
+ header, table = _build_table(
+ window_entries, delta_by_id, now, width, limit
+ )
+ lines += ["", _fit(" Recent samples", width), _fit(header, width)]
+ lines += [_fit(row, width) for row in table]
+
+ lines += footer
+
+ if rows is not None and len(lines) > rows:
+ # Sehr kleines Terminal: Kerninfo kürzen, Footer unten halten.
+ lines = lines[:max(0, rows - 1)] + [lines[-1]]
+
+ return lines
+
+
+# --------------------------------------------------------------------------
+# Interaktiv / einmalig
+# --------------------------------------------------------------------------
+
+def _blocks_for(args):
+ if args.ascii:
+ return ASCII_BLOCKS
+
+ encoding = (getattr(sys.stdout, "encoding", None) or "").lower()
+ return SPARK_BLOCKS if "utf" in encoding else ASCII_BLOCKS
+
+
+def _render(args, entries, now, width, rows=None):
+ return render_dashboard(
+ entries,
+ window=args.window,
+ now=now,
+ width=width,
+ rows=rows,
+ source=args.file,
+ blocks=_blocks_for(args),
+ )
+
+
+def emit_once(args, stream=None, errors=None):
+ """Momentaufnahme ausgeben (kein curses). Rückgabe: Exit-Code."""
+ stream = stream or sys.stdout
+ errors = errors or sys.stderr
+
+ try:
+ entries = load_entries(args.file)
+ except StatsDataError as exc:
+ errors.write(f"error: {exc}\n")
+ return 1
+
+ width = args.width or shutil.get_terminal_size((80, 24)).columns
+ lines = _render(args, entries, datetime.now(), width)
+ stream.write("\n".join(lines) + "\n")
+ return 0
+
+
+def next_window(window):
+ try:
+ index = WINDOW_ORDER.index(window)
+ except ValueError:
+ index = 0
+ return WINDOW_ORDER[(index + 1) % len(WINDOW_ORDER)]
+
+
+def run_interactive(args):
+ try:
+ curses.wrapper(_interactive_loop, args)
+ except KeyboardInterrupt:
+ pass
+ except curses.error as exc:
+ print(
+ f"Terminal-Dashboard nicht möglich ({exc}); "
+ f"gebe Momentaufnahme aus.",
+ file=sys.stderr,
+ )
+ return emit_once(args)
+
+ return 0
+
+
+def _interactive_loop(stdscr, args):
+ try:
+ curses.curs_set(0)
+ except curses.error:
+ pass
+
+ stdscr.timeout(max(50, int(args.interval * 1000)))
+
+ entries = []
+ error = None
+ loaded_at = 0.0
+ window = args.window
+
+ while True:
+ monotonic_now = time.monotonic()
+
+ if monotonic_now - loaded_at >= args.interval:
+ try:
+ entries = load_entries(args.file)
+ error = None
+ except StatsDataError as exc:
+ error = str(exc)
+ loaded_at = monotonic_now
+
+ height, width = stdscr.getmaxyx()
+ lines = render_dashboard(
+ entries,
+ window=window,
+ now=datetime.now(),
+ width=width,
+ rows=height,
+ source=args.file,
+ error=error,
+ blocks=_blocks_for(args),
+ )
+
+ stdscr.erase()
+ for index, line in enumerate(lines[:height]):
+ attributes = curses.A_BOLD if index == 0 else 0
+ try:
+ stdscr.addnstr(index, 0, line, max(0, width - 1), attributes)
+ except curses.error:
+ pass
+ stdscr.refresh()
+
+ key = stdscr.getch()
+
+ if key in (ord("q"), 27):
+ return
+ if key == ord("r"):
+ loaded_at = 0.0
+ elif key == ord("t"):
+ window = next_window(window)
+
+
+def main(argv=None):
+ parser = argparse.ArgumentParser(
+ description=(
+ "Terminal-Dashboard für den Statserver "
+ "(liest web/page_counts.json)."
+ ),
+ )
+ parser.add_argument(
+ "--file",
+ default=str(DB_DATA),
+ help=f"Statistikdatei (Default: {DB_DATA})",
+ )
+ parser.add_argument(
+ "--window",
+ choices=WINDOW_ORDER,
+ default=DEFAULT_WINDOW,
+ help="Zeitfenster für Verlauf und Tabelle (Default: 24h)",
+ )
+ parser.add_argument(
+ "--once",
+ action="store_true",
+ help="nur einmal ausgeben und beenden",
+ )
+ parser.add_argument(
+ "--width",
+ type=int,
+ default=None,
+ help="Breite für --once (Default: Terminalbreite)",
+ )
+ parser.add_argument(
+ "--interval",
+ type=float,
+ default=REFRESH_SECONDS,
+ help=f"Reload-Intervall in Sekunden (Default: {REFRESH_SECONDS})",
+ )
+ parser.add_argument(
+ "--ascii",
+ action="store_true",
+ help="ASCII-Sparkline statt Unicode-Blöcke",
+ )
+
+ args = parser.parse_args(argv)
+
+ try:
+ locale.setlocale(locale.LC_ALL, "")
+ except locale.Error:
+ pass
+
+ if args.once or not (sys.stdin.isatty() and sys.stdout.isatty()):
+ return emit_once(args)
+
+ return run_interactive(args)
+
+
+if __name__ == "__main__":
+ sys.exit(main())