ros_control/app/migrations.py

"""Миграции схемы SQLite. Версия хранится в PRAGMA user_version.

v1 — глобально уникальные ID: числовые ID таблиц (devices, device_groups, jobs, backups) заменяются на
<префикс>_<uuid7> (см. app/ids.py) с пересчётом ссылок; порядок записей сохраняется (время в UUIDv7 берётся из
created_at/requested_at). Перед изменениями делается копия файла БД (<файл>.bak-<метка>); всё выполняется
одной транзакцией на «сыром» соединении (DDL в pysqlite иначе не транзакционен) и при любом расхождении в числе строк откатывается.
"""
import json
import logging
import sqlite3
from datetime import datetime, timezone
from pathlib import Path

from sqlalchemy import Engine
from sqlalchemy.dialects import sqlite as sqlite_dialect
from sqlalchemy.schema import CreateTable

from app.ids import new_id

log = logging.getLogger("ros_control.migrations")
SCHEMA_VERSION = 2  # v2: таблица app_settings (создаётся create_all), данные не меняются
TABLES = ("device_groups", "devices", "backups", "jobs")


def _dt(value) -> datetime | None:
    try:
        return datetime.fromisoformat(str(value)) if value else None
    except ValueError:
        return None


def _int(value) -> int | None:
    try:
        return int(value)
    except (TypeError, ValueError):
        return None


def _is_legacy(con: sqlite3.Connection) -> bool:
    cols = {r[1]: (r[2] or "").upper() for r in con.execute("PRAGMA table_info(devices)")}
    return bool(cols) and cols.get("id", "").startswith("INT")


def run(engine: Engine, db_path: Path | None) -> None:
    """Приводит БД к текущей версии схемы. Идемпотентна."""
    if db_path is None:  # не файловая БД (:memory:) — старых данных быть не может
        with engine.begin() as c:
            c.exec_driver_sql(f"PRAGMA user_version = {SCHEMA_VERSION}")
        return
    engine.dispose()  # закрыть пул: дальше работаем отдельным соединением
    con = sqlite3.connect(db_path, isolation_level=None)
    try:
        version = con.execute("PRAGMA user_version").fetchone()[0]
        if version >= SCHEMA_VERSION:
            return
        if version < 1 and _is_legacy(con):
            _to_v1(con, db_path)
        con.execute(f"PRAGMA user_version = {SCHEMA_VERSION}")
    finally:
        con.close()


def _to_v1(con: sqlite3.Connection, db_path: Path) -> None:
    from app.db import Base  # noqa: F401
    from app import models

    stamp = datetime.now(timezone.utc).strftime("%Y%m%d-%H%M%S")
    bak = db_path.with_name(f"{db_path.name}.bak-{stamp}")
    dst = sqlite3.connect(bak)
    con.backup(dst)
    dst.close()
    log.warning("Миграция ID: копия БД сохранена в %s", bak)

    con.row_factory = sqlite3.Row
    con.execute("PRAGMA foreign_keys=OFF")
    con.execute("BEGIN IMMEDIATE")
    try:
        old = {t: con.execute(f"SELECT * FROM {t} ORDER BY id").fetchall() for t in TABLES}
        for t in TABLES:
            con.execute(f"ALTER TABLE {t} RENAME TO {t}_old")
        for tbl in (models.Group.__table__, models.Device.__table__, models.Backup.__table__, models.Job.__table__):
            con.execute(str(CreateTable(tbl).compile(dialect=sqlite_dialect.dialect())))

        grp: dict[int, str] = {}
        for r in old["device_groups"]:
            grp[r["id"]] = new_id("grp")
            con.execute("INSERT INTO device_groups (id, name) VALUES (?, ?)", (grp[r["id"]], r["name"]))

        dev: dict[int, str] = {}
        names: dict[int, str] = {}
        for r in old["devices"]:
            dev[r["id"]] = new_id("dev", _dt(r["created_at"]))
            names[r["id"]] = r["name"]
            con.execute(
                "INSERT INTO devices (id, name, host, port, username, password_enc, verify_tls, use_tls, group_id, note,"
                " created_at, online, status_json, status_at, last_error, last_backup_requested_at)"
                " VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)",
                (dev[r["id"]], r["name"], r["host"], r["port"], r["username"], r["password_enc"], r["verify_tls"],
                 r["use_tls"], grp.get(_int(r["group_id"])), r["note"], r["created_at"], r["online"], r["status_json"],
                 r["status_at"], r["last_error"], r["last_backup_requested_at"]))

        def dev_id(old_id):  # ссылка на уже удалённое устройство сохраняет идентичность (один новый ID на старый)
            return None if old_id is None else dev.setdefault(old_id, new_id("dev"))

        for r in old["backups"]:
            key = r["key_binary"] or r["key_rsc"] or ""
            parts = key.split("/")
            name = names.get(_int(r["device_id"])) or (parts[1] if len(parts) >= 3 else "")
            con.execute(
                "INSERT INTO backups (id, device_id, device_name, job_id, deleted_at, requested_at, status, key_binary, key_rsc, error)"
                " VALUES (?,?,?,?,?,?,?,?,?,?)",
                (new_id("bkp", _dt(r["requested_at"])), dev_id(_int(r["device_id"])), name, None, None, r["requested_at"],
                 r["status"], r["key_binary"], r["key_rsc"], r["error"]))
        for r in old["jobs"]:
            con.execute(
                "INSERT INTO jobs (id, device_id, device_name, type, status, message, created_at, finished_at)"
                " VALUES (?,?,?,?,?,?,?,?)",
                (new_id("job", _dt(r["created_at"])), dev_id(_int(r["device_id"])), r["device_name"], r["type"], r["status"],
                 r["message"], r["created_at"], r["finished_at"]))

        counts = {t: len(old[t]) for t in TABLES}
        for t in TABLES:  # сверка числа строк до коммита
            n = con.execute(f"SELECT count(*) FROM {t}").fetchone()[0]
            if n != counts[t]:
                raise RuntimeError(f"Миграция ID: в {t} было {counts[t]} строк, стало {n}")
        for t in TABLES:
            con.execute(f"DROP TABLE {t}_old")
        con.execute(
            "INSERT INTO events (id, ts, type, entity_type, entity_id, device_id, job_id, actor, message, data)"
            " VALUES (?,?,?,?,?,?,?,?,?,?)",
            (new_id("evt"), datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S.%f"), "system.migrated", "system", None,
             None, None, "system", "Числовые ID заменены на глобально уникальные (dev_/grp_/job_/bkp_)",
             json.dumps({"rows": counts, "backup_file": bak.name})))
        con.execute("COMMIT")
    except Exception:
        con.execute("ROLLBACK")
        log.exception("Миграция ID не выполнена, БД не изменена (копия: %s)", bak)
        raise
    log.warning("Миграция ID выполнена: %s", counts)