import os from datetime import datetime, timedelta, timezone from pathlib import Path from typing import List import yaml try: from yaml import CLoader as Loader, CDumper as Dumper except ImportError: from yaml import Loader, Dumper class FileVersion: def __init__( self, date: datetime = None, filename: str = None, author: str = None, is_approve: bool = False, is_public: bool = False, is_backup: bool = None, dir_path: Path = None, st_size: int = None, cmf_deleted: bool = False, version: int = 0, md5sum: str = '', base_path: str = '' ): # В старых версиях нет оффсета, считаем что там был utc if not date.tzinfo: date = date.replace(tzinfo=timezone.utc) self.date = date self.filename = filename self.author = author self.is_approve = is_approve self.is_public = is_public self.is_backup = is_backup self.dir_path = dir_path self.st_size = st_size self.cmf_deleted = cmf_deleted self.version = version self.md5sum = md5sum self.base_path = base_path dump_fieldnames = ['author', 'date', 'filename', 'is_approve', 'is_public', 'is_backup', 'cmf_deleted', 'st_size', 'version', 'md5sum'] def to_pyobj(self): return {k: v for k, v in vars(self).items() if k in self.dump_fieldnames} @property def abspath(self): return os.path.join(self.dir_path, self.filename) @property def path(self): return os.path.join(self.dir_path, self.filename).replace(self.base_path, "", 1) class FileVersions: def __init__(self, file_path: str = None, versions: List[FileVersion] = None, _rdisk=None): self.file_path = file_path self.versions = versions or [] self._rdisk = _rdisk @property def meta_path(self) -> str: return f"{self.file_path.rstrip('/')}.meta" @property def version_file_path(self) -> str: return os.path.join(self.meta_path, "versions.yaml") @property def versions_dir_path(self) -> str: return os.path.join(self.meta_path, "versions") @classmethod def for_file_path(cls, file_path: Path, _rdisk=None) -> "FileVersions": fvs = cls(file_path=file_path, _rdisk=_rdisk) fvs._rdisk.io.io_touch(fvs.version_file_path, exist_ok=True) fvs._rdisk.io.io_mkdir(fvs.versions_dir_path, exist_ok=True) versions = [] versions_file_data = fvs._rdisk.io.io_read_bytes(fvs.version_file_path) if versions_file_data: # safe_load приводил к тормозам # TODO1: вернуть safe_load, когда сделаем очистку versions-файла for v in yaml.load(versions_file_data, Loader=Loader): if v.get('cmf_deleted'): continue versions.append(FileVersion(**v, dir_path=fvs.versions_dir_path, base_path=fvs._rdisk.base_path)) versions.sort(key=lambda v: v.version) fvs.versions = versions return fvs def last_version_date(self): if not self.versions: return None return max(i.date for i in self.versions if not i.is_approve) def last_version(self): if not self.versions: return None return sorted(self.versions, key=lambda x: x.version, reverse=True)[0] @staticmethod def get_filename(name: str) -> str: from . import cmf_hashlib import re import math import mimetypes rus_simbol = re.findall(r'[А-Яа-я]', name) os_max_length = 127 file_name_length = len(name) - len(rus_simbol) + len(rus_simbol) * 2 if file_name_length > os_max_length: ext = '' mimetype, encoding = mimetypes.guess_type(name) if mimetype: ext = name.split('.')[-1] slice_length = math.floor(os_max_length/2) name = f'{name[:slice_length - (5 + len(ext))].strip()}-{cmf_hashlib.short_str_enc(name, 4)}' if ext: name = f'{name}.{ext}' print(f'Слишком длинное название. {file_name_length} > {os_max_length}. Меняем на {name}') return name def backup(self, **kwargs): date = datetime.now(timezone.utc).replace(microsecond=0) filename = self.get_filename(date.strftime("%Y%m%d%H%M%S") + '-' + self._rdisk.io.io_name(self.file_path)) new_version = FileVersion( date=date, filename=filename, author=kwargs.get("author", None), is_approve=False, is_backup=True, dir_path=self.versions_dir_path, st_size= self._rdisk.io.io_stat(self.file_path).st_size, base_path=self._rdisk.base_path ) # https://bcrm.carbonsoft.ru/desk/cards?obj=Task:TEM-1625015582#na-eve-ne-skachivayutsya-nekotorye-fajly md5_path = os.path.join(self.meta_path, "md5") if self._rdisk.io.io_exists(md5_path): new_version.md5sum = self._rdisk.io.io_read_text(md5_path) # проверяем изменился ли файл if self.versions and self.versions[-1].st_size == new_version.st_size and self.versions[-1].md5sum == new_version.md5sum: return self.safe_copy(self.file_path, new_version.abspath) new_version.st_size = self._rdisk.io.io_stat(new_version.abspath).st_size if not self.versions: new_version.version = 1 else: new_version.version = self.versions[-1].version + 1 self.versions.append(new_version) self.dump_versions_file() def backup_smart(self, **kwargs): last_version = self.last_version() if last_version and last_version.date and (datetime.now(timezone.utc) - last_version.date < timedelta(minutes=30)): md5_path = os.path.join(self.meta_path, "md5") if self._rdisk.io.io_exists(md5_path): md5sum = self._rdisk.io.io_read_text(md5_path) st_size = self._rdisk.io.io_stat(self.file_path).st_size # проверяем изменился ли файл if last_version.st_size == st_size and last_version.md5sum == md5sum: return # последнюю версию заменить на новую self.remove(last_version) self.backup(**kwargs) def dump_versions_file(self): pyobj = [i.to_pyobj() for i in self.versions] data = yaml.safe_dump(pyobj, allow_unicode=True) self.safe_write_text(data, self.version_file_path) def restore(self, version: FileVersion): src = version.abspath dst = self.file_path self.safe_copy(src, dst) def remove(self, version: FileVersion): """Фейковое удаление, просто не загружаем в список""" version.cmf_deleted = True self.dump_versions_file() self.versions.remove(version) def safe_write_text(self, data: str, dst: str): dst_tmp = f"{dst.rstrip('/')}.tmp" self._rdisk.io.io_write_text(dst_tmp, data) self._rdisk.io.io_unlink(dst) self._rdisk.io.io_rename(dst_tmp, dst) def safe_copy(self, src: str, dst: str): dst_tmp = f"{dst.rstrip('/')}.tmp" self._rdisk.io.io_touch(dst_tmp) self._rdisk.io.io_copy(src, dst_tmp) self._rdisk.io.io_rename(dst_tmp, dst)