certain/backend/app/services/backup_data.py

"""Voll-Backup als JSON: Export und Wiederherstellung.

Warum ein eigenes Format neben ``/export/all.json``: jener Export ist zum
Ansehen gedacht und referenziert alles über Namen. Kategorienamen sind aber
nicht eindeutig, mehrere Tabellen fehlen ganz, und eine Version steht auch nicht
drin – zurückladen lässt sich das nicht verlustfrei.

Hier wird stattdessen tabellengetrieben gearbeitet: jede Zeile bringt ihre
Quell-ID als ``local_id`` mit, Fremdschlüssel bleiben IDs. Beim Einspielen wird
eine Abbildung ``local_id -> neue ID`` aufgebaut und jeder Verweis übersetzt.
Das ist unabhängig von Namen und überlebt auch doppelte Bezeichnungen.

Zwei Reichweiten:

``user``    – die Daten eines Benutzers. Jeder kann sein eigenes Backup ziehen.
``system``  – alle Benutzer inklusive Passwort-Hash und TOTP-Secret, dazu die
              globalen Tabellen. Nur für Admins; die Datei ist wie ein Passwort
              zu behandeln.
"""

from __future__ import annotations

import json
from dataclasses import dataclass, field
from datetime import date as date_t, datetime, time as time_t
from decimal import Decimal
from typing import Callable

from sqlalchemy import Boolean, Date, DateTime, Numeric, Time
from sqlalchemy import Enum as SAEnum
from sqlalchemy import inspect as sa_inspect
from sqlalchemy.orm import Session

from ..branding import VERSION
from ..models import (
    Account,
    AccountIdentifier,
    ApiToken,
    AppSetting,
    BudgetPeriod,
    BudgetPeriodLine,
    BudgetTemplateLine,
    Category,
    CategoryGroup,
    CategorizationRule,
    DashboardLayout,
    ExchangeRate,
    ImportBatch,
    ImportTemplate,
    InboxItem,
    Subscription,
    LinkIgnored,
    PlannedIncomeLine,
    RecoveryCode,
    Transaction,
    TransferLink,
    User,
    UserSetting,
)

FORMAT_NAME = "certain-finance-backup"
SCHEMA_VERSION = 2
ALEMBIC_REVISION = "0008"


class BackupError(ValueError):
    """Fachlicher Fehler beim Lesen oder Einspielen eines Backups."""


# --- Tabellenbeschreibung ---------------------------------------------------


@dataclass
class TableSpec:
    key: str                       # Name im JSON
    model: type                    # ORM-Klasse
    scope: str                     # owner | user | account | period | global
    refs: dict[str, str] = field(default_factory=dict)  # Spalte -> Tabellen-key
    has_id: bool = True            # False bei Tabellen mit natürlichem Schlüssel
    label: str = ""                # Anzeigename für die Prüfansicht

    def __post_init__(self) -> None:
        if not self.label:
            self.label = self.key


# Reihenfolge = Einfügereihenfolge. Verweise dürfen nur auf bereits
# eingefügte Tabellen zeigen; gelöscht wird in umgekehrter Reihenfolge.
USER_TABLES: list[TableSpec] = [
    # Kategorien VOR den Konten: Ein Konto kann auf eine Spar-Kategorie zeigen,
    # und ein Verweis darf nur auf eine schon eingefuegte Tabelle gehen.
    # Umgekehrt zeigt keine Kategorie je auf ein Konto.
    TableSpec("category_groups", CategoryGroup, "owner", label="Kategoriegruppen"),
    TableSpec("categories", Category, "owner",
              {"group_id": "category_groups"}, label="Kategorien"),
    TableSpec("accounts", Account, "owner",
              {"savings_category_id": "categories"}, label="Konten"),
    TableSpec("account_identifiers", AccountIdentifier, "account",
              {"account_id": "accounts"}, label="Konto-Merkmale"),
    TableSpec("import_templates", ImportTemplate, "owner", label="Import-Vorlagen"),
    TableSpec("import_batches", ImportBatch, "owner",
              {"account_id": "accounts", "template_id": "import_templates"},
              label="Import-Läufe"),
    TableSpec("transactions", Transaction, "owner", {
        "account_id": "accounts",
        "to_account_id": "accounts",
        "category_id": "categories",
        "budget_category_id": "categories",
        "import_batch_id": "import_batches",
    }, label="Buchungen"),
    TableSpec("budget_template_lines", BudgetTemplateLine, "owner",
              {"category_id": "categories"}, label="Budget-Vorlage"),
    TableSpec("planned_income_lines", PlannedIncomeLine, "owner",
              {"category_id": "categories", "target_account_id": "accounts"},
              label="Geplante Einnahmen"),
    TableSpec("budget_periods", BudgetPeriod, "owner", label="Budget-Monate"),
    TableSpec("budget_period_lines", BudgetPeriodLine, "period",
              {"budget_period_id": "budget_periods", "category_id": "categories"},
              label="Budget-Monatswerte"),
    TableSpec("dashboard_layouts", DashboardLayout, "owner", label="Dashboard-Layout"),
    TableSpec("categorization_rules", CategorizationRule, "owner",
              {"category_id": "categories"}, label="Schlagwortregeln"),
    TableSpec("user_settings", UserSetting, "user", has_id=False,
              label="Persönliche Einstellungen"),
    # Der Eingang MUSS mitgesichert werden: Er hält fest, welche Zahlungen
    # bereits von aussen erfasst wurden. Fehlt er nach dem Wiedereinspielen,
    # erkennt der nächste Kontoauszug sie nicht wieder und bucht sie ein
    # zweites Mal – genau der Fehler, den diese Tabelle verhindern soll.
    TableSpec("inbox_items", InboxItem, "owner", {
        "account_id": "accounts",
        "category_id": "categories",
        "transaction_id": "transactions",
    }, label="Eingang"),
    # Abos & Vertraege: reine Stammdaten, aber muehsam nachzutragen – und sie
    # verweisen auf Konten und Kategorien, stehen deshalb nach beiden.
    TableSpec("subscriptions", Subscription, "owner", {
        "account_id": "accounts",
        "savings_account_id": "accounts",
        "category_id": "categories",
    }, label="Abos & Vertraege"),
    TableSpec("transfer_links", TransferLink, "owner",
              {"from_txn_id": "transactions", "to_txn_id": "transactions"},
              label="Übertrags-Verknüpfungen"),
    TableSpec("link_ignored", LinkIgnored, "owner",
              {"a_txn_id": "transactions", "b_txn_id": "transactions"},
              label="Abgelehnte Übertrags-Vorschläge"),
    # Ohne die Geräteschlüssel hört der Kurzbefehl auf dem Telefon nach einem
    # Wiedereinspielen stillschweigend auf zu funktionieren. Gesichert wird nur
    # der Hash – der Klartext existiert nirgends.
    TableSpec("api_tokens", ApiToken, "owner",
              {"default_account_id": "accounts"}, label="Geräteschlüssel"),
]

# Bewusst NICHT gesichert:
#
# ``mail_sources``/``mail_senders`` – die Quelle trägt das verschlüsselte
# IMAP-Passwort. Ein Backup ist eine Datei, die der Nutzer herumreicht; dort
# gehört keine Zugangsdatei hinein, auch keine verschlüsselte. Nach einem
# Wiedereinspielen richtet man das Postfach neu ein, das dauert zwei Minuten.
#
# ``pending_uploads`` – dort liegen rohe Dateien, die noch niemand geprüft hat.
# Sie sind flüchtig, blähen ein JSON-Backup um Megabytes auf und lassen sich
# jederzeit erneut vom Telefon schicken.
#
# ``bank_apps``/``bank_connections``/``bank_account_links`` – dieselbe
# Überlegung wie beim Postfach, nur schärfer: ``bank_apps`` trägt den privaten
# Schlüssel, mit dem jede Anfrage an die Bank signiert wird, und
# ``bank_connections`` eine laufende Sitzung. Beides gehört in keine Datei, die
# man weitergibt. Die Sitzung liesse sich ohnehin nicht wiederbeleben – die
# Zustimmung hängt an der Bank, nicht an unserer Datenbank. Nach einem
# Wiedereinspielen verbindet man die Bank neu; die abgerufenen Buchungen selbst
# sind längst gesichert.

# Nur im systemweiten Backup, je Benutzer.
SYSTEM_USER_TABLES: list[TableSpec] = [
    TableSpec("recovery_codes", RecoveryCode, "user", label="Wiederherstellungscodes"),
]

# Globale Tabellen (nicht benutzergebunden), nur im systemweiten Backup.
GLOBAL_TABLES: list[TableSpec] = [
    TableSpec("app_settings", AppSetting, "global", has_id=False, label="Globale Einstellungen"),
    TableSpec("exchange_rates", ExchangeRate, "global", has_id=False, label="Wechselkurse"),
]

# Spalten, die nie exportiert werden: sie werden beim Einspielen neu gesetzt.
_SKIP_COLUMNS = {"id", "owner_id", "user_id"}


# --- Werte umwandeln --------------------------------------------------------


def _dump_value(col, value):
    if value is None:
        return None
    if isinstance(col.type, SAEnum):
        return value.value if hasattr(value, "value") else str(value)
    if isinstance(col.type, DateTime) or isinstance(value, datetime):
        return value.isoformat()
    if isinstance(col.type, Date) or isinstance(value, date_t):
        return value.isoformat()
    if isinstance(col.type, Time) or isinstance(value, time_t):
        return value.isoformat()
    if isinstance(col.type, Numeric) or isinstance(value, Decimal):
        return str(value)
    if isinstance(col.type, Boolean):
        return bool(value)
    return value


def _load_value(col, value):
    if value is None:
        return None
    try:
        if isinstance(col.type, SAEnum) and col.type.enum_class is not None:
            return col.type.enum_class(value)
        if isinstance(col.type, DateTime):
            return datetime.fromisoformat(value)
        if isinstance(col.type, Date):
            return date_t.fromisoformat(value)
        if isinstance(col.type, Time):
            return time_t.fromisoformat(value)
        if isinstance(col.type, Numeric):
            return Decimal(str(value))
        if isinstance(col.type, Boolean):
            return bool(value)
    except (ValueError, TypeError) as exc:
        raise BackupError(f"Unlesbarer Wert für '{col.key}': {value!r}") from exc
    return value


def _columns(spec: TableSpec):
    """Zu sichernde Spalten – ohne technische IDs, die neu vergeben werden."""
    return [c for c in sa_inspect(spec.model).columns if c.key not in _SKIP_COLUMNS]


# --- Export -----------------------------------------------------------------


def _query(db: Session, spec: TableSpec, user: User | None):
    q = db.query(spec.model)
    if spec.scope == "owner":
        q = q.filter(spec.model.owner_id == user.id)
    elif spec.scope == "user":
        q = q.filter(spec.model.user_id == user.id)
    elif spec.scope == "account":
        q = q.join(Account, spec.model.account_id == Account.id).filter(
            Account.owner_id == user.id
        )
    elif spec.scope == "period":
        q = q.join(BudgetPeriod, spec.model.budget_period_id == BudgetPeriod.id).filter(
            BudgetPeriod.owner_id == user.id
        )
    return q


def _dump_table(db: Session, spec: TableSpec, user: User | None) -> list[dict]:
    cols = _columns(spec)
    rows = []
    for obj in _query(db, spec, user).all():
        row = {c.key: _dump_value(c, getattr(obj, c.key)) for c in cols}
        if spec.has_id:
            row["local_id"] = obj.id
        rows.append(row)
    return rows


def _dump_user_data(db: Session, user: User, specs: list[TableSpec]) -> dict:
    return {spec.key: _dump_table(db, spec, user) for spec in specs}


def _envelope(scope: str, payload: dict) -> dict:
    return {
        "format": FORMAT_NAME,
        "schema_version": SCHEMA_VERSION,
        "app_version": VERSION,
        "alembic_revision": ALEMBIC_REVISION,
        "exported_at": datetime.now().astimezone().isoformat(timespec="seconds"),
        "scope": scope,
        **payload,
    }


def export_user(db: Session, user: User) -> dict:
    """Backup der Daten eines Benutzers."""
    return _envelope("user", {
        "username": user.username,
        "data": _dump_user_data(db, user, USER_TABLES),
    })


def export_system(db: Session) -> dict:
    """Backup der ganzen Instanz inklusive Anmeldegeheimnissen (nur Admin)."""
    user_cols = _columns(TableSpec("users", User, "global"))
    benutzer = []
    for u in db.query(User).order_by(User.id).all():
        satz = {c.key: _dump_value(c, getattr(u, c.key)) for c in user_cols}
        satz["local_id"] = u.id
        satz["data"] = _dump_user_data(db, u, USER_TABLES + SYSTEM_USER_TABLES)
        benutzer.append(satz)
    return _envelope("system", {
        "users": benutzer,
        "global": {spec.key: _dump_table(db, spec, None) for spec in GLOBAL_TABLES},
    })


def to_bytes(obj: dict) -> bytes:
    return json.dumps(obj, ensure_ascii=False, indent=2).encode("utf-8")


# --- Umschlag prüfen --------------------------------------------------------


def parse(raw: bytes) -> dict:
    """Liest eine hochgeladene Datei und prüft den Umschlag."""
    try:
        data = json.loads(raw.decode("utf-8"))
    except (UnicodeDecodeError, json.JSONDecodeError) as exc:
        raise BackupError("Die Datei ist kein lesbares JSON.") from exc
    if not isinstance(data, dict):
        raise BackupError("Unerwarteter Dateiinhalt – erwartet wird ein JSON-Objekt.")
    if data.get("format") != FORMAT_NAME:
        # Der alte, nur zum Ansehen gedachte Export – klar benennen statt
        # halbherzig zu raten.
        if "accounts" in data and "category_groups" in data:
            raise BackupError(
                "Das ist der Ansichts-Export (all.json). Er lässt sich nicht "
                "wiederherstellen – bitte ein Backup über „Voll-Backup“ erzeugen."
            )
        raise BackupError("Das ist kein Backup von Certain Finance.")
    version = data.get("schema_version")
    if version != SCHEMA_VERSION:
        raise BackupError(
            f"Backup-Version {version} wird nicht unterstützt "
            f"(diese Fassung liest Version {SCHEMA_VERSION})."
        )
    if data.get("scope") not in ("user", "system"):
        raise BackupError("Unbekannte Reichweite im Backup.")
    return data


def summarize(data: dict) -> list[dict]:
    """Zählt je Tabelle, was in der Datei steckt – für die Prüfansicht."""
    if data["scope"] == "user":
        haeufen = [data.get("data", {})]
    else:
        haeufen = [u.get("data", {}) for u in data.get("users", [])]

    zeilen = []
    for spec in USER_TABLES + SYSTEM_USER_TABLES:
        anzahl = sum(len(h.get(spec.key) or []) for h in haeufen)
        if anzahl:
            zeilen.append({"key": spec.key, "label": spec.label, "count": anzahl})
    if data["scope"] == "system":
        zeilen.insert(0, {"key": "users", "label": "Benutzer", "count": len(data.get("users", []))})
        for spec in GLOBAL_TABLES:
            anzahl = len(data.get("global", {}).get(spec.key) or [])
            if anzahl:
                zeilen.append({"key": spec.key, "label": spec.label, "count": anzahl})
    return zeilen


# --- Einspielen -------------------------------------------------------------

# Wie im Modus „Ergänzen" entschieden wird, ob eine Zeile schon da ist.
# Rückgabe: Schlüssel, unter dem verglichen wird – oder None für „immer neu".
_MERGE_KEYS: dict[str, Callable[[dict], tuple | None]] = {
    "accounts": lambda r: ("name", (r.get("name") or "").strip().lower()),
    # Ohne diesen Eintrag würde ein zweiter Ergänzen-Lauf die Eindeutigkeit
    # (account_id, kind, value_norm) verletzen.
    "account_identifiers": lambda r: ("ident", r.get("account_id"), r.get("kind"),
                                      r.get("value_norm")),
    "category_groups": lambda r: ("name", (r.get("name") or "").strip().lower()),
    "categories": lambda r: ("group", r.get("group_id"), (r.get("name") or "").strip().lower()),
    "import_templates": lambda r: ("name", (r.get("name") or "").strip().lower()),
    "import_batches": lambda r: ("batch", r.get("account_id"), r.get("filename"),
                                 r.get("created_at")),
    # Die Kennung der Quelle zuerst: Sie ist die einzige EXAKTE Aussage, dass
    # zwei Zeilen dieselbe Zahlung sind – und ohne sie liefe „Ergaenzen" gegen
    # die Eindeutigkeit von (owner_id, source_uid), statt die Zeile zu
    # erkennen. Erst danach der Aehnlichkeits-Schluessel.
    "subscriptions": lambda r: ("abo", (r.get("name") or "").strip().lower()),
    "transactions": lambda r: (
        ("uid", r["source_uid"]) if r.get("source_uid")
        else ("hash", r["import_hash"]) if r.get("import_hash")
        else ("nat", r.get("date"), r.get("account_id"), r.get("amount_cents"),
              (r.get("payee") or "").strip().lower())
    ),
}

# Tabellen, die im Modus „Ergänzen" nur dann übernommen werden, wenn der
# Benutzer davon noch gar nichts hat. Zeilenweise mischen ergäbe hier Unsinn
# (halbe Budgets, doppelte Layouts).
_ALL_OR_NOTHING = {
    "budget_template_lines",
    "planned_income_lines",
    "budget_periods",
    "budget_period_lines",
    "dashboard_layouts",
    "categorization_rules",
    "user_settings",
}


def _existing_keys(db: Session, spec: TableSpec, user: User) -> dict[tuple, int]:
    """Vergleichsschlüssel der schon vorhandenen Zeilen -> deren ID."""
    schluessel = _MERGE_KEYS.get(spec.key)
    if schluessel is None:
        return {}
    cols = _columns(spec)
    treffer: dict[tuple, int] = {}
    for obj in _query(db, spec, user).all():
        roh = {c.key: _dump_value(c, getattr(obj, c.key)) for c in cols}
        k = schluessel(roh)
        if k is not None:
            treffer.setdefault(k, obj.id)
    return treffer


def _delete_user_data(db: Session, user: User, specs: list[TableSpec]) -> None:
    """Löscht die Daten eines Benutzers in fremdschlüsselsicherer Reihenfolge."""
    for spec in reversed(specs):
        for obj in _query(db, spec, user).all():
            db.delete(obj)
        db.flush()


def _insert_rows(
    db: Session, spec: TableSpec, user: User, rows: list[dict], maps: dict[str, dict[int, int]],
    merge: bool,
) -> int:
    cols = {c.key: c for c in _columns(spec)}
    id_map = maps.setdefault(spec.key, {})
    vorhanden = _existing_keys(db, spec, user) if merge else {}
    schluessel = _MERGE_KEYS.get(spec.key)
    neu = 0

    for row in rows:
        # Verweise zuerst übersetzen – sie fließen auch in den Vergleichsschlüssel ein.
        aufgeloest = dict(row)
        for spalte, ziel in spec.refs.items():
            alt = row.get(spalte)
            aufgeloest[spalte] = maps.get(ziel, {}).get(alt) if alt is not None else None

        if merge and schluessel is not None:
            k = schluessel(aufgeloest)
            if k is not None and k in vorhanden:
                # Nicht anlegen, aber den Verweis auf die bestehende Zeile merken.
                if spec.has_id and row.get("local_id") is not None:
                    id_map[row["local_id"]] = vorhanden[k]
                continue

        werte = {}
        for name, col in cols.items():
            if name in spec.refs:
                werte[name] = aufgeloest[name]
            else:
                werte[name] = _load_value(col, row.get(name))
        if spec.scope in ("owner",):
            werte["owner_id"] = user.id
        elif spec.scope == "user":
            werte["user_id"] = user.id

        obj = spec.model(**werte)
        db.add(obj)
        db.flush()
        if spec.has_id and row.get("local_id") is not None:
            id_map[row["local_id"]] = obj.id
        if merge and schluessel is not None:
            k = schluessel(aufgeloest)
            if k is not None:
                vorhanden[k] = obj.id
        neu += 1
    return neu


def restore_user_data(
    db: Session, user: User, data: dict, merge: bool, specs: list[TableSpec] | None = None
) -> dict[str, int]:
    """Spielt den Datenblock eines Benutzers ein. Ruft **kein** commit."""
    specs = specs or USER_TABLES
    if not merge:
        _delete_user_data(db, user, specs)

    maps: dict[str, dict[int, int]] = {}
    bilanz: dict[str, int] = {}
    for spec in specs:
        rows = data.get(spec.key) or []
        if merge and spec.key in _ALL_OR_NOTHING:
            # Nur übernehmen, wenn dazu noch nichts existiert.
            if _query(db, spec, user).first() is not None:
                bilanz[spec.key] = 0
                continue
        bilanz[spec.key] = _insert_rows(db, spec, user, rows, maps, merge)
    return bilanz


def restore_system(db: Session, data: dict) -> dict[str, int]:
    """Baut die ganze Instanz aus einem systemweiten Backup neu auf.

    Immer ersetzend – ein teilweises Verschmelzen fremder Benutzerkonten wäre
    nicht sinnvoll deutbar. Ruft **kein** commit; das übernimmt der Aufrufer,
    damit im Fehlerfall nichts halb eingespielt zurückbleibt."""
    benutzer = data.get("users") or []
    if not any(u.get("role") == "admin" for u in benutzer):
        raise BackupError(
            "Das Backup enthält keinen Administrator – damit käme niemand mehr hinein."
        )

    # Benutzer weg: alles Benutzergebundene hängt per ON DELETE CASCADE daran.
    for u in db.query(User).all():
        db.delete(u)
    db.flush()
    for spec in GLOBAL_TABLES:
        for obj in db.query(spec.model).all():
            db.delete(obj)
    db.flush()

    user_cols = {c.key: c for c in _columns(TableSpec("users", User, "global"))}
    bilanz: dict[str, int] = {"users": 0}
    for satz in benutzer:
        werte = {k: _load_value(col, satz.get(k)) for k, col in user_cols.items()}
        neuer = User(**werte)
        db.add(neuer)
        db.flush()
        bilanz["users"] += 1
        teil = restore_user_data(
            db, neuer, satz.get("data") or {}, merge=False,
            specs=USER_TABLES + SYSTEM_USER_TABLES,
        )
        for key, anzahl in teil.items():
            bilanz[key] = bilanz.get(key, 0) + anzahl

    global_data = data.get("global") or {}
    for spec in GLOBAL_TABLES:
        cols = {c.key: c for c in _columns(spec)}
        rows = global_data.get(spec.key) or []
        for row in rows:
            db.add(spec.model(**{k: _load_value(col, row.get(k)) for k, col in cols.items()}))
        bilanz[spec.key] = len(rows)
    db.flush()
    return bilanz