Daniel/Plugin_SN_Basis
1
0
Fork 0
forked from AG_QGIS/Plugin_SN_Basis
Code Pull Requests Activity

Überarbeitung für Pufferlayer-Fachdaten laden und gpkg-speichern/laden

Browse Source
This commit is contained in:
Daniel Helbig
2026-03-11 20:56:02 +01:00
parent 0ec24029d8
commit ae956b0046
4 changed files with 122 additions and 225 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
functions/qt_wrapper.py
View File

@@ -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)

16
modules/DataGrabber.py
View File

@@ -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

274
modules/Datenschreiber.py
View File

@@ -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:
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 Überschreiben von {layer_name}: {write_err}",
meldung=f"Fehler beim Schreiben des Layers {layer_name}: {err_msg}",
aktion="save_exception",
kontext={"ident": ident, "thema": thema, "error": write_err},
kontext={"ident": ident, "layer_name": layer_name},
)
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
# 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):
if layer is None or not hasattr(layer, "isValid") or not layer.isValid():
return "Ungültiger Layer zum Schreiben übergeben"
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":
# 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
err = qgiscore.QgsVectorFileWriter.writeAsVectorFormatV3(
mem_layer,
layer,
self.gpkg_path,
qgiscore.QgsProject.instance().transformContext(),
opts
opts,
)
if err != qgiscore.QgsVectorFileWriter.NoError:
return f"Fehler beim Schreiben (Code {err})"
# 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 = ""
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)
return "Keine Schreib-Funktion verfügbar (Wrapper nicht implementiert)"

23
modules/Pruefmanager.py
View File

@@ -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"