eventmap/scripts/add_places_from_ai.py

#!/usr/bin/env python3
"""Fügt neue Ausflugsziele aus einem KI-Import-JSON (Format wie in
anforderung.md "Vorgabe für das KI-JSON") zur eventmap_seed.json auf dem
Server hinzu (per SMB-Mount erreichbar, Default: gvfs-Freigabe unter
smb://172.16.1.4/web).

Eingabe wahlweise per Datei oder Zwischenablage:
    python3 add_places_from_ai.py                 # liest aus der Zwischenablage
    python3 add_places_from_ai.py --file ki.json   # liest aus einer Datei
    python3 add_places_from_ai.py --file - < ki.json

Duplikat-Check: Name + Ort (case-insensitive, whitespace-normalisiert),
sowohl gegen bestehende Einträge als auch innerhalb der neuen Eingabe.

Vor jedem Schreibvorgang wird ein Backup der Zieldatei angelegt
(eventmap_seed.json.bak-<timestamp>), da es sich um eine gemeinsam
genutzte Datei auf dem Server handelt.
"""

import argparse
import datetime
import json
import re
import subprocess
import sys
import uuid
from pathlib import Path

DEFAULT_TARGET = (
    "/run/user/1000/gvfs/smb-share:server=172.16.1.4,share=web/eventmap_seed.json"
)
DEFAULT_SMB_URL = "smb://172.16.1.4/web"

# Muss mit js/config/categories.js in Sync bleiben.
CATEGORIES = [
    ("naturpark", "Naturpark"),
    ("wald-wanderweg", "Wald/Wanderweg"),
    ("see-badestelle", "See/Badestelle"),
    ("erlebnispark", "Erlebnispark"),
    ("freizeitpark", "Freizeitpark"),
    ("tierpark-zoo", "Tierpark/Zoo"),
    ("spielplatz", "Spielplatz"),
    ("museum-kinder", "Museum für Kinder"),
    ("indoor-spielplatz", "Indoor-Spielplatz"),
    ("bauernhof", "Bauernhof"),
    ("kletterpark", "Kletterpark"),
    ("aussichtspunkt", "Aussichtspunkt"),
    ("cafe-restaurant", "Familienfreundliches Café/Restaurant"),
    ("sonstige-aktivitaet", "Sonstige kinderfreundliche Aktivität"),
    ("sonstige", "Sonstige"),
]
CATEGORY_IDS = {cid for cid, _ in CATEGORIES}
CATEGORY_ID_BY_LABEL = {label.lower(): cid for cid, label in CATEGORIES}


def resolve_category(raw):
    if isinstance(raw, str):
        if raw in CATEGORY_IDS:
            return raw
        return CATEGORY_ID_BY_LABEL.get(raw.strip().lower(), "sonstige")
    return "sonstige"


def default_place_fields():
    return {
        "name": "",
        "category": "sonstige",
        "shortDescription": "",
        "description": "",
        "latitude": None,
        "longitude": None,
        "address": "",
        "city": "",
        "region": "",
        "ageRecommendation": {"min": None, "max": None},
        "cost": {"type": "unknown", "description": ""},
        "openingHours": "",
        "durationMinutes": None,
        "facilities": {
            "strollerAccessible": False,
            "wheelchairAccessible": False,
            "toilets": False,
            "parking": False,
            "restaurant": False,
        },
        "environment": "outdoor",
        "website": "",
        "image": "",
        "rating": None,
        "tags": [],
        "source": "",
        "sourceVerified": False,
    }


def iso_now():
    now = datetime.datetime.now(datetime.timezone.utc)
    return now.strftime("%Y-%m-%dT%H:%M:%S.") + f"{now.microsecond // 1000:03d}Z"


def normalize_place(raw):
    defaults = default_place_fields()
    now_iso = iso_now()
    return {
        "id": raw.get("id") or str(uuid.uuid4()),
        "name": raw.get("name") or defaults["name"],
        "category": resolve_category(raw.get("category")),
        "shortDescription": raw.get("shortDescription") or defaults["shortDescription"],
        "description": raw.get("description") or defaults["description"],
        "latitude": raw.get("latitude", defaults["latitude"]),
        "longitude": raw.get("longitude", defaults["longitude"]),
        "address": raw.get("address") or defaults["address"],
        "city": raw.get("city") or defaults["city"],
        "region": raw.get("region") or defaults["region"],
        "ageRecommendation": {**defaults["ageRecommendation"], **(raw.get("ageRecommendation") or {})},
        "cost": {**defaults["cost"], **(raw.get("cost") or {})},
        "openingHours": raw.get("openingHours") or defaults["openingHours"],
        "durationMinutes": raw.get("durationMinutes", defaults["durationMinutes"]),
        "facilities": {**defaults["facilities"], **(raw.get("facilities") or {})},
        "environment": raw.get("environment") or defaults["environment"],
        "website": raw.get("website") or defaults["website"],
        "image": raw.get("image") or defaults["image"],
        "rating": raw.get("rating", defaults["rating"]),
        "tags": list(raw["tags"]) if isinstance(raw.get("tags"), list) else [],
        "source": raw.get("source") or defaults["source"],
        "sourceVerified": bool(raw.get("sourceVerified", defaults["sourceVerified"])),
        "createdAt": raw.get("createdAt") or now_iso,
        "updatedAt": raw.get("updatedAt") or now_iso,
    }


def extract_json_text(text):
    text = text.strip()
    fence = re.match(r"^```(?:json)?\s*\n(.*)\n```$", text, re.DOTALL)
    return fence.group(1) if fence else text


def extract_candidates(data):
    if isinstance(data, dict) and isinstance(data.get("places"), list):
        return data["places"]
    if isinstance(data, list):
        return data
    if isinstance(data, dict) and "name" in data:
        return [data]
    return None


def read_input(file_arg):
    if file_arg is None:
        try:
            import pyperclip
        except ImportError:
            sys.exit(
                "pyperclip ist nicht installiert (für die Zwischenablage nötig).\n"
                "Installieren mit: pip install pyperclip\n"
                "Unter Linux zusätzlich ein Clipboard-Tool: 'xclip' oder 'xsel' (X11) "
                "bzw. 'wl-clipboard' (Wayland).\n"
                "Alternativ das JSON in eine Datei speichern und mit --file übergeben."
            )
        text = pyperclip.paste()
        if not text or not text.strip():
            sys.exit("Zwischenablage ist leer.")
        print("Lese KI-JSON aus der Zwischenablage ...")
        return text
    if file_arg == "-":
        print("Lese KI-JSON von stdin ...")
        return sys.stdin.read()
    print(f"Lese KI-JSON aus Datei: {file_arg}")
    return Path(file_arg).read_text(encoding="utf-8")


def ensure_mounted(target, smb_url, automount):
    parent = Path(target).parent
    if parent.is_dir():
        return
    if not automount:
        sys.exit(
            f"Fehler: {parent} existiert nicht. Bitte die SMB-Freigabe zuerst mounten "
            f"(z.B. im Dateimanager öffnen oder: gio mount {smb_url})"
        )
    print(f"SMB-Freigabe scheint nicht gemountet zu sein, versuche: gio mount {smb_url}")
    try:
        subprocess.run(["gio", "mount", smb_url], check=True, timeout=60)
    except (OSError, subprocess.SubprocessError) as err:
        sys.exit(f"Automatisches Mounten fehlgeschlagen ({err}).")
    if not parent.is_dir():
        sys.exit(f"Fehler: {parent} existiert immer noch nicht nach dem Mount-Versuch.")


def norm_key_part(value):
    return re.sub(r"\s+", " ", str(value or "")).strip().lower()


def dedup_key(place):
    return (norm_key_part(place["name"]), norm_key_part(place["city"]))


def main():
    parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
    parser.add_argument("--file", "-f", help="Pfad zur KI-JSON-Datei ('-' für stdin). Ohne Angabe: Zwischenablage.")
    parser.add_argument("--target", "-t", default=DEFAULT_TARGET, help=f"Pfad zur eventmap_seed.json (Default: {DEFAULT_TARGET})")
    parser.add_argument("--smb-url", default=DEFAULT_SMB_URL, help=f"SMB-URL zum Automount (Default: {DEFAULT_SMB_URL})")
    parser.add_argument("--no-automount", action="store_true", help="Nicht automatisch versuchen, die Freigabe zu mounten.")
    parser.add_argument("--no-backup", action="store_true", help="Kein Backup der Zieldatei vor dem Schreiben anlegen.")
    parser.add_argument("--dry-run", action="store_true", help="Nur anzeigen, was passieren würde, nichts schreiben.")
    args = parser.parse_args()

    raw_text = read_input(args.file)
    try:
        data = json.loads(extract_json_text(raw_text))
    except json.JSONDecodeError as err:
        sys.exit(f"Ungültiges JSON: {err}")

    candidates = extract_candidates(data)
    if not candidates:
        sys.exit('Kein "places"-Array bzw. keine Orte im JSON gefunden.')

    ensure_mounted(args.target, args.smb_url, automount=not args.no_automount)

    target_path = Path(args.target)
    if target_path.exists():
        existing = json.loads(target_path.read_text(encoding="utf-8"))
    else:
        print(f"Hinweis: {target_path} existiert noch nicht, starte mit leerer Liste.")
        existing = []

    key_owner = {dedup_key(p): p.get("name", "?") for p in existing}

    to_add = []
    skipped = []
    for raw in candidates:
        if not isinstance(raw, dict) or not str(raw.get("name") or "").strip():
            skipped.append(("(ohne Namen)", "kein Name angegeben"))
            continue
        place = normalize_place(raw)
        key = dedup_key(place)
        if key in key_owner:
            skipped.append((place["name"], f"Duplikat von {key_owner[key]!r}"))
            continue
        key_owner[key] = place["name"]
        to_add.append(place)

    print(f"Gefunden: {len(candidates)}, neu: {len(to_add)}, übersprungen: {len(skipped)}")
    for name, reason in skipped:
        print(f"  - übersprungen: {name} ({reason})")

    if not to_add:
        print("Nichts zu tun.")
        return

    print("Neu hinzuzufügen:")
    for p in to_add:
        print(f"  - {p['name']} ({p['category']}, {p['city'] or 'ohne Ort'})")

    if args.dry_run:
        print("--dry-run: Zieldatei wird nicht verändert.")
        return

    if target_path.exists() and not args.no_backup:
        backup_path = target_path.with_name(
            target_path.name + f".bak-{datetime.datetime.now().strftime('%Y%m%dT%H%M%S')}"
        )
        # Reines Byte-Kopieren statt shutil.copy2/copy: der gvfs-SMB-Mount
        # unterstuetzt kein chmod, das shutil beim Metadaten-Kopieren aufruft.
        backup_path.write_bytes(target_path.read_bytes())
        print(f"Backup angelegt: {backup_path}")

    merged = existing + to_add
    target_path.write_text(
        json.dumps(merged, indent=2, ensure_ascii=False) + "\n", encoding="utf-8"
    )
    print(f"{target_path} aktualisiert: {len(to_add)} neue Einträge hinzugefügt (gesamt: {len(merged)}).")


if __name__ == "__main__":
    main()