Wegwichtel/tools/python/moderate_routes.py

#!/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 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 = table_text(row["name"])
        print(f'{row["id"]}\t{row["status"]}\t{row["activated_at"] or "nicht freigegeben"}\t{name}')
        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())