"""Загрузка, сохранение и версионирование схемы базы. Ключевая деталь — определение «изменилось ли что-нибудь». Шифрование недетерминировано, поэтому сравнивать байты файла бессмысленно: одни и те же данные каждый раз дают разный шифртекст. Вместо этого храним sha256 от канонического (отсортированного, без пробелов) открытого JSON. Без этого git получал бы новый коммит на каждое срабатывание автосохранения. """ from __future__ import annotations import hashlib import json import os import shutil from pathlib import Path from . import crypto from .models import SCHEMA_VERSION, Document def canonical_bytes(raw: dict) -> bytes: """Байты, по которым считается хеш и которые уходят в шифрование. sort_keys обязателен: без него порядок ключей мог бы плавать и создавать «изменения» на ровном месте. """ return json.dumps( raw, ensure_ascii=False, sort_keys=True, separators=(",", ":") ).encode("utf-8") def content_hash(raw: dict) -> str: return hashlib.sha256(canonical_bytes(raw)).hexdigest() # --- Миграции схемы ------------------------------------------------------- # # Каждая функция получает сырой словарь версии N и возвращает версию N+1. # Нужны потому, что база уже лежит в git: старый файл должен открываться # новой программой без ручных плясок. def _migrate_0_to_1(raw: dict) -> dict: raw["schema_version"] = 1 raw.setdefault("journal", []) return raw def _migrate_1_to_2(raw: dict) -> dict: """Появились фасовки товара. Всё, что заведено раньше, считается записанным в базовых единицах: фасовка пустая, множитель единица. Поведение старых документов при этом не меняется ни на копейку. """ for product in raw.get("products", []): product.setdefault("packs", []) for document, key in ( *((b, "lines") for b in raw.get("batches", [])), *((s, "lines") for s in raw.get("sales", [])), ): for line in document.get(key, []): line.setdefault("uom", "") line.setdefault("uom_size", "1") raw["schema_version"] = 2 return raw _MIGRATIONS = {0: _migrate_0_to_1, 1: _migrate_1_to_2} def migrate(raw: dict) -> dict: version = int(raw.get("schema_version", 0)) if version > SCHEMA_VERSION: raise crypto.UnsupportedFormat( f"База сохранена более новой версией программы (схема {version}). " "Обнови приложение, иначе данные можно потерять." ) while version < SCHEMA_VERSION: migration = _MIGRATIONS.get(version) if migration is None: raise crypto.VaultError(f"Нет миграции со схемы {version}.") raw = migration(raw) version = int(raw["schema_version"]) return raw class Vault: """Открытая база: документ в памяти плюс всё, что нужно для записи.""" def __init__(self, path: Path, password: str, doc: Document, saved_hash: str = ""): self.path = Path(path) self._password = password self.doc = doc self._saved_hash = saved_hash # Растёт при каждой мутации. UI смотрит на него, чтобы дёшево понять, # что пора перерисоваться и запустить отложенное автосохранение. self.revision = 0 # --- открытие и создание --- @classmethod def create(cls, path: Path, password: str) -> "Vault": path = Path(path) if path.exists(): raise crypto.VaultError(f"Файл {path} уже существует.") path.parent.mkdir(parents=True, exist_ok=True) vault = cls(path, password, Document()) vault.save(force=True) return vault @classmethod def open(cls, path: Path, password: str) -> "Vault": path = Path(path) blob = path.read_bytes() plaintext = crypto.decrypt(blob, password) try: raw = json.loads(plaintext.decode("utf-8")) except (UnicodeDecodeError, json.JSONDecodeError) as exc: raise crypto.VaultError( "База расшифровалась, но её содержимое повреждено." ) from exc raw = migrate(raw) doc = Document.from_dict(raw) # Хеш считаем от того, что получилось после разбора и миграции, а не от # прочитанного сырья: иначе миграция или нормализация Decimal сразу же # выглядели бы как несохранённое изменение. return cls(path, password, doc, saved_hash=content_hash(doc.to_dict())) # --- состояние --- def touch(self) -> None: """Отметить, что документ изменили. Зовётся из journal.apply.""" self.revision += 1 @property def is_dirty(self) -> bool: return content_hash(self.doc.to_dict()) != self._saved_hash # --- запись --- def save(self, force: bool = False) -> bool: """Записать базу на диск. Возвращает True, только если реально писали. Именно этот возврат решает, будет ли git-коммит: если открытые данные не изменились, файл не трогаем и коммита не будет. """ raw = self.doc.to_dict() digest = content_hash(raw) if not force and digest == self._saved_hash: return False blob = crypto.encrypt(canonical_bytes(raw), self._password) _atomic_write(self.path, blob) self._saved_hash = digest return True def reload(self) -> None: """Перечитать файл тем же паролем. Нужно после того, как файл на диске подменил git: подтянули изменения с другой машины или откатились на серверную версию. """ fresh = Vault.open(self.path, self._password) self.doc = fresh.doc self._saved_hash = fresh._saved_hash self.revision += 1 def change_password(self, new_password: str) -> None: self._password = new_password self.save(force=True) def check_password(self, password: str) -> bool: return password == self._password def export_plain_json(self, path: Path) -> Path: """Выгрузить базу в открытый JSON. Единственная страховка от забытого пароля, поэтому она есть в UI явной кнопкой. Файл содержит и git-токен — его место не в общей папке. """ path = Path(path) path.parent.mkdir(parents=True, exist_ok=True) path.write_text( json.dumps(self.doc.to_dict(), ensure_ascii=False, indent=2), encoding="utf-8", ) return path def _atomic_write(path: Path, blob: bytes) -> None: """Записать файл так, чтобы обрыв питания не оставил его наполовину. Пишем во временный файл, сбрасываем на диск, только потом подменяем боевой. Предыдущая версия уезжает в .bak — дешёвая страховка от того, что свежий файл окажется битым. """ path.parent.mkdir(parents=True, exist_ok=True) tmp = path.with_suffix(path.suffix + ".tmp") with open(tmp, "wb") as fh: fh.write(blob) fh.flush() os.fsync(fh.fileno()) if path.exists(): try: shutil.copy2(path, path.with_suffix(path.suffix + ".bak")) except OSError: # Бэкап — приятный бонус, но не повод не сохранить данные. pass os.replace(tmp, path)