Daniel/Plugin_SN_Basis
1
0
Fork 0
forked from AG_QGIS/Plugin_SN_Basis
Code Pull Requests Activity
Files
main
Plugin_SN_Basis/modules/Datenschreiber.py

436 lines
18 KiB
Python
Raw Permalink Blame History

# 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)"
View Git Blame Copy Permalink