"""Datenbank-Umstellungen mit Versionsnummer (PRAGMA user_version). SCHEMA in db.py beschreibt immer den neuesten Stand. Bei einer Änderung am Schema: 1. SCHEMA in db.py anpassen – neue Datenbanken entstehen direkt so und bekommen die höchste Nummer. 2. Unten an STEPS eine Funktion anhängen, die eine bestehende Datenbank vom vorigen Stand dorthin bringt. Jeder Schritt läuft genau einmal, in einer Transaktion mit dem Hochzählen der Nummer. Tests: tests/test_migrationen.py (alte Stände nachbauen und umstellen). """ import logging import re from .bilder import hat_metadaten, ohne_metadaten from .common import ValidationError, now_iso from .config import DATA_DIR from .db import HOUSEHOLD_TABLE, KANBAN_PROJECTS_TABLE, KANBAN_TASKS_TABLE from .htmlclean import sanitize log = logging.getLogger(__name__) def start(conn) -> None: """Neue, leere Datenbank: erste Liste anlegen und gleich auf den neuesten Stand setzen.""" conn.execute("INSERT INTO shopping_lists (name, everyone, position, created_at) VALUES ('Zuhause', 1, 1, ?)", (now_iso(),)) conn.execute(f"PRAGMA user_version = {len(STEPS)}") def upgrade(conn) -> None: """Bestehende Datenbank: alle Schritte nach ihrer Nummer nacheinander ausführen.""" version = conn.execute("PRAGMA user_version").fetchone()[0] for number, step in enumerate(STEPS[version:], start=version + 1): step(conn) conn.execute(f"PRAGMA user_version = {number}") # ---------- Schritt 1: Altbestand aus der Zeit vor der Versionsnummer ---------- # Diese Umstellungen prüfen noch selbst, ob sie nötig sind – eine Datenbank ohne Nummer kann # auf jedem früheren Stand sein. Reihenfolge wie früher beim Start. def _v1_altbestand(conn) -> None: _household(conn) _household_checked(conn) _travel_columns(conn) _monitor_token(conn) _kanban_without_categories(conn) _kanban_completed(conn) _kanban_projects(conn) _kanban_priority(conn) _kanban_done_at(conn) def _household(conn) -> None: """Vorrat/Einkauf aus früheren Versionen (ohne Einheiten, ohne Listen oder Vorrat ohne Liste) neu aufbauen. Alles, was noch keiner Liste gehört, kommt in die erste Liste bzw. in eine neue Liste "Zuhause", die alle Konten sehen. """ info = {r["name"]: r for r in conn.execute("PRAGMA table_info(household_items)")} if "list_id" in info and info["list_id"]["notnull"]: return row = conn.execute("SELECT id FROM shopping_lists ORDER BY position, id LIMIT 1").fetchone() if row: default = row["id"] # Die Zwischenversion nannte die gemeinsame Liste "Einkaufsliste"; jetzt enthält sie auch den Vorrat conn.execute("UPDATE shopping_lists SET name = 'Zuhause' WHERE id = ? AND name = 'Einkaufsliste' AND created_by IS NULL", (default,)) else: default = conn.execute( "INSERT INTO shopping_lists (name, everyone, position, created_at) VALUES ('Zuhause', 1, 1, ?)", (now_iso(),) ).lastrowid unit = "unit" if "unit" in info else "'stk'" list_id = "COALESCE(list_id, ?)" if "list_id" in info else "?" conn.execute("DROP INDEX IF EXISTS household_vorrat") conn.execute("DROP INDEX IF EXISTS household_einkauf") conn.execute("ALTER TABLE household_items RENAME TO household_items_alt") conn.execute("CREATE TABLE household_items " + HOUSEHOLD_TABLE) conn.execute( f"""INSERT INTO household_items (id, key, name, list, list_id, quantity, unit, updated_at, updated_by) SELECT id, key, name, list, {list_id}, quantity, {unit}, updated_at, updated_by FROM household_items_alt""", (default,), ) conn.execute("DROP TABLE household_items_alt") def _household_checked(conn) -> None: # Einkauf ohne Haken-Spalte: ergänzen, bestehende Einträge sind nicht abgehakt columns = {r["name"] for r in conn.execute("PRAGMA table_info(household_items)")} if "checked" not in columns: conn.execute("ALTER TABLE household_items ADD COLUMN checked INTEGER NOT NULL DEFAULT 0") def _travel_columns(conn) -> None: # Reise-Städte aus einer Vorversion ohne Höhe/Zeitzone/Hauptstadt: Spalten ergänzen (Hauptstadt füllt reisen.py nach) columns = {r["name"] for r in conn.execute("PRAGMA table_info(travel_cities)")} for column, kind in (("elevation", "REAL"), ("timezone", "TEXT"), ("capital", "INTEGER")): if column not in columns: conn.execute(f"ALTER TABLE travel_cities ADD COLUMN {column} {kind}") def _kanban_without_categories(conn) -> None: """Kategorien wieder entfernt: kanban_projects ohne category_id neu aufbauen. Aufgaben und Artikel-Links verweisen per ON DELETE CASCADE auf kanban_projects. Ein normales DROP würde sie mitlöschen, deshalb das SQLite-Verfahren: Fremdschlüssel aus, neue Tabelle anlegen, kopieren, alte löschen, umbenennen, Verweise prüfen. """ columns = {r["name"] for r in conn.execute("PRAGMA table_info(kanban_projects)")} if "category_id" not in columns: conn.execute("DROP TABLE IF EXISTS kanban_categories") return conn.commit() conn.execute("PRAGMA foreign_keys = OFF") # wirkt nur außerhalb einer Transaktion try: conn.execute("BEGIN") conn.execute("CREATE TABLE kanban_projects_neu " + KANBAN_PROJECTS_TABLE) conn.execute( """INSERT INTO kanban_projects_neu (id, name, description, position, created_at, created_by, updated_at, updated_by) SELECT id, name, description, position, created_at, created_by, updated_at, updated_by FROM kanban_projects""" ) conn.execute("DROP TABLE kanban_projects") conn.execute("ALTER TABLE kanban_projects_neu RENAME TO kanban_projects") conn.execute("DROP TABLE kanban_categories") if conn.execute("PRAGMA foreign_key_check").fetchall(): raise RuntimeError("Kanban-Migration: ungültige Verweise, Änderungen werden zurückgenommen.") conn.commit() except Exception: conn.rollback() raise finally: conn.execute("PRAGMA foreign_keys = ON") def _kanban_priority(conn) -> None: # Aufgaben ohne Priorität: Spalte ergänzen, bestehende Aufgaben werden "mittel" columns = {r["name"] for r in conn.execute("PRAGMA table_info(kanban_tasks)")} if "priority" not in columns: conn.execute( "ALTER TABLE kanban_tasks ADD COLUMN priority INTEGER NOT NULL DEFAULT 2 CHECK (priority IN (1, 2, 3))" ) def _kanban_done_at(conn) -> None: """Aufgaben ohne "erledigt seit": Spalte ergänzen. Schon erledigte Aufgaben zählen ab jetzt, damit beim Update nichts sofort archiviert oder gelöscht wird.""" columns = {r["name"] for r in conn.execute("PRAGMA table_info(kanban_tasks)")} if "done_at" not in columns: conn.execute("ALTER TABLE kanban_tasks ADD COLUMN done_at TEXT") conn.execute("UPDATE kanban_tasks SET done_at = ? WHERE status = 'erledigt'", (now_iso(),)) def _monitor_token(conn) -> None: # Monitore aus der Zeit vor P42-S.M.: Spalte für den Token ergänzen columns = {r["name"] for r in conn.execute("PRAGMA table_info(monitors)")} if "token" not in columns: conn.execute("ALTER TABLE monitors ADD COLUMN token TEXT") def _kanban_completed(conn) -> None: # Projekte "abgeschlossen": Spalte ergänzen, falls sie fehlt columns = {r["name"] for r in conn.execute("PRAGMA table_info(kanban_projects)")} if "completed_at" not in columns: conn.execute("ALTER TABLE kanban_projects ADD COLUMN completed_at TEXT") def _kanban_projects(conn) -> None: # Aufgaben aus der Zeit vor den Projekten: in "Allgemeines Board" übernehmen columns = {r["name"] for r in conn.execute("PRAGMA table_info(kanban_tasks)")} if "project_id" in columns: return conn.execute("ALTER TABLE kanban_tasks RENAME TO kanban_tasks_alt") conn.execute("CREATE TABLE kanban_tasks " + KANBAN_TASKS_TABLE) if conn.execute("SELECT COUNT(*) FROM kanban_tasks_alt").fetchone()[0]: now = now_iso() project = conn.execute( "INSERT INTO kanban_projects (name, position, created_at, updated_at) VALUES ('Allgemeines Board', 1, ?, ?)", (now, now), ).lastrowid conn.execute( """INSERT INTO kanban_tasks (id, project_id, title, description, status, assignee, position, created_at, created_by, updated_at, updated_by) SELECT id, ?, title, description, status, assignee, position, created_at, created_by, updated_at, updated_by FROM kanban_tasks_alt""", (project,), ) conn.execute("DROP TABLE kanban_tasks_alt") # ---------- Schritt 2: Datenschutz ---------- def _v2_datenschutz(conn) -> None: """Vorschläge pro Liste, Artikel ohne fremde Bilder, hochgeladene Bilder ohne Metadaten (GPS usw.).""" # Vorschläge den Listen zuordnen: was gerade auf einer Liste steht, gehört zu dieser Liste; # alles Übrige stammt aus der Zeit vor den privaten Listen und gehört zu den Listen, die alle sehen conn.execute( """INSERT OR IGNORE INTO list_products (list_id, key, uses, last_used) SELECT DISTINCT i.list_id, p.key, p.uses, p.last_used FROM household_items i JOIN products p ON p.key = i.key""" ) conn.execute( """INSERT OR IGNORE INTO list_products (list_id, key, uses, last_used) SELECT l.id, p.key, p.uses, p.last_used FROM products p JOIN shopping_lists l ON l.everyone = 1 WHERE NOT EXISTS (SELECT 1 FROM household_items i WHERE i.key = p.key)""" ) # Artikel mit den neuen Regeln bereinigen (Bilder nur noch aus den eigenen Uploads) for row in conn.execute("SELECT id, content FROM projects").fetchall(): cleaned = sanitize(row["content"] or "") if cleaned != (row["content"] or ""): conn.execute("UPDATE projects SET content = ? WHERE id = ?", (cleaned, row["id"])) # Hochgeladene Bilder: nur solche mit Metadaten neu schreiben (die anderen bleiben unverändert). # Ein einzelnes Problembild (Rechte, kaputte Datei) darf den Serverstart nie verhindern. uploads = DATA_DIR / "uploads" if uploads.is_dir(): for f in uploads.iterdir(): if not re.fullmatch(r"[a-f0-9]{24}\.(jpg|png|webp)", f.name): continue try: data = f.read_bytes() if hat_metadaten(data): f.write_bytes(ohne_metadaten(data, f.suffix[1:])) except (OSError, ValidationError) as e: log.warning("Bild %s konnte nicht bereinigt werden: %s", f.name, e) # Reihenfolge = Versionsnummer (Index + 1). Nur anhängen, nie umsortieren oder entfernen. STEPS = [ _v1_altbestand, _v2_datenschutz, ]