"""Selbst-Update der Integration direkt von Gitea. Ersetzt install.ps1 als Update-Weg - nicht als Erstinstallations-Weg, die Integration muss bereits laufen, um sich selbst aktualisieren zu können. Grund: Windows Smart App Control blockiert die Ausführung von install.ps1 zuverlässig und lässt sich, anders als der SmartScreen davor, nicht per "trotzdem ausführen" umgehen. SICHERHEIT: dieses Modul läuft, anders als install.ps1, INNERHALB des Prozesses, den es aktualisiert - der eigene Ordner wird ersetzt, während der Dienstaufruf noch läuft. Deshalb strenger als install.ps1s "leeren-dann-kopieren": 1. Laden und Prüfen passiert komplett in einem Geschwisterordner (audi_dashboard_update_staging). Der Live-Ordner bleibt unangetastet, bis feststeht, dass der Download vollständig und plausibel ist. 2. Erst danach ein atomarer Tausch per os.rename, nicht Löschen+Kopieren: der alte Ordner wird zu audi_dashboard_backup umbenannt statt gelöscht - reversibel, falls das Update Probleme macht. Dessen manifest.json wird dabei zu manifest.json.bak umbenannt: HA erkennt Integrationen an jedem manifest.json unter custom_components/, unabhängig vom Ordnernamen - ohne diesen Schritt lädt ein Neustart mit vorhandenem Backup-Ordner eine zweite Integration mit derselben Domain. 3. Schlägt der zweite Rename fehl, wird der erste zurückgenommen, statt einen halb ersetzten Ordner zurückzulassen. 4. Kein automatischer Reload oder Neustart. Ein Reload aus dem eigenen, noch laufenden Dienstaufruf heraus anzustoßen wäre riskant - der Handler könnte mitten in der Ausführung durch den eigenen Reload unterbrochen werden. Die Oberfläche zeigt stattdessen "Home Assistant neu starten" an, genau wie install.ps1 es über seine Restliste tut. 5. Nichts außerhalb des eigenen Ordners: configuration.yaml, .storage/, www/ und der separate Fahrzeugdaten-Ordner werden nirgends geöffnet. Das Umbenennen des eigenen, gerade ausgeführten Ordners ist auf Linux (Home Assistant OS/Docker, worauf dieses Projekt läuft) unkritisch: Python hält nach dem Import keine offenen Dateizugriffe auf die .py-Quelltexte mehr, und der Verzeichnisname selbst ist nur eine Zeichenkette in os.rename() - kein Handle, das durch die Umbenennung ungültig würde. """ from __future__ import annotations import base64 import datetime import hashlib import json import logging import os import shutil import zipfile from io import BytesIO from typing import Any from homeassistant.core import HomeAssistant from homeassistant.helpers.aiohttp_client import async_get_clientsession _LOGGER = logging.getLogger(__name__) _REPO_BESITZER = "paul" _REPO_NAME = "audi-app" _ZWEIG = "main" _GITEA_BASIS = "https://gitea.nothaft.cloud/api/v1" _MANIFEST_PFAD_IM_REPO = "custom_components/audi_dashboard/manifest.json" _INTEGRATIONS_PFAD_IM_REPO = "custom_components/audi_dashboard/" # Absolute Pfade, einmal zur Modul-Ladezeit bestimmt - __file__ liegt selbst # im Live-Ordner, dessen Elternordner (custom_components/) ist der Ort für # Staging- und Backup-Geschwisterordner. INTEGRATIONSORDNER = os.path.dirname(os.path.abspath(__file__)) _STAGING_ORDNER = os.path.join( os.path.dirname(INTEGRATIONSORDNER), "audi_dashboard_update_staging" ) _BACKUP_ORDNER = os.path.join(os.path.dirname(INTEGRATIONSORDNER), "audi_dashboard_backup") class AktualisierungsFehler(Exception): """Jeder Fehlerfall beim Prüfen oder Installieren - Netzwerk, falscher Token, kaputtes Zip, falsche Domain, fehlgeschlagener Tausch. Eine eigene Klasse statt roher Exceptions, damit der Dienstaufruf sie gezielt abfangen und der Oberfläche eine verständliche Meldung zeigen kann, statt eines unbehandelten Fehlers.""" def _header(token: str) -> dict[str, str]: return {"Authorization": f"token {token}"} def _versionsteile(version: str) -> list[int] | None: """Zerlegt `2026.9.4.17` in [2026, 9, 4, 17]. `None`, sobald ein Teil keine reine Ziffernfolge ist - dann wird nicht sortiert. Bewusst kein `int()` mit Vorabschnitt: das läse "4b" klaglos als 4 und erzeugte genau die scheingenaue Reihenfolge, die hier vermieden werden soll. Wortgleich mit `teile()` in companion-app/src/daten/appVersion.ts.""" if not version: return None zahlen: list[int] = [] for stueck in version.split("."): if not stueck.isdigit(): return None zahlen.append(int(stueck)) return zahlen def ist_neuer(kandidat: str, bisher: str) -> bool | None: """Ob `kandidat` nachweislich neuer ist als `bisher`. `None` heißt "nicht zu entscheiden" - verschieden, aber nicht in eine Reihenfolge zu bringen (Formatwechsel). Fehlende Stellen zählen als 0, damit `2026.9.4` und `2026.9.4.0` denselben Stand bezeichnen. Gegenstück zu `versionOrdnung()` in appVersion.ts. Der Grund ist derselbe und dort ausführlich vermerkt: bis zum 05.09.2026 stand hier `remote_version != eigene_version`, also reine Ungleichheit ohne Richtung. Lief die Instanz einer Veröffentlichung voraus - während der Entwicklung der Normalfall -, bot die Integration an, sich auf die ÄLTERE Fassung zu "aktualisieren". Vom Eigentümer am 05.09.2026 gemeldet: „Warum wird eine ältere Version angeboten? .16 ist älter als .18".""" if not kandidat or not bisher: return None if kandidat == bisher: return False links = _versionsteile(kandidat) rechts = _versionsteile(bisher) if links is None or rechts is None: return None for i in range(max(len(links), len(rechts))): l = links[i] if i < len(links) else 0 r = rechts[i] if i < len(rechts) else 0 if l != r: return l > r return False async def version_pruefen(hass: HomeAssistant, token: str, eigene_version: str) -> dict[str, Any]: """Fragt nur die manifest.json von Gitea ab (ein kleiner Request, kein Repo-Download) und vergleicht die Version gegen die installierte. Läuft ausschließlich auf Tastendruck aus der Oberfläche - niemals aus dem Koordinator-Takt heraus, sonst würde jeder normale Veröffentlichungs- Zyklus einen Netzwerkaufruf zu Gitea auslösen.""" if not token: raise AktualisierungsFehler( "Kein Gitea-Token hinterlegt - unter Einstellungen -> Geräte & Dienste -> " "Audi Dashboard -> Konfigurieren eintragen." ) url = f"{_GITEA_BASIS}/repos/{_REPO_BESITZER}/{_REPO_NAME}/contents/{_MANIFEST_PFAD_IM_REPO}" session = async_get_clientsession(hass) try: async with session.get(url, headers=_header(token), params={"ref": _ZWEIG}) as antwort: if antwort.status == 401: raise AktualisierungsFehler("Token ungültig oder abgelaufen.") if antwort.status == 404: raise AktualisierungsFehler("Repository oder Pfad bei Gitea nicht gefunden.") if antwort.status != 200: raise AktualisierungsFehler(f"Gitea antwortete mit Status {antwort.status}.") rumpf = await antwort.json() except AktualisierungsFehler: raise except Exception as fehler: # aiohttp-/Verbindungsfehler aller Art raise AktualisierungsFehler(f"Gitea nicht erreichbar: {fehler}") from fehler try: roh = base64.b64decode(rumpf["content"]) remote_version = json.loads(roh)["version"] except (KeyError, ValueError, TypeError) as fehler: raise AktualisierungsFehler(f"Antwort von Gitea nicht lesbar: {fehler}") from fehler return { # Nur eine NACHWEISLICH neuere Fassung ist ein Update. Ist sie nicht # in eine Reihenfolge zu bringen (None), gilt sie als verfuegbar - # dann sagt die Oberflaeche, dass etwas abweicht, aber nicht in # welche Richtung, statt eine Aenderung zu verschweigen. "verfuegbar": ist_neuer(remote_version, eigene_version) is not False, "version": remote_version, "geprueft_am": datetime.datetime.now(datetime.timezone.utc).isoformat(), } # Dateien, deren Aenderung KEINEN Neustart braucht. # # frontend/ wird direkt von der Platte ausgeliefert (StaticPathConfig mit # cache_headers=False, siehe __init__.py) - eine ersetzte .js oder .css ist # sofort wirksam, es braucht nur ein Neuladen im Browser. Das gilt auch fuer # frontend/app/, das OTA-Buendel der Companion-App. # # manifest.json steht dabei, weil sich seine Versionsnummer bei JEDER # Veroeffentlichung aendert - ohne diese Ausnahme waere jedes Update ein # Neustart-Update, und die Unterscheidung waere wertlos. Home Assistant haelt # das Manifest fuer die Laufzeit fest (loader.py: hass.data[DATA_INTEGRATIONS]), # die von HA selbst angezeigte Versionsnummer bleibt also bis zum naechsten # Neustart die alte. Das ist kosmetisch; unsere eigene Anzeige liest die # Version direkt von der Platte (siehe __init__.py). OHNE_NEUSTART = ("frontend/",) OHNE_NEUSTART_DATEIEN = ("manifest.json",) def version_von_platte() -> str | None: """Die Versionsnummer aus der manifest.json neben diesem Modul. Home Assistant haelt das Manifest fuer die Laufzeit fest (loader.py: `cache = hass.data[DATA_INTEGRATIONS]`) - nach einem Update meldet `async_get_integration()` also weiter die alte Nummer, selbst wenn der Config-Eintrag neu geladen wird. Fuer ein Update ohne Neustart braucht es deshalb eine Quelle, die wirklich von der Platte liest. `None`, wenn die Datei fehlt oder unlesbar ist - dann bleibt der Aufrufer bei dem Wert, den er schon hat, statt eine erfundene Nummer zu zeigen.""" pfad = os.path.join(INTEGRATIONSORDNER, "manifest.json") try: with open(pfad, encoding="utf-8") as datei: return json.load(datei).get("version") or None except (OSError, ValueError) as fehler: _LOGGER.warning("manifest.json nicht lesbar (%s)", fehler) return None # Endungen, bei denen Zeilenenden nichts bedeuten. # # Am 05.09.2026 am echten Update-Weg gemessen: die installierte Fassung lag mit # CRLF auf der Platte (Windows-Arbeitsstand), das Gitea-Archiv liefert LF - # geokodierung.py, screening.py und veroeffentlichung.py galten dadurch als # geaendert, obwohl sie nach Entfernen der CR byteweise identisch waren. Die # Erkennung meldete "Neustart noetig" fuer einen rein kosmetischen Unterschied. # # Binaerdateien (Schriften, Bilder, das OTA-Zip) bleiben bewusst aussen vor: # dort waere 0x0D0A echter Inhalt, und ein normalisierter Vergleich koennte # einen echten Unterschied verschlucken. TEXTENDUNGEN = (".py", ".json", ".yaml", ".yml", ".md", ".txt", ".js", ".css", ".html", ".svg") def _dateihash(pfad: str) -> str: """sha256 einer Datei - bei Textdateien ueber auf LF normalisierten Inhalt.""" hasher = hashlib.sha256() if pfad.lower().endswith(TEXTENDUNGEN): try: with open(pfad, "rb") as datei: roh = datei.read() hasher.update(roh.replace(b"\r\n", b"\n").replace(b"\r", b"\n")) return hasher.hexdigest() except OSError: raise with open(pfad, "rb") as datei: for block in iter(lambda: datei.read(65536), b""): hasher.update(block) return hasher.hexdigest() def _dateihashes(ordner: str) -> dict[str, str]: """Alle Dateien unter `ordner` als {relativer Pfad: sha256}. __pycache__ bleibt aussen vor: es entsteht beim Laufen und sagt nichts ueber die ausgelieferte Fassung.""" aus: dict[str, str] = {} for wurzel, ordnerliste, dateien in os.walk(ordner): ordnerliste[:] = [o for o in ordnerliste if o != "__pycache__"] for name in dateien: pfad = os.path.join(wurzel, name) rel = os.path.relpath(pfad, ordner).replace(os.sep, "/") aus[rel] = _dateihash(pfad) return aus def neustart_noetig(alter_ordner: str, neuer_ordner: str) -> bool: """Ob die neue Fassung einen Home-Assistant-Neustart braucht. Bewusst pessimistisch: gemeldet wird `False` nur, wenn JEDE geaenderte, hinzugekommene oder entfallene Datei unter frontend/ liegt oder die manifest.json ist. Alles andere - .py, services.yaml, translations/, vorlage/ - gilt als neustartpflichtig, auch wenn es das im Einzelfall vielleicht nicht waere. Der Grund fuer die Schieflage: sagt diese Funktion faelschlich "kein Neustart noetig", laeuft neuer Python-Code nie an, und der Fehler wird an einer ganz anderen Stelle gesucht. Ein ueberfluessiger Neustart kostet dagegen eine Minute. Im Zweifel also Neustart. Laesst sich der alte Ordner nicht lesen, gilt ebenfalls Neustart.""" try: alt = _dateihashes(alter_ordner) neu = _dateihashes(neuer_ordner) except OSError as fehler: _LOGGER.warning( "Konnte alte und neue Fassung nicht vergleichen (%s) - Neustart angenommen.", fehler, ) return True geaendert = { pfad for pfad in set(alt) | set(neu) if alt.get(pfad) != neu.get(pfad) } ohne_neustart = { pfad for pfad in geaendert if pfad.startswith(OHNE_NEUSTART) or pfad in OHNE_NEUSTART_DATEIEN } rest = sorted(geaendert - ohne_neustart) if rest: _LOGGER.info( "Update braucht einen Neustart - geaendert ausserhalb von frontend/: %s", ", ".join(rest[:8]) + (" …" if len(rest) > 8 else ""), ) return True _LOGGER.info( "Update betrifft nur die Oberflaeche (%s Datei(en)) - kein Neustart noetig.", len(geaendert), ) return False async def update_installieren(hass: HomeAssistant, token: str) -> dict[str, Any]: """Lädt das komplette Repo-Archiv und delegiert Entpacken/Prüfen/Tauschen an eine blockierende Funktion im Executor - Datei- und Zip-Operationen gehören nicht in den Event-Loop.""" if not token: raise AktualisierungsFehler( "Kein Gitea-Token hinterlegt - unter Einstellungen -> Geräte & Dienste -> " "Audi Dashboard -> Konfigurieren eintragen." ) url = f"{_GITEA_BASIS}/repos/{_REPO_BESITZER}/{_REPO_NAME}/archive/{_ZWEIG}.zip" session = async_get_clientsession(hass) try: async with session.get(url, headers=_header(token)) as antwort: if antwort.status == 401: raise AktualisierungsFehler("Token ungültig oder abgelaufen.") if antwort.status != 200: raise AktualisierungsFehler(f"Gitea antwortete mit Status {antwort.status}.") zip_bytes = await antwort.read() except AktualisierungsFehler: raise except Exception as fehler: raise AktualisierungsFehler(f"Gitea nicht erreichbar: {fehler}") from fehler ergebnis = await hass.async_add_executor_job( entpacken_pruefen_tauschen, zip_bytes, INTEGRATIONSORDNER, _STAGING_ORDNER, _BACKUP_ORDNER ) return ergebnis def entpacken_pruefen_tauschen( zip_bytes: bytes, integrationsordner: str, staging_ordner: str, backup_ordner: str ) -> str: """Blockierend: entpackt nur custom_components/audi_dashboard/ aus dem Archiv in staging_ordner, prüft das dortige manifest.json, und tauscht per os.rename mit integrationsordner. Wirft AktualisierungsFehler bei jedem Problem - und zwar VOR jedem Tausch, siehe Moduldocstring. Bewusst ohne hass-Abhängigkeit und mit allen Pfaden als Parameter statt fester Modul-Konstanten, damit sie sich isoliert mit einem selbstgebauten Zip gegen einen Temp-Ordner testen lässt, ohne den echten Integrationsordner anzufassen.""" if os.path.isdir(staging_ordner): shutil.rmtree(staging_ordner) try: with zipfile.ZipFile(BytesIO(zip_bytes)) as archiv: namen = archiv.namelist() if not namen: raise AktualisierungsFehler("Leeres Archiv erhalten.") # Gitea packt alles unter einem Wurzelordner wie "audi-app-main/" # - das wird hier gesucht statt geraten. treffer = [ n for n in namen if _INTEGRATIONS_PFAD_IM_REPO in n and not n.endswith("/") ] if not treffer: raise AktualisierungsFehler( f"{_INTEGRATIONS_PFAD_IM_REPO} nicht im heruntergeladenen Archiv gefunden." ) wurzel_marker = treffer[0].split(_INTEGRATIONS_PFAD_IM_REPO, 1)[0] \ + _INTEGRATIONS_PFAD_IM_REPO os.makedirs(staging_ordner, exist_ok=True) for name in treffer: if wurzel_marker not in name: continue rel = name.split(wurzel_marker, 1)[1] if not rel: continue ziel = os.path.join(staging_ordner, rel.replace("/", os.sep)) os.makedirs(os.path.dirname(ziel), exist_ok=True) with archiv.open(name) as quelle, open(ziel, "wb") as senke: shutil.copyfileobj(quelle, senke) except zipfile.BadZipFile as fehler: shutil.rmtree(staging_ordner, ignore_errors=True) raise AktualisierungsFehler(f"Kein gültiges Zip-Archiv: {fehler}") from fehler manifest_pfad = os.path.join(staging_ordner, "manifest.json") if not os.path.exists(manifest_pfad): shutil.rmtree(staging_ordner, ignore_errors=True) raise AktualisierungsFehler("Entpackter Ordner enthält keine manifest.json.") with open(manifest_pfad, encoding="utf-8") as datei: manifest = json.load(datei) if manifest.get("domain") != "audi_dashboard": shutil.rmtree(staging_ordner, ignore_errors=True) raise AktualisierungsFehler( f"Heruntergeladenes Manifest gehört zu Domain '{manifest.get('domain')}', " "nicht 'audi_dashboard' - Abbruch, nichts wurde angefasst." ) neue_version = manifest.get("version") if not neue_version: shutil.rmtree(staging_ordner, ignore_errors=True) raise AktualisierungsFehler("Heruntergeladenes Manifest hat kein version-Feld.") # __pycache__ aus dem Staging-Ordner entfernen - dieselbe Sorgfalt wie # install.ps1s Aufräumschritt nach dem Kopieren. for wurzel, ordner, _dateien in os.walk(staging_ordner): if "__pycache__" in ordner: shutil.rmtree(os.path.join(wurzel, "__pycache__"), ignore_errors=True) # Vergleich VOR dem Tausch - danach gibt es die alte Fassung unter diesem # Pfad nicht mehr. neustart = neustart_noetig(integrationsordner, staging_ordner) if os.path.isdir(backup_ordner): shutil.rmtree(backup_ordner) try: os.rename(integrationsordner, backup_ordner) except OSError as fehler: shutil.rmtree(staging_ordner, ignore_errors=True) raise AktualisierungsFehler( f"Konnte den bestehenden Ordner nicht sichern: {fehler}" ) from fehler # HA scannt JEDEN Ordner unter custom_components/ mit manifest.json als # eigene Integration - nach der Domain im Manifest, nicht nach dem # Ordnernamen. Ohne diesen Schritt wäre der Backup-Ordner beim nächsten # Neustart eine zweite Integration mit domain "audi_dashboard": HA # importiert dann auch deren dienste.py als eigenes Modul # (custom_components.audi_dashboard_backup.dienste), dessen __file__ im # Backup-Ordner liegt - INTEGRATIONSORDNER und _BACKUP_ORDNER fallen für # diese Kopie auf denselben Pfad zusammen, und ihre Dienst-Registrierung # kollidiert mit der echten Integration. Das Manifest bleibt als .bak # erhalten, falls die Fassung zurückgerollt werden muss. backup_manifest = os.path.join(backup_ordner, "manifest.json") if os.path.isfile(backup_manifest): os.rename(backup_manifest, backup_manifest + ".bak") try: os.rename(staging_ordner, integrationsordner) except OSError as fehler: # Rollback: erst das Manifest wiederherstellen, dann die alte # Fassung zurückbenennen - sonst stünde HA mit einem Ordner ohne # manifest.json da. backup_manifest_bak = backup_manifest + ".bak" if os.path.isfile(backup_manifest_bak): os.rename(backup_manifest_bak, backup_manifest) os.rename(backup_ordner, integrationsordner) raise AktualisierungsFehler( f"Tausch fehlgeschlagen, alte Fassung wiederhergestellt: {fehler}" ) from fehler return {"version": neue_version, "neustart_noetig": neustart}