Updater: Prüfung beim Start und unter Hilfe, Setup herunterladen, DB sichern, still installieren, Neustart
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
1 parent
968c68ac1a
commit
b85a75857e
6 files changed
+260
-2
No files matched your search
+95
-1
@@ -5,14 +5,17 @@ import os
|
||||
import re
|
||||
import subprocess
|
||||
import sys
|
||||
import tempfile
|
||||
import threading
|
||||
import tkinter as tk
|
||||
import traceback
|
||||
import webbrowser
|
||||
from collections import defaultdict
|
||||
from datetime import date
|
||||
from pathlib import Path
|
||||
from tkinter import filedialog
|
||||
|
||||
from provtool import config
|
||||
from provtool import __version__, config, updater
|
||||
from provtool import export as exp
|
||||
from provtool import regeln
|
||||
from provtool.db import EIGEN, UNZUGEORDNET, AbrechnungErledigt, Datenbank
|
||||
@@ -235,6 +238,7 @@ class App:
|
||||
hi.add_command(label="Kurzanleitung", accelerator="F1", command=self.hilfe)
|
||||
hi.add_command(label="Rundgang durch das Programm", command=self.rundgang)
|
||||
hi.add_command(label="Tipp des Tages...", command=self.tipp_des_tages)
|
||||
hi.add_command(label="Nach Updates suchen...", command=self.updates_pruefen)
|
||||
hi.add_separator()
|
||||
hi.add_command(label="Einrichtungs-Assistent...", command=self.assistent)
|
||||
hi.add_separator()
|
||||
@@ -1247,9 +1251,95 @@ class App:
|
||||
self.db_wechseln(res["pfad"])
|
||||
except Exception as exc:
|
||||
w.fehler(self.root, "Einstellungen", f"Datenbank konnte nicht geöffnet werden:\n{exc}")
|
||||
self.settings["update_pruefen"] = res["update"]
|
||||
config.save(self.settings)
|
||||
if res["design"] != w.DESIGN:
|
||||
self.design_wechseln(res["design"])
|
||||
|
||||
# ================================================================= Updates
|
||||
@staticmethod
|
||||
def _installiert() -> bool:
|
||||
"""Läuft die per Setup installierte Windows-Version? Nur die kann sich selbst aktualisieren."""
|
||||
return (bool(getattr(sys, "frozen", False)) and sys.platform.startswith("win")
|
||||
and Path(sys.executable).name.lower() == "provisionsmanager.exe")
|
||||
|
||||
def _im_hintergrund(self, arbeit, fertig):
|
||||
"""arbeit() im Thread (nur Netzwerk/Dateien, kein Tk), danach fertig(ergebnis) im Tk-Thread."""
|
||||
erg: dict = {}
|
||||
|
||||
def lauf():
|
||||
try:
|
||||
erg["wert"] = arbeit()
|
||||
except Exception as exc:
|
||||
erg["fehler"] = exc
|
||||
threading.Thread(target=lauf, daemon=True).start()
|
||||
|
||||
def warten():
|
||||
if self._beendet or self._schliesst:
|
||||
return
|
||||
if not erg:
|
||||
self.root.after(200, warten)
|
||||
else:
|
||||
fertig(erg)
|
||||
self.root.after(200, warten)
|
||||
|
||||
def updates_pruefen(self, still: bool = False):
|
||||
"""Auf Gitea nach einer neueren Version suchen. still = automatisch beim Start: nur melden, wenn es
|
||||
eine gibt (keine Meldung bei Fehler oder wenn aktuell)."""
|
||||
if getattr(self, "_update_laeuft", False):
|
||||
return
|
||||
self._update_laeuft = True
|
||||
|
||||
def fertig(erg):
|
||||
self._update_laeuft = False
|
||||
self._update_melden(erg, still)
|
||||
self._im_hintergrund(updater.neueste, fertig)
|
||||
|
||||
def _update_melden(self, erg: dict, still: bool):
|
||||
if "fehler" in erg:
|
||||
if not still:
|
||||
w.warnung(self.root, "Updates", f"Konnte nicht nach Updates suchen:\n{erg['fehler']}")
|
||||
return
|
||||
info = erg["wert"]
|
||||
if not updater.ist_neuer(info["version"], __version__):
|
||||
if not still:
|
||||
w.info(self.root, "Updates", f"Sie haben die neueste Version ({__version__}).")
|
||||
return
|
||||
if still and modal.offen(): # z. B. Tipp des Tages offen – später noch einmal
|
||||
self.root.after(3000, lambda: self._update_melden(erg, still))
|
||||
return
|
||||
text = f"Version {info['version']} ist verfügbar (installiert: {__version__})."
|
||||
if self._installiert() and info["setup_url"]:
|
||||
if w.frage(self.root, "Update verfügbar",
|
||||
text + "\n\nJetzt herunterladen und installieren? Die Datenbank wird vorher gesichert, das "
|
||||
"Programm beendet sich kurz und startet danach von selbst wieder.",
|
||||
ja="Jetzt installieren", nein="Später"):
|
||||
self._update_installieren(info)
|
||||
elif w.frage(self.root, "Update verfügbar", text + "\n\nDownload-Seite öffnen?",
|
||||
ja="Seite öffnen", nein="Später"):
|
||||
webbrowser.open(info["seite"])
|
||||
|
||||
def _update_installieren(self, info: dict):
|
||||
self.status.set("info", f" Lade Version {info['version']} herunter ...")
|
||||
ordner = Path(tempfile.gettempdir()) / "Provisionsmanager-Update"
|
||||
|
||||
def fertig(erg):
|
||||
if "fehler" in erg:
|
||||
self.status.set("info", "")
|
||||
w.fehler(self.root, "Update", f"Download fehlgeschlagen:\n{erg['fehler']}")
|
||||
return
|
||||
try:
|
||||
self.db.backup(config.app_dir() / f"vor-update-{__version__}.db")
|
||||
except Exception as exc:
|
||||
if not w.frage(self.root, "Update", f"Die Datenbank konnte nicht gesichert werden:\n{exc}\n\n"
|
||||
"Trotzdem installieren?", ja="Installieren", nein="Abbrechen", default_ja=False):
|
||||
return
|
||||
# NEUSTART=1: das Setup startet das Programm danach wieder (installer.iss)
|
||||
subprocess.Popen([str(erg["wert"]), "/VERYSILENT", "/SUPPRESSMSGBOXES", "/NORESTART",
|
||||
"/CLOSEAPPLICATIONS", "/NEUSTART=1"], close_fds=True)
|
||||
self.beenden()
|
||||
self._im_hintergrund(lambda: updater.herunterladen(info, ordner), fertig)
|
||||
|
||||
def design_umschalten(self):
|
||||
self.design_wechseln("modern" if w.DUNKEL else "dunkel")
|
||||
|
||||
@@ -1450,12 +1540,16 @@ def main():
|
||||
except Exception:
|
||||
pass
|
||||
dateien = [a for a in sys.argv[1:] if Path(a).exists()]
|
||||
erster = True
|
||||
while True:
|
||||
root = tk.Tk(className="Provisionsmanager95")
|
||||
app = App(root)
|
||||
if dateien:
|
||||
root.after(200, lambda: app.importieren(dateien))
|
||||
dateien = []
|
||||
if erster and app.settings.get("update_pruefen", True): # nur im echten Start, nicht in Tests
|
||||
root.after(3000, lambda: app.updates_pruefen(still=True))
|
||||
erster = False
|
||||
root.mainloop()
|
||||
if not app.neustart: # sonst: Design wurde umgeschaltet – Fenster neu aufbauen
|
||||
break
|
||||
|
||||
@@ -1035,6 +1035,10 @@ class EinstellungenDialog(w.Dialog):
|
||||
tk.Radiobutton(g2, text="Klassisch (Windows 95, Legacy)", variable=self.v_design, value="win95").pack(
|
||||
anchor="w")
|
||||
tk.Label(g2, text="Das Fenster wird nach OK mit dem neuen Design neu aufgebaut.", fg=w.SHADOW).pack(anchor="w")
|
||||
g3 = w.gruppe(self, "Updates")
|
||||
g3.pack(fill="x", padx=10, pady=(0, 10))
|
||||
self.v_update = tk.BooleanVar(value=settings.get("update_pruefen", True))
|
||||
tk.Checkbutton(g3, text="Beim Start nach neuen Versionen suchen", variable=self.v_update).pack(anchor="w")
|
||||
self.knopfleiste([("OK", self._ok), ("Abbrechen", self.abbrechen)]).pack(pady=(0, 10))
|
||||
|
||||
def _waehlen(self):
|
||||
@@ -1045,7 +1049,7 @@ class EinstellungenDialog(w.Dialog):
|
||||
self.v.set(p)
|
||||
|
||||
def _ok(self):
|
||||
self.result = dict(pfad=self.v.get().strip(), design=self.v_design.get())
|
||||
self.result = dict(pfad=self.v.get().strip(), design=self.v_design.get(), update=self.v_update.get())
|
||||
self.destroy()
|
||||
|
||||
|
||||
|
||||
@@ -0,0 +1,82 @@
|
||||
"""Updates: neuestes Release auf Gitea finden und das Windows-Setup herunterladen.
|
||||
|
||||
Nur Netzwerk und Dateien – kein Tk (läuft im Hintergrund-Thread). Das Installieren startet die App (gui/app.py).
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import re
|
||||
import urllib.request
|
||||
from pathlib import Path
|
||||
from urllib.parse import urlparse
|
||||
|
||||
HOST = "git.handy-insel.de"
|
||||
API = f"https://{HOST}/api/v1/repos/kerimatilgan/provisionsmanager95/releases/latest"
|
||||
SETUP = re.compile(r"Provisionsmanager-Setup-[\d.]+\.exe")
|
||||
MAX_BYTES = 200 * 1024 * 1024
|
||||
|
||||
|
||||
class UpdateFehler(Exception):
|
||||
pass
|
||||
|
||||
|
||||
def version_tupel(v: str) -> tuple[int, ...]:
|
||||
"""'v1.10.0' -> (1, 10, 0); Unlesbares -> ()"""
|
||||
m = re.fullmatch(r"v?(\d+(?:\.\d+)*)", (v or "").strip())
|
||||
return tuple(int(x) for x in m[1].split(".")) if m else ()
|
||||
|
||||
|
||||
def ist_neuer(neu: str, alt: str) -> bool:
|
||||
return version_tupel(neu) > version_tupel(alt) and bool(version_tupel(alt))
|
||||
|
||||
|
||||
def _eigene_url(url: str) -> bool:
|
||||
"""Nur über HTTPS vom eigenen Gitea laden (die Antwort des Servers ist hier die Vertrauensgrenze)."""
|
||||
u = urlparse(url)
|
||||
return u.scheme == "https" and u.hostname == HOST
|
||||
|
||||
|
||||
def neueste(timeout: float = 8) -> dict:
|
||||
"""Neuestes Release: {"version", "seite", "setup_url", "groesse"} (setup_url None, wenn kein Setup dabei)."""
|
||||
with urllib.request.urlopen(API, timeout=timeout) as r:
|
||||
daten = json.loads(r.read(1024 * 1024).decode("utf-8"))
|
||||
setup = next((a for a in daten.get("assets", []) if SETUP.fullmatch(a.get("name", ""))), None)
|
||||
url = setup.get("browser_download_url") if setup else None
|
||||
if url and not _eigene_url(url):
|
||||
url = None
|
||||
seite = daten.get("html_url") or ""
|
||||
return {"version": str(daten.get("tag_name", "")).lstrip("v"),
|
||||
"seite": seite if _eigene_url(seite) else f"https://{HOST}/kerimatilgan/provisionsmanager95/releases",
|
||||
"setup_url": url, "groesse": int(setup.get("size") or 0) if setup else 0}
|
||||
|
||||
|
||||
def herunterladen(info: dict, ordner: Path, timeout: float = 60) -> Path:
|
||||
"""Setup nach `ordner` laden und prüfen (Größe wie angekündigt, Windows-Programm). Gibt den Pfad zurück."""
|
||||
url = info.get("setup_url")
|
||||
if not url or not _eigene_url(url):
|
||||
raise UpdateFehler("Für diese Version gibt es kein Setup zum Herunterladen.")
|
||||
ordner.mkdir(parents=True, exist_ok=True)
|
||||
ziel = ordner / Path(urlparse(url).path).name
|
||||
if not SETUP.fullmatch(ziel.name):
|
||||
raise UpdateFehler(f"Unerwarteter Dateiname: {ziel.name}")
|
||||
teil = ziel.with_suffix(".teil")
|
||||
n = 0
|
||||
try:
|
||||
with urllib.request.urlopen(url, timeout=timeout) as r, teil.open("wb") as f:
|
||||
if not _eigene_url(r.geturl()): # Weiterleitung auf einen fremden Server
|
||||
raise UpdateFehler("Download wurde auf einen fremden Server umgeleitet.")
|
||||
while block := r.read(1 << 16):
|
||||
n += len(block)
|
||||
if n > MAX_BYTES:
|
||||
raise UpdateFehler("Download ist unerwartet groß.")
|
||||
f.write(block)
|
||||
except BaseException:
|
||||
teil.unlink(missing_ok=True)
|
||||
raise
|
||||
with teil.open("rb") as f:
|
||||
kopf = f.read(2)
|
||||
if kopf != b"MZ" or (info.get("groesse") and n != info["groesse"]):
|
||||
teil.unlink(missing_ok=True)
|
||||
raise UpdateFehler("Download ist unvollständig oder beschädigt – bitte später erneut versuchen.")
|
||||
teil.replace(ziel)
|
||||
return ziel
|
||||
Reference in new issue
Block a user