qt_wrapper, dialog;wrapper, Pruef_ergebnis und Pruefmanager überarbeitet, so dass die Übergaben jetzt stimmen. Nutzerabfragen werden tatsächlich ausgelöst- Nutzerabfrage Datei überschreiebn... ist noch Blödsinn
This commit is contained in:
@@ -1,40 +1,27 @@
|
||||
# sn_basis/modules/DataGrabber.py
|
||||
"""
|
||||
DataGrabber module
|
||||
==================
|
||||
|
||||
Leichter Orchestrator, der eine Quelle (Datei, Einzellink, Datenbank)
|
||||
analysiert, passende Prüfer aufruft und die Ergebnisse an den
|
||||
:class:`sn_basis.modules.Pruefmanager.Pruefmanager` delegiert.
|
||||
UI‑freier Orchestrator für die Prüfung und Klassifikation von Datenquellen.
|
||||
|
||||
Dieses vereinfachte Modul geht davon aus, dass alle benötigten Prüfer
|
||||
und der ExcelImporter vorhanden und importierbar sind. Es enthält
|
||||
keine Fallbacks oder defensive Exception-Handling-Pfade für fehlende
|
||||
Prüfer-Module — fehlende Komponenten führen zu Import- oder Laufzeitfehlern,
|
||||
die bewusst nicht unterdrückt werden.
|
||||
Der DataGrabber:
|
||||
- klassifiziert die übergebene Quelle (Datei, Dienst, Datenbank, Excel),
|
||||
- ruft passende Prüfer (Dateipruefer, Linkpruefer, Layerpruefer, Stilpruefer) auf,
|
||||
- sammelt alle rohen ``pruef_ergebnis``‑Objekte,
|
||||
- aggregiert diese zu einem zusammenfassenden Ergebnis,
|
||||
- **löst selbst keinerlei UI‑Interaktion aus**.
|
||||
|
||||
Alle Nutzerinteraktionen (MessageBar, QMessageBox, Logging) erfolgen
|
||||
ausschließlich über den ``Pruefmanager`` im aufrufenden Kontext (UI / Pipeline).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import (
|
||||
Optional,
|
||||
Any,
|
||||
Mapping,
|
||||
Iterable,
|
||||
Dict,
|
||||
Protocol,
|
||||
Literal,
|
||||
Tuple,
|
||||
List,
|
||||
)
|
||||
from pathlib import Path
|
||||
import sqlite3
|
||||
from typing import Any, Dict, List, Mapping, Optional, Tuple, Literal
|
||||
|
||||
from sn_basis.modules.pruef_ergebnis import pruef_ergebnis
|
||||
from sn_basis.modules.Pruefmanager import Pruefmanager
|
||||
from sn_basis.modules.pruef_ergebnis import pruef_ergebnis, PruefAktion
|
||||
|
||||
# In dieser vereinfachten Variante werden die Prüfer und der ExcelImporter
|
||||
# direkt importiert. Fehlende Module führen zu ImportError (gewollt).
|
||||
from sn_basis.modules.Dateipruefer import Dateipruefer
|
||||
from sn_basis.modules.linkpruefer import Linkpruefer
|
||||
from sn_basis.modules.layerpruefer import Layerpruefer
|
||||
@@ -42,383 +29,144 @@ from sn_basis.modules.stilpruefer import Stilpruefer
|
||||
from sn_basis.modules.excel_importer import ExcelImporter
|
||||
|
||||
|
||||
SourceType = Literal["file", "link", "database", "unknown"]
|
||||
SourceType = Literal["service", "database", "excel", "unknown"]
|
||||
|
||||
|
||||
class LinklistAdapter(Protocol):
|
||||
"""
|
||||
Minimal-Protokoll für Adapter, die Linklisten liefern/normalisieren.
|
||||
|
||||
Implementierende Klassen sollten:
|
||||
- load() -> Iterable[Mapping[str, Any]]
|
||||
- normalize(raw_item) -> Mapping[str, Any]
|
||||
"""
|
||||
def load(self) -> Iterable[Mapping[str, Any]]:
|
||||
...
|
||||
def normalize(self, raw_item: Mapping[str, Any]) -> Mapping[str, Any]:
|
||||
...
|
||||
SourceDict = Dict[str, List[Mapping[str, Any]]]
|
||||
|
||||
|
||||
class DataGrabber:
|
||||
"""
|
||||
DataGrabber orchestriert das Einlesen einer Quelle und die Übergabe an Prüfer.
|
||||
Analysiert und prüft Datenquellen für den Fachdatenabruf.
|
||||
|
||||
Diese vereinfachte Implementierung erwartet, dass alle Prüferklassen und
|
||||
der ExcelImporter vorhanden sind. Es gibt keine defensive Logik für
|
||||
fehlende Komponenten.
|
||||
|
||||
Konstruktor-Parameter
|
||||
--------------------
|
||||
:param pruefmanager: Instanz des Pruefmanagers (verpflichtend).
|
||||
:param datei_pruefer_cls: Klasse des Dateipruefers (Standard: Dateipruefer).
|
||||
:param link_pruefer: Instanz des Linkpruefers.
|
||||
:param layer_pruefer: Instanz des Layerpruefers.
|
||||
:param stil_pruefer: Instanz des Stilpruefers.
|
||||
Der DataGrabber ist **UI‑frei**. Er erzeugt ausschließlich rohe
|
||||
``pruef_ergebnis``‑Objekte und überlässt deren Verarbeitung
|
||||
vollständig dem aufrufenden Code.
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
pruefmanager: Pruefmanager,
|
||||
*,
|
||||
datei_pruefer_cls=Dateipruefer,
|
||||
link_pruefer: Linkpruefer,
|
||||
layer_pruefer: Layerpruefer,
|
||||
stil_pruefer: Stilpruefer,
|
||||
datei_pruefer_cls: type[Dateipruefer] = Dateipruefer,
|
||||
link_pruefer: Optional[Linkpruefer] = None,
|
||||
layer_pruefer: Optional[Layerpruefer] = None,
|
||||
stil_pruefer: Optional[Stilpruefer] = None,
|
||||
excel_importer_cls: type[ExcelImporter] = ExcelImporter,
|
||||
) -> None:
|
||||
# Pruefmanager ist verpflichtend
|
||||
self.pruefmanager: Pruefmanager = pruefmanager
|
||||
|
||||
# Dateipruefer-Klasse (wird zur Laufzeit mit einem Pfad instanziert)
|
||||
self.pruefmanager = pruefmanager
|
||||
self._datei_pruefer_cls = datei_pruefer_cls
|
||||
self.link_pruefer = link_pruefer
|
||||
self.layer_pruefer = layer_pruefer
|
||||
self.stil_pruefer = stil_pruefer
|
||||
self._excel_importer_cls = excel_importer_cls
|
||||
|
||||
# Prüfer-Instanzen (werden direkt verwendet)
|
||||
self.link_pruefer: Linkpruefer = link_pruefer
|
||||
self.layer_pruefer: Layerpruefer = layer_pruefer
|
||||
self.stil_pruefer: Stilpruefer = stil_pruefer
|
||||
self._source: Optional[str] = None
|
||||
|
||||
# Quelle (wird später gesetzt)
|
||||
self.source: Optional[str] = None
|
||||
|
||||
# ------------------------------------------------------------------ #
|
||||
# Source Management
|
||||
# ------------------------------------------------------------------ #
|
||||
# ------------------------------------------------------------------
|
||||
# Öffentliche API
|
||||
# ------------------------------------------------------------------
|
||||
def set_source(self, source: str) -> None:
|
||||
"""Setzt die aktuell zu untersuchende Rohquelle."""
|
||||
self._source = source
|
||||
|
||||
def analyze_source_type(self, source: str) -> SourceType:
|
||||
"""
|
||||
Setzt die Quelle für den DataGrabber.
|
||||
Klassifiziert die Quelle.
|
||||
|
||||
Die Quelle ist ein String, der entweder ein lokaler Dateipfad,
|
||||
ein Einzellink (URL/URI) oder ein Pfad zu einer Datenbank/GeoPackage ist.
|
||||
Aktuell Platzhalter – liefert ``"unknown"``.
|
||||
"""
|
||||
self.source = source
|
||||
|
||||
def analyze_source(self, source: str) -> SourceType:
|
||||
"""
|
||||
Klassifiziert die angegebene Quelle ausschließlich anhand des Dateipruefers.
|
||||
|
||||
Ablauf
|
||||
------
|
||||
1. Instanziere den Dateipruefer mit `pfad=source` und `temporaer_erlaubt=False`.
|
||||
2. Rufe `pruefe()` auf und werte das zurückgegebene :class:`pruef_ergebnis` aus.
|
||||
3. Bei `ok==True` wird anhand der Dateiendung zwischen "database" (gpkg/sqlite/db)
|
||||
und "file" unterschieden.
|
||||
4. Bei `ok==False` werden typische Aktionen wie "datei_nicht_gefunden" als "link"
|
||||
interpretiert; bei "falsche_endung" wird anhand der Endung klassifiziert.
|
||||
"""
|
||||
dp = self._datei_pruefer_cls(pfad=source, temporaer_erlaubt=False)
|
||||
pe: pruef_ergebnis = dp.pruefe()
|
||||
|
||||
if getattr(pe, "ok", False):
|
||||
suffix = Path(source).suffix.lower()
|
||||
if suffix in (".gpkg", ".sqlite", ".db"):
|
||||
return "database"
|
||||
return "file"
|
||||
|
||||
aktion = getattr(pe, "aktion", None)
|
||||
if aktion in ("datei_nicht_gefunden", "pfad_nicht_gefunden", "kein_dateipfad"):
|
||||
return "link"
|
||||
if aktion == "falsche_endung":
|
||||
lower = source.lower()
|
||||
for db_ext in (".gpkg", ".sqlite", ".db"):
|
||||
if lower.endswith(db_ext):
|
||||
return "database"
|
||||
for file_ext in (".xlsx", ".xls", ".csv"):
|
||||
if lower.endswith(file_ext):
|
||||
return "file"
|
||||
return "unknown"
|
||||
|
||||
# ------------------------------------------------------------------ #
|
||||
# Excel-Verarbeitung
|
||||
#Es werden alle Werte mit gültigem Link übernommen. Die restliche Struktur
|
||||
#wird nicht überprüft, da alle Fachplugins unterschiedliche Strukturen haben können
|
||||
# ------------------------------------------------------------------ #
|
||||
def process_excel_source(
|
||||
self,
|
||||
filepath: str
|
||||
) -> Tuple[Optional[Dict[str, List[Mapping[str, Any]]]], Any]:
|
||||
def run(self, source: str) -> Tuple[SourceDict, pruef_ergebnis]:
|
||||
"""
|
||||
Liest eine Excel-Datei ein und übernimmt ausschließlich die Zeilen,
|
||||
deren Link durch den Linkpruefer als gültig eingestuft wurde.
|
||||
Führt die vollständige Quellprüfung aus.
|
||||
|
||||
Ablauf
|
||||
------
|
||||
1. Die Excel-Datei wird mit dem ``ExcelImporter`` eingelesen.
|
||||
Erwartet wird eine Liste von Mappings (z.B. dicts), die jeweils
|
||||
die Linkparameter enthalten.
|
||||
|
||||
2. Für jede Zeile wird der Wert ``row["Link"]`` extrahiert und durch
|
||||
``self.link_pruefer.pruefe(...)`` geprüft.
|
||||
|
||||
3. Das Prüfergebnis wird durch ``self.pruefmanager.verarbeite(...)``
|
||||
geleitet, der UI-Interaktion, Logging und finale Entscheidung übernimmt.
|
||||
|
||||
4. Nur Zeilen, deren verarbeitete Prüfergebnisse ``ok == True`` liefern,
|
||||
werden in die Ergebnisliste übernommen.
|
||||
|
||||
5. Wenn mindestens eine Zeile gültig ist, wird ein Dict der Form::
|
||||
|
||||
{"rows": [row1, row2, ...]}
|
||||
|
||||
zurückgegeben.
|
||||
Wenn keine Zeile gültig ist, wird ``None`` zurückgegeben.
|
||||
|
||||
Parameter
|
||||
---------
|
||||
filepath:
|
||||
Pfad zur Excel-Datei, die eingelesen werden soll.
|
||||
|
||||
Returns
|
||||
-------
|
||||
Tuple[Optional[Dict[str, List[Mapping[str, Any]]]], pruef_ergebnis]
|
||||
- ``data``: ``{"rows": [...]} `` wenn gültige Zeilen existieren,
|
||||
sonst ``None``.
|
||||
- ``pruef_ergebnis``: ein zusammenfassendes Prüfergebnis, das
|
||||
den Lesevorgang beschreibt (nicht die Einzelprüfungen).
|
||||
|
||||
Hinweise
|
||||
--------
|
||||
- Diese Methode führt **keine Normalisierung** durch.
|
||||
- Die Verantwortung für die Struktur der Excel-Zeilen liegt beim Fachplugin.
|
||||
- Der Linkpruefer prüft ausschließlich den Wert ``row["Link"]``.
|
||||
Diese Methode ist **UI‑frei**. Sie gibt rohe Ergebnisse zurück,
|
||||
die vom Aufrufer über den ``Pruefmanager`` verarbeitet werden.
|
||||
"""
|
||||
self.set_source(source)
|
||||
source_type = self.analyze_source_type(source)
|
||||
|
||||
# 1) Excel einlesen
|
||||
importer = ExcelImporter(filepath=filepath, pruefmanager=self.pruefmanager)
|
||||
rows = importer.import_xlsx() # erwartet: List[Mapping[str, Any]]
|
||||
source_dict: SourceDict = {}
|
||||
partial_results: List[pruef_ergebnis] = []
|
||||
|
||||
valid_rows: List[Mapping[str, Any]] = []
|
||||
|
||||
# 2) Jede Zeile einzeln prüfen
|
||||
for row in rows:
|
||||
raw_link = row.get("Link")
|
||||
|
||||
# 2a) Fachliche Prüfung
|
||||
pe = self.link_pruefer.pruefe(raw_link)
|
||||
|
||||
# 2b) Verarbeitung durch den Pruefmanager
|
||||
processed = self.pruefmanager.verarbeite(pe)
|
||||
|
||||
# 2c) Nur gültige Zeilen übernehmen
|
||||
if getattr(processed, "ok", False):
|
||||
valid_rows.append(row)
|
||||
|
||||
# 3) Zusammenfassendes Prüfergebnis erzeugen
|
||||
if valid_rows:
|
||||
pe_ok = pruef_ergebnis(
|
||||
ok=True,
|
||||
meldung=f"{len(valid_rows)} gültige Zeilen aus Excel gelesen",
|
||||
aktion="ok",
|
||||
kontext=filepath,
|
||||
if source_type == "excel":
|
||||
source_dict, partial_results = self._process_excel_source(source)
|
||||
elif source_type == "database":
|
||||
source_dict, partial_results = self._process_database_source(source)
|
||||
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},
|
||||
)
|
||||
)
|
||||
processed_summary = self.pruefmanager.verarbeite(pe_ok)
|
||||
return {"rows": valid_rows}, processed_summary
|
||||
|
||||
# Keine gültigen Zeilen
|
||||
pe_fail = pruef_ergebnis(
|
||||
ok=False,
|
||||
meldung="Keine gültigen Links in der Excel-Datei gefunden",
|
||||
aktion="read_error",
|
||||
kontext=filepath,
|
||||
)
|
||||
processed_summary = self.pruefmanager.verarbeite(pe_fail)
|
||||
return None, processed_summary
|
||||
summary = self._aggregate_results(source, source_dict, partial_results)
|
||||
return source_dict, summary
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Excel‑Quellen
|
||||
# ------------------------------------------------------------------
|
||||
def _process_excel_source(
|
||||
self, filepath: str
|
||||
) -> Tuple[SourceDict, List[pruef_ergebnis]]:
|
||||
source_dict: SourceDict = {}
|
||||
results: List[pruef_ergebnis] = []
|
||||
return source_dict, results
|
||||
|
||||
# ------------------------------------------------------------------ #
|
||||
# Einzellink-Verarbeitung
|
||||
# ------------------------------------------------------------------ #
|
||||
def process_single_link(
|
||||
# ------------------------------------------------------------------
|
||||
# Datenbank‑Quellen
|
||||
# ------------------------------------------------------------------
|
||||
def _process_database_source(
|
||||
self, db_path: str
|
||||
) -> Tuple[SourceDict, List[pruef_ergebnis]]:
|
||||
source_dict: SourceDict = {}
|
||||
results: List[pruef_ergebnis] = []
|
||||
return source_dict, results
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Dienst‑Quellen
|
||||
# ------------------------------------------------------------------
|
||||
def _process_service_source(
|
||||
self, link: str
|
||||
) -> Tuple[SourceDict, List[pruef_ergebnis]]:
|
||||
source_dict: SourceDict = {}
|
||||
results: List[pruef_ergebnis] = []
|
||||
return source_dict, results
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Aggregation
|
||||
# ------------------------------------------------------------------
|
||||
def _aggregate_results(
|
||||
self,
|
||||
link: Mapping[str, Any]
|
||||
) -> Tuple[Optional[Dict[str, List[Mapping[str, Any]]]], Any]:
|
||||
source: str,
|
||||
source_dict: SourceDict,
|
||||
partial_results: List[pruef_ergebnis],
|
||||
) -> pruef_ergebnis:
|
||||
"""
|
||||
Prüft einen einzelnen Link anhand der im Link-Dict enthaltenen Link-URL.
|
||||
Aggregiert Einzelprüfungen zu einem Gesamt‑``pruef_ergebnis``.
|
||||
|
||||
Ablauf
|
||||
------
|
||||
1. Erwartet wird ein Mapping (z.B. dict), das die Linkparameter enthält.
|
||||
Mindestens der Schlüssel ``"Link"`` muss vorhanden sein.
|
||||
|
||||
2. Der eigentliche Link (z.B. URL) wird aus ``link["Link"]`` extrahiert
|
||||
und an ``self.link_pruefer.pruefe(...)`` übergeben.
|
||||
|
||||
3. Das Prüfergebnis wird anschließend durch ``self.pruefmanager.verarbeite(...)``
|
||||
geleitet, der UIInteraktion, Logging und finale Entscheidung übernimmt.
|
||||
|
||||
4. Wenn das verarbeitete Prüfergebnis **nicht OK** ist, wird
|
||||
``(None, pruef_ergebnis)`` zurückgegeben.
|
||||
|
||||
5. Wenn das Prüfergebnis **OK** ist, wird das unveränderte LinkDict
|
||||
in der Struktur ``{"rows": [link]}`` zurückgegeben.
|
||||
|
||||
Parameter
|
||||
---------
|
||||
link:
|
||||
Ein Mapping mit den Linkparametern (z.B. id, Thema, Gruppe, Link,
|
||||
Anbieter, Stildatei). Diese Methode verändert das Mapping nicht.
|
||||
|
||||
Returns
|
||||
-------
|
||||
Tuple[Optional[Dict[str, List[Mapping[str, Any]]]], pruef_ergebnis]
|
||||
- ``data``: ``{"rows": [link]}`` wenn gültig, sonst ``None``
|
||||
- ``pruef_ergebnis``: das vom Pruefmanager verarbeitete Ergebnis
|
||||
|
||||
Hinweise
|
||||
--------
|
||||
- Diese Methode führt **keine Normalisierung** durch.
|
||||
- Die Verantwortung für die Struktur des Link-Dicts liegt beim Fachplugin.
|
||||
- Der Linkpruefer prüft ausschließlich den Wert ``link["Link"]``.
|
||||
**Keine UI‑Interaktion.**
|
||||
"""
|
||||
if source_dict:
|
||||
return pruef_ergebnis(
|
||||
ok=True,
|
||||
meldung="Quelle erfolgreich geprüft",
|
||||
aktion="ok",
|
||||
kontext={
|
||||
"source": source,
|
||||
"valid_entries": sum(len(v) for v in source_dict.values()),
|
||||
},
|
||||
)
|
||||
|
||||
# 1) Link extrahieren (Fachplugin garantiert, dass "Link" existiert)
|
||||
raw_link = link.get("Link")
|
||||
|
||||
# 2) Fachliche Prüfung durch den Linkpruefer
|
||||
pruef_ergebnis = self.link_pruefer.pruefe(raw_link)
|
||||
|
||||
# 3) Verarbeitung durch den Pruefmanager
|
||||
processed = self.pruefmanager.verarbeite(pruef_ergebnis)
|
||||
|
||||
# 4) Wenn Prüfung nicht OK → keine Daten zurückgeben
|
||||
if not getattr(processed, "ok", False):
|
||||
return None, processed
|
||||
|
||||
# 5) Prüfung OK → unverändertes Link-Dict zurückgeben
|
||||
data = {"rows": [link]}
|
||||
|
||||
return data, processed
|
||||
|
||||
|
||||
|
||||
|
||||
# ------------------------------------------------------------------ #
|
||||
# Datenbank-Verarbeitung
|
||||
# ------------------------------------------------------------------ #
|
||||
#def process_database_table(self, db_path: str, table: Optional[str] = None) -> Tuple[Optional[Dict[str, List[Mapping[str, Any]]]], pruef_ergebnis]:
|
||||
#noch nicht implementiert
|
||||
"""
|
||||
Liest eine Tabelle aus einer SQLite/GeoPackage-Datei.
|
||||
|
||||
Verhalten
|
||||
---------
|
||||
1. Validiert die Datei mit dem Dateipruefer.
|
||||
2. Falls OK, versucht es, die angegebene Tabelle zu lesen; falls keine Tabelle
|
||||
angegeben ist, wird nach einer typischen Metadaten-Tabelle 'layer_metadaten'
|
||||
gesucht und diese gelesen.
|
||||
3. Gibt die Zeilen als Liste von Dicts zurück.
|
||||
"""
|
||||
dp = self._datei_pruefer_cls(pfad=db_path, temporaer_erlaubt=False)
|
||||
pe = dp.pruefe()
|
||||
processed = self.pruefmanager.verarbeite(pe)
|
||||
if not getattr(processed, "ok", False):
|
||||
return None, processed
|
||||
|
||||
conn = sqlite3.connect(db_path)
|
||||
cur = conn.cursor()
|
||||
if table:
|
||||
cur.execute(f"SELECT * FROM {table}")
|
||||
cols = [d[0] for d in cur.description]
|
||||
rows = [dict(zip(cols, r)) for r in cur.fetchall()]
|
||||
else:
|
||||
cur.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='layer_metadaten'")
|
||||
if cur.fetchone():
|
||||
cur.execute("SELECT * FROM layer_metadaten")
|
||||
cols = [d[0] for d in cur.description]
|
||||
rows = [dict(zip(cols, r)) for r in cur.fetchall()]
|
||||
else:
|
||||
rows = []
|
||||
conn.close()
|
||||
|
||||
pe_ok = pruef_ergebnis(ok=True, meldung="DB gelesen", aktion="ok", kontext=db_path)
|
||||
processed_ok = self.pruefmanager.verarbeite(pe_ok)
|
||||
return {"rows": rows}, processed_ok
|
||||
|
||||
# ------------------------------------------------------------------ #
|
||||
# Hauptlauf / Dispatch
|
||||
# ------------------------------------------------------------------ #
|
||||
def run(self) -> Dict[str, Any]:
|
||||
"""
|
||||
Hauptmethode des DataGrabbers.
|
||||
|
||||
Ablauf
|
||||
------
|
||||
1. Prüft, ob eine Quelle gesetzt ist.
|
||||
2. Klassifiziert die Quelle via :meth:`analyze_source`.
|
||||
3. Dispatch:
|
||||
- file (.xlsx/.xls) -> :meth:`process_excel_source`
|
||||
- link -> :meth:`process_single_link`
|
||||
- database -> :meth:`process_database_table`
|
||||
- unknown -> Fehler
|
||||
4. Aggregiert geladene Einträge in einem Ergebnis-Dict und gibt dieses zurück.
|
||||
|
||||
Rückgabeformat
|
||||
-------------
|
||||
Ein Dict mit den Schlüsseln:
|
||||
- 'geladen' : Liste der geladenen Themen/Namen
|
||||
- 'fehler' : Mapping Thema -> Fehlermeldung
|
||||
- 'ausserhalb': Liste der als ausserhalb klassifizierten Themen
|
||||
- 'relevant' : Liste der relevanten Themen
|
||||
- 'details' : zusätzliche Detailinformationen (z. B. Anzahl Zeilen)
|
||||
"""
|
||||
result: Dict[str, Any] = {"geladen": [], "fehler": {}, "ausserhalb": [], "relevant": [], "details": {}}
|
||||
|
||||
if not self.source:
|
||||
pe = pruef_ergebnis(ok=False, meldung="Keine Quelle gesetzt", aktion="kein_dateipfad", kontext=None)
|
||||
processed = self.pruefmanager.verarbeite(pe)
|
||||
result["fehler"]["source"] = getattr(processed, "meldung", "Keine Quelle")
|
||||
return result
|
||||
|
||||
src_type = self.analyze_source(self.source)
|
||||
|
||||
if src_type == "file":
|
||||
suffix = Path(self.source).suffix.lower()
|
||||
if suffix in (".xlsx", ".xls"):
|
||||
data_dict, pe = self.process_excel_source(self.source)
|
||||
else:
|
||||
pe = pruef_ergebnis(ok=False, meldung="Dateityp nicht unterstützt", aktion="falsche_endung", kontext=self.source)
|
||||
pe = self.pruefmanager.verarbeite(pe)
|
||||
data_dict = None
|
||||
|
||||
elif src_type == "link":
|
||||
data_dict, pe = self.process_single_link(self.source)
|
||||
|
||||
#elif src_type == "database":
|
||||
#data_dict, pe = self.process_database_table(self.source, table=None)
|
||||
|
||||
else:
|
||||
pe = pruef_ergebnis(ok=False, meldung="Quelle unbekannt", aktion="kein_dateipfad", kontext=self.source)
|
||||
pe = self.pruefmanager.verarbeite(pe)
|
||||
data_dict = None
|
||||
|
||||
# Falls Daten vorhanden: fülle result['geladen'] und details
|
||||
if data_dict and "rows" in data_dict:
|
||||
rows = data_dict["rows"]
|
||||
for r in rows:
|
||||
thema = r.get("Inhalt") or r.get("ident") or r.get("Link") or "unbenannt"
|
||||
result["geladen"].append(thema)
|
||||
result["details"]["source_rows"] = len(rows)
|
||||
|
||||
# Falls das letzte pruef_ergebnis einen Fehler enthält, übernehme es
|
||||
if not getattr(pe, "ok", False):
|
||||
result["fehler"]["source"] = getattr(pe, "meldung", "Fehler bei Quelle")
|
||||
|
||||
return result
|
||||
return pruef_ergebnis(
|
||||
ok=False,
|
||||
meldung="Keine gültigen Einträge in der Quelle gefunden",
|
||||
aktion="read_error",
|
||||
kontext={"source": source},
|
||||
)
|
||||
|
||||
@@ -1,98 +1,175 @@
|
||||
"""
|
||||
sn_basis/modules/Dateipruefer.py – Prüfung von Dateieingaben für das Plugin.
|
||||
Verwendet sys_wrapper und gibt pruef_ergebnis an den Pruefmanager zurück.
|
||||
sn_basis/modules/Dateipruefer.py
|
||||
|
||||
Erweiterter Dateiprüfer für Verfahrens-DB-Workflows mit vollständiger Unterstützung
|
||||
der Anforderungen 1-2.e (leerer Pfad, fehlende Datei, bestehende Datei).
|
||||
"""
|
||||
|
||||
from pathlib import Path
|
||||
from typing import Optional
|
||||
|
||||
from sn_basis.functions.sys_wrapper import (
|
||||
join_path,
|
||||
file_exists,
|
||||
)
|
||||
|
||||
from sn_basis.modules.Pruefmanager import pruef_ergebnis
|
||||
from sn_basis.functions.sys_wrapper import join_path, file_exists
|
||||
from sn_basis.modules.pruef_ergebnis import pruef_ergebnis, PruefAktion
|
||||
|
||||
|
||||
class Dateipruefer:
|
||||
"""
|
||||
Prüft Dateieingaben und liefert ein pruef_ergebnis zurück.
|
||||
Die eigentliche Nutzerinteraktion übernimmt der Pruefmanager.
|
||||
Prüft Dateieingaben für Verfahrens-DB-Workflows und liefert :class:`pruef_ergebnis`.
|
||||
|
||||
**Funktionsweise (deine Anforderungen 1-2.e):**
|
||||
|
||||
+---------------------+------------------------------------------+---------------+
|
||||
| **Fall** | **Ergebnis** | **ok** |
|
||||
+=====================+==========================================+===============+
|
||||
| 1. Leerer Pfad | ``temporaer_erlaubt`` | False |
|
||||
+---------------------+------------------------------------------+---------------+
|
||||
| 2.a Leerer Pfad | Pruefmanager fragt → ``temporaer_erzeugen`` | True |
|
||||
+---------------------+------------------------------------------+---------------+
|
||||
| 2.b Datei existiert | ``ok`` | True |
|
||||
+---------------------+------------------------------------------+---------------+
|
||||
| 2.c Ungültiger Pfad | ``datei_nicht_gefunden`` | False |
|
||||
+---------------------+------------------------------------------+---------------+
|
||||
| **2.d Datei fehlt** | **``datei_wird_erzeugt``** | **True** |
|
||||
+---------------------+------------------------------------------+---------------+
|
||||
| **2.e Datei da** | **``datei_existiert``** | **False** |
|
||||
+---------------------+------------------------------------------+---------------+
|
||||
|
||||
Der Dateiprüfer führt **keine UI-Interaktion** durch.
|
||||
Entscheidungen werden ausschließlich vom :class:`Pruefmanager` getroffen.
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
pfad: str,
|
||||
pfad: Optional[str],
|
||||
basis_pfad: str = "",
|
||||
leereingabe_erlaubt: bool = False,
|
||||
standarddatei: str | None = None,
|
||||
standarddatei: Optional[str] = None,
|
||||
temporaer_erlaubt: bool = False,
|
||||
):
|
||||
*,
|
||||
verfahrens_db_modus: bool = True, # 🆕 Verfahrens-DB-spezifische Logik
|
||||
) -> None:
|
||||
"""
|
||||
Parameters
|
||||
----------
|
||||
pfad : Optional[str]
|
||||
Vom UI gelieferter Dateipfad (kann leer oder Whitespace sein).
|
||||
basis_pfad : str, optional
|
||||
Basisverzeichnis für relative Pfade (default: "").
|
||||
leereingabe_erlaubt : bool, optional
|
||||
Ob leere Eingabe grundsätzlich erlaubt ist (default: False).
|
||||
standarddatei : Optional[str], optional
|
||||
Optionaler Standardpfad (default: None).
|
||||
temporaer_erlaubt : bool, optional
|
||||
Ob bei leerer Eingabe temporäre Layer erlaubt sind (default: False).
|
||||
verfahrens_db_modus : bool, optional
|
||||
Aktiviert Verfahrens-DB-spezifische Logik (2.d, 2.e) (default: True).
|
||||
"""
|
||||
self.pfad = pfad
|
||||
self.basis_pfad = basis_pfad
|
||||
self.leereingabe_erlaubt = leereingabe_erlaubt
|
||||
self.standarddatei = standarddatei
|
||||
self.temporaer_erlaubt = temporaer_erlaubt
|
||||
self.verfahrens_db_modus = verfahrens_db_modus
|
||||
|
||||
# ---------------------------------------------------------
|
||||
# Hilfsfunktion
|
||||
# ---------------------------------------------------------
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Hilfsfunktionen
|
||||
# ------------------------------------------------------------------
|
||||
def _pfad(self, relativer_pfad: str) -> Path:
|
||||
"""
|
||||
Erzeugt einen OS-unabhängigen Pfad relativ zum Basisverzeichnis.
|
||||
"""
|
||||
"""Erzeugt OS-unabhängigen Pfad relativ zum Basisverzeichnis."""
|
||||
return join_path(self.basis_pfad, relativer_pfad)
|
||||
|
||||
# ---------------------------------------------------------
|
||||
# Hauptfunktion
|
||||
# ---------------------------------------------------------
|
||||
def _ist_leer(self) -> bool:
|
||||
"""
|
||||
Prüft robust, ob Eingabe als „leer" zu behandeln ist.
|
||||
|
||||
Returns
|
||||
-------
|
||||
bool
|
||||
True bei None, leerem String oder reinem Whitespace.
|
||||
"""
|
||||
if self.pfad is None:
|
||||
return True
|
||||
if not isinstance(self.pfad, str):
|
||||
return True
|
||||
return not self.pfad.strip()
|
||||
|
||||
def _ist_gueltiger_gpkg_pfad(self, pfad: Path) -> bool:
|
||||
"""
|
||||
Prüft, ob Pfad für GPKG geeignet ist (Endung + Schreibrechte).
|
||||
|
||||
Returns
|
||||
-------
|
||||
bool
|
||||
True wenn `.gpkg`-Endung und Verzeichnis beschreibbar.
|
||||
"""
|
||||
if not str(pfad).lower().endswith('.gpkg'):
|
||||
return False
|
||||
|
||||
# Verzeichnis muss beschreibbar sein
|
||||
return pfad.parent.exists() and pfad.parent.is_dir()
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Hauptlogik: deine Anforderungen 1-2.e
|
||||
# ------------------------------------------------------------------
|
||||
def pruefe(self) -> pruef_ergebnis:
|
||||
"""
|
||||
Prüft eine Dateieingabe und liefert ein pruef_ergebnis zurück.
|
||||
Der Pruefmanager entscheidet später, wie der Nutzer gefragt wird.
|
||||
"""
|
||||
🆕 Prüft Dateieingabe gemäß Anforderungen 1-2.e.
|
||||
|
||||
# -----------------------------------------------------
|
||||
# 1. Fall: Eingabe ist leer
|
||||
# -----------------------------------------------------
|
||||
if not self.pfad:
|
||||
**Workflow:**
|
||||
1. **Leere Eingabe** → ``temporaer_erlaubt`` (Pruefmanager fragt)
|
||||
2. **Pfad prüfen**:
|
||||
- **Ungültig** → 2.c ``datei_nicht_gefunden``
|
||||
- **Gültig, fehlt** → **2.d** ``datei_wird_erzeugt`` (ok=True!)
|
||||
- **Gültig, existiert** → **2.e** ``datei_existiert`` (Pruefmanager fragt)
|
||||
3. **Datei OK** → 2.b ``ok``
|
||||
|
||||
Returns
|
||||
-------
|
||||
pruef_ergebnis
|
||||
Mit korrekter Aktion für jeden Fall.
|
||||
"""
|
||||
# 1. 🎯 ANFORDERUNG 1: Leere Eingabe
|
||||
if self._ist_leer():
|
||||
return self._handle_leere_eingabe()
|
||||
|
||||
# -----------------------------------------------------
|
||||
# 2. Fall: Eingabe ist nicht leer → Datei prüfen
|
||||
# -----------------------------------------------------
|
||||
pfad = self._pfad(self.pfad)
|
||||
# 2. Pfad normalisieren
|
||||
pfad = self._pfad(self.pfad.strip())
|
||||
|
||||
if not file_exists(pfad):
|
||||
# 🆕 2.c: Ungültiger GPKG-Pfad?
|
||||
if not self.verfahrens_db_modus or not self._ist_gueltiger_gpkg_pfad(pfad):
|
||||
return pruef_ergebnis(
|
||||
ok=False,
|
||||
meldung=f"Die Datei '{self.pfad}' wurde nicht gefunden.",
|
||||
meldung=f"Der Pfad '{self.pfad}' ist kein gültiger GPKG-Pfad.",
|
||||
aktion="datei_nicht_gefunden",
|
||||
kontext=pfad,
|
||||
)
|
||||
|
||||
# -----------------------------------------------------
|
||||
# 3. Datei existiert → Erfolg
|
||||
# -----------------------------------------------------
|
||||
# 🆕 2.d: Gültiger Pfad, Datei fehlt → DIREKT WEITER (ok=True!)
|
||||
if not file_exists(pfad):
|
||||
return pruef_ergebnis(
|
||||
ok=True, # 🎯 WICHTIG: Pipeline fortsetzen!
|
||||
meldung=f"Datei '{self.pfad}' wird erzeugt.",
|
||||
aktion="datei_wird_erzeugt",
|
||||
kontext=pfad,
|
||||
)
|
||||
|
||||
# 🆕 2.e: Datei existiert → Pruefmanager fragt Überschreiben/etc.
|
||||
return pruef_ergebnis(
|
||||
ok=True,
|
||||
meldung="Datei gefunden.",
|
||||
aktion="ok",
|
||||
ok=False, # 🎯 Pruefmanager soll 4-Optionen-Dialog zeigen
|
||||
meldung=f"Datei '{self.pfad}' existiert bereits.",
|
||||
aktion="datei_existiert",
|
||||
kontext=pfad,
|
||||
)
|
||||
|
||||
# ---------------------------------------------------------
|
||||
# Behandlung leerer Eingaben
|
||||
# ---------------------------------------------------------
|
||||
# 2.b: Wird nicht erreicht (durch 2.e abgefangen)
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Leere Eingabe (ANFORDERUNG 1, 2.a)
|
||||
# ------------------------------------------------------------------
|
||||
def _handle_leere_eingabe(self) -> pruef_ergebnis:
|
||||
"""
|
||||
Liefert ein pruef_ergebnis für den Fall, dass das Dateifeld leer ist.
|
||||
Der Pruefmanager fragt später den Nutzer.
|
||||
Behandelt leere Eingaben (Priorität: leereingabe → Standard → temporär → Fehler).
|
||||
"""
|
||||
|
||||
# 1. Leereingabe erlaubt → Nutzer fragen, ob das beabsichtigt war
|
||||
if self.leereingabe_erlaubt:
|
||||
return pruef_ergebnis(
|
||||
ok=False,
|
||||
@@ -101,19 +178,17 @@ class Dateipruefer:
|
||||
kontext=None,
|
||||
)
|
||||
|
||||
# 2. Standarddatei verfügbar → Nutzer fragen, ob sie verwendet werden soll
|
||||
if self.standarddatei:
|
||||
return pruef_ergebnis(
|
||||
ok=False,
|
||||
meldung=(
|
||||
f"Es wurde keine Datei angegeben. "
|
||||
"Es wurde keine Datei angegeben. "
|
||||
f"Soll die Standarddatei '{self.standarddatei}' verwendet werden?"
|
||||
),
|
||||
aktion="standarddatei_vorschlagen",
|
||||
kontext=self._pfad(self.standarddatei),
|
||||
)
|
||||
|
||||
# 3. Temporäre Datei erlaubt → Nutzer fragen, ob temporär gearbeitet werden soll
|
||||
if self.temporaer_erlaubt:
|
||||
return pruef_ergebnis(
|
||||
ok=False,
|
||||
@@ -125,7 +200,6 @@ class Dateipruefer:
|
||||
kontext=None,
|
||||
)
|
||||
|
||||
# 4. Leereingabe nicht erlaubt → Fehler
|
||||
return pruef_ergebnis(
|
||||
ok=False,
|
||||
meldung="Es wurde keine Datei angegeben.",
|
||||
|
||||
@@ -1,66 +1,45 @@
|
||||
"""
|
||||
sn_basis/modules/Pruefmanager.py
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
from typing import Optional, Any
|
||||
|
||||
from sn_basis.functions import (
|
||||
ask_yes_no,
|
||||
info,
|
||||
warning,
|
||||
error,
|
||||
set_layer_visible,
|
||||
)
|
||||
|
||||
from sn_basis.functions import ask_yes_no, info, warning, error
|
||||
from sn_basis.modules.pruef_ergebnis import pruef_ergebnis, PruefAktion
|
||||
|
||||
print("DEBUG: Pruefmanager DATEI GELADEN:", __file__)
|
||||
|
||||
class Pruefmanager:
|
||||
"""
|
||||
Zentrale Verarbeitung von pruef_ergebnis-Objekten.
|
||||
|
||||
Erwartete öffentliche API (verwendet von Core-Komponenten wie DataGrabber):
|
||||
- report_error(thema, meldung, *, aktion: Optional[PruefAktion]=None, kontext=None) -> None
|
||||
- request_decision(pruef_res) -> str
|
||||
- report_summary(summary: dict) -> None
|
||||
- verarbeite(ergebnis: pruef_ergebnis) -> pruef_ergebnis
|
||||
"""
|
||||
|
||||
def __init__(self, ui_modus: str = "qgis", parent: Optional[Any] = None):
|
||||
def __init__(self, ui_modus: str = "qgis", parent: Optional[Any] = None) -> None:
|
||||
self.ui_modus = ui_modus
|
||||
self.parent = parent
|
||||
|
||||
# ---------------------------------------------------------------------
|
||||
# Basis-API: Meldungen / Zusammenfassungen
|
||||
# ---------------------------------------------------------------------
|
||||
def report_error(self, thema: str, meldung: str, *, aktion: Optional[PruefAktion] = None, kontext: Optional[Any] = None) -> None:
|
||||
"""
|
||||
Einheitliche Meldung für Fehler/Warnungen aus dem Core.
|
||||
Keine Rückgabe; dient als zentraler Hook für Logging/UI.
|
||||
"""
|
||||
# ------------------------------------------------------------------
|
||||
# Meldungen / Zusammenfassungen
|
||||
# ------------------------------------------------------------------
|
||||
def report_error(
|
||||
self,
|
||||
thema: str,
|
||||
meldung: str,
|
||||
*,
|
||||
aktion: Optional[PruefAktion] = None,
|
||||
kontext: Optional[Any] = None,
|
||||
) -> None:
|
||||
critical_actions = {
|
||||
"netzwerkfehler",
|
||||
"pruefe_exception",
|
||||
"save_exception",
|
||||
"layer_create_failed",
|
||||
"read_error",
|
||||
"open_error",
|
||||
"netzwerkfehler", "pruefe_exception", "save_exception",
|
||||
"layer_create_failed", "read_error", "open_error",
|
||||
}
|
||||
warn_actions = {
|
||||
"datei_nicht_gefunden",
|
||||
"pfad_nicht_gefunden",
|
||||
"url_nicht_erreichbar",
|
||||
"falsche_endung",
|
||||
"kein_header",
|
||||
"kein_arbeitsblatt",
|
||||
"datei_nicht_gefunden", "pfad_nicht_gefunden", "url_nicht_erreichbar",
|
||||
"falsche_endung", "kein_header", "kein_arbeitsblatt",
|
||||
}
|
||||
|
||||
if aktion in critical_actions:
|
||||
error(thema, meldung)
|
||||
return
|
||||
|
||||
if aktion in warn_actions:
|
||||
warning(thema, meldung)
|
||||
return
|
||||
|
||||
# Default: informative Warnung
|
||||
warning(thema, meldung)
|
||||
|
||||
def report_summary(self, summary: dict) -> None:
|
||||
@@ -75,202 +54,159 @@ class Pruefmanager:
|
||||
f"Dienste ausserhalb: {len(ausserhalb)}\n"
|
||||
f"Fehler: {len(fehler)}"
|
||||
)
|
||||
|
||||
info("DataGrabber Zusammenfassung", message)
|
||||
|
||||
# ---------------------------------------------------------------------
|
||||
# Entscheidungs-API
|
||||
# ---------------------------------------------------------------------
|
||||
def request_decision(self, pruef_res: Any) -> str:
|
||||
"""
|
||||
Synchronously request a decision from the user (or return a default in headless mode).
|
||||
# ------------------------------------------------------------------
|
||||
# VERFAHRENS-DB-spezifische Entscheidungen
|
||||
# ------------------------------------------------------------------
|
||||
def _handle_datei_existiert(self, ergebnis: pruef_ergebnis) -> pruef_ergebnis:
|
||||
if self.ui_modus != "qgis":
|
||||
return ergebnis
|
||||
|
||||
Returns one of:
|
||||
- "abort"
|
||||
- "continue"
|
||||
- "temporaer_erzeugen"
|
||||
- "ignore"
|
||||
"""
|
||||
aktion = getattr(pruef_res, "aktion", None)
|
||||
meldung = getattr(pruef_res, "meldung", str(pruef_res))
|
||||
pfad = ergebnis.kontext
|
||||
pfad_str = str(pfad) if pfad else "unbekannt"
|
||||
|
||||
titel = "Verfahrens-DB existiert bereits"
|
||||
meldung = (
|
||||
f"Die Datei '{pfad_str}' existiert bereits.\n\n"
|
||||
"Was soll geschehen?\n\n"
|
||||
"• **Überschreiben**: Bestehende Layer ersetzen\n"
|
||||
"• **Anhängen**: Neue Layer hinzufügen\n"
|
||||
"• **Überspringen**: Nur temporäre Layer erzeugen"
|
||||
)
|
||||
|
||||
# Vereinfacht: Erst Überschreiben? → Dann Anhängen? → Überspringen
|
||||
if ask_yes_no(
|
||||
titel,
|
||||
f"{meldung}\n\n**Überschreiben** (alle Layer ersetzen)?",
|
||||
default=False,
|
||||
parent=self.parent
|
||||
):
|
||||
return pruef_ergebnis(
|
||||
ok=True,
|
||||
aktion="datei_existiert_ueberschreiben",
|
||||
kontext=ergebnis.kontext,
|
||||
)
|
||||
|
||||
if ask_yes_no(
|
||||
titel,
|
||||
f"{meldung}\n\n**Anhängen** (neue Layer hinzufügen)?",
|
||||
default=False,
|
||||
parent=self.parent
|
||||
):
|
||||
return pruef_ergebnis(
|
||||
ok=True,
|
||||
aktion="datei_existiert_anhaengen",
|
||||
kontext=ergebnis.kontext,
|
||||
)
|
||||
|
||||
if ask_yes_no(
|
||||
titel,
|
||||
f"{meldung}\n\n**Überspringen** (nur temporäre Layer)?",
|
||||
default=True,
|
||||
parent=self.parent
|
||||
):
|
||||
return pruef_ergebnis(
|
||||
ok=True,
|
||||
aktion="datei_existiert_ueberspringen",
|
||||
kontext=ergebnis.kontext,
|
||||
)
|
||||
|
||||
return ergebnis
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Basis-Entscheidungen (KORREKT: → pruef_ergebnis)
|
||||
# ------------------------------------------------------------------
|
||||
def _handle_basic_decision(self, ergebnis: pruef_ergebnis) -> pruef_ergebnis:
|
||||
"""Basis-Entscheidung für einfache Ja/Nein-Fragen."""
|
||||
print(f"DEBUG _handle_basic_decision: aktion='{ergebnis.aktion}', ui_modus='{self.ui_modus}'")
|
||||
|
||||
if self.ui_modus != "qgis":
|
||||
print("DEBUG: Nicht QGIS → ergebnis unverändert")
|
||||
return ergebnis
|
||||
|
||||
title_map = {
|
||||
"leereingabe_erlaubt": "Ohne Eingabe fortfahren",
|
||||
"standarddatei_vorschlagen": "Standarddatei verwenden",
|
||||
"temporaer_erlaubt": "Temporäre Layer erzeugen",
|
||||
"layer_unsichtbar": "Layer einblenden",
|
||||
}
|
||||
|
||||
title = title_map.get(ergebnis.aktion, "Entscheidung erforderlich")
|
||||
meldung = ergebnis.meldung or ""
|
||||
|
||||
try:
|
||||
print(f"DEBUG ask_yes_no: title='{title}', meldung='{meldung[:50]}...'")
|
||||
yes = ask_yes_no(title, meldung, default=False, parent=self.parent)
|
||||
print(f"DEBUG ask_yes_no: yes={yes}")
|
||||
except Exception as e:
|
||||
print(f"DEBUG ask_yes_no Exception: {e}")
|
||||
return ergebnis
|
||||
|
||||
if not yes:
|
||||
print("DEBUG: Nutzer sagte Nein → ok=False")
|
||||
return ergebnis
|
||||
|
||||
# Nutzer sagte Ja
|
||||
if ergebnis.aktion == "temporaer_erlaubt":
|
||||
print("DEBUG: temporaer_erlaubt bestätigt → ok=True")
|
||||
return pruef_ergebnis(
|
||||
ok=True,
|
||||
aktion="temporaer_erlaubt",
|
||||
kontext=ergebnis.kontext
|
||||
)
|
||||
|
||||
print("DEBUG: Andere Aktion bestätigt → ok=True, aktion='ok'")
|
||||
return pruef_ergebnis(
|
||||
ok=True,
|
||||
aktion="ok",
|
||||
kontext=ergebnis.kontext
|
||||
)
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Hauptlogik: verarbeite() (KORRIGIERT!)
|
||||
# ------------------------------------------------------------------
|
||||
def verarbeite(self, ergebnis: pruef_ergebnis) -> pruef_ergebnis:
|
||||
print("🔥 verarbeite() START")
|
||||
print("DEBUG Pruefmanager:", ergebnis.ok, ergebnis.aktion)
|
||||
print("DEBUG ergebnis.aktion TYPE:", type(ergebnis.aktion), repr(ergebnis.aktion))
|
||||
|
||||
# 1. Erfolg → direkt weiter
|
||||
print("🔍 Schritt 1: Prüfe ergebnis.ok =", ergebnis.ok)
|
||||
if ergebnis.ok:
|
||||
print("✅ Schritt 1: ok=True → return")
|
||||
return ergebnis
|
||||
|
||||
# 2. VERFAHRENS-DB: Bestehende Datei
|
||||
print("🔍 Schritt 2: Prüfe datei_existiert =", ergebnis.aktion == "datei_existiert")
|
||||
if ergebnis.aktion == "datei_existiert":
|
||||
print("✅ Schritt 2: _handle_datei_existiert")
|
||||
return self._handle_datei_existiert(ergebnis)
|
||||
|
||||
# 3. Basis interaktive Aktionen
|
||||
print("🔍 Schritt 3: Definiere interactive_actions")
|
||||
interactive_actions = {
|
||||
"leereingabe_erlaubt",
|
||||
"standarddatei_vorschlagen",
|
||||
"standarddatei_vorschlagen",
|
||||
"temporaer_erlaubt",
|
||||
"layer_unsichtbar",
|
||||
}
|
||||
print("DEBUG interactive_actions:", repr(interactive_actions))
|
||||
print("DEBUG ergebnis.aktion in interactive_actions?", ergebnis.aktion in interactive_actions)
|
||||
|
||||
if ergebnis.aktion in interactive_actions:
|
||||
print("✅ Schritt 3: Interaktive Aktion → _handle_basic_decision")
|
||||
decision = self._handle_basic_decision(ergebnis)
|
||||
print(f"DEBUG: _handle_basic_decision Ergebnis: ok={decision.ok}, aktion='{decision.aktion}'")
|
||||
return decision
|
||||
|
||||
if aktion in interactive_actions:
|
||||
if self.ui_modus == "qgis":
|
||||
title_map = {
|
||||
"leereingabe_erlaubt": "Ohne Eingabe fortfahren",
|
||||
"standarddatei_vorschlagen": "Standarddatei verwenden",
|
||||
"temporaer_erlaubt": "Temporäre Datei erzeugen",
|
||||
"layer_unsichtbar": "Layer einblenden",
|
||||
}
|
||||
title = title_map.get(aktion, "Entscheidung erforderlich")
|
||||
try:
|
||||
yes = ask_yes_no(title, meldung, default=False, parent=self.parent)
|
||||
except Exception:
|
||||
return "abort"
|
||||
if yes:
|
||||
if aktion == "temporaer_erlaubt":
|
||||
return "temporaer_erzeugen"
|
||||
return "continue"
|
||||
return "abort"
|
||||
|
||||
if self.ui_modus == "headless":
|
||||
return "abort"
|
||||
|
||||
informational_actions = {
|
||||
"leer",
|
||||
"datei_nicht_gefunden",
|
||||
"pfad_nicht_gefunden",
|
||||
"url_nicht_erreichbar",
|
||||
"netzwerkfehler",
|
||||
"falscher_geotyp",
|
||||
"layer_leer",
|
||||
"falscher_layertyp",
|
||||
"falsches_crs",
|
||||
"felder_fehlen",
|
||||
"datenquelle_unerwartet",
|
||||
"layer_nicht_editierbar",
|
||||
"kein_header",
|
||||
"kein_arbeitsblatt",
|
||||
"read_error",
|
||||
"open_error",
|
||||
"pflichtfelder_fehlen",
|
||||
}
|
||||
if aktion in informational_actions:
|
||||
return "abort"
|
||||
|
||||
return "abort"
|
||||
|
||||
# ---------------------------------------------------------------------
|
||||
# Höhere Abstraktion: verarbeite
|
||||
# ---------------------------------------------------------------------
|
||||
def verarbeite(self, ergebnis: pruef_ergebnis) -> pruef_ergebnis:
|
||||
"""
|
||||
Verarbeitet ein pruef_ergebnis-Objekt und führt ggf. Nutzerinteraktion durch.
|
||||
Liefert ein ggf. modifiziertes pruef_ergebnis zurück.
|
||||
"""
|
||||
if ergebnis.ok:
|
||||
return ergebnis
|
||||
|
||||
aktion = ergebnis.aktion
|
||||
kontext = ergebnis.kontext
|
||||
meldung = ergebnis.meldung
|
||||
|
||||
# Zentrale Meldung
|
||||
self.report_error(aktion or "pruefung", meldung or "", aktion=aktion, kontext=kontext)
|
||||
|
||||
# Interaktive Entscheidungen
|
||||
if aktion in ("leereingabe_erlaubt", "standarddatei_vorschlagen", "temporaer_erlaubt", "layer_unsichtbar"):
|
||||
decision = self.request_decision(ergebnis)
|
||||
if decision == "temporaer_erzeugen":
|
||||
return pruef_ergebnis(ok=True, meldung="Temporäre Datei soll erzeugt werden.", aktion="temporaer_erzeugen", kontext=None)
|
||||
if decision == "continue":
|
||||
return pruef_ergebnis(ok=True, meldung="Fortgefahren.", aktion="ok", kontext=kontext)
|
||||
return ergebnis # abort / unverändert
|
||||
|
||||
# Spezielle Excel/Importer-Fälle: klare Meldungen, keine interaktive Entscheidung
|
||||
if aktion == "kein_header":
|
||||
warning("Excel-Import", meldung or "")
|
||||
return ergebnis
|
||||
|
||||
if aktion == "kein_arbeitsblatt":
|
||||
warning("Excel-Import", meldung or "")
|
||||
return ergebnis
|
||||
|
||||
if aktion in ("read_error", "open_error"):
|
||||
error("Excel-Import", meldung or "")
|
||||
return ergebnis
|
||||
|
||||
if aktion == "datei_nicht_gefunden":
|
||||
warning("Datei nicht gefunden", meldung or "")
|
||||
return ergebnis
|
||||
|
||||
# Spezieller Fall: layer_unsichtbar (falls nicht interaktiv behandelt)
|
||||
if aktion == "layer_unsichtbar":
|
||||
if kontext is not None:
|
||||
try:
|
||||
set_layer_visible(kontext, True)
|
||||
return pruef_ergebnis(ok=True, meldung="Layer wurde eingeblendet.", aktion="ok", kontext=kontext)
|
||||
except Exception:
|
||||
return ergebnis
|
||||
return ergebnis
|
||||
|
||||
# Standard: keine Änderung
|
||||
# 4. Fehler behandeln
|
||||
print("❌ Schritt 4: FEHLER BEHANDELN")
|
||||
self.report_error(
|
||||
thema=ergebnis.aktion or "pruefung",
|
||||
meldung=ergebnis.meldung or "",
|
||||
aktion=ergebnis.aktion,
|
||||
kontext=ergebnis.kontext,
|
||||
)
|
||||
print("🔥 verarbeite() ENDE mit ok=False")
|
||||
return ergebnis
|
||||
def ask_overwrite_append_cancel(self, layer_name: str, default: str = "overwrite") -> str:
|
||||
"""
|
||||
Zeigt dem Nutzer eine Auswahl für einen bereits existierenden Layer an.
|
||||
|
||||
Rückgabe
|
||||
-------
|
||||
str
|
||||
Einer der Werte: "overwrite", "append", "cancel".
|
||||
|
||||
Verhalten
|
||||
--------
|
||||
- Verwendet bevorzugt die UI-Wrapper-Funktion `qt_wrapper` / `qgisui_wrapper`,
|
||||
falls vorhanden (z. B. ein QMessageBox-Dialog mit drei Buttons).
|
||||
- Im Mock- oder Headless-Modus (kein Qt/QGIS verfügbar) wird der übergebene
|
||||
`default`-Wert zurückgegeben.
|
||||
- Alle Nutzerinteraktionen laufen über diese zentrale Methode, damit das
|
||||
Plugin an einer Stelle gesteuert und ggf. getested werden kann.
|
||||
|
||||
Parameter
|
||||
---------
|
||||
layer_name:
|
||||
Anzeigename des Layers, der bereits existiert (wird im Dialog angezeigt).
|
||||
default:
|
||||
Rückgabewert im Headless/Mock-Modus oder wenn der Dialog nicht verfügbar ist.
|
||||
Gültige Werte: "overwrite", "append", "cancel". Standard: "overwrite".
|
||||
"""
|
||||
# Validierung des Defaults
|
||||
if default not in ("overwrite", "append", "cancel"):
|
||||
default = "overwrite"
|
||||
|
||||
# Versuche, eine UI-Wrapper-Funktion zu verwenden, falls vorhanden
|
||||
try:
|
||||
# qgisui_wrapper kann eine spezialisierte Dialogfunktion bereitstellen
|
||||
from sn_basis.functions import qgisui_wrapper as qgisui
|
||||
ask_fn = getattr(qgisui, "ask_overwrite_append_cancel", None)
|
||||
if callable(ask_fn):
|
||||
# Die Wrapper-Funktion soll genau die drei Strings zurückgeben
|
||||
choice = ask_fn(layer_name)
|
||||
if choice in ("overwrite", "append", "cancel"):
|
||||
return choice
|
||||
except Exception:
|
||||
# Falls Import/Wrapper fehlschlägt, weiter zum Qt-Fallback
|
||||
pass
|
||||
|
||||
# Fallback: direkte Qt-Dialoge über qt_wrapper (wenn verfügbar)
|
||||
try:
|
||||
from sn_basis.functions import qt_wrapper as qt
|
||||
QMessageBox = getattr(qt, "QMessageBox", None)
|
||||
if QMessageBox is not None:
|
||||
# Erzeuge und konfiguriere Dialog
|
||||
msg = QMessageBox()
|
||||
msg.setWindowTitle("Layer bereits vorhanden")
|
||||
msg.setText(f"Der Layer '{layer_name}' existiert bereits. Was möchten Sie tun?")
|
||||
overwrite_btn = msg.addButton("Überschreiben", QMessageBox.AcceptRole)
|
||||
append_btn = msg.addButton("Anhängen", QMessageBox.AcceptRole)
|
||||
cancel_btn = msg.addButton("Abbrechen", QMessageBox.RejectRole)
|
||||
msg.setDefaultButton(overwrite_btn)
|
||||
# Blockierend anzeigen
|
||||
msg.exec_()
|
||||
clicked = msg.clickedButton()
|
||||
if clicked == overwrite_btn:
|
||||
return "overwrite"
|
||||
if clicked == append_btn:
|
||||
return "append"
|
||||
return "cancel"
|
||||
except Exception:
|
||||
# Qt nicht verfügbar oder Fehler beim Dialogaufbau
|
||||
pass
|
||||
|
||||
# Headless / Mock: gib Default zurück
|
||||
return default
|
||||
|
||||
@@ -1,9 +1,21 @@
|
||||
"""
|
||||
sn_basis/modules/pruef_ergebnis.py
|
||||
|
||||
Erweitertes Ergebnisobjekt für Dateiprüfungen mit Verfahrens-DB-spezifischen Aktionen.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
from dataclasses import dataclass
|
||||
from typing import Any, Optional, Literal
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
# =============================================================================
|
||||
# Erweiterte PruefAktionen für Verfahrens-DB-Workflow
|
||||
# =============================================================================
|
||||
|
||||
# Erweitertes Literal mit allen erlaubten Aktionen (PruefAktion)
|
||||
PruefAktion = Literal[
|
||||
# Basis-Aktionen (bestehend)
|
||||
"ok",
|
||||
"leer",
|
||||
"leereingabe_erlaubt",
|
||||
@@ -16,6 +28,8 @@ PruefAktion = Literal[
|
||||
"pfad_nicht_gefunden",
|
||||
"url_nicht_erreichbar",
|
||||
"netzwerkfehler",
|
||||
|
||||
# Layer-spezifisch
|
||||
"layer_nicht_gefunden",
|
||||
"layer_unsichtbar",
|
||||
"falscher_geotyp",
|
||||
@@ -25,15 +39,26 @@ PruefAktion = Literal[
|
||||
"felder_fehlen",
|
||||
"datenquelle_unerwartet",
|
||||
"layer_nicht_editierbar",
|
||||
|
||||
# Dateiendung/Format
|
||||
"falsche_endung",
|
||||
"pflichtfelder_fehlen",
|
||||
# Excel / Import-spezifische Aktionen
|
||||
|
||||
# Excel/Import
|
||||
"kein_header",
|
||||
"kein_arbeitsblatt",
|
||||
"read_error",
|
||||
"open_error",
|
||||
"datenabruf",
|
||||
# Generische Prüf-/Speicher-Aktionen
|
||||
|
||||
# 🆕 VERFAHRENS-DB SPEZIFISCH (deine Anforderungen 2.d, 2.e)
|
||||
"datei_wird_erzeugt", # 2.d: Pfad gültig, Datei fehlt → weiter
|
||||
"datei_existiert", # Datei vorhanden → Layer-Entscheidung
|
||||
"datei_existiert_ueberschreiben", # 2.e: Nutzer wählt "Überschreiben"
|
||||
"datei_existiert_anhaengen", # 2.e: Nutzer wählt "Anhängen"
|
||||
"datei_existiert_ueberspringen", # 2.e: Nutzer wählt "Überspringen"
|
||||
|
||||
# Generisch
|
||||
"pruefe_exception",
|
||||
"save_exception",
|
||||
"save_not_implemented",
|
||||
@@ -42,22 +67,113 @@ PruefAktion = Literal[
|
||||
"needs_user_action",
|
||||
]
|
||||
|
||||
|
||||
@dataclass
|
||||
class pruef_ergebnis:
|
||||
"""
|
||||
Einheitliches Ergebnisobjekt für Prüfer.
|
||||
- ok: True wenn Prüfung bestanden
|
||||
- meldung: menschenlesbare Meldung
|
||||
- aktion: maschinenlesbarer Aktionscode (PruefAktion)
|
||||
- kontext: optionaler Zusatzkontext (z. B. Pfad, Layer-Objekt)
|
||||
Einheitliches Ergebnisobjekt für Prüfer im Verfahrens-DB-Workflow.
|
||||
|
||||
Attributes
|
||||
----------
|
||||
ok : bool
|
||||
True wenn Prüfung bestanden und Pipeline fortgesetzt werden kann.
|
||||
False signalisiert Fehler oder Nutzerentscheidung erforderlich.
|
||||
meldung : Optional[str], optional
|
||||
Menschenlesbare Meldung für UI-Dialoge (BY: Pruefmanager).
|
||||
aktion : Optional[PruefAktion], optional
|
||||
Maschinenlesbarer Aktionscode für nachfolgende Pipeline-Schritte.
|
||||
kontext : Optional[Any], optional
|
||||
Zusatzkontext: meist `pathlib.Path` für Dateipfade oder Layer-Objekte.
|
||||
|
||||
Verfahrens-DB-spezifische Aktionen:
|
||||
|
||||
+-----------------------------+-------------------------------------------------+
|
||||
| Aktion | Bedeutung |
|
||||
+=============================+=================================================+
|
||||
| ``datei_wird_erzeugt`` | 2.d: Neues GPKG wird angelegt (Pfad gültig) |
|
||||
+-----------------------------+-------------------------------------------------+
|
||||
| ``datei_existiert`` | Datei vorhanden → Layer-Überschreibung prüfen |
|
||||
+-----------------------------+-------------------------------------------------+
|
||||
| ``datei_existiert_*`` | 2.e: Nutzerentscheidung für bestehende Datei |
|
||||
+-----------------------------+-------------------------------------------------+
|
||||
"""
|
||||
|
||||
ok: bool
|
||||
meldung: Optional[str] = None
|
||||
aktion: Optional[PruefAktion] = None
|
||||
kontext: Optional[Any] = None
|
||||
|
||||
def __init__(self, ok: bool, meldung: Optional[str] = None, aktion: Optional[PruefAktion] = None, kontext: Optional[Any] = None):
|
||||
def __init__(
|
||||
self,
|
||||
ok: bool,
|
||||
meldung: Optional[str] = None,
|
||||
aktion: Optional[PruefAktion] = None,
|
||||
kontext: Optional[Any] = None,
|
||||
) -> None:
|
||||
"""
|
||||
Erstellt ein neues Prüfergebnis.
|
||||
|
||||
Parameters
|
||||
----------
|
||||
ok : bool
|
||||
True für "weiter mit Pipeline", False für "Entscheidung/Fehler".
|
||||
meldung : Optional[str]
|
||||
UI-Text für Nutzerdialoge.
|
||||
aktion : Optional[PruefAktion]
|
||||
Maschinenaktion für nachfolgende Verarbeitung.
|
||||
kontext : Optional[Any]
|
||||
Typischerweise `pathlib.Path` (Dateipfad) oder `QgsVectorLayer`.
|
||||
"""
|
||||
self.ok = ok
|
||||
self.meldung = meldung
|
||||
self.aktion = aktion
|
||||
self.kontext = kontext
|
||||
|
||||
@property
|
||||
def ist_verfahrens_db_aktion(self) -> bool:
|
||||
"""
|
||||
Prüft, ob es sich um eine Verfahrens-DB-spezifische Aktion handelt.
|
||||
|
||||
Returns
|
||||
-------
|
||||
bool
|
||||
True für ``datei_wird_erzeugt`` oder ``datei_existiert*``.
|
||||
"""
|
||||
return self.aktion in {
|
||||
"datei_wird_erzeugt",
|
||||
"datei_existiert",
|
||||
"datei_existiert_ueberschreiben",
|
||||
"datei_existiert_anhaengen",
|
||||
"datei_existiert_ueberspringen",
|
||||
}
|
||||
|
||||
@property
|
||||
def dateipfad(self) -> Optional[Path]:
|
||||
"""
|
||||
Extrahiert den Dateipfad aus dem Kontext (falls vorhanden).
|
||||
|
||||
Returns
|
||||
-------
|
||||
Optional[Path]
|
||||
`Path`-Objekt oder None.
|
||||
"""
|
||||
if isinstance(self.kontext, Path):
|
||||
return self.kontext
|
||||
return None
|
||||
|
||||
@property
|
||||
def erlaubte_persistierung(self) -> bool:
|
||||
"""
|
||||
Prüft, ob die Pipeline Daten persistieren darf.
|
||||
|
||||
Returns
|
||||
-------
|
||||
bool
|
||||
True für ``datei_wird_erzeugt``, ``datei_existiert_ueberschreiben``,
|
||||
``datei_existiert_anhaengen``.
|
||||
"""
|
||||
return self.aktion in {
|
||||
"datei_wird_erzeugt",
|
||||
"datei_existiert_ueberschreiben",
|
||||
"datei_existiert_anhaengen",
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user