Auf Wrapper umgestellt, Prüfarchitektur QT6-kompatibel gemacht (Nicht lauffähig)

This commit is contained in:
2025-12-18 22:00:31 +01:00
parent f64d56d4bc
commit e8fea163b5
31 changed files with 2791 additions and 889 deletions

View File

@@ -1,77 +1,141 @@
# Linkpruefer.py Qt5/Qt6-kompatibel über qt_compat
"""
sn_basis/modules/linkpruefer.py Prüfung von URLs und lokalen Links.
Verwendet syswrapper und qgisqt_wrapper.
Gibt pruef_ergebnis an den Pruefmanager zurück.
"""
from modules.qt_compat import (
QEventLoop,
QUrl,
QNetworkRequest,
QNetworkReply
from sn_basis.functions.syswrapper import (
file_exists,
is_file,
join_path,
)
from qgis.core import QgsNetworkAccessManager
from modules.pruef_ergebnis import PruefErgebnis
from sn_basis.functions.qgisqt_wrapper import (
network_head,
)
from sn_basis.modules.Pruefmanager import pruef_ergebnis
class Linkpruefer:
"""Prüft den Link mit QgsNetworkAccessManager und klassifiziert Anbieter nach Attribut."""
"""
Prüft URLs und lokale Pfade.
Die eigentliche Nutzerinteraktion übernimmt der Pruefmanager.
"""
ANBIETER_TYPEN: dict[str, str] = {
"REST": "REST",
"WFS": "WFS",
"WMS": "WMS",
"OGR": "OGR"
}
def __init__(self, basis_pfad: str | None = None):
"""
basis_pfad: optionaler Basisordner für relative Pfade.
"""
self.basis = basis_pfad
def __init__(self, link: str, anbieter: str):
self.link = link
self.anbieter = anbieter.upper().strip() if anbieter else ""
self.network_manager = QgsNetworkAccessManager()
# ---------------------------------------------------------
# Hilfsfunktionen
# ---------------------------------------------------------
def klassifiziere_anbieter(self):
typ = self.ANBIETER_TYPEN.get(self.anbieter, self.anbieter)
quelle = "remote" if self.link.startswith(("http://", "https://")) else "local"
return {"typ": typ, "quelle": quelle}
def _pfad(self, relativer_pfad: str) -> str:
"""Erzeugt einen OS-unabhängigen Pfad relativ zum Basisverzeichnis."""
if not self.basis:
return relativer_pfad
return join_path(self.basis, relativer_pfad)
def pruefe_link(self):
fehler = []
warnungen = []
def _ist_url(self, text: str) -> bool:
"""Einfache URL-Erkennung."""
return text.startswith("http://") or text.startswith("https://")
if not self.link:
fehler.append("Link fehlt.")
return PruefErgebnis(False, fehler=fehler, warnungen=warnungen)
# ---------------------------------------------------------
# Hauptfunktion
# ---------------------------------------------------------
if not self.anbieter or not self.anbieter.strip():
fehler.append("Anbieter muss gesetzt werden und darf nicht leer sein.")
def pruefe(self, eingabe: str) -> pruef_ergebnis:
"""
Prüft einen Link (URL oder lokalen Pfad).
Rückgabe: pruef_ergebnis
"""
# Remote-Links prüfen
if self.link.startswith(("http://", "https://")):
request = QNetworkRequest(QUrl(self.link))
reply = self.network_manager.head(request)
if not eingabe:
return pruef_ergebnis(
ok=False,
meldung="Es wurde kein Link angegeben.",
aktion="leer",
pfad=None,
)
loop = QEventLoop()
reply.finished.connect(loop.quit)
loop.exec() # Qt5/Qt6-kompatibel über qt_compat
# -----------------------------------------------------
# 1. Fall: URL
# -----------------------------------------------------
if self._ist_url(eingabe):
return self._pruefe_url(eingabe)
# Fehlerprüfung Qt5/Qt6-kompatibel
if reply.error() != QNetworkReply.NetworkError.NoError:
fehler.append(f"Verbindungsfehler: {reply.errorString()}")
else:
status = reply.attribute(QNetworkRequest.HttpStatusCodeAttribute)
if status is None or status < 200 or status >= 400:
fehler.append(f"Link nicht erreichbar: HTTP {status}")
# -----------------------------------------------------
# 2. Fall: lokaler Pfad
# -----------------------------------------------------
return self._pruefe_dateipfad(eingabe)
reply.deleteLater()
# ---------------------------------------------------------
# URLPrüfung
# ---------------------------------------------------------
else:
# Lokale Pfade: Plausibilitätscheck
if "." not in self.link.split("/")[-1]:
warnungen.append("Der lokale Link sieht ungewöhnlich aus.")
def _pruefe_url(self, url: str) -> pruef_ergebnis:
"""
Prüft eine URL über einen HEADRequest.
"""
return PruefErgebnis(
len(fehler) == 0,
daten=self.klassifiziere_anbieter(),
fehler=fehler,
warnungen=warnungen
reply = network_head(url)
if reply is None:
return pruef_ergebnis(
ok=False,
meldung=f"Die URL '{url}' konnte nicht geprüft werden.",
aktion="netzwerkfehler",
pfad=url,
)
if reply.error != 0:
return pruef_ergebnis(
ok=False,
meldung=f"Die URL '{url}' ist nicht erreichbar.",
aktion="url_nicht_erreichbar",
pfad=url,
)
return pruef_ergebnis(
ok=True,
meldung="URL ist erreichbar.",
aktion="ok",
pfad=url,
)
def ausfuehren(self):
return self.pruefe_link()
# ---------------------------------------------------------
# Lokale Datei/Pfadprüfung
# ---------------------------------------------------------
def _pruefe_dateipfad(self, eingabe: str) -> pruef_ergebnis:
"""
Prüft einen lokalen Pfad.
"""
pfad = self._pfad(eingabe)
if not file_exists(pfad):
return pruef_ergebnis(
ok=False,
meldung=f"Der Pfad '{eingabe}' wurde nicht gefunden.",
aktion="pfad_nicht_gefunden",
pfad=pfad,
)
if not is_file(pfad):
return pruef_ergebnis(
ok=False,
meldung=f"Der Pfad '{eingabe}' ist keine Datei.",
aktion="kein_dateipfad",
pfad=pfad,
)
return pruef_ergebnis(
ok=True,
meldung="Dateipfad ist gültig.",
aktion="ok",
pfad=pfad,
)