sources/uWebDavSync.py - main - VolcanoByte/vbNotes

sources/uWebDavSync.py
162 строки5 KB
Eugene Gavrilyuk
Sync on start and finish
01 апр 2026, 15:32
01 апр 2026, 15:32f2a8bd9
"""Синхронизация с WebDav сервером"""
import os
import shutil
import tempfile
from fnmatch import fnmatch

from PyQt6.QtCore import Qt, QThread, QObject, pyqtSignal

from uDB import db
from uGlobal import store, MessageLanguageConsts as mlc
from uUtils import timestamp2FileName, zipDB, unzipDB
from uConsts import AppConsts, StoreConsts, SyncConsts

from webdav3.client import Client
from webdav3.exceptions import WebDavException


class WebDavSync(QObject):
    finished = pyqtSignal()
    alreadyStarted = False
    result = (True, None)

    def __init__(self, syncURL, syncUsername, syncPassword):
        super().__init__()

        self.syncURL = syncURL
        self.syncUsername = syncUsername
        self.syncPassword = syncPassword

        self.client = None
        WebDavSync.alreadyStarted = True
        self.timestamp = db.timestamp()
        self.syncId = db.syncId
        self.filename = db.filename

    def check(self) -> int | None:
        """Проверка резервной копии. Возвращает константу SyncConsts или
        None, если нет интернета или другая ошибка"""

        dbLastChange = timestamp2FileName(self.timestamp, self.syncId)

        self.client = self.getClient()
        try:
            self.client.check()


            files = self.getFiles(f'*.{self.syncId}_{AppConsts.CollectionKind}')

            whoYounger, _ = self.whoYounger(dbLastChange, files)

            return whoYounger
        except WebDavException as e:
            return None

    def run(self):
        dbLastChange = timestamp2FileName(self.timestamp, self.syncId)

        self.client = self.getClient()
        try:
            self.client.check()

            files = self.getFiles(
                f'*.{self.syncId}_{AppConsts.CollectionKind}')
            whoYounger, backup = self.whoYounger(dbLastChange, files)

            if whoYounger == SyncConsts.YoungerDB:
                # Копируем базу на бэкап
                self.copyDBTo(dbLastChange)
                self.delete(files)
                WebDavSync.result = (True, mlc.syncYoungerDB)
            elif whoYounger == SyncConsts.YoungerBackup:
                # Копируем бэкап на базу
                self.copyFrom(backup)
                WebDavSync.result = (True, mlc.syncYoungerBackup)
            else:
                WebDavSync.result = (True, mlc.syncIdentical)

        except WebDavException as e:
            WebDavSync.result = (False, e)

        self.finished.emit()

    @staticmethod
    def whoYounger(dtDB: str | None, backups: tuple) -> tuple:
        """Проверка, кто свежее, база или бэкап. dtDB - это дата и время
        последнего изменения базы, backups - это список файлов-бэкапов.
        Возвращается кортеж из константы (моложе/старше/одинаковы) и имя
        свежайшего файла-бэкапа, если таковой есть."""

        # Ищем самый свежий бэкап
        _filename = None
        for filename in backups:
            if _filename is None or filename > _filename:
                _filename = filename

        #  Если дата базы неопределенна
        if dtDB is None:
            if _filename is None:
                return SyncConsts.YoungerIdentical, None
            else:
                return SyncConsts.YoungerBackup, _filename

        # Если бэкап не определен
        if _filename is None:
            return SyncConsts.YoungerDB, None

        if dtDB == _filename:
            return SyncConsts.YoungerIdentical, None
        # Бэкап свежее
        elif _filename > dtDB:
            return SyncConsts.YoungerBackup, _filename
        #  Одинаковы
        elif dtDB == _filename:
            return SyncConsts.YoungerIdentical, None
        else:
            # База свежее
            return SyncConsts.YoungerDB, None

    def getClient(self) -> Client:
        """Создание клиента WebDav"""
        options = {
            'webdav_hostname': self.syncURL,
            'webdav_login': self.syncUsername,
            'webdav_password': self.syncPassword
        }

        return Client(options)

    def getFiles(self, wildcards: str = '*.*') -> tuple:
        """Получение списка файлов-бэкапов"""
        return tuple(file for file in self.client.list() if fnmatch(file,
                                                                    wildcards))

    def copyDBTo(self, zipFileName: str):
        """Копирует сжатую базу на сервер WebDav"""

        tempFolder = tempfile.mkdtemp()

        _zip = os.path.join(tempFolder, zipFileName)

        zipDB(self.filename, _zip)

        self.client.upload_sync(zipFileName, _zip)

        shutil.rmtree(tempFolder)

    def copyFrom(self, zipFileName: str):
        """Копирует сжатую базу с сервера WebDav"""
        tempFolder = tempfile.mkdtemp()
        _zip = os.path.join(tempFolder, zipFileName)

        self.client.download(zipFileName, _zip)

        _path, _filename = os.path.split(self.filename)
        unzipDB(_path, _zip)

        shutil.rmtree(tempFolder)

    def delete(self, files: tuple):
        """Удаление бэкапов"""
        for file in files:
            self.client.clean(file)