# -*- coding: utf-8 -*- """ AZA Desktop Updater — Startprompt + sichere Installation. Beim Startpanel-Start: freundliche Nachfrage bei neuer Version. Kein sichtbarer Update-Button, kein stilles Auto-Update. """ from __future__ import annotations import argparse import json import shutil import subprocess import sys import threading import tkinter as tk from pathlib import Path from tkinter import messagebox, ttk from typing import Any from aza_update_core import ( check_update_for_startup, check_update_from_manifest, configure_test_install_dir, create_pre_update_backup, compute_sha256, download_update_package, evaluate_update, extract_update_zip, fetch_remote_manifest, format_version_label, get_install_dir, get_project_root, get_test_install_dir, list_available_backups, load_local_version, load_manifest_from_source, rollback_from_backup, save_local_version, validate_update_install_ready, ) def _log(msg: str) -> None: print(f"[AZA Updater] {msg}") def check_update_status(*, startup: bool = False) -> dict[str, Any]: """Manifest pruefen — startup=True nutzt nur Update-Kanal (kein Legacy-Fallback).""" if startup: return check_update_for_startup() local = load_local_version() remote, err = fetch_remote_manifest() if err or not remote: return { "status": "error", "message": "Manifest nicht erreichbar.", "detail": err, "local": local, } result = evaluate_update(local, remote) result["detail"] = err return result def _stop_aza_processes() -> None: if sys.platform != "win32": return for name in ("aza_desktop.exe", "aza_controller.exe", "AZA_EmpfangShell.exe"): try: subprocess.run( ["taskkill", "/F", "/IM", name], capture_output=True, text=True, timeout=15, ) except Exception: pass def _apply_downloaded_update(package: Path, result: dict[str, Any]) -> tuple[bool, str]: install_dir = get_install_dir() backup_dir = create_pre_update_backup(install_dir) _log(f"Backup erstellt: {backup_dir}") suffix = package.suffix.lower() if suffix == ".zip": ok, msg = extract_update_zip(package, install_dir) if not ok: rollback_from_backup(backup_dir, install_dir) return False, f"Installation fehlgeschlagen.\nRollback durchgefuehrt." elif suffix == ".exe": target = install_dir / package.name try: import shutil shutil.copy2(package, target) except Exception as exc: rollback_from_backup(backup_dir, install_dir) return False, f"Installation fehlgeschlagen.\nRollback durchgefuehrt." else: return False, "Unbekanntes Update-Format." new_version = { "version": result.get("latest_version"), "build": result.get("latest_build"), "channel": (result.get("local") or {}).get("channel", "stable"), } try: save_local_version({k: v for k, v in new_version.items() if v}) except Exception as exc: _log(f"version.json write failed: {exc}") return True, "Das Update wurde installiert.\nBitte starten Sie AZA neu." def perform_confirmed_update( result: dict[str, Any], *, parent: tk.Misc | None = None, ) -> tuple[bool, str]: """Download + SHA256 + Backup + Installation — nur bei gueltigem Manifest.""" ready, reason = validate_update_install_ready(result) if not ready: return False, ( "Das Update ist noch nicht vollstaendig verfuegbar.\n" "Bitte versuchen Sie es spaeter erneut." ) files = result.get("files") or [] file_info = files[0] _log("Download starten") package, err = download_update_package(file_info) if not package: _log(f"Download fehlgeschlagen: {err}") return False, ( "Das Update konnte nicht heruntergeladen werden.\n" "Bitte versuchen Sie es spaeter erneut." ) _stop_aza_processes() return _apply_downloaded_update(package, result) def _show_friendly_update_dialog( result: dict[str, Any], *, parent: tk.Misc | None = None, ) -> bool: """Freundlicher Dialog — True = Jetzt aktualisieren.""" owns = False root = parent if root is None: root = tk.Tk() root.withdraw() owns = True choice = {"install": False} dlg = tk.Toplevel(root) dlg.title("Update verfügbar") dlg.resizable(False, False) dlg.transient(root) dlg.grab_set() frm = ttk.Frame(dlg, padding=18) frm.pack(fill="both", expand=True) ttk.Label( frm, text="Eine neue AZA-Version ist verfügbar.", font=("Segoe UI", 11, "bold"), wraplength=360, ).pack(anchor="w", pady=(0, 8)) ttk.Label( frm, text="Möchten Sie das Update jetzt installieren?", font=("Segoe UI", 10), wraplength=360, ).pack(anchor="w", pady=(0, 6)) ttk.Label( frm, text="Ihre aktuellen Einstellungen und Praxisdaten bleiben erhalten.", font=("Segoe UI", 9), foreground="#5a6d7d", wraplength=360, ).pack(anchor="w", pady=(0, 14)) btn_row = ttk.Frame(frm) btn_row.pack(fill="x") def on_install() -> None: choice["install"] = True dlg.destroy() def on_later() -> None: dlg.destroy() ttk.Button(btn_row, text="Jetzt aktualisieren", command=on_install).pack( side="left", padx=(0, 8) ) ttk.Button(btn_row, text="Später", command=on_later).pack(side="left") dlg.protocol("WM_DELETE_WINDOW", on_later) dlg.update_idletasks() try: sw = dlg.winfo_screenwidth() sh = dlg.winfo_screenheight() dlg.geometry( f"+{(sw - dlg.winfo_reqwidth()) // 2}+{(sh - dlg.winfo_reqheight()) // 2}" ) except Exception: pass if owns: dlg.wait_window() try: root.destroy() except Exception: pass else: root.wait_window(dlg) return bool(choice["install"]) def maybe_prompt_update_on_startup(*, parent: tk.Misc | None = None) -> None: """ Beim Startpanel: still pruefen, bei Update freundlich fragen. 404 / kein Manifest / aktuell => nichts anzeigen. """ result = check_update_status(startup=True) status = result.get("status") _log(f"startup status={status} detail={result.get('detail')}") if status in ("manifest_unavailable", "current", "channel_mismatch", "error"): return if status != "update_available": return install_now = _show_friendly_update_dialog(result, parent=parent) if not install_now: _log("startup update deferred by user") return ready, _reason = validate_update_install_ready(result) if not ready: if parent is not None: messagebox.showinfo( "Update", "Das Update-Modul wird vorbereitet.\n" "Bitte starten Sie die Aktualisierung spaeter erneut.", parent=parent, ) else: root = tk.Tk() root.withdraw() messagebox.showinfo( "Update", "Das Update-Modul wird vorbereitet.\n" "Bitte starten Sie die Aktualisierung spaeter erneut.", parent=root, ) root.destroy() return ok, msg = perform_confirmed_update(result, parent=parent) if parent is not None: if ok: messagebox.showinfo("Update", msg, parent=parent) else: messagebox.showinfo("Update", msg, parent=parent) else: root = tk.Tk() root.withdraw() messagebox.showinfo("Update", msg, parent=root) root.destroy() def run_startup_update_check_in_background(*, delay_seconds: float = 1.2) -> None: """Nicht-blockierende Startpruefung fuer aza_start_panel.py.""" def worker() -> None: import time time.sleep(max(0.0, delay_seconds)) try: maybe_prompt_update_on_startup() except Exception as exc: _log(f"startup check skipped: {type(exc).__name__}") threading.Thread( target=worker, daemon=True, name="aza-startup-update", ).start() def check_updates_interactive(*, parent: tk.Misc | None = None) -> None: """Manuelle Pruefung (technisch) — fuer aza_controller.py.""" result = check_update_status(startup=False) status = result.get("status") if status == "update_available": install_now = _show_friendly_update_dialog(result, parent=parent) if install_now: ok, msg = perform_confirmed_update(result, parent=parent) messagebox.showinfo("Update", msg, parent=parent) return if status == "current": messagebox.showinfo( "Update", f"AZA ist aktuell ({format_version_label()}).", parent=parent, ) return messagebox.showinfo( "Update", "Derzeit ist kein Update verfuegbar.", parent=parent, ) def show_app_info(*, parent: tk.Misc | None = None) -> None: local = load_local_version() messagebox.showinfo( "AzA", f"Version: {format_version_label(local)}\n\nAzA Medwork", parent=parent, ) def _read_marker(install_dir: Path) -> str: p = install_dir / "app_marker.txt" if p.is_file(): return p.read_text(encoding="utf-8").strip() return "" def _reset_test_install_dir(install_dir: Path) -> None: install_dir.mkdir(parents=True, exist_ok=True) backups = install_dir / "_update_backups" if backups.is_dir(): shutil.rmtree(backups, ignore_errors=True) (install_dir / "version.json").write_text( json.dumps( { "version": "1.2.0", "build": "TEST_OLD", "channel": "stable", "app": "AZA Desktop", }, indent=2, ) + "\n", encoding="utf-8", ) (install_dir / "app_marker.txt").write_text("OLD_VERSION\n", encoding="utf-8") for name in ( "aza_controller.exe", "aza_office.exe", "aza_praxis_chat.exe", "aza_updater.exe", ): p = install_dir / name if not p.is_file(): p.write_bytes(b"OLD_DUMMY_EXE\n") def _build_test_package(project_root: Path) -> tuple[Path, Path, str, int]: import zipfile pkg_dir = project_root / "_updater_test_package" zip_path = project_root / "_updater_test_package.zip" if pkg_dir.is_dir(): shutil.rmtree(pkg_dir) pkg_dir.mkdir(parents=True) (pkg_dir / "version.json").write_text( json.dumps( { "version": "9.9.9", "build": "TEST_LOCAL", "channel": "stable", "app": "AZA Desktop", }, indent=2, ) + "\n", encoding="utf-8", ) (pkg_dir / "app_marker.txt").write_text("NEW_VERSION\n", encoding="utf-8") for name in ( "aza_controller.exe", "aza_office.exe", "aza_praxis_chat.exe", "aza_updater.exe", ): (pkg_dir / name).write_bytes(b"NEW_DUMMY_EXE\n") if zip_path.is_file(): zip_path.unlink() with zipfile.ZipFile(zip_path, "w", zipfile.ZIP_DEFLATED) as zf: for fp in pkg_dir.rglob("*"): if fp.is_file(): zf.write(fp, fp.relative_to(pkg_dir).as_posix()) sha = compute_sha256(zip_path) return pkg_dir, zip_path, sha, zip_path.stat().st_size def _write_test_manifest( project_root: Path, zip_path: Path, sha256: str, size_bytes: int, *, bad_sha: str | None = None, missing_sha: bool = False, missing_zip: bool = False, ) -> Path: from datetime import date manifest_path = project_root / "test_update_manifest.json" url = str(zip_path.resolve()) if not missing_zip else str( (project_root / "_missing_update.zip").resolve() ) entry = { "name": zip_path.name, "url": url, "size_bytes": size_bytes, } if not missing_sha: entry["sha256"] = bad_sha or sha256 payload = { "product": "AZA Desktop", "channel": "stable", "latest_version": "9.9.9", "latest_build": "TEST_LOCAL", "min_supported_version": "1.0.0", "mandatory": False, "release_date": date.today().isoformat(), "notes_de": ["Lokaler Updater-Test"], "files": [entry], } manifest_path.write_text( json.dumps(payload, indent=2, ensure_ascii=False) + "\n", encoding="utf-8", ) return manifest_path def run_local_e2e_tests(project_root: Path | None = None) -> dict[str, Any]: """Automatisierter lokaler End-to-End-Test ohne GUI und ohne Projektroot.""" root = (project_root or get_project_root()).resolve() install_dir = root / "_updater_test_install" configure_test_install_dir(install_dir) _reset_test_install_dir(install_dir) _, zip_path, sha256, size_bytes = _build_test_package(root) manifest_path = _write_test_manifest(root, zip_path, sha256, size_bytes) report: dict[str, Any] = { "install_dir": str(install_dir), "zip_path": str(zip_path), "manifest_path": str(manifest_path), "sha256": sha256, } result = check_update_from_manifest(manifest_path) report["update_detected"] = result.get("status") == "update_available" ok, msg = perform_confirmed_update(result) report["update_applied"] = ok report["update_message"] = msg report["marker_after_update"] = _read_marker(install_dir) report["version_after_update"] = load_local_version().get("version") backups = list_available_backups() report["backup_created"] = bool(backups) if backups: rb_ok, rb_msg = rollback_from_backup(backups[0], install_dir) report["rollback_tested"] = rb_ok report["rollback_message"] = rb_msg report["marker_after_rollback"] = _read_marker(install_dir) report["version_after_rollback"] = load_local_version().get("version") _reset_test_install_dir(install_dir) bad_manifest = _write_test_manifest( root, zip_path, sha256, size_bytes, bad_sha="0" * 64 ) bad_result = check_update_from_manifest(bad_manifest) bad_ok, _ = perform_confirmed_update(bad_result) report["bad_sha_blocked"] = (not bad_ok) and _read_marker(install_dir) == "OLD_VERSION" _reset_test_install_dir(install_dir) no_sha_manifest = _write_test_manifest( root, zip_path, sha256, size_bytes, missing_sha=True ) no_sha_result = check_update_from_manifest(no_sha_manifest) no_sha_ready, _ = validate_update_install_ready(no_sha_result) report["missing_sha_blocked"] = not no_sha_ready _reset_test_install_dir(install_dir) missing_zip_manifest = _write_test_manifest( root, zip_path, sha256, size_bytes, missing_zip=True ) missing_result = check_update_from_manifest(missing_zip_manifest) missing_ok, _ = perform_confirmed_update(missing_result) report["missing_zip_blocked"] = (not missing_ok) and _read_marker(install_dir) == "OLD_VERSION" _reset_test_install_dir(install_dir) later_result = check_update_from_manifest(manifest_path) report["later_unchanged"] = ( later_result.get("status") == "update_available" and _read_marker(install_dir) == "OLD_VERSION" and load_local_version().get("version") == "1.2.0" ) project_version = json.loads((root / "version.json").read_text(encoding="utf-8-sig")) report["sha256_verified"] = report.get("update_applied", False) report["project_root_version"] = project_version.get("version") report["project_root_unchanged"] = project_version.get("version") == "1.2.0" configure_test_install_dir(None) _write_test_manifest(root, zip_path, sha256, size_bytes) return report def _parse_cli() -> argparse.Namespace: import argparse p = argparse.ArgumentParser(description="AZA Desktop Updater") p.add_argument( "--e2e-local-test", action="store_true", help="Lokalen End-to-End-Test ausfuehren (ohne GUI)", ) p.add_argument( "--install-dir", help="Test-Installationsordner (setzt AZA_UPDATE_TEST_INSTALL_DIR)", ) p.add_argument( "--manifest", help="Manifest-Pfad fuer manuelle Pruefung", ) p.add_argument( "--yes", action="store_true", help="Update ohne zusaetzlichen Dialog installieren (nur Testmodus)", ) return p.parse_args() def main() -> int: args = _parse_cli() if args.e2e_local_test: report = run_local_e2e_tests() print(json.dumps(report, indent=2, ensure_ascii=False)) ok = all( report.get(k) for k in ( "update_detected", "update_applied", "backup_created", "rollback_tested", "bad_sha_blocked", "missing_sha_blocked", "missing_zip_blocked", "later_unchanged", "project_root_unchanged", ) ) and report.get("marker_after_update") == "NEW_VERSION" and report.get( "marker_after_rollback" ) == "OLD_VERSION" return 0 if ok else 1 if args.install_dir: configure_test_install_dir(args.install_dir) if args.manifest and args.yes: result = check_update_from_manifest(args.manifest) ok, msg = perform_confirmed_update(result) print(msg) return 0 if ok else 1 owns = False parent: tk.Misc | None = None if not tk._default_root: # type: ignore[attr-defined] parent = tk.Tk() parent.withdraw() owns = True check_updates_interactive(parent=parent) if owns and parent is not None: parent.destroy() return 0 if __name__ == "__main__": raise SystemExit(main())