# sn_basis/modules/Datenschreiber.py """ Modul Datenschreiber Enthält die Klasse Datenschreiber mit drei Hauptmethoden: - schreibe_Daten: schreibt die abgerufenen Daten in die Ziel-GPKG/Dateien, fragt bei vorhandenen Layern nach Überschreiben/Anhängen/Abbrechen und legt Stile in der Datenbank ab. - lade_Layer: lädt die erzeugten/aktualisierten Layer ins Projekt und wendet die Vorgabestile an; sortiert abschließend die Layer. - schreibe_log: schreibt die verarbeiteten Pruefergebnisse strukturiert in eine Log-Datei im angegebenen Speicherort. Die Implementierung verwendet die Wrapper-APIs: - qgiscore_wrapper als qgiscore - qgisui_wrapper als qgisui (nur wenn nötig) - qt_wrapper als qt Wichtig ------ Alle Nutzerinteraktionen (z. B. Überschreiben / Anhängen / Abbrechen) werden zentral über den Pruefmanager gebündelt. Die Methode `ask_overwrite_append_cancel` des Pruefmanagers wird verwendet, damit UI-Interaktionen an einer Stelle konsolidiert und testbar sind. """ from __future__ import annotations from typing import Any, Dict, List, Optional import os import json import datetime from sn_basis.functions import qgiscore_wrapper as qgiscore from sn_basis.modules.pruef_ergebnis import pruef_ergebnis class Datenschreiber: """ Schreibt abgerufene Fachdaten in die Zieldatenbank/Dateien und lädt die Layer ins Projekt. Konstruktor ---------- pruefmanager: Instanz des Pruefmanagers; wird verwendet, um Pruefergebnisse zu verarbeiten und Nutzerinteraktionen zu zentralisieren. gpkg_path: Pfad zur Ziel-GPKG-Datei (oder Verzeichnis). Wenn None, muss der Aufrufer einen Speicherort übergeben. """ def __init__(self, pruefmanager: Any, gpkg_path: Optional[str] = None) -> None: self.pruefmanager = pruefmanager self.gpkg_path = gpkg_path # ------------------------------------------------------------------ # # Schreibe Daten # ------------------------------------------------------------------ # def schreibe_Daten( self, daten_dict: Dict[str, Any], processed_results: List[Any], speicherort: str, ) -> List[Dict[str, Any]]: """ Schreibt die abgerufenen Daten in die Zieldatenbank/Dateien. Ablauf ------ Für jede Zeile (ident) in ``daten_dict["daten"]``: 1. Bestimme Ziel-Layername (z. B. Thema oder ident). 2. Prüfe, ob ein Layer mit diesem Namen bereits existiert (Wrapper). 3. Falls vorhanden, frage den Benutzer (Überschreiben / Anhängen / Abbrechen) über die zentrale Pruefmanager-Methode `ask_overwrite_append_cancel`. 4. Führe die gewählte Operation aus oder schreibe den Layer, wenn er noch nicht existiert. 5. Schreibe ggf. den Stil in die GPKG und setze ihn als Vorgabe. 6. Sammle und gib eine Liste der angelegten/geänderten Layer zurück. Returns ------- List[Dict[str, Any]] Liste von Dicts mit Informationen zu jedem angelegten/geänderten Layer. """ if not speicherort: raise ValueError("Ein gültiger Speicherort (speicherort) muss übergeben werden.") # Setze gpkg_path falls noch nicht vorhanden if not self.gpkg_path: self.gpkg_path = speicherort results: List[Dict[str, Any]] = [] daten_map: Dict[str, List[Any]] = daten_dict.get("daten", {}) # Iteriere über alle Einträge for ident, features in daten_map.items(): # Thema/Name ableiten (falls vorhanden in processed_results oder ident) thema = None for pe in processed_results: try: kontext = getattr(pe, "kontext", None) or {} if kontext and kontext.get("ident") == ident: thema = kontext.get("thema") break except Exception: continue if not thema: thema = str(ident) layer_name = thema # Prüfe, ob Layer bereits existiert in der Ziel-GPKG layer_exists = False try: layer_exists_fn = getattr(qgiscore, "layer_exists_in_gpkg", None) if callable(layer_exists_fn): layer_exists = layer_exists_fn(self.gpkg_path, layer_name) else: # Fallback: QGIS-Fallback-Check via QgsVectorLayer if getattr(qgiscore, "QgsVectorLayer", None) is not None and qgiscore.QGIS_AVAILABLE: uri = f"{self.gpkg_path}|layername={layer_name}" layer = qgiscore.QgsVectorLayer(uri, layer_name, "ogr") layer_exists = bool(layer and getattr(layer, "isValid", lambda: False)()) except Exception: layer_exists = False operation = "created" if layer_exists: # Zentrale Nutzerabfrage über Pruefmanager # Erwartet Rückgabe: "overwrite" | "append" | "cancel" try: user_choice = self.pruefmanager.ask_overwrite_append_cancel(layer_name) except Exception: # Fallback: overwrite, falls Pruefmanager nicht verfügbar user_choice = "overwrite" if user_choice == "cancel": operation = "skipped" results.append({ "ident": ident, "thema": thema, "operation": operation, "layer_path": f"{self.gpkg_path}|layername={layer_name}", "feature_count": 0, }) continue if user_choice == "overwrite": write_err = self._write_layer_to_gpkg(layer_name, features, mode="overwrite") if write_err: pe_err = pruef_ergebnis( ok=False, meldung=f"Fehler beim Überschreiben von {layer_name}: {write_err}", aktion="save_exception", kontext={"ident": ident, "thema": thema, "error": write_err}, ) self.pruefmanager.verarbeite(pe_err) operation = "skipped" results.append({ "ident": ident, "thema": thema, "operation": operation, "layer_path": f"{self.gpkg_path}|layername={layer_name}", "feature_count": 0, }) continue else: operation = "overwritten" elif user_choice == "append": write_err = self._write_layer_to_gpkg(layer_name, features, mode="append") if write_err: pe_err = pruef_ergebnis( ok=False, meldung=f"Fehler beim Anhängen an {layer_name}: {write_err}", aktion="save_exception", kontext={"ident": ident, "thema": thema, "error": write_err}, ) self.pruefmanager.verarbeite(pe_err) operation = "skipped" results.append({ "ident": ident, "thema": thema, "operation": operation, "layer_path": f"{self.gpkg_path}|layername={layer_name}", "feature_count": 0, }) continue else: operation = "appended" else: # Layer existiert nicht -> neu anlegen write_err = self._write_layer_to_gpkg(layer_name, features, mode="create") if write_err: pe_err = pruef_ergebnis( ok=False, meldung=f"Fehler beim Erstellen von {layer_name}: {write_err}", aktion="save_exception", kontext={"ident": ident, "thema": thema, "error": write_err}, ) self.pruefmanager.verarbeite(pe_err) operation = "skipped" results.append({ "ident": ident, "thema": thema, "operation": operation, "layer_path": f"{self.gpkg_path}|layername={layer_name}", "feature_count": 0, }) continue else: operation = "created" # Stilbehandlung (falls in processed_results referenziert) style_written = False style_path = None for pe in processed_results: try: kontext = getattr(pe, "kontext", None) or {} if kontext and kontext.get("ident") == ident: style_path = kontext.get("stildatei") or kontext.get("Stildatei") break except Exception: continue if style_path: if not os.path.isabs(style_path): base_dir = os.path.dirname(__file__) style_path = os.path.join(base_dir, style_path) write_style_fn = getattr(qgiscore, "write_style_to_gpkg", None) if callable(write_style_fn): try: write_style_fn(self.gpkg_path, style_path, layer_name) style_written = True except Exception: style_written = False feature_count = len(features) if isinstance(features, list) else 0 results.append({ "ident": ident, "thema": thema, "operation": operation, "layer_path": f"{self.gpkg_path}|layername={layer_name}", "feature_count": feature_count, "style_written": style_written, }) return results # ------------------------------------------------------------------ # # Lade Layer ins Projekt # ------------------------------------------------------------------ # def lade_Layer(self, layer_infos: List[Dict[str, Any]]) -> None: """ Lädt die in schreibe_Daten erzeugten/aktualisierten Layer ins Projekt und wendet die Vorgabestile an. """ loaded_layers = [] for info in layer_infos: layer_path = info.get("layer_path") thema = info.get("thema") if not layer_path: continue try: layer = qgiscore.QgsVectorLayer(layer_path, thema, "ogr") if not layer or not getattr(layer, "isValid", lambda: False)(): pe_err = pruef_ergebnis( ok=False, meldung=f"Layer {thema} konnte nicht geladen werden", aktion="layer_nicht_gefunden", kontext={"layer_path": layer_path}, ) self.pruefmanager.verarbeite(pe_err) continue except Exception as exc: pe_err = pruef_ergebnis( ok=False, meldung=f"Fehler beim Erzeugen des Layers {thema}: {exc}", aktion="layer_nicht_gefunden", kontext={"layer_path": layer_path, "error": str(exc)}, ) self.pruefmanager.verarbeite(pe_err) continue try: apply_style_fn = getattr(qgiscore, "apply_default_style_from_gpkg", None) if callable(apply_style_fn): apply_style_fn(self.gpkg_path, layer) except Exception: pe_warn = pruef_ergebnis( ok=True, meldung=f"Style konnte für {thema} nicht automatisch angewendet werden", aktion="stil_not_implemented", kontext={"thema": thema}, ) self.pruefmanager.verarbeite(pe_warn) try: # qgisui wrapper wird hier nicht direkt für die Abfrage verwendet; # qgisui.add_layer_to_project sollte aber vorhanden sein. from sn_basis.functions import qgisui_wrapper as qgisui add_fn = getattr(qgisui, "add_layer_to_project", None) if callable(add_fn): add_fn(layer) else: # Fallback: falls wrapper nicht vorhanden, versuche QGIS-API direkt if getattr(qgiscore, "QgsProject", None) is not None and qgiscore.QGIS_AVAILABLE: qgiscore.QgsProject.instance().addMapLayer(layer) loaded_layers.append(layer) except Exception: pe_err = pruef_ergebnis( ok=False, meldung=f"Layer {thema} konnte nicht ins Projekt geladen werden", aktion="layer_nicht_gefunden", kontext={"thema": thema}, ) self.pruefmanager.verarbeite(pe_err) continue # Sortiere Layer im Projekt nach ID (Wrapper-Funktion bevorzugt) sort_fn = getattr(qgiscore, "sort_layers_by_id", None) if callable(sort_fn): try: sort_fn() except Exception: pass # ------------------------------------------------------------------ # # Schreibe Log # ------------------------------------------------------------------ # def schreibe_log(self, processed_results: List[Any], speicherort: str) -> str: """ Schreibt die verarbeiteten Pruefergebnisse strukturiert in eine Log-Datei. """ if not speicherort: raise ValueError("Ein gültiger Speicherort muss übergeben werden.") log_dir = speicherort os.makedirs(log_dir, exist_ok=True) timestamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S") log_path = os.path.join(log_dir, f"datenabruf_log_{timestamp}.json") serializable: List[Dict[str, Any]] = [] for pe in processed_results: try: entry = {} entry["ok"] = getattr(pe, "ok", None) if hasattr(pe, "ok") else None entry["meldung"] = getattr(pe, "meldung", None) if hasattr(pe, "meldung") else None kontext = getattr(pe, "kontext", None) if hasattr(pe, "kontext") else None entry["kontext"] = kontext serializable.append(entry) except Exception: serializable.append({"raw": str(pe)}) with open(log_path, "w", encoding="utf-8") as fh: json.dump(serializable, fh, ensure_ascii=False, indent=2) pe_log = pruef_ergebnis( ok=True, meldung=f"Log geschrieben: {os.path.basename(log_path)}", aktion="standarddatei_vorschlagen", kontext={"log_path": log_path}, ) self.pruefmanager.verarbeite(pe_log) return log_path # ------------------------------------------------------------------ # # Hilfsfunktionen intern # ------------------------------------------------------------------ # def _write_layer_to_gpkg(self, layer_name: str, features: List[Any], mode: str = "create") -> Optional[str]: """ Interne Hilfsfunktion zum Schreiben eines Layers in das GPKG. Erwartete qgiscore-Funktion: qgiscore.write_features_to_gpkg(gpkg_path, layer_name, features, mode) """ write_fn = getattr(qgiscore, "write_features_to_gpkg", None) if callable(write_fn): try: write_fn(self.gpkg_path, layer_name, features, mode) return None except Exception as exc: return str(exc) # Fallback: Verwende QgsVectorFileWriter, falls QGIS verfügbar if getattr(qgiscore, "QGIS_AVAILABLE", False) and getattr(qgiscore, "QgsVectorFileWriter", None) is not None: try: # Minimaler Fallback: erwarte, dass 'features' eine Liste von QgsFeature ist if not features: # Erstelle leeren Layer-Eintrag (GPKG erlaubt leere Layer) # Hier vereinfachen wir: writeAsVectorFormatV3 benötigt ein Layer-Objekt. return None # Versuche, ein Memory-Layer aus dem ersten Feature zu ermitteln first = features[0] mem_layer = None if hasattr(first, "fields") and hasattr(first, "geometry"): # Wenn Features QgsFeature sind, versuchen wir, das zugehörige Layer zu nutzen try: mem_layer = first.layer() if hasattr(first, "layer") else None except Exception: mem_layer = None if mem_layer is None: return "Keine Feld-/Geometrie-Informationen zum Schreiben vorhanden" opts = qgiscore.QgsVectorFileWriter.SaveVectorOptions() opts.driverName = "GPKG" opts.layerName = layer_name opts.fileEncoding = "UTF-8" if mode == "overwrite": opts.actionOnExistingFile = qgiscore.QgsVectorFileWriter.CreateOrOverwriteFile else: opts.actionOnExistingFile = qgiscore.QgsVectorFileWriter.CreateOrOverwriteLayer err = qgiscore.QgsVectorFileWriter.writeAsVectorFormatV3( mem_layer, self.gpkg_path, qgiscore.QgsProject.instance().transformContext(), opts ) if err != qgiscore.QgsVectorFileWriter.NoError: return f"Fehler beim Schreiben (Code {err})" return None except Exception as exc: return str(exc) return "Keine Schreib-Funktion verfügbar (Wrapper nicht implementiert)"