"""
|
Update-Checker fuer die Server-Bibliothek.
|
|
Unterstuetzt drei Update-Quellen pro Software:
|
- winget : Windows Paketverwaltung, deckt die meiste bekannte Standard-Software ab
|
- github : GitHub-Releases, fuer Open-Source-Tools ohne winget-Paket
|
- direkt : ein fester Download-Link, der immer zur neuesten Datei fuehrt
|
|
WICHTIGES PRINZIP - Ordner-Konvention:
|
Das Feld "Installer-Pfad" kann entweder auf eine konkrete Datei zeigen
|
(z. B. '...\\7-Zip\\7zip.exe' - alte Schreibweise, funktioniert weiter) oder
|
einfach nur auf einen ORDNER (z. B. '...\\7-Zip\\' - empfohlen). Zeigt es auf
|
einen Ordner, kuemmert sich dieses Modul selbst um einen sauberen, stabilen
|
Dateinamen darin (aus dem Tool-Namen abgeleitet) - niemand muss sich einen
|
Dateinamen ausdenken oder eintragen.
|
|
Jede Pruef-Funktion gibt ein UpdateResult zurueck und aendert NIE die
|
Datenbank direkt - das uebernimmt app.py, nachdem das Ergebnis vorliegt.
|
"""
|
|
import json
|
import os
|
import re
|
import shutil
|
import subprocess
|
import urllib.request
|
import urllib.error
|
from datetime import datetime
|
|
USER_AGENT = "ServerBibliothek-UpdateChecker/1.0"
|
DATEI_ENDUNGEN = (".exe", ".msi", ".msix", ".appx", ".zip")
|
|
|
class UpdateResult:
|
def __init__(self, erfolg, alte_version=None, neue_version=None, meldung=""):
|
self.erfolg = erfolg
|
self.alte_version = alte_version
|
self.neue_version = neue_version
|
self.meldung = meldung
|
|
def unveraendert(self):
|
return self.erfolg and self.alte_version == self.neue_version
|
|
|
# ---------------------------------------------------------------------------
|
# Ordner-Konvention: Dateiname automatisch verwalten
|
# ---------------------------------------------------------------------------
|
|
def sanitisiere_dateiname(name):
|
"""Macht aus einem Tool-Namen einen sicheren Dateinamen-Stamm."""
|
name = (name or "tool").strip()
|
name = re.sub(r'[<>:"/\\|?*]', "", name)
|
name = re.sub(r"\s+", "-", name)
|
return name or "tool"
|
|
|
def ist_datei_ziel(pfad):
|
"""True, wenn der Pfad bereits auf eine konkrete Datei zeigt (erkennbar an
|
der Endung)."""
|
return (pfad or "").lower().endswith(DATEI_ENDUNGEN)
|
|
|
def _roh_ordner_und_wunschname(installer_pfad):
|
"""Zerlegt den eingetragenen Pfad in (Ordneranteil, bevorzugter Dateiname
|
oder None). Egal ob eine Datei-Endung dabei ist oder nicht - der
|
Ordneranteil wird in jedem Fall separat ermittelt, damit anschliessend
|
IMMER in einem Tool-eigenen Unterordner gelandet wird."""
|
pfad = (installer_pfad or "").rstrip("\\/")
|
if ist_datei_ziel(pfad):
|
return os.path.dirname(pfad), os.path.basename(pfad)
|
return pfad, None
|
|
|
def effektiver_ablageordner(installer_pfad, name):
|
"""Ermittelt den tatsaechlichen Ordner, in dem die Datei(en) fuer DIESES
|
Tool liegen sollen - egal ob im Feld ein reiner Ordner oder ein
|
kompletter Dateipfad eingetragen ist (auch aeltere Eintraege). Haengt
|
automatisch einen Unterordner mit dem Tool-Namen an, falls der Ordner
|
noch nicht selbst so heisst. Dadurch landen mehrere Tools nie gemeinsam
|
'nackt' im selben Ordner, selbst wenn ueberall derselbe Basisordner
|
eingetragen wurde oder frueher schon eine Datei direkt dort abgelegt war."""
|
roh_ordner, _ = _roh_ordner_und_wunschname(installer_pfad)
|
stamm = sanitisiere_dateiname(name)
|
letzter_teil = os.path.basename(roh_ordner)
|
if letzter_teil.lower() == stamm.lower():
|
return roh_ordner
|
return os.path.join(roh_ordner, stamm)
|
|
|
def verwaltete_datei_pfad(installer_pfad, name):
|
"""Gibt den Pfad zur automatisch verwalteten Installer-Datei zurueck,
|
falls eine existiert (im Tool-eigenen Unterordner). None, wenn (noch)
|
keine Datei da ist."""
|
if not installer_pfad:
|
return None
|
ordner = effektiver_ablageordner(installer_pfad, name)
|
_, wunschname = _roh_ordner_und_wunschname(installer_pfad)
|
kandidaten = ([wunschname] if wunschname else []) + [
|
sanitisiere_dateiname(name) + endung for endung in DATEI_ENDUNGEN
|
]
|
for kandidat in kandidaten:
|
pfad = os.path.join(ordner, kandidat)
|
if os.path.isfile(pfad):
|
return pfad
|
# Uebergangsweise: eine aeltere, noch nicht verschachtelte Datei direkt
|
# am urspruenglich eingetragenen Pfad findet der Download-Button auch
|
# noch - bis der naechste automatische Check eine frische Kopie im
|
# Tool-eigenen Unterordner anlegt.
|
if ist_datei_ziel(installer_pfad) and os.path.isfile(installer_pfad):
|
return installer_pfad
|
return None
|
|
|
def _ziel_datei_ermitteln(installer_pfad, name, quell_dateiname):
|
"""Bestimmt den tatsaechlichen Ziel-Dateipfad fuer einen Download: immer
|
im Tool-eigenen Unterordner. Der Dateiname folgt einer schon frueher
|
explizit eingetragenen Datei-Angabe, sonst wird er automatisch aus dem
|
Tool-Namen + Endung der heruntergeladenen Datei gebildet."""
|
ordner = effektiver_ablageordner(installer_pfad, name)
|
_, wunschname = _roh_ordner_und_wunschname(installer_pfad)
|
if wunschname:
|
return os.path.join(ordner, wunschname)
|
endung = os.path.splitext(quell_dateiname)[1].lower()
|
if endung not in DATEI_ENDUNGEN:
|
endung = ".exe"
|
return os.path.join(ordner, sanitisiere_dateiname(name) + endung)
|
|
|
def _endung_aus_url(url):
|
pfad_teil = url.split("?")[0].rstrip("/")
|
_, ext = os.path.splitext(pfad_teil)
|
return ext.lower() if ext.lower() in DATEI_ENDUNGEN else ".exe"
|
|
|
# ---------------------------------------------------------------------------
|
# Hilfsfunktionen
|
# ---------------------------------------------------------------------------
|
|
def _http_get_json(url):
|
req = urllib.request.Request(
|
url, headers={"User-Agent": USER_AGENT, "Accept": "application/vnd.github+json"}
|
)
|
with urllib.request.urlopen(req, timeout=20) as resp:
|
return json.loads(resp.read().decode("utf-8"))
|
|
|
def _http_download(url, timeout=120):
|
"""Laedt eine Datei in eine temporaere Datei und gibt den Pfad zurueck."""
|
req = urllib.request.Request(url, headers={"User-Agent": USER_AGENT})
|
tmp_pfad = os.path.join(
|
os.environ.get("TEMP", "."), f"_sb_download_{datetime.now().strftime('%Y%m%d%H%M%S%f')}"
|
)
|
with urllib.request.urlopen(req, timeout=timeout) as resp, open(tmp_pfad, "wb") as f:
|
shutil.copyfileobj(resp, f)
|
return tmp_pfad
|
|
|
def sichern_und_ersetzen(ziel_pfad, neue_datei_pfad):
|
"""Legt eine Sicherheitskopie der alten Datei an (Ordner '_sicherungen'
|
neben dem Installer) und verschiebt dann die neue Datei an den Zielort.
|
ziel_pfad muss eine konkrete Datei sein (kein Ordner)."""
|
if os.path.isdir(ziel_pfad):
|
raise ValueError(
|
f"Interner Fehler: Zielpfad ist ein Ordner statt einer Datei: '{ziel_pfad}'."
|
)
|
|
ziel_ordner = os.path.dirname(ziel_pfad) or "."
|
os.makedirs(ziel_ordner, exist_ok=True)
|
|
if os.path.isfile(ziel_pfad):
|
sicherungs_ordner = os.path.join(ziel_ordner, "_sicherungen")
|
os.makedirs(sicherungs_ordner, exist_ok=True)
|
zeitstempel = datetime.now().strftime("%Y%m%d_%H%M%S")
|
sicherungs_ziel = os.path.join(
|
sicherungs_ordner, f"{os.path.basename(ziel_pfad)}.{zeitstempel}.bak"
|
)
|
shutil.copy2(ziel_pfad, sicherungs_ziel)
|
|
shutil.move(neue_datei_pfad, ziel_pfad)
|
|
|
# ---------------------------------------------------------------------------
|
# Quelle: GitHub Releases
|
# ---------------------------------------------------------------------------
|
|
def _normalisiere_github_pfad(wert):
|
"""Macht das Feld tolerant gegenueber versehentlich mitkopierten
|
URL-Bestandteilen, z. B. 'github.com/ip7z/7zip' -> 'ip7z/7zip'."""
|
wert = (wert or "").strip()
|
for praefix in ("https://github.com/", "http://github.com/", "www.github.com/", "github.com/"):
|
if wert.lower().startswith(praefix):
|
wert = wert[len(praefix):]
|
break
|
wert = wert.strip("/")
|
if wert.lower().endswith(".git"):
|
wert = wert[:-4]
|
return wert
|
|
|
def pruefe_github(name, update_wert, aktuelle_version, installer_pfad):
|
"""update_wert Format: 'besitzer/projekt', z. B. 'videolan/vlc' """
|
update_wert = _normalisiere_github_pfad(update_wert)
|
if not update_wert or "/" not in update_wert:
|
return UpdateResult(False, aktuelle_version, None,
|
"Ungueltiges Format. Erwartet: besitzer/projekt")
|
|
try:
|
daten = _http_get_json(f"https://api.github.com/repos/{update_wert}/releases/latest")
|
except (urllib.error.URLError, urllib.error.HTTPError) as e:
|
return UpdateResult(False, aktuelle_version, None, f"GitHub nicht erreichbar: {e}")
|
|
neue_version = str(daten.get("tag_name", "")).lstrip("vV")
|
if not neue_version:
|
return UpdateResult(False, aktuelle_version, None, "Keine Version im GitHub-Release gefunden.")
|
|
assets = daten.get("assets", [])
|
passendes_asset = next(
|
(a for a in assets if a.get("name", "").lower().endswith((".exe", ".msi"))), None
|
)
|
if not passendes_asset:
|
return UpdateResult(False, aktuelle_version, neue_version,
|
"Neue Version gefunden, aber keine .exe/.msi-Datei im Release vorhanden.")
|
|
if not installer_pfad:
|
return UpdateResult(False, aktuelle_version, neue_version,
|
"Kein Ablageort hinterlegt, Download uebersprungen.")
|
|
ziel_pfad = _ziel_datei_ermitteln(installer_pfad, name, passendes_asset["name"])
|
datei_fehlt = not os.path.isfile(ziel_pfad)
|
|
if neue_version == (aktuelle_version or "") and not datei_fehlt:
|
return UpdateResult(True, aktuelle_version, aktuelle_version, "Bereits aktuell.")
|
|
try:
|
tmp = _http_download(passendes_asset["browser_download_url"])
|
sichern_und_ersetzen(ziel_pfad, tmp)
|
except Exception as e:
|
return UpdateResult(False, aktuelle_version, neue_version, f"Download fehlgeschlagen: {e}")
|
|
meldung = "Datei fehlte, wurde neu heruntergeladen." if datei_fehlt else "Erfolgreich aktualisiert."
|
return UpdateResult(True, aktuelle_version, neue_version, meldung)
|
|
|
# ---------------------------------------------------------------------------
|
# Quelle: winget (Windows Paketverwaltung)
|
# ---------------------------------------------------------------------------
|
|
def _winget_verfuegbar():
|
return shutil.which("winget") is not None
|
|
|
def pruefe_winget(name, update_wert, aktuelle_version, installer_pfad):
|
"""update_wert = winget-Paket-ID, z. B. '7zip.7zip' """
|
update_wert = (update_wert or "").strip()
|
if not update_wert:
|
return UpdateResult(False, aktuelle_version, None, "Keine winget-Paket-ID hinterlegt.")
|
|
if not _winget_verfuegbar():
|
return UpdateResult(False, aktuelle_version, None, "winget ist auf diesem Server nicht verfuegbar.")
|
|
try:
|
ausgabe = subprocess.run(
|
["winget", "show", "--id", update_wert, "--exact", "--accept-source-agreements"],
|
capture_output=True, text=True, timeout=60,
|
)
|
except Exception as e:
|
return UpdateResult(False, aktuelle_version, None, f"winget-Aufruf fehlgeschlagen: {e}")
|
|
treffer = re.search(r"^Version:\s*(.+)$", ausgabe.stdout, re.MULTILINE)
|
if not treffer:
|
return UpdateResult(False, aktuelle_version, None,
|
f"Version ueber winget nicht auslesbar (Paket-ID korrekt? '{update_wert}')")
|
neue_version = treffer.group(1).strip()
|
|
if not installer_pfad:
|
return UpdateResult(False, aktuelle_version, neue_version,
|
"Kein Ablageort hinterlegt, Download uebersprungen.")
|
|
datei_fehlt = verwaltete_datei_pfad(installer_pfad, name) is None
|
if neue_version == (aktuelle_version or "") and not datei_fehlt:
|
return UpdateResult(True, aktuelle_version, aktuelle_version, "Bereits aktuell.")
|
|
staging_ordner = os.path.join(os.environ.get("TEMP", "."), f"_sb_winget_{update_wert.replace('.', '_')}")
|
if os.path.isdir(staging_ordner):
|
shutil.rmtree(staging_ordner, ignore_errors=True)
|
os.makedirs(staging_ordner, exist_ok=True)
|
|
try:
|
subprocess.run(
|
["winget", "download", "--id", update_wert, "--exact", "-d", staging_ordner,
|
"--accept-source-agreements", "--accept-package-agreements"],
|
capture_output=True, text=True, timeout=300, check=True,
|
)
|
except Exception as e:
|
return UpdateResult(False, aktuelle_version, neue_version, f"Download ueber winget fehlgeschlagen: {e}")
|
|
gefundene_dateien = []
|
for wurzel, _, dateien in os.walk(staging_ordner):
|
for datei in dateien:
|
if datei.lower().endswith((".exe", ".msi")):
|
gefundene_dateien.append(os.path.join(wurzel, datei))
|
|
if not gefundene_dateien:
|
shutil.rmtree(staging_ordner, ignore_errors=True)
|
return UpdateResult(False, aktuelle_version, neue_version,
|
"winget-Download hat keine .exe/.msi-Datei geliefert.")
|
|
neueste_datei = max(gefundene_dateien, key=os.path.getmtime)
|
ziel_pfad = _ziel_datei_ermitteln(installer_pfad, name, os.path.basename(neueste_datei))
|
try:
|
sichern_und_ersetzen(ziel_pfad, neueste_datei)
|
finally:
|
shutil.rmtree(staging_ordner, ignore_errors=True)
|
|
meldung = "Datei fehlte, wurde neu heruntergeladen." if datei_fehlt else "Ueber winget aktualisiert."
|
return UpdateResult(True, aktuelle_version, neue_version, meldung)
|
|
|
# ---------------------------------------------------------------------------
|
# Quelle: direkter Download-Link
|
# ---------------------------------------------------------------------------
|
|
def pruefe_direkt(name, update_wert, aktuelle_version, installer_pfad):
|
"""update_wert = feste URL, die immer zur aktuellen Datei fuehrt.
|
Da kein Versionsstand aus der URL ablesbar ist, wird ETag/Last-Modified
|
des Servers als technische Kennung genutzt, um Aenderungen zu erkennen."""
|
update_wert = (update_wert or "").strip()
|
if not update_wert:
|
return UpdateResult(False, aktuelle_version, None, "Kein Download-Link hinterlegt.")
|
|
try:
|
req = urllib.request.Request(update_wert, method="HEAD", headers={"User-Agent": USER_AGENT})
|
with urllib.request.urlopen(req, timeout=20) as resp:
|
kennung = resp.headers.get("ETag") or resp.headers.get("Last-Modified") or ""
|
except Exception as e:
|
return UpdateResult(False, aktuelle_version, None, f"Link nicht erreichbar: {e}")
|
|
if not kennung:
|
kennung = "unbekannt-" + datetime.now().strftime("%Y%m%d")
|
|
if not installer_pfad:
|
return UpdateResult(False, aktuelle_version, kennung, "Kein Ablageort hinterlegt, Download uebersprungen.")
|
|
datei_fehlt = verwaltete_datei_pfad(installer_pfad, name) is None
|
if kennung == (aktuelle_version or "") and not datei_fehlt:
|
return UpdateResult(True, aktuelle_version, aktuelle_version, "Bereits aktuell (laut Server unveraendert).")
|
|
ziel_pfad = _ziel_datei_ermitteln(installer_pfad, name, "download" + _endung_aus_url(update_wert))
|
try:
|
tmp = _http_download(update_wert)
|
sichern_und_ersetzen(ziel_pfad, tmp)
|
except Exception as e:
|
return UpdateResult(False, aktuelle_version, kennung, f"Download fehlgeschlagen: {e}")
|
|
meldung = ("Datei fehlte, wurde neu heruntergeladen." if datei_fehlt
|
else "Neu heruntergeladen (Aenderung am Link erkannt).")
|
return UpdateResult(True, aktuelle_version, kennung, meldung)
|
|
|
# ---------------------------------------------------------------------------
|
# Einstiegspunkt
|
# ---------------------------------------------------------------------------
|
|
def pruefe_software(name, quelle, update_wert, aktuelle_version, installer_pfad):
|
if quelle == "winget":
|
return pruefe_winget(name, update_wert, aktuelle_version, installer_pfad)
|
if quelle == "github":
|
return pruefe_github(name, update_wert, aktuelle_version, installer_pfad)
|
if quelle == "direkt":
|
return pruefe_direkt(name, update_wert, aktuelle_version, installer_pfad)
|
return UpdateResult(False, aktuelle_version, None, "Keine gueltige Update-Quelle hinterlegt.")
|