161 lines
6.5 KiB
Python
161 lines
6.5 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 re
|
||
import shlex
|
||
import sqlite3
|
||
import sys
|
||
from urllib.parse import urlsplit
|
||
|
||
PROJECT_ROOT = Path(__file__).resolve().parents[2]
|
||
|
||
|
||
def environment_settings(project_root):
|
||
"""Liest die .env, ohne die Prozessumgebung zu verändern."""
|
||
settings = {}
|
||
env_file = project_root / ".env"
|
||
try:
|
||
contents = env_file.read_text(encoding="utf-8-sig")
|
||
except FileNotFoundError:
|
||
contents = ""
|
||
for number, line in enumerate(contents.splitlines(), 1):
|
||
line = re.sub(r'^\s*export\s+', '', line).strip()
|
||
if not line or line.startswith("#"):
|
||
continue
|
||
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
|
||
return settings
|
||
|
||
|
||
def database_path(project_root):
|
||
"""Bereits gesetzte Umgebungsvariablen haben Vorrang vor der .env."""
|
||
settings = environment_settings(project_root)
|
||
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 review_settings(project_root):
|
||
settings = environment_settings(project_root)
|
||
url = os.environ.get("MODERATION_REVIEW_URL", settings.get("MODERATION_REVIEW_URL", ""))
|
||
parsed = urlsplit(url)
|
||
if url and (parsed.scheme != "https" or not parsed.netloc or parsed.fragment or parsed.query
|
||
or parsed.username or parsed.password):
|
||
raise ValueError("MODERATION_REVIEW_URL muss eine HTTPS-Adresse ohne Zugangsdaten, Query oder Fragment sein.")
|
||
return url.rstrip("/")
|
||
|
||
|
||
def table_text(value):
|
||
plain = re.sub(r'[\x00-\x1f\x7f-\x9f]', ' ', str(value or ""))
|
||
return " ".join(plain.split())
|
||
|
||
|
||
def print_table(headers, rows):
|
||
cells = [[table_text(value) for value in row] for row in [headers, *rows]]
|
||
widths = [max(len(row[index]) for row in cells) for index in range(len(headers))]
|
||
for row in cells:
|
||
print(" | ".join(value.ljust(width) for value, width in zip(row, widths)))
|
||
|
||
|
||
def show_pois(database, identifier, review_url, active):
|
||
rows = database.execute("""SELECT id, sequence, title, description, created_at, audio_path
|
||
FROM pois WHERE route_id=? ORDER BY sequence, id""", (identifier,)).fetchall()
|
||
print("Stationen:")
|
||
if not rows:
|
||
print("Keine Stationen vorhanden.")
|
||
return
|
||
details = []
|
||
for number, poi in enumerate(rows, 1):
|
||
access = []
|
||
if review_url and active:
|
||
base = f"{review_url}/routes/{identifier}/pois/{poi['id']}"
|
||
images = database.execute("SELECT id FROM poi_images WHERE poi_id=? ORDER BY sequence, id",
|
||
(poi["id"],)).fetchall()
|
||
access.extend(f"{base}/pictures/{image['id']}" for image in images)
|
||
if poi["audio_path"]:
|
||
access.append(f"{base}/audio")
|
||
description = poi["description"] or poi["title"]
|
||
details.append([poi["id"], number, description, poi["created_at"] or "–", " ; ".join(access) or "–"])
|
||
print_table(["ID", "Platz", "Beschreibung", "Erstellt am (UTC)", "Bilder / Ton (Anmeldung)"], details)
|
||
|
||
|
||
def list_routes(database, review_url=""):
|
||
rows = database.execute("SELECT id, name, status, activated_at, creator_id FROM routes ORDER BY id").fetchall()
|
||
if not rows:
|
||
print("Keine Strecken vorhanden.")
|
||
return
|
||
print("ID\tStatus\tFreigegeben am (UTC)\tName\tErsteller-ID")
|
||
for row in rows:
|
||
name = table_text(row["name"])
|
||
print(f'{row["id"]}\t{row["status"]}\t{row["activated_at"] or "nicht freigegeben"}\t{name}\t{row["creator_id"] or "–"}')
|
||
show_pois(database, row["id"], review_url, row["status"] == "active")
|
||
print()
|
||
|
||
|
||
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)
|
||
listing = args.enable is None and args.disable is None
|
||
review_url = review_settings(project_root or PROJECT_ROOT) if listing else ""
|
||
mode = "ro" if listing else "rw"
|
||
database = sqlite3.connect(f"{path.as_uri()}?mode={mode}", uri=True, timeout=10)
|
||
try:
|
||
database.row_factory = sqlite3.Row
|
||
if listing:
|
||
list_routes(database, review_url)
|
||
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())
|