audi-app/custom_components/audi_dashboard/sicherung.py

"""Sicherung von Fahrzeugprofil, Fahrten, Tankvorgängen und Sensor-Zuordnung.

Manuell über "Backup jetzt erstellen" in den Einstellungen oder automatisch je
nach einstellungen.backup_intervall im Fahrzeugprofil (aus/täglich/
wöchentlich/monatlich) - immer um 04:00 Uhr, wenn eingestellt. Ein einzelner
täglicher Termin genügt für alle drei Intervalle: wöchentlich läuft nur
montags, monatlich nur am Monatsersten, täglich jeden Tag.

Jede Sicherung ist ein eigener Zeitstempel-Unterordner mit Kopien der fünf
Bestände, keine ZIP-Datei - eine Sicherung, in die man ohne Werkzeug
hineinschauen und aus der man eine einzelne Datei zurückkopieren kann, ist im
Ernstfall mehr wert als ein paar gesparte Kilobyte.

batteriespannung.jsonl gehört ausdrücklich dazu: batterie.py führt sie
bewusst als mehrjährigen Verlauf, um eine über die Zeit absinkende
Ruhespannung erkennen zu können - ohne Sicherung wäre dieser Verlauf bei
einem Festplattenausfall unwiederbringlich weg, anders als Profil/Fahrten/
Tankvorgänge, die dieselbe Sicherung bereits abdeckte.

entitaeten.json gehört ausdrücklich dazu: ohne sie wäre nach einer
Wiederherstellung die komplette Sensor-Zuordnung aus dem Setup-Menü verloren,
und weil sie sonst nirgends abgelegt ist, gäbe es nichts zurückzuholen.

"Backup exportieren"/"importieren" in der Oberfläche laufen dagegen rein über
den Browser (aktuell geladenes Profil + Fahrten + Tankvorgänge als eine
JSON-Datei); wiederherstellen() hier ist das Gegenstück zum Import, das die
Bestände wieder komplett überschreibt.
"""

from __future__ import annotations

import datetime
import json
import logging
import os
import shutil
from typing import TYPE_CHECKING

if TYPE_CHECKING:
    from .koordinator import Koordinator

_LOGGER = logging.getLogger(__name__)

DATEIEN = (
    "fahrzeugprofil.json",
    "fahrten.jsonl",
    "tankvorgaenge.jsonl",
    "entitaeten.json",
    "batteriespannung.jsonl",
)


def _kopieren(basis: str, ziel: str) -> str:
    os.makedirs(ziel, exist_ok=True)
    for name in DATEIEN:
        quelle = os.path.join(basis, name)
        if os.path.exists(quelle):
            shutil.copy(quelle, os.path.join(ziel, name))
    return ziel


async def jetzt(k: Koordinator) -> str:
    zeitstempel = datetime.datetime.now().strftime("%Y%m%d_%H%M%S")
    ziel = os.path.join(k.ablage.backup_ordner, zeitstempel)
    await k.hass.async_add_executor_job(_kopieren, k.ablage.basis, ziel)
    await _vermerken(k)
    _LOGGER.info("Sicherung erstellt unter %s", ziel)
    return ziel


async def _vermerken(k: Koordinator) -> None:
    profil = await k.ablage.profil_lesen()
    if profil is None:
        return
    profil.setdefault("einstellungen", {})["letztes_backup"] = (
        datetime.datetime.now().isoformat()
    )
    await k.ablage.profil_schreiben(profil)
    await k.profil_veroeffentlichen(profil)


async def automatisch(k: Koordinator) -> None:
    """Der tägliche 04:00-Termin - prüft selbst, ob heute überhaupt eine
    Sicherung fällig ist."""
    profil = await k.ablage.profil_lesen()
    if profil is None:
        return
    intervall = profil.get("einstellungen", {}).get("backup_intervall", "aus")
    heute = datetime.date.today()
    faellig = (
        intervall == "taeglich"
        or (intervall == "woechentlich" and heute.weekday() == 0)
        or (intervall == "monatlich" and heute.day == 1)
    )
    if not faellig:
        return
    ziel = await jetzt(k)
    _LOGGER.info("Automatische Sicherung (%s) erstellt unter %s", intervall, ziel)


def _als_objekt(wert: object) -> object:
    return json.loads(wert) if isinstance(wert, str) else wert


async def wiederherstellen(k: Koordinator, daten: dict) -> None:
    """Überschreibt Profil, Fahrten, Tankvorgänge und die Sensor-Zuordnung
    komplett.

    entitaeten_json ist optional: ältere Sicherungen, die vor dem Setup-Menü
    entstanden sind, enthalten den Abschnitt nicht - dann bleibt die aktuelle
    Zuordnung unangetastet, statt sie zu leeren."""
    if daten.get("profil_json"):
        await k.ablage.profil_schreiben(_als_objekt(daten["profil_json"]))  # type: ignore[arg-type]
    if daten.get("fahrten_json") is not None:
        await k.ablage.fahrten_schreiben(_als_objekt(daten["fahrten_json"]))  # type: ignore[arg-type]
    if daten.get("tankvorgaenge_json") is not None:
        await k.ablage.tankvorgaenge_schreiben(_als_objekt(daten["tankvorgaenge_json"]))  # type: ignore[arg-type]
    if daten.get("entitaeten_json"):
        await k.zuordnung.speichern(_als_objekt(daten["entitaeten_json"]))  # type: ignore[arg-type]
        await k.trigger_neu_binden()
    await k.alles_veroeffentlichen()
    _LOGGER.info("Sicherung wiederhergestellt")