audi-app/custom_components/audi_dashboard/aktualisierung.py

"""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.
  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 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}"}


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 {
        "verfuegbar": remote_version != eigene_version,
        "version": remote_version,
        "geprueft_am": datetime.datetime.now(datetime.timezone.utc).isoformat(),
    }


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

    neue_version = await hass.async_add_executor_job(
        entpacken_pruefen_tauschen, zip_bytes, INTEGRATIONSORDNER, _STAGING_ORDNER, _BACKUP_ORDNER
    )
    return {"version": neue_version}


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)

    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

    try:
        os.rename(staging_ordner, integrationsordner)
    except OSError as fehler:
        # Rollback: die alte Fassung zurückbenennen, statt HA ohne
        # Integrationsordner dastehen zu lassen.
        os.rename(backup_ordner, integrationsordner)
        raise AktualisierungsFehler(
            f"Tausch fehlgeschlagen, alte Fassung wiederhergestellt: {fehler}"
        ) from fehler

    return neue_version