""" sn_plan41/modules/vln_api_logic.py – Fachlogik für die VLN-API-Integration in Tab A. Kapselt: - Anmeldung / Abmeldung (KartenApiClient aus vln_karten) - Persistierung des API-Keys in QSettings - Verfahrensliste laden (GET /tgen) - Persistierung der VKZ-Auswahl in der Projektdatei - Plan 41 laden (2 Layer + Stile aus sn_plan41/assets/) - Aktiven Layer hochladen Voraussetzung: Das Plugin `vln_karten` muss installiert sein. Wenn nicht vorhanden, liefert :attr:`is_available` False; die UI kann dann einen entsprechenden Hinweis anzeigen. Sitzungsablauf bei AuthError: - :meth:`handle_session_expired` aufrufen → API-Key löschen + logout - In der UI danach :meth:`_update_vln_ui_state` aufrufen und QMessageBox.warning anzeigen. Kein automatischer Login-Dialog. """ from __future__ import annotations from typing import Optional from sn_basis.functions.qt_wrapper import QSettings from sn_basis.functions.qgiscore_wrapper import QgsProject from sn_basis.functions.sys_wrapper import get_plugin_root, join_path from sn_basis.functions.ly_style_wrapper import apply_style_from_path # Lokale API-Module — fallenübrig, sobald QGIS verfügbar ist. # In der reinen Python-Testumgebung (kein qgis-Paket) schlagen die # qgis-Imports in den Untermodulen fehl; daher der try/except-Guard. try: from sn_plan41.modules.vln_api_client import KartenApiClient, ApiError, AuthError from sn_plan41.modules.vln_layer_manager import ( feature_collection_to_layers, plugin_layers, layers_to_feature_collection, layer_api_path, PROP_DATASET, PROP_VERFAHREN, ) VLN_KARTEN_AVAILABLE = True except (ImportError, ModuleNotFoundError): KartenApiClient = None # type: ignore[assignment,misc] ApiError = Exception # type: ignore[assignment,misc] AuthError = Exception # type: ignore[assignment,misc] VLN_KARTEN_AVAILABLE = False # Geometrietyp-Konstanten (QGIS intern: 1 = Linie, 2 = Fläche) try: from qgis.core import Qgis as _Qgis _GEOM_LINE = _Qgis.GeometryType.Line _GEOM_POLYGON = _Qgis.GeometryType.Polygon except (ImportError, AttributeError): try: from qgis.core import QgsWkbTypes as _QgsWkbTypes _GEOM_LINE = _QgsWkbTypes.LineGeometry _GEOM_POLYGON = _QgsWkbTypes.PolygonGeometry except (ImportError, AttributeError): _GEOM_LINE = 1 _GEOM_POLYGON = 2 # Konstanten SETTINGS_GROUP = "vln_karten" PROJECT_SCOPE = "vln_karten" PROJECT_KEY_VKZ = "/vkz" P41_DATASET_KEY = "p41" P41_LABEL = "Plan 41 (Wege- und Gewässerplan)" P41_PATH_TEMPLATE = "/maps/p41/{vkz}" P41_GEOMETRY = "MultiPolygon" STYLE_FLAECHE = "QGIS_P41_API_Layer_flaeche.qml" STYLE_LINIE = "QGIS_P41_API_Layer_Linie.qml" class VlnApiLogic: """ Kapselt die VLN-API-Fachlogik für den Plan41-Tab. Alle Methoden, die Netzwerkkommunikation erfordern, können ``ApiError`` oder ``AuthError`` werfen – die UI ist für die Fehlerdarstellung zuständig. """ def __init__(self, pruefmanager=None) -> None: self.pruefmanager = pruefmanager self._client: Optional[object] = None # KartenApiClient | None # API-Client aus gespeicherten Credentials initialisieren if VLN_KARTEN_AVAILABLE: api_key, mail = self.load_stored_credentials() if api_key: self._client = KartenApiClient(api_key=api_key, mail=mail) # ------------------------------------------------------------------ # Verfügbarkeit # ------------------------------------------------------------------ @staticmethod def is_available() -> bool: """True, wenn die lokalen API-Module geladen werden konnten (erfordert eine QGIS-Laufzeitumgebung).""" return VLN_KARTEN_AVAILABLE # ------------------------------------------------------------------ # Authentifizierungsstatus # ------------------------------------------------------------------ @property def is_authenticated(self) -> bool: """True, wenn ein gültiger API-Key vorhanden ist.""" if self._client is None: return False return bool(getattr(self._client, "is_authenticated", False)) @property def mail(self) -> Optional[str]: """E-Mail-Adresse des angemeldeten Benutzers oder None.""" if self._client is None: return None return getattr(self._client, "mail", None) # ------------------------------------------------------------------ # QSettings: API-Key persistieren # ------------------------------------------------------------------ def load_stored_credentials(self) -> tuple: """Liest API-Key und Mail aus QSettings. Gibt ``(api_key, mail)`` zurück – leere Strings wenn nicht vorhanden. """ settings = QSettings() settings.beginGroup(SETTINGS_GROUP) api_key = settings.value("api_key", "") mail = settings.value("mail", "") settings.endGroup() return api_key or None, mail or None def save_api_key(self, api_key: str, mail: str) -> None: """Speichert API-Key und Mail in QSettings.""" settings = QSettings() settings.beginGroup(SETTINGS_GROUP) settings.setValue("api_key", api_key) settings.setValue("mail", mail or "") settings.endGroup() def clear_api_key(self) -> None: """Entfernt API-Key aus QSettings.""" settings = QSettings() settings.beginGroup(SETTINGS_GROUP) settings.remove("api_key") settings.endGroup() # ------------------------------------------------------------------ # Login / Logout # ------------------------------------------------------------------ def login(self, mail: str, password: str) -> None: """ Meldet den Benutzer an und speichert den API-Key. :raises ApiError: Bei ungültigen Zugangsdaten oder Netzwerkproblem. :raises RuntimeError: Wenn QGIS nicht verfügbar ist (kein qgis-Paket). """ if not VLN_KARTEN_AVAILABLE: raise RuntimeError( "VLN-API-Module nicht verfügbar (QGIS-Laufzeitumgebung benötigt)." ) client = KartenApiClient() client.login(mail, password) # wirft ApiError bei Fehler self._client = client self.save_api_key(client.api_key, mail) def logout(self) -> None: """Meldet den Benutzer ab (lokal, kein API-Aufruf).""" if self._client is not None: getattr(self._client, "logout", lambda: None)() self._client = None def handle_session_expired(self) -> None: """ Behandelt einen abgelaufenen API-Key: entfernt ihn aus QSettings und setzt den Client zurück. """ self.clear_api_key() self.logout() # ------------------------------------------------------------------ # Verfahrensliste # ------------------------------------------------------------------ def get_verfahren(self) -> list: """ Lädt alle Verfahren (TGs) vom Server. :returns: Sortierte Liste von ``{"vkz": ..., "name": ...}``-Dicts. :raises AuthError: Wenn nicht angemeldet oder Session abgelaufen. :raises ApiError: Bei sonstigem Netzwerkfehler. """ if self._client is None: raise AuthError("Nicht angemeldet.") return self._client.get_verfahren() # ------------------------------------------------------------------ # VKZ-Persistenz in Projektdatei # ------------------------------------------------------------------ def load_stored_vkz(self) -> Optional[str]: """Liest die zuletzt gewählte VKZ aus der Projektdatei.""" try: vkz, _ok = QgsProject.instance().readEntry( PROJECT_SCOPE, PROJECT_KEY_VKZ, "" ) return vkz if vkz else None except Exception: return None def save_vkz(self, vkz: Optional[str]) -> None: """Schreibt die gewählte VKZ in die Projektdatei.""" if not vkz: return try: QgsProject.instance().writeEntry(PROJECT_SCOPE, PROJECT_KEY_VKZ, vkz) except Exception: pass # ------------------------------------------------------------------ # Plan 41 laden # ------------------------------------------------------------------ def get_p41_layers(self, vkz: str) -> list: """Gibt alle bereits geladenen P41-Layer für diese VKZ zurück.""" return plugin_layers(P41_DATASET_KEY, vkz) def remove_p41_layers(self, vkz: str) -> None: """Entfernt alle P41-Layer dieser VKZ aus dem Projekt.""" existing = plugin_layers(P41_DATASET_KEY, vkz) if existing: QgsProject.instance().removeMapLayers([l.id() for l in existing]) def load_p41(self, vkz: str) -> list: """ Lädt den Plan-41-Datensatz für eine VKZ vom Server und legt Memory-Layer im Projekt an. Auf die entstandenen Layer werden die passenden QML-Stile aus ``sn_plan41/assets/`` angewendet. :returns: Liste der erzeugten ``QgsVectorLayer``. :raises AuthError: Wenn nicht angemeldet oder Session abgelaufen. :raises ApiError: Bei sonstigem Netzwerkfehler. :raises RuntimeError: Wenn QGIS nicht verfügbar ist. """ if not VLN_KARTEN_AVAILABLE: raise RuntimeError( "VLN-API-Module nicht verfügbar (QGIS-Laufzeitumgebung benötigt)." ) if self._client is None: raise AuthError("Nicht angemeldet.") path = P41_PATH_TEMPLATE.format(vkz=vkz) feature_collection = self._client.load_layer_complete( P41_DATASET_KEY, vkz ) base_name = "%s %s" % (P41_LABEL, vkz) layers = feature_collection_to_layers( feature_collection, base_name, P41_GEOMETRY, P41_DATASET_KEY, vkz, path, ) self._apply_p41_styles(layers) return layers def _apply_p41_styles(self, layers: list) -> None: """Wendet den passenden QML-Stil auf jeden Layer an.""" assets_dir = join_path(get_plugin_root(), "sn_plan41", "assets") for layer in layers: try: gtype = layer.geometryType() except Exception: continue if gtype == _GEOM_POLYGON: style_file = STYLE_FLAECHE elif gtype == _GEOM_LINE: style_file = STYLE_LINIE else: continue # Punkt- und None-Layer erhalten keinen Stil style_path = join_path(assets_dir, style_file) apply_style_from_path(layer, style_path) # ------------------------------------------------------------------ # Upload # ------------------------------------------------------------------ def upload_active_layer(self, active_layer) -> tuple: """ Lädt alle Teil-Layer des aktiven Layers zum Server hoch. :param active_layer: Aktiver ``QgsVectorLayer`` (aus ``iface.activeLayer()``). :returns: ``(success: bool, message: str)`` :raises AuthError: Wenn Session abgelaufen. :raises ApiError: Bei Netzwerkfehler. """ if not VLN_KARTEN_AVAILABLE: return False, ( "VLN-API-Module nicht verfügbar (QGIS-Laufzeitumgebung benötigt)." ) if self._client is None: raise AuthError("Nicht angemeldet.") path = layer_api_path(active_layer) if active_layer else None if not path: return ( False, "Der aktive Layer wurde nicht über 'vln_karten' geladen. " "Bitte einen Plan41-Layer aus dem Projekt auswählen.", ) dataset_key = active_layer.customProperty(PROP_DATASET) vkz = active_layer.customProperty(PROP_VERFAHREN) siblings = plugin_layers(dataset_key, vkz) for layer in siblings: if layer.isEditable() and not layer.commitChanges(): return ( False, "Die Bearbeitungssitzung von '%s' konnte nicht " "gespeichert werden." % layer.name(), ) feature_collection = layers_to_feature_collection(siblings) self._client.save_feature_collection(path, feature_collection) total = sum(layer.featureCount() for layer in siblings) return ( True, "%d Objekte aus %d Layer(n) erfolgreich hochgeladen." % (total, len(siblings)), ) def upload_summary(self, active_layer) -> Optional[dict]: """ Gibt eine Zusammenfassung der zu hochladenden Layer zurück (für den Bestätigungs-Dialog in der UI), ohne den Upload tatsächlich auszuführen. :returns: Dict mit ``dataset_label``, ``vkz``, ``siblings`` (Layer-Liste), ``total`` (Objektanzahl), ``listing`` (str) oder ``None`` wenn kein gültiger Layer. """ if not VLN_KARTEN_AVAILABLE or active_layer is None: return None path = layer_api_path(active_layer) if not path: return None dataset_key = active_layer.customProperty(PROP_DATASET) vkz = active_layer.customProperty(PROP_VERFAHREN) siblings = plugin_layers(dataset_key, vkz) total = sum(l.featureCount() for l in siblings) listing = "\n".join( "• %s (%d Objekte)" % (l.name(), l.featureCount()) for l in siblings ) return { "dataset_label": P41_LABEL if dataset_key == P41_DATASET_KEY else dataset_key, "vkz": vkz, "siblings": siblings, "total": total, "listing": listing, }