From 2e0b91243d4b3547eb3ec67ddd0835b9711a1e77 Mon Sep 17 00:00:00 2001 From: Florian Zumpe Date: Sat, 3 Oct 2026 11:57:27 +0200 Subject: [PATCH] add route moderation script --- package.json | 3 +- test/python/test_moderate_routes.py | 106 ++++++++++++++++++++++++++++ tools/python/README.md | 36 ++++++++++ tools/python/moderate_routes.py | 103 +++++++++++++++++++++++++++ 4 files changed, 247 insertions(+), 1 deletion(-) create mode 100644 test/python/test_moderate_routes.py create mode 100644 tools/python/moderate_routes.py diff --git a/package.json b/package.json index dc78649..2590bd0 100644 --- a/package.json +++ b/package.json @@ -19,7 +19,8 @@ "test:coverage": "npm run test:coverage:js && npm run test:coverage:python", "test:python": "python3 -m unittest discover -s test/python -p 'test_*.py'", "test:coverage:js": "c8 --exclude='test/**' --exclude='test-support/**' --reporter=text-summary --reporter=lcov --reports-dir=coverage/js node --test", - "test:coverage:python": "python3 test-support/python_coverage_runner.py" + "test:coverage:python": "python3 test-support/python_coverage_runner.py", + "moderate": "python3 tools/python/moderate_routes.py" }, "dependencies": { "@file-type/av": "0.2.0", diff --git a/test/python/test_moderate_routes.py b/test/python/test_moderate_routes.py new file mode 100644 index 0000000..759cbe6 --- /dev/null +++ b/test/python/test_moderate_routes.py @@ -0,0 +1,106 @@ +"""Tests für lokale Freigaben mit echter SQLite-Datenbank.""" + +from contextlib import redirect_stdout, redirect_stderr +from datetime import datetime, timezone +import io +from pathlib import Path +import sqlite3 +import sys +import tempfile +import unittest +from unittest.mock import patch + +ROOT = Path(__file__).resolve().parents[2] +sys.path.insert(0, str(ROOT / "tools" / "python")) +import moderate_routes + + +class ModerationTests(unittest.TestCase): + def setUp(self): + self.temporary = tempfile.TemporaryDirectory() + self.addCleanup(self.temporary.cleanup) + self.root = Path(self.temporary.name) + self.environment = patch.dict("os.environ", {}, clear=True) + self.environment.start() + self.addCleanup(self.environment.stop) + self.path = self.root / "data" / "wegwichtel.sqlite" + self.path.parent.mkdir() + with sqlite3.connect(self.path) as database: + database.executescript((ROOT / "database" / "schema.sql").read_text()) + database.execute("INSERT INTO routes (name, slug, gpx_path) VALUES (?, ?, ?)", + ("Teststrecke", "test", "test.gpx")) + database.execute("INSERT INTO routes (name, slug, gpx_path, status) VALUES (?, ?, ?, ?)", + ("Gelöschte Strecke", "deleted", "deleted.gpx", "deleted")) + self.out = io.StringIO() + self.error = io.StringIO() + + def run_cli(self, args): + with redirect_stdout(self.out), redirect_stderr(self.error): + return moderate_routes.main(args, self.root) + + def activation(self, identifier=1): + with sqlite3.connect(self.path) as database: + return database.execute("SELECT activated_at FROM routes WHERE id = ?", (identifier,)).fetchone()[0] + + def test_listing_enable_and_disable(self): + self.assertEqual(self.run_cli([]), 0) + self.assertIn("1\tactive\tnicht freigegeben\tTeststrecke", self.out.getvalue()) + before = datetime.now(timezone.utc) + self.assertEqual(self.run_cli(["-e", "1"]), 0) + stamp = self.activation() + self.assertTrue(stamp.endswith("Z")) + self.assertGreaterEqual(datetime.fromisoformat(stamp.replace("Z", "+00:00")), + before.replace(microsecond=before.microsecond // 1000 * 1000)) + self.assertEqual(self.run_cli([]), 0) + self.assertIn(stamp, self.out.getvalue()) + self.assertEqual(self.run_cli(["--disable", "1"]), 0) + self.assertIsNone(self.activation()) + self.assertIsNone(self.activation(2)) + + def test_aliases(self): + for flag in ["--enable", "--enanble"]: + self.assertEqual(self.run_cli([flag, "1"]), 0) + self.assertIsNotNone(self.activation()) + self.assertEqual(self.run_cli(["-d", "1"]), 0) + self.assertIsNone(self.activation()) + + def test_missing_and_deleted_routes_do_not_change_database(self): + self.assertEqual(self.run_cli(["-e", "999"]), 1) + self.assertEqual(self.run_cli(["-e", "2"]), 1) + self.assertIsNone(self.activation()) + self.assertIsNone(self.activation(2)) + self.assertIn("existiert nicht", self.error.getvalue()) + self.assertIn("gelöscht", self.error.getvalue()) + + def test_env_and_environment_precedence(self): + (self.root / ".env").write_text('# Kommentar\nexport DATA_DIR="anderer Ordner" # Kommentar\n', encoding="utf-8") + self.assertEqual(moderate_routes.database_path(self.root), self.root / "anderer Ordner" / "wegwichtel.sqlite") + with patch.dict("os.environ", {"DATA_DIR": str(self.path.parent)}): + self.assertEqual(self.run_cli([]), 0) + (self.root / ".env").write_text("DATA_DIR=\n", encoding="utf-8") + self.assertEqual(moderate_routes.database_path(self.root), self.path) + + def test_missing_database_is_not_created(self): + self.path.unlink() + for args in [[], ["-e", "1"]]: + self.assertEqual(self.run_cli(args), 1) + self.assertFalse(self.path.exists()) + + def test_bad_env_reports_line_without_secrets(self): + for value in ["secret-without-equals", 'DATA_DIR="secret']: + (self.root / ".env").write_text(value, encoding="utf-8") + self.assertEqual(self.run_cli([]), 1) + self.assertNotIn("secret", self.error.getvalue()) + self.assertIn("Zeile 1", self.error.getvalue()) + + def test_empty_database(self): + with sqlite3.connect(self.path) as database: + database.execute("DELETE FROM routes") + self.assertEqual(self.run_cli([]), 0) + self.assertIn("Keine Strecken", self.out.getvalue()) + + def test_invalid_arguments(self): + for args in [["-e", "0"], ["-d", "abc"], ["-e", "1", "-d", "1"], ["-e"]]: + with redirect_stderr(self.error), self.assertRaises(SystemExit) as result: + moderate_routes.main(args, self.root) + self.assertEqual(result.exception.code, 2) diff --git a/tools/python/README.md b/tools/python/README.md index 1cd005e..2e509e7 100644 --- a/tools/python/README.md +++ b/tools/python/README.md @@ -17,6 +17,42 @@ Die API-Basisadresse ist standardmäßig `https://wegwichtel.de`. Alle Skripte u ## Routen +### Lokale Moderation auf dem Server + +`moderate_routes.py` greift direkt auf die bestehende SQLite-Datenbank zu und +benötigt keine API-Verbindung. Ohne Parameter zeigt es alle Strecken mit ID, +Name, Status und Freigabezeitpunkt an: + +```bash +python3 tools/python/moderate_routes.py +python3 tools/python/moderate_routes.py --enable 12 +python3 tools/python/moderate_routes.py --disable 12 +``` + +Alternativ aus dem Projektverzeichnis über npm: + +```bash +npm run moderate +npm run moderate -- -e 12 +npm run moderate -- -d 12 +``` + +Das `--` trennt die npm-Optionen von den Parametern des Moderationsskripts. + +Kurzformen: `-e 12` und `-d 12`. `--enanble` wird als Alias für `--enable` +unterstützt. Freigeben setzt `activated_at` auf den aktuellen UTC-Zeitstempel; +Zurückziehen setzt es auf `NULL`. Gelöschte Strecken werden nicht freigegeben. + +Das Skript liest `.env` im Projektverzeichnis, unabhängig vom Arbeitsverzeichnis. +`DATA_DIR` bestimmt das Datenverzeichnis, standardmäßig `data`; darin liegt +`wegwichtel.sqlite`. Bereits gesetzte Umgebungsvariablen haben Vorrang wie beim +App-Start. Die `.env` unterstützt einzeilige Werte, Anführungszeichen, Kommentare +und `export`; sie wird nicht als Shell-Skript ausgeführt. Eine fehlende Datenbank +wird nicht neu angelegt. Der aufrufende Benutzer benötigt die entsprechenden +Dateirechte für Datenbank und Datenverzeichnis. + +### Bearbeitung über die API + ```bash python3 tools/python/create_route.py python3 tools/python/update_route.py diff --git a/tools/python/moderate_routes.py b/tools/python/moderate_routes.py new file mode 100644 index 0000000..f97dd60 --- /dev/null +++ b/tools/python/moderate_routes.py @@ -0,0 +1,103 @@ +#!/usr/bin/env python3 +"""Strecken direkt in der lokalen Wegwichtel-Datenbank moderieren.""" + +import argparse +from datetime import datetime, timezone +import os +from pathlib import Path +import shlex +import sqlite3 +import sys + +PROJECT_ROOT = Path(__file__).resolve().parents[2] + + +def database_path(project_root): + """Liest die .env; bereits gesetzte Umgebungsvariablen haben Vorrang.""" + settings = {} + env_file = project_root / ".env" + if env_file.is_file(): + for number, line in enumerate(env_file.read_text(encoding="utf-8-sig").splitlines(), 1): + line = line.strip() + if not line or line.startswith("#"): + continue + if line.startswith("export "): + line = line[7:].lstrip() + key, separator, value = line.partition("=") + if not separator: + raise ValueError(f"Ungültige .env-Angabe in Zeile {number}.") + try: + settings[key.strip()] = " ".join(shlex.split(value, comments=True)) + except ValueError as error: + raise ValueError(f"Ungültige .env-Anführungszeichen in Zeile {number}.") from error + data_dir = os.environ.get("DATA_DIR", settings.get("DATA_DIR", "")) or "data" + return (project_root / data_dir / "wegwichtel.sqlite").resolve() + + +def route_id(value): + result = int(value) + if result < 1: + raise argparse.ArgumentTypeError("Die Strecken-ID muss eine positive Zahl sein.") + return result + + +def arguments(argv): + parser = argparse.ArgumentParser(description=__doc__) + group = parser.add_mutually_exclusive_group() + group.add_argument("--enable", "--enanble", "-e", type=route_id, metavar="ID", + help="Strecke mit dem aktuellen Zeitstempel freigeben") + group.add_argument("--disable", "-d", type=route_id, metavar="ID", + help="Freigabe zurückziehen (activated_at wird NULL)") + return parser.parse_args(argv) + + +def list_routes(database): + rows = database.execute("SELECT id, name, status, activated_at FROM routes ORDER BY id").fetchall() + if not rows: + print("Keine Strecken vorhanden.") + return + print("ID\tStatus\tFreigegeben am (UTC)\tName") + for row in rows: + name = row["name"].replace("\n", " ").replace("\r", " ").replace("\t", " ") + print(f'{row["id"]}\t{row["status"]}\t{row["activated_at"] or "nicht freigegeben"}\t{name}') + + +def moderate(database, identifier, enable): + activated_at = datetime.now(timezone.utc).isoformat(timespec="milliseconds").replace("+00:00", "Z") if enable else None + with database: + database.execute("BEGIN IMMEDIATE") + row = database.execute("SELECT status FROM routes WHERE id = ?", (identifier,)).fetchone() + if row is None: + raise ValueError(f"Strecke {identifier} existiert nicht.") + if enable and row["status"] != "active": + raise ValueError(f"Strecke {identifier} ist gelöscht und kann nicht freigegeben werden.") + database.execute("UPDATE routes SET activated_at = ? WHERE id = ?", (activated_at, identifier)) + if enable: + print(f"Strecke {identifier} freigegeben: {activated_at}") + else: + print(f"Freigabe von Strecke {identifier} zurückgezogen.") + + +def main(argv=None, project_root=None): + args = arguments(argv) + try: + path = database_path(project_root or PROJECT_ROOT) + mode = "ro" if args.enable is None and args.disable is None else "rw" + database = sqlite3.connect(f"{path.as_uri()}?mode={mode}", uri=True, timeout=10) + try: + database.row_factory = sqlite3.Row + if mode == "ro": + list_routes(database) + else: + identifier = args.enable if args.enable is not None else args.disable + moderate(database, identifier, args.enable is not None) + finally: + database.close() + except (OSError, ValueError, sqlite3.Error) as error: + print(f"Moderation fehlgeschlagen: {error}", file=sys.stderr) + return 1 + return 0 + + +if __name__ == "__main__": + raise SystemExit(main())