forked from AG_QGIS/Plugin_SN_Plan41
254 lines
8.6 KiB
Python
254 lines
8.6 KiB
Python
"""
|
||
sn_plan41/ui/tab_a_logic.py – Fachlogik für Tab A (Daten)
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
from typing import Any, Dict, List, Optional, Tuple
|
||
from collections.abc import Mapping as _Mapping
|
||
|
||
from sn_basis.functions.variable_wrapper import ( # type: ignore
|
||
get_variable,
|
||
set_variable,
|
||
)
|
||
from sn_basis.functions.sys_wrapper import ( # type: ignore
|
||
file_exists,
|
||
write_text,
|
||
)
|
||
from sn_basis.functions.ly_existence_wrapper import layer_exists # type: ignore
|
||
from sn_basis.functions.ly_metadata_wrapper import get_layer_type # type: ignore
|
||
|
||
# Prüfer-Typen (werden als Instanzen erwartet)
|
||
from sn_basis.modules.Pruefmanager import Pruefmanager # type: ignore
|
||
from sn_basis.modules.linkpruefer import Linkpruefer # type: ignore
|
||
from sn_basis.modules.stilpruefer import Stilpruefer # type: ignore
|
||
|
||
# Typalias für Klarheit
|
||
Row = Dict[str, Any]
|
||
DataDict = Dict[str, List[Row]]
|
||
|
||
|
||
class TabALogic:
|
||
"""
|
||
Kapselt die komplette Logik von Tab A:
|
||
- Verfahrens-Datenbank
|
||
- optionale Linkliste
|
||
- Verfahrensgebiet-Layer
|
||
|
||
Diese Klasse erwartet beim Erzeugen Instanzen von Pruefmanager, Linkpruefer
|
||
und Stilpruefer. Die validate_and_filter_rows-Methode verwendet diese
|
||
Instanzen über self.
|
||
"""
|
||
|
||
def __init__(self, pruefmanager: Pruefmanager, link_pruefer: Linkpruefer, stil_pruefer: Stilpruefer) -> None:
|
||
"""
|
||
:param pruefmanager: Instanz des Pruefmanagers (für verarbeite/report)
|
||
:param link_pruefer: Instanz des Linkpruefers (mit methode pruefe)
|
||
:param stil_pruefer: Instanz des Stilpruefers (mit methode pruefe)
|
||
"""
|
||
self.pruefmanager = pruefmanager
|
||
self.link_pruefer = link_pruefer
|
||
self.stil_pruefer = stil_pruefer
|
||
|
||
# -------------------------------
|
||
# Verfahrens-Datenbank
|
||
# -------------------------------
|
||
|
||
def load_verfahrens_db(self) -> Optional[str]:
|
||
"""
|
||
Lädt die gespeicherte Verfahrens-Datenbank.
|
||
"""
|
||
path = get_variable("verfahrens_db", scope="project")
|
||
if path and file_exists(path):
|
||
return path
|
||
return None
|
||
|
||
def set_verfahrens_db(self, path: Optional[str]) -> None:
|
||
"""
|
||
Speichert oder löscht die Verfahrens-Datenbank.
|
||
"""
|
||
if path:
|
||
set_variable("verfahrens_db", path, scope="project")
|
||
else:
|
||
set_variable("verfahrens_db", "", scope="project")
|
||
|
||
def create_new_verfahrens_db(self, path: str) -> bool:
|
||
"""
|
||
Legt eine neue leere GPKG-Datei an.
|
||
"""
|
||
if not path:
|
||
return False
|
||
|
||
try:
|
||
write_text(path, "")
|
||
except Exception:
|
||
return False
|
||
|
||
self.set_verfahrens_db(path)
|
||
return True
|
||
|
||
# -------------------------------
|
||
# Lokale Linkliste
|
||
# -------------------------------
|
||
|
||
def load_linkliste(self) -> Optional[str]:
|
||
"""
|
||
Lädt die gespeicherte lokale Linkliste.
|
||
"""
|
||
path = get_variable("linkliste", scope="project")
|
||
if path and file_exists(path):
|
||
return path
|
||
return None
|
||
|
||
def set_linkliste(self, path: Optional[str]) -> None:
|
||
"""
|
||
Speichert oder löscht die lokale Linkliste.
|
||
"""
|
||
if path:
|
||
set_variable("linkliste", path, scope="project")
|
||
else:
|
||
set_variable("linkliste", "", scope="project")
|
||
|
||
# -------------------------------
|
||
# Verfahrensgebiet-Layer
|
||
# -------------------------------
|
||
|
||
def save_verfahrensgebiet_layer(self, layer) -> None:
|
||
"""
|
||
Speichert die ID des Verfahrensgebiet-Layers.
|
||
Ungültige Layer werden ignoriert.
|
||
"""
|
||
if layer is None:
|
||
set_variable("verfahrensgebiet_layer", "", scope="project")
|
||
return
|
||
|
||
if not hasattr(layer, "id") or not callable(layer.id):
|
||
set_variable("verfahrensgebiet_layer", "", scope="project")
|
||
return
|
||
|
||
try:
|
||
layer_id = layer.id()
|
||
except Exception:
|
||
set_variable("verfahrensgebiet_layer", "", scope="project")
|
||
return
|
||
|
||
if not layer_id:
|
||
set_variable("verfahrensgebiet_layer", "", scope="project")
|
||
return
|
||
|
||
set_variable("verfahrensgebiet_layer", layer_id, scope="project")
|
||
|
||
def load_verfahrensgebiet_layer_id(self) -> Optional[str]:
|
||
"""
|
||
Lädt die gespeicherte Layer-ID.
|
||
"""
|
||
value = get_variable("verfahrensgebiet_layer", scope="project")
|
||
return value or None
|
||
|
||
def is_valid_verfahrensgebiet_layer(self, layer) -> bool:
|
||
"""
|
||
Prüft, ob ein Layer als Verfahrensgebiet geeignet ist.
|
||
"""
|
||
if not layer_exists(layer):
|
||
return False
|
||
|
||
layer_type = get_layer_type(layer)
|
||
return layer_type == "vector"
|
||
|
||
# -------------------------------
|
||
# Validierung und Filterung von data_dict
|
||
# -------------------------------
|
||
|
||
def validate_and_filter_rows(self, data_dict: DataDict) -> Tuple[DataDict, List[Any]]:
|
||
"""
|
||
Validiert und filtert die Zeilen aus `data_dict`.
|
||
|
||
Erwartete Struktur von `data_dict`: {'rows': [ {attr}, ... ]}.
|
||
|
||
Für jede Zeile werden die folgenden Attribute gelesen:
|
||
ident = attr['ident'] (Pflicht)
|
||
thema = attr['Inhalt'] (optional)
|
||
url = attr['Link'] (Pflicht)
|
||
stildatei = attr['Stildatei'] (optional)
|
||
provider = attr['Provider'] (Pflicht, wird uppercased)
|
||
|
||
Verhalten
|
||
- Pflichtfelder (ident, Link, Provider) müssen vorhanden und nicht-leer sein,
|
||
sonst wird die Zeile verworfen.
|
||
- Wenn Link nicht leer ist, wird self.link_pruefer.pruefe(url) aufgerufen.
|
||
- Ist das Ergebnis ok: Zeile wird behalten.
|
||
- Ist das Ergebnis nicht ok: Zeile wird verworfen; das verarbeitete
|
||
pruef_ergebnis wird gesammelt.
|
||
- Wenn Stildatei nicht leer ist, wird self.stil_pruefer.pruefe(stildatei) aufgerufen.
|
||
- Ist das Ergebnis ok: der Wert bleibt erhalten.
|
||
- Ist das Ergebnis nicht ok: das Feld `Stildatei` wird in der zurückgegebenen
|
||
Zeile auf None gesetzt; das verarbeitete pruef_ergebnis wird gesammelt.
|
||
- Alle pruef_ergebnis-Objekte werden an self.pruefmanager.verarbeite(...) übergeben.
|
||
Die verarbeiteten Ergebnisse werden in der Rückgabe-Liste gesammelt.
|
||
|
||
Rückgabe
|
||
- (valid_data_dict, processed_results)
|
||
valid_data_dict: {'rows': [valid_row1, valid_row2, ...]}
|
||
processed_results: Liste der vom Pruefmanager verarbeiteten pruef_ergebnis-Objekte
|
||
"""
|
||
processed_results: List[Any] = []
|
||
valid_rows: List[Row] = []
|
||
|
||
# Grundstruktur prüfen
|
||
if not isinstance(data_dict, dict):
|
||
return {"rows": []}, processed_results
|
||
|
||
rows = data_dict.get("rows", [])
|
||
if not isinstance(rows, (list, tuple)):
|
||
return {"rows": []}, processed_results
|
||
|
||
for raw in rows:
|
||
# Sicherstellen, dass raw ein Mapping ist
|
||
if not isinstance(raw, _Mapping):
|
||
continue
|
||
|
||
ident = raw.get("ident")
|
||
inhalt = raw.get("Inhalt")
|
||
link = raw.get("Link")
|
||
stildatei = raw.get("Stildatei")
|
||
provider = raw.get("Provider")
|
||
|
||
# Pflichtfelder prüfen
|
||
if not ident or not link or not provider:
|
||
continue
|
||
|
||
# Provider normalisieren
|
||
provider_norm = str(provider).upper()
|
||
|
||
# Link prüfen
|
||
pe_link = self.link_pruefer.pruefe(link)
|
||
processed_link = self.pruefmanager.verarbeite(pe_link)
|
||
if not getattr(processed_link, "ok", False):
|
||
processed_results.append(processed_link)
|
||
continue # Zeile verwerfen
|
||
|
||
# Stil prüfen (falls vorhanden)
|
||
if stildatei:
|
||
pe_stil = self.stil_pruefer.pruefe(stildatei)
|
||
processed_stil = self.pruefmanager.verarbeite(pe_stil)
|
||
if not getattr(processed_stil, "ok", False):
|
||
processed_results.append(processed_stil)
|
||
stildatei_value: Optional[str] = None
|
||
else:
|
||
stildatei_value = stildatei
|
||
else:
|
||
stildatei_value = None
|
||
|
||
# Validierte Zeile zusammenbauen
|
||
validated_row: Row = {
|
||
"ident": ident,
|
||
"Inhalt": inhalt,
|
||
"Link": link,
|
||
"Stildatei": stildatei_value,
|
||
"Provider": provider_norm,
|
||
}
|
||
valid_rows.append(validated_row)
|
||
|
||
result_dict: DataDict = {"rows": valid_rows}
|
||
return result_dict, processed_results
|