diff --git a/functions/qt_wrapper.py b/functions/qt_wrapper.py index d0c325f..e2bb90f 100644 --- a/functions/qt_wrapper.py +++ b/functions/qt_wrapper.py @@ -10,6 +10,8 @@ YES: Optional[Any] = None NO: Optional[Any] = None CANCEL: Optional[Any] = None ICON_QUESTION: Optional[Any] = None +QVariant: Type[Any] = object + # Qt-Klassen (werden dynamisch gesetzt) diff --git a/modules/DataGrabber.py b/modules/DataGrabber.py index 8bccbfc..8d922b5 100644 --- a/modules/DataGrabber.py +++ b/modules/DataGrabber.py @@ -17,6 +17,7 @@ ausschließlich über den ``Pruefmanager`` im aufrufenden Kontext (UI / Pipeline from __future__ import annotations +from pathlib import Path from typing import Any, Dict, List, Mapping, Optional, Tuple, Literal from sn_basis.modules.pruef_ergebnis import pruef_ergebnis @@ -66,10 +67,6 @@ class DataGrabber: """Setzt die aktuell zu untersuchende Rohquelle.""" self._source = source - from pathlib import Path - from typing import Tuple - - SourceType = str # "excel" | "datenbank" | "dienst" | "unbekannt" @@ -123,7 +120,7 @@ class DataGrabber: die vom Aufrufer über den ``Pruefmanager`` verarbeitet werden. """ self.set_source(source) - source_type = self.analyze_source_type(source) + source_type, source_result = self.analyze_source_type(source) source_dict: SourceDict = {} partial_results: List[pruef_ergebnis] = [] @@ -135,14 +132,7 @@ class DataGrabber: elif source_type == "service": source_dict, partial_results = self._process_service_source(source) else: - partial_results.append( - pruef_ergebnis( - ok=False, - meldung="Quelle konnte nicht klassifiziert werden", - aktion="kein_dateipfad", - kontext={"source": source}, - ) - ) + partial_results.append(source_result) summary = self._aggregate_results(source, source_dict, partial_results) return source_dict, summary diff --git a/modules/Datenschreiber.py b/modules/Datenschreiber.py index 143e090..9884b22 100644 --- a/modules/Datenschreiber.py +++ b/modules/Datenschreiber.py @@ -53,7 +53,7 @@ class Datenschreiber: def __init__(self, pruefmanager: Any, gpkg_path: Optional[str] = None) -> None: self.pruefmanager = pruefmanager - self.gpkg_path = gpkg_path + self.gpkg_path = str(gpkg_path) if gpkg_path else None # ------------------------------------------------------------------ # # Schreibe Daten @@ -65,192 +65,82 @@ class Datenschreiber: speicherort: str, ) -> List[Dict[str, Any]]: """ - Schreibt die abgerufenen Daten in die Zieldatenbank/Dateien. + Schreibt die übergebenen Layer in die Ziel-GPKG. - 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. + Erwartung: + - daten_dict["daten"] enthält Einträge der Form: + ident -> {"layer": QgsVectorLayer} + - self.gpkg_path ist ein str """ + if not speicherort: raise ValueError("Ein gültiger Speicherort (speicherort) muss übergeben werden.") - # Setze gpkg_path falls noch nicht vorhanden + # gpkg_path einmalig setzen / normalisieren if not self.gpkg_path: - self.gpkg_path = speicherort + self.gpkg_path = str(speicherort) results: List[Dict[str, Any]] = [] - daten_map: Dict[str, List[Any]] = daten_dict.get("daten", {}) + daten_map: Dict[str, 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) + for ident, entry in daten_map.items(): + layer = None + + # ----------------------------- + # Layer extrahieren + # ----------------------------- + if isinstance(entry, dict) and "layer" in entry: + layer = entry["layer"] + + if layer is None or not hasattr(layer, "isValid") or not layer.isValid(): + pe_err = pruef_ergebnis( + ok=False, + meldung=f"Ungültiger Layer für {ident}", + aktion="save_exception", + kontext={"ident": ident}, + ) + self.pruefmanager.verarbeite(pe_err) + continue + + # ----------------------------- + # Layername bestimmen + # ----------------------------- thema = None for pe in processed_results: try: kontext = getattr(pe, "kontext", None) or {} - if kontext and kontext.get("ident") == ident: + if 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 + layer_name = thema or str(ident) + # Layer in GPKG schreiben + err_msg = self._write_layer_to_gpkg(layer_name=layer_name, layer=layer) + if err_msg is not None: + pe_err = pruef_ergebnis( + ok=False, + meldung=f"Fehler beim Schreiben des Layers {layer_name}: {err_msg}", + aktion="save_exception", + kontext={"ident": ident, "layer_name": layer_name}, + ) + self.pruefmanager.verarbeite(pe_err) + continue + # Erfolgsfall: Info für lade_Layer sammeln + layer_path = f"{self.gpkg_path}|layername={layer_name}" results.append({ + "layer_path": layer_path, + "thema": layer_name, "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 # ------------------------------------------------------------------ # @@ -374,62 +264,54 @@ class Datenschreiber: # ------------------------------------------------------------------ # # Hilfsfunktionen intern # ------------------------------------------------------------------ # - def _write_layer_to_gpkg(self, layer_name: str, features: List[Any], mode: str = "create") -> Optional[str]: + def _write_layer_to_gpkg( + self, + layer_name: str, + layer: Any, + ) -> Optional[str]: """ - Interne Hilfsfunktion zum Schreiben eines Layers in das GPKG. + Schreibt einen QgsVectorLayer in die Ziel-GPKG. - Erwartete qgiscore-Funktion: - qgiscore.write_features_to_gpkg(gpkg_path, layer_name, features, mode) + Voraussetzungen: + - self.gpkg_path ist ein str + - layer ist ein gültiger QgsVectorLayer """ - 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 + if layer is None or not hasattr(layer, "isValid") or not layer.isValid(): + return "Ungültiger Layer zum Schreiben übergeben" - # 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 + try: + opts = qgiscore.QgsVectorFileWriter.SaveVectorOptions() + opts.driverName = "GPKG" + opts.layerName = layer_name + opts.fileEncoding = "UTF-8" - if mem_layer is None: - return "Keine Feld-/Geometrie-Informationen zum Schreiben vorhanden" + # Datei existiert → Layer überschreiben + # Datei existiert nicht → neue GPKG anlegen + if not os.path.exists(self.gpkg_path): + opts.actionOnExistingFile = qgiscore.QgsVectorFileWriter.CreateOrOverwriteFile + else: + opts.actionOnExistingFile = qgiscore.QgsVectorFileWriter.CreateOrOverwriteLayer - 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( + layer, + self.gpkg_path, + qgiscore.QgsProject.instance().transformContext(), + opts, + ) - 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) + # QGIS ≥3 liefert ein Tupel: (error_code, error_message, new_filename, new_layer_name) + if isinstance(err, tuple): + error_code = err[0] + error_msg = err[1] if len(err) > 1 else "" + else: + error_code = err + error_msg = "" - return "Keine Schreib-Funktion verfügbar (Wrapper nicht implementiert)" + if error_code != qgiscore.QgsVectorFileWriter.NoError: + return f"Fehler beim Schreiben (Code {error_code}, msg='{error_msg}')" + + return None + + except Exception as exc: + return str(exc) diff --git a/modules/Pruefmanager.py b/modules/Pruefmanager.py index 96bb8b5..aa979d4 100644 --- a/modules/Pruefmanager.py +++ b/modules/Pruefmanager.py @@ -216,3 +216,26 @@ class Pruefmanager: ) print("🔥 verarbeite() ENDE mit ok=False") return ergebnis + + def _ask_use_or_replace_pufferlayer(self) -> str: + """ + Fragt den Nutzer, ob ein vorhandener Pufferlayer verwendet + oder ersetzt werden soll. + + Returns + ------- + str + "verwenden", "ersetzen" oder "abbrechen" + """ + ergebnis = pruef_ergebnis( + ok=False, + aktion="layer_existiert", + meldung="Ein Pufferlayer ist bereits vorhanden.", + ) + + ergebnis = self.pruefmanager.verarbeite(ergebnis) + + if not ergebnis.ok: + return "abbrechen" + + return "verwenden" if ergebnis.aktion == "ok" else "ersetzen"