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

Diensteabruf integriert

Browse Source
This commit is contained in:
Daniel Helbig
2026-03-12 16:14:02 +01:00
parent ae956b0046
commit 9829ac9c81
10 changed files with 1136 additions and 89 deletions
Download Patch File Download Diff File Expand all files Collapse all files

164
modules/Datenschreiber.py
View File

@@ -30,9 +30,13 @@ from __future__ import annotations
from typing import Any, Dict, List, Optional
import os
import json
import re
import datetime
import sqlite3
from sn_basis.functions import qgiscore_wrapper as qgiscore
from sn_basis.functions.os_wrapper import normalize_path, is_absolute_path
from sn_basis.functions.sys_wrapper import get_plugin_root, join_path, file_exists
from sn_basis.modules.pruef_ergebnis import pruef_ergebnis
@@ -55,8 +59,95 @@ class Datenschreiber:
self.pruefmanager = pruefmanager
self.gpkg_path = str(gpkg_path) if gpkg_path else None
# ------------------------------------------------------------------ #
# Schreibe Daten
def _resolve_style_path(self, style_path: Optional[str]) -> Optional[str]:
if not style_path:
return None
style_path_str = str(style_path).strip()
if not style_path_str:
return None
if not is_absolute_path(style_path_str):
plugin_root = get_plugin_root()
style_path_str = str(join_path(plugin_root, "sn_plan41", "assets", style_path_str))
style_path_str = str(normalize_path(style_path_str))
return style_path_str if file_exists(style_path_str) else None
def _store_style_in_gpkg(self, layer_name: str, style_path: str, layer: Optional[Any] = None) -> None:
"""Stellt sicher, dass der Stil in der layer_styles-Tabelle der GPKG gespeichert wird."""
try:
with open(style_path, "r", encoding="utf-8") as fh:
style_qml = fh.read()
f_geometry_column = ''
if layer is not None:
try:
if hasattr(layer, 'geometryColumn'):
f_geometry_column = str(layer.geometryColumn())
elif hasattr(layer, 'dataProvider') and hasattr(layer.dataProvider(), 'geometryColumnName'):
f_geometry_column = str(layer.dataProvider().geometryColumnName())
except Exception:
f_geometry_column = ''
with sqlite3.connect(self.gpkg_path) as conn:
cur = conn.cursor()
cur.execute(
"""
CREATE TABLE IF NOT EXISTS layer_styles (
id INTEGER PRIMARY KEY AUTOINCREMENT,
f_table_catalog TEXT,
f_table_schema TEXT,
f_table_name TEXT NOT NULL,
f_geometry_column TEXT,
styleName TEXT,
styleQML TEXT,
styleSLD TEXT,
useAsDefault BOOLEAN,
description TEXT,
owner TEXT,
ui TEXT,
update_time DATETIME DEFAULT CURRENT_TIMESTAMP
)
"""
)
# Das aktuelle QGIS-Style-Verhalten: bestehenden Style für denselben Layer nicht löschen (nur appenden)
# Wir wollen aber Default-Style setzen: alte Default-Styles entfernen.
cur.execute(
"UPDATE layer_styles SET useAsDefault = 0 WHERE f_table_name = ?",
(layer_name,),
)
# Fülle die bekannten QGIS-Kolonnen
style_name = os.path.basename(style_path)
cur.execute(
"INSERT INTO layer_styles (f_table_catalog, f_table_schema, f_table_name, f_geometry_column, styleName, styleQML, styleSLD, useAsDefault, description, owner, ui) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
(
'',
'',
layer_name,
f_geometry_column,
style_name,
style_qml,
None,
1,
'',
'',
'',
),
)
conn.commit()
except Exception as exc:
self.pruefmanager.verarbeite(
pruef_ergebnis(
ok=False,
meldung=f"Fehler beim Speichern des Layer-Stils in GPKG: {exc}",
aktion="style_gpkg_speichern_fehlgeschlagen",
kontext={"layer_name": layer_name, "style_path": style_path},
)
)
# ------------------------------------------------------------------ #
def schreibe_Daten(
self,
@@ -85,12 +176,14 @@ class Datenschreiber:
for ident, entry in daten_map.items():
layer = None
style_path = None
# -----------------------------
# Layer extrahieren
# -----------------------------
if isinstance(entry, dict) and "layer" in entry:
layer = entry["layer"]
if isinstance(entry, dict):
layer = entry.get("layer")
style_path = self._resolve_style_path(entry.get("style_path"))
if layer is None or not hasattr(layer, "isValid") or not layer.isValid():
pe_err = pruef_ergebnis(
@@ -115,7 +208,11 @@ class Datenschreiber:
except Exception:
continue
layer_name = thema or str(ident)
layer_name_raw = thema or str(ident)
layer_name = re.sub(r"[^A-Za-z0-9_]+", "_", layer_name_raw).strip("_")
if not layer_name:
layer_name = f"layer_{ident}"
# Layer in GPKG schreiben
err_msg = self._write_layer_to_gpkg(layer_name=layer_name, layer=layer)
if err_msg is not None:
@@ -128,12 +225,17 @@ class Datenschreiber:
self.pruefmanager.verarbeite(pe_err)
continue
# Wenn der Stil vorhanden und valide ist, als Default in GPKG-Style-Tabelle ablegen
if style_path:
self._store_style_in_gpkg(layer_name, style_path, layer)
# 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,
"style_path": style_path,
})
return results
@@ -178,18 +280,33 @@ class Datenschreiber:
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)
style_path = info.get("style_path")
resolved_style_path = self._resolve_style_path(style_path)
if resolved_style_path:
try:
layer.loadNamedStyle(resolved_style_path)
layer.triggerRepaint()
except Exception as exc:
pe_warn = pruef_ergebnis(
ok=True,
meldung=f"Style konnte für {thema} nicht geladen werden: {exc}",
aktion="stil_laden_fehlgeschlagen",
kontext={"thema": thema, "style_path": resolved_style_path},
)
self.pruefmanager.verarbeite(pe_warn)
else:
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;
@@ -286,6 +403,19 @@ class Datenschreiber:
opts.layerName = layer_name
opts.fileEncoding = "UTF-8"
# Style in der GPKG speichern, wenn möglich
if hasattr(opts, "symbologyExport"):
try:
# QGIS: SymbologyExport-Wert z.B. QgsVectorFileWriter.SaveVectorOptions.Symbology
saveOpts = qgiscore.QgsVectorFileWriter.SaveVectorOptions
sym_val = getattr(saveOpts, "Symbology", None)
if sym_val is None:
sym_val = getattr(saveOpts, "SymbologyExport", None)
if sym_val is not None:
opts.symbologyExport = sym_val
except Exception:
pass
# Datei existiert → Layer überschreiben
# Datei existiert nicht → neue GPKG anlegen
if not os.path.exists(self.gpkg_path):