Add in-app self-update via git (Hilfe menu)

New Updater class + "Hilfe" menubar:
- "Nach Update suchen": git fetch + count of new upstream commits.
- Quiet background check on start; surfaces via status bar + menu label.
- Install: git merge --ff-only, pip install if requirements.txt changed,
  then restart via os.execv. data/ is git-ignored and untouched.
- Fast-forward only; diverged history or offline -> clear message, no action.
- Inert unless run from a git clone with git on PATH.

Also: "Version…" menu item shows the installed commit.

Verified against throwaway git repos (check / ff-update / diverged / no-op /
non-clone) and via GUI build on Python 3.14 / Tk 9.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
2026-09-07 22:06:51 +02:00
parent e3bbd4d264
commit 1bed723363
7 changed files with 330 additions and 10 deletions
+238
View File
@@ -12,6 +12,8 @@ import math
import os
import re
import shutil
import subprocess
import sys
import threading
import time
import webbrowser
@@ -451,6 +453,90 @@ def generate_map(store: DataStore) -> Path:
return MAP_PATH
# ── Selbst-Update (git) ──────────────────────────────────────────────────────
class UpdateError(Exception):
"""Ein Schritt beim Aktualisieren ist fehlgeschlagen (Text ist benutzertauglich)."""
class Updater:
"""Aktualisiert die App per ``git`` nur wenn sie aus einem Git-Klon läuft.
Fährt bewusst konservativ: es wird ausschließlich per *fast-forward* auf den
Stand des Servers gezogen. Lokale Änderungen oder abweichende Historie führen
zu einer klaren Fehlermeldung statt zu einem automatischen Merge.
"""
def __init__(self, base_dir: Path) -> None:
self.base_dir = base_dir
self.available = bool(shutil.which("git")) and (base_dir / ".git").is_dir()
def _git(self, *args: str, timeout: int = 30) -> subprocess.CompletedProcess:
return subprocess.run(
["git", "-C", str(self.base_dir), *args],
capture_output=True, text=True, timeout=timeout,
)
def _upstream_ref(self) -> str:
r = self._git("rev-parse", "--abbrev-ref", "--symbolic-full-name", "@{u}")
return r.stdout.strip() or "origin/main"
def current_version(self) -> str:
r = self._git("log", "-1", "--format=%h · %cs")
return r.stdout.strip() or "unbekannt"
def check(self) -> int:
"""Anzahl neuer Commits auf dem Server. Wirft ``UpdateError`` bei Problemen."""
f = self._git("fetch", "--quiet", timeout=60)
if f.returncode != 0:
raise UpdateError(f.stderr.strip() or "Server nicht erreichbar (git fetch).")
ref = self._upstream_ref()
c = self._git("rev-list", "--count", f"HEAD..{ref}")
if c.returncode != 0:
raise UpdateError(c.stderr.strip() or "Versionsvergleich fehlgeschlagen.")
return int(c.stdout.strip() or "0")
def update(self) -> bool:
"""Zieht den neuen Stand (fast-forward) und installiert ggf. neue Pakete.
Gibt ``True`` zurück, wenn tatsächlich etwas aktualisiert wurde.
"""
f = self._git("fetch", "--quiet", timeout=60)
if f.returncode != 0:
raise UpdateError(f.stderr.strip() or "Server nicht erreichbar (git fetch).")
ref = self._upstream_ref()
before = self._git("rev-parse", "HEAD").stdout.strip()
m = self._git("merge", "--ff-only", ref, timeout=60)
if m.returncode != 0:
raise UpdateError(
(m.stderr.strip() or "Automatisches Update nicht möglich.")
+ "\n\nBitte Patrick Bescheid geben."
)
after = self._git("rev-parse", "HEAD").stdout.strip()
if before == after:
return False
changed = self._git("diff", "--name-only", before, after).stdout.split()
if "requirements.txt" in changed:
self._install_requirements()
log.info("Update: %s%s", before[:9], after[:9])
return True
def _install_requirements(self) -> None:
req = self.base_dir / "requirements.txt"
if not req.exists():
return
r = subprocess.run(
[sys.executable, "-m", "pip", "install", "-r", str(req)],
capture_output=True, text=True, timeout=600,
)
if r.returncode != 0:
raise UpdateError(
"Neue Programmbibliotheken konnten nicht installiert werden:\n"
+ (r.stderr.strip()[-400:] or "pip-Fehler")
)
# ── Bearbeiten-Dialog ─────────────────────────────────────────────────────────
@@ -612,6 +698,9 @@ class App(ttk.Window):
self.geocoder.start()
self._service_error_shown = False
self.updater = Updater(BASE_DIR)
self._update_busy = False
self._queue: list[dict] = []
self._next_id = 0
self._status_after_id: str | None = None
@@ -623,6 +712,9 @@ class App(ttk.Window):
self.protocol("WM_DELETE_WINDOW", self._on_close)
log.info("App gestartet (%d Einträge).", self.store.total_count())
if self.updater.available:
threading.Thread(target=self._auto_check_update, daemon=True).start()
# ── Aussehen ──────────────────────────────────────────────────────────────
def _scale_fonts(self) -> None:
@@ -684,6 +776,150 @@ class App(ttk.Window):
except OSError:
log.warning("Fensterposition konnte nicht gespeichert werden.")
# ── Menüleiste & Selbst-Update ───────────────────────────────────────────
def _build_menubar(self) -> None:
menubar = tk.Menu(self)
help_menu = tk.Menu(menubar, tearoff=0)
if self.updater.available:
self._update_menu = help_menu
self._update_menu_index = 0
help_menu.add_command(
label="Nach Update suchen", command=lambda: self._check_update(manual=True)
)
help_menu.add_separator()
else:
self._update_menu = None
help_menu.add_command(label="Version…", command=self._show_version)
menubar.add_cascade(label="Hilfe", menu=help_menu)
self.configure(menu=menubar)
def _show_version(self) -> None:
if self.updater.available:
info = f"Installierte Version:\n{self.updater.current_version()}"
else:
info = "Version: unbekannt (kein Git-Klon Update über das Menü nicht möglich)."
messagebox.showinfo("DRK Blutspende Arbeitsorte", info, parent=self)
def _auto_check_update(self) -> None:
"""Stiller Hintergrund-Check beim Start meldet sich nur, wenn es etwas gibt."""
time.sleep(2)
try:
n = self.updater.check()
except Exception as exc: # offline o. Ä. nicht stören
log.info("Automatischer Update-Check übersprungen: %s", exc)
return
if n > 0:
self.after(0, lambda: self._on_update_available(n, manual=False))
def _check_update(self, *, manual: bool) -> None:
if self._update_busy:
return
if not self.updater.available:
messagebox.showinfo(
"Update", "Diese Installation kann sich nicht selbst aktualisieren.",
parent=self,
)
return
self._update_busy = True
self._set_status("Suche nach Updates…", sticky=True)
def work() -> None:
try:
n = self.updater.check()
err = None
except Exception as exc:
n, err = 0, str(exc)
self.after(0, self._on_update_checked, n, err, manual)
threading.Thread(target=work, daemon=True).start()
def _on_update_checked(self, n: int, err: str | None, manual: bool) -> None:
self._update_busy = False
if err is not None:
self._set_status("Update-Prüfung fehlgeschlagen.", "warning")
if manual:
messagebox.showwarning(
"Update",
f"Es konnte nicht nach Updates gesucht werden:\n\n{err}\n\n"
"Ist der Computer mit dem Internet verbunden?",
parent=self,
)
return
if n == 0:
self._set_status("Die App ist auf dem neuesten Stand.", "success")
if manual:
messagebox.showinfo("Update", "Die App ist auf dem neuesten Stand.", parent=self)
return
self._on_update_available(n, manual=manual)
def _on_update_available(self, n: int, *, manual: bool) -> None:
if self._update_menu is not None:
self._update_menu.entryconfigure(
self._update_menu_index, label=f"Update installieren ({n} verfügbar)"
)
self._set_status(
f"{n} Aktualisierung(en) verfügbar Menü „Hilfe“ Update installieren.",
"info", sticky=True,
)
count = f"{n} Aktualisierung" + ("en" if n != 1 else "")
if messagebox.askyesno(
"Update verfügbar",
f"Es gibt {count} für diese App.\n\n"
"Jetzt installieren? Die App wird dabei kurz neu gestartet.\n"
"Deine gespeicherten Einträge bleiben unverändert.",
parent=self,
):
self._do_update()
def _do_update(self) -> None:
if self._update_busy:
return
self._update_busy = True
self._set_status("App wird aktualisiert bitte warten…", sticky=True)
def work() -> None:
try:
changed = self.updater.update()
err = None
except Exception as exc:
changed, err = False, str(exc)
self.after(0, self._on_update_done, changed, err)
threading.Thread(target=work, daemon=True).start()
def _on_update_done(self, changed: bool, err: str | None) -> None:
self._update_busy = False
if err is not None:
self._set_status("Update fehlgeschlagen.", "error")
messagebox.showerror("Update fehlgeschlagen", err, parent=self)
return
if not changed:
self._set_status("Nichts zu aktualisieren.", "info")
return
messagebox.showinfo(
"Update installiert",
"Die Aktualisierung wurde installiert. Die App startet jetzt neu.",
parent=self,
)
self._restart()
def _restart(self) -> None:
log.info("Neustart nach Update.")
self._save_window_state()
self.geocoder.stop()
try:
os.chdir(BASE_DIR)
os.execv(sys.executable, [sys.executable, str(Path(__file__).resolve())])
except OSError:
log.exception("Neustart fehlgeschlagen.")
messagebox.showwarning(
"Neustart nötig",
"Bitte die App einmal schließen und neu öffnen, "
"damit das Update wirksam wird.",
parent=self,
)
def _on_geocode_service_error(self) -> None:
"""Einmalige Warnung, wenn der Standortdienst wiederholt nicht erreichbar ist."""
self._set_status(
@@ -704,6 +940,8 @@ class App(ttk.Window):
# ── UI-Aufbau ─────────────────────────────────────────────────────────────
def _build_ui(self) -> None:
self._build_menubar()
# Kopfzeile mit DRK-Rot
header = ttk.Frame(self, bootstyle="danger", padding=(14, 8))
header.pack(fill=X, side=TOP)