105 lines
4.1 KiB
Python
105 lines
4.1 KiB
Python
#!/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
|
|
import re
|
|
|
|
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"
|
|
try:
|
|
for number, line in enumerate(env_file.read_text(encoding="utf-8-sig").splitlines(), 1):
|
|
line = re.sub(r'^\s*export\s+', '', 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())
|