forked from Daniel/Plugin_SN_Plan41
251 lines
8.3 KiB
Python
251 lines
8.3 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
|
||
from collections.abc import Mapping as _Mapping
|
||
import os
|
||
|
||
from sn_basis.functions.qgiscore_wrapper import (
|
||
QgsVectorFileWriter,
|
||
QgsVectorLayer,
|
||
QgsProject,
|
||
|
||
)
|
||
|
||
from sn_basis.functions.variable_wrapper import (
|
||
get_variable,
|
||
set_variable,
|
||
)
|
||
from sn_basis.functions.sys_wrapper import file_exists
|
||
from sn_basis.functions.ly_existence_wrapper import layer_exists
|
||
from sn_basis.functions.ly_metadata_wrapper import get_layer_type
|
||
from sn_basis.functions.qgisui_wrapper import QgsFileWidget
|
||
|
||
# Prüfer-Typen
|
||
from sn_basis.modules.Pruefmanager import Pruefmanager
|
||
from sn_basis.modules.linkpruefer import Linkpruefer
|
||
from sn_basis.modules.stilpruefer import Stilpruefer
|
||
from sn_basis.modules.Dateipruefer import Dateipruefer
|
||
from sn_basis.modules.pruef_ergebnis import pruef_ergebnis
|
||
from sn_basis.modules.DataGrabber import DataGrabber, SourceType, SourceDict
|
||
|
||
Row = Dict[str, Any]
|
||
DataDict = Dict[str, List[Row]]
|
||
|
||
class TabALogic:
|
||
"""
|
||
Kapselt die Fachlogik von Tab A. Verfahrens-DB wird **nicht** bei Pfad-Auswahl,
|
||
sondern erst beim ersten Layer-Schreiben angelegt.
|
||
"""
|
||
|
||
def __init__(self, pruefmanager: Pruefmanager, link_pruefer: Linkpruefer, stil_pruefer: Stilpruefer) -> None:
|
||
self.pruefmanager = pruefmanager
|
||
self.link_pruefer = link_pruefer
|
||
self.stil_pruefer = stil_pruefer
|
||
self.data_grabber: Optional[DataGrabber] = None
|
||
|
||
# -------------------------------
|
||
# Verfahrens-Datenbank (Pfad-Management)
|
||
# -------------------------------
|
||
|
||
def load_verfahrens_db(self) -> Optional[str]:
|
||
"""Lädt den gespeicherten Verfahrens-DB-Pfad (Datei muss nicht existieren)."""
|
||
path = get_variable("verfahrens_db", scope="project")
|
||
return path or None
|
||
|
||
def set_verfahrens_db(self, path: Optional[str]) -> None:
|
||
"""Speichert den Verfahrens-DB-Pfad (Datei wird später angelegt)."""
|
||
if path:
|
||
set_variable("verfahrens_db", path, scope="project")
|
||
else:
|
||
set_variable("verfahrens_db", "", scope="project")
|
||
|
||
# -------------------------------
|
||
# Layer → Verfahrens-DB schreiben (alte Logik!)
|
||
# -------------------------------
|
||
|
||
def write_layer_to_verfahrens_db(
|
||
self,
|
||
source_layer: QgsVectorLayer,
|
||
zielpfad: str,
|
||
layer_name: str,
|
||
) -> bool:
|
||
"""
|
||
Schreibt einen Layer in die Verfahrens-DB.
|
||
Legt GPKG **bei Bedarf neu an** (wie puffer_setzen im alten Code).
|
||
|
||
Args:
|
||
source_layer: Layer zum Exportieren (z.B. aus DataGrabber)
|
||
zielpfad: Vom Dateiprüfer geprüfter Ziel-GPKG-Pfad
|
||
layer_name: Name des Layers in der GPKG
|
||
|
||
Returns:
|
||
True wenn erfolgreich
|
||
"""
|
||
if not zielpfad or not source_layer or not source_layer.isValid():
|
||
return False
|
||
|
||
# Optionen wie im alten puffer_setzen
|
||
opts = QgsVectorFileWriter.SaveVectorOptions()
|
||
opts.driverName = "GPKG"
|
||
opts.fileEncoding = "UTF-8"
|
||
opts.layerName = layer_name
|
||
|
||
# Alte Logik: bei neuem Pfad komplett neue GPKG, sonst Layer überschreiben
|
||
if not os.path.exists(zielpfad):
|
||
opts.actionOnExistingFile = QgsVectorFileWriter.CreateOrOverwriteFile
|
||
else:
|
||
opts.actionOnExistingFile = QgsVectorFileWriter.CreateOrOverwriteLayer
|
||
|
||
transform_context = QgsProject.instance().transformContext()
|
||
|
||
error = QgsVectorFileWriter.writeAsVectorFormatV3(
|
||
source_layer,
|
||
zielpfad,
|
||
transform_context,
|
||
opts,
|
||
)
|
||
|
||
if error != QgsVectorFileWriter.NoError:
|
||
print(f"Fehler beim Schreiben nach {zielpfad}: {error}")
|
||
return False
|
||
|
||
# Pfad jetzt auch als "Verfahrens-DB" merken
|
||
self.set_verfahrens_db(zielpfad)
|
||
return True
|
||
|
||
# -------------------------------
|
||
# Lokale Linkliste
|
||
# -------------------------------
|
||
|
||
def load_linkliste(self) -> Optional[str]:
|
||
path = get_variable("linkliste", scope="project")
|
||
if path and file_exists(path):
|
||
return path
|
||
return None
|
||
|
||
def set_linkliste(self, path: Optional[str]) -> None:
|
||
if path:
|
||
set_variable("linkliste", path, scope="project")
|
||
else:
|
||
set_variable("linkliste", "", scope="project")
|
||
|
||
# -------------------------------
|
||
# Verfahrensgebiet-Layer
|
||
# -------------------------------
|
||
|
||
def save_verfahrensgebiet_layer(self, layer) -> None:
|
||
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]:
|
||
value = get_variable("verfahrensgebiet_layer", scope="project")
|
||
return value or None
|
||
|
||
def is_valid_verfahrensgebiet_layer(self, layer) -> bool:
|
||
if not layer_exists(layer):
|
||
return False
|
||
|
||
layer_type = get_layer_type(layer)
|
||
return layer_type == "vector"
|
||
|
||
# === PIPELINE ===
|
||
def _on_run_pipeline(self,source: str) -> None:
|
||
"""DEBUG: Pipeline mit maximaler Ausgabe."""
|
||
print("\n" + "="*60)
|
||
print("🚀 _on_run_pipeline GESTARTET")
|
||
print("="*60)
|
||
# 🔥 DEBUG QT STATUS
|
||
from sn_basis.functions import qt_wrapper
|
||
qt_wrapper.debug_qt_status() # ← Zeigt Version an
|
||
|
||
|
||
|
||
# 1. Services prüfen
|
||
print(f"pruefmanager: {self.pruefmanager is not None}")
|
||
print(f"data_grabber: {self.data_grabber is not None}")
|
||
print(f"logic: {hasattr(self, 'logic')}")
|
||
|
||
if not self.pruefmanager:
|
||
print("❌ FEHLER: self.pruefmanager fehlt!")
|
||
return
|
||
if not self.data_grabber:
|
||
print("❌ FEHLER: self.data_grabber fehlt!")
|
||
return
|
||
|
||
print("✅ Services OK")
|
||
|
||
# 2. FileWidget
|
||
#source = self.file_widget.filePath()
|
||
print(f"📁 Eingabe: '{source}' (len={len(source or '')})")
|
||
|
||
# 3. Dateipruefer
|
||
print("🔍 Dateipruefer starte...")
|
||
try:
|
||
pruefer = Dateipruefer(
|
||
source,
|
||
basis_pfad="",
|
||
leereingabe_erlaubt=False,
|
||
standarddatei=None,
|
||
temporaer_erlaubt=True, # ✅ Explizit True
|
||
verfahrens_db_modus=True # ✅ Keyword-only
|
||
)
|
||
ergebnis1 = pruefer.pruefe()
|
||
print(f" → ok={ergebnis1.ok}, aktion='{ergebnis1.aktion}', kontext={ergebnis1.kontext}")
|
||
except Exception as e:
|
||
print(f"💥 Dateipruefer FEHLER: {e}")
|
||
import traceback
|
||
traceback.print_exc()
|
||
return
|
||
|
||
# 4. Pruefmanager
|
||
print("🤖 Pruefmanager starte...")
|
||
try:
|
||
ergebnis2 = self.pruefmanager.verarbeite(ergebnis1)
|
||
|
||
print(f" → ok={ergebnis2.ok}, aktion='{ergebnis2.aktion}', kontext={ergebnis2.kontext}")
|
||
except Exception as e:
|
||
print(f"💥 Pruefmanager FEHLER: {e}")
|
||
import traceback
|
||
traceback.print_exc()
|
||
return
|
||
|
||
# 5. Entscheidung
|
||
weiter = ergebnis2.ok
|
||
print(f"➡️ Weiter? {weiter} (aktion='{ergebnis2.aktion}')")
|
||
|
||
if weiter:
|
||
final_pfad = ergebnis2.kontext if ergebnis2.kontext else source
|
||
print(f"🚀 DataGrabber mit: '{final_pfad}'")
|
||
try:
|
||
self.data_grabber.run(final_pfad)
|
||
print("✅ DataGrabber aufgerufen!")
|
||
except Exception as e:
|
||
print(f"💥 DataGrabber FEHLER: {e}")
|
||
import traceback
|
||
traceback.print_exc()
|
||
else:
|
||
print("⏹️ Pipeline gestoppt (erwartet bei leerem Pfad)")
|
||
|
||
print("="*60 + "\n")
|