Исходный код EJIO.user.media.facade

## /user/media/facade.py

import csv
import io
import shutil
import datetime
from pathlib import Path
from threading import Lock
from loguru import logger
from EJIO.utils import encrypt_payload, decrypt_payload, generate_timestamped_filename
from .model import FileNode

__all__ = ['FileFacade']




class FileFacade:
    """
    Универсальный фасад управления физическими файлами и их метаданными.
    Инициализируется индивидуальными путями для изоляции контекстов (аватары, логи, отчеты).
    """

    def __init__(self, working_dir: str | Path, db_path: str | Path, crypto_key: str, file_base_name: str) -> None:
        self._lock: Lock = Lock()
        self._working_dir: Path = Path(working_dir).resolve()
        self._db_path: Path = Path(db_path).resolve()
        self._crypto_key: str = crypto_key
        self._registry: dict[str, FileNode] = {}
        self._csv_headers: list[str] = ["id", "original_name", "system_name", "created_at", "updated_at"]
        self._file_base_name: str = file_base_name

        self._working_dir.mkdir(parents=True, exist_ok=True)
        self._db_path.parent.mkdir(parents=True, exist_ok=True)
        self._load_registry()

    def _generate_incremental_id(self, fill_gaps: bool = True) -> str:
        """Находит следующий свободный числовой ID в таблице файлов с заполнением дыр."""
        if not self._registry:
            return "1"

        existing_ids = sorted([int(fid) for idx, fid in enumerate(self._registry.keys()) if fid.isdigit()])
        if not existing_ids:
            return "1"

        if fill_gaps:
            if existing_ids[0] > 1:
                return "1"
            for i in range(len(existing_ids) - 1):
                if existing_ids[i + 1] - existing_ids[i] > 1:
                    return str(existing_ids[i] + 1)

        return str(existing_ids[-1] + 1)

    def _load_registry(self) -> None:
        if not self._db_path.exists():
            return

        try:
            with open(self._db_path, "rb") as f:
                encrypted_bytes = f.read()

            if not encrypted_bytes:
                return

            # Дешифруем двоичный CSV-поток через AES-256
            decrypted_bytes = decrypt_payload(encrypted_bytes, self._crypto_key)
            csv_text = decrypted_bytes.decode('utf-8')

            # Читаем CSV таблицу из текстового буфера в памяти
            csv_buffer = io.StringIO(csv_text)
            reader = csv.DictReader(csv_buffer)

            headers = next(reader, None)
            if headers is None:
                return

            for row in reader:
                if len(row) < len(self._csv_headers):
                    continue
                row_dict = dict(zip(self._csv_headers, row))
                node = FileNode.from_dict(row_dict)
                self._registry[node.id] = node

            logger.debug(f"CSV file registry successfully decrypted. Indices loaded: {len(self._registry)}")
        except Exception as e:
            logger.error(f"Critical read failure for the tabular file registry: {e}")

    def _save_registry(self) -> None:
        try:
            # Создаем текстовый буфер для сборки CSV таблицы в ОЗУ
            csv_buffer = io.StringIO()
            writer = csv.DictWriter(csv_buffer, fieldnames=self._csv_headers, lineterminator='\n')
            writer.writeheader()

            for node in self._registry.values():
                writer.writerow(node.to_dict())

            # Переводим таблицу в байты и шифруем
            raw_csv_bytes = csv_buffer.getvalue().encode('utf-8')
            encrypted_payload = encrypt_payload(raw_csv_bytes, self._crypto_key)

            # Атомарная транзакционная запись, защищающая реестр от повреждений
            def _write(temp_path):
                with open(temp_path, "wb") as f:
                    f.write(encrypted_payload)

            from EJIO.utils import save_atomic
            save_atomic(self._db_path, _write)
        except Exception as e:
            logger.error(f"File registry write error: {e}")



    def store_file(self, source_path: str | Path) -> str:
        """
        Копирует внешний файл в хранилище, присваивает ему уникальное имя по DateTime,
        фиксирует метаданные в псевдо-таблицу и возвращает строковый ID.
        """
        src = Path(source_path)
        if not src.exists() or not src.is_file():
            raise FileNotFoundError(f"File '{source_path}' does not exist.")

        with self._lock:
            file_id = self._generate_incremental_id(fill_gaps=True)

            # Генерация уникального системного имени файла по DateTime
            ext = src.suffix.lstrip('.') or "png"
            system_name = generate_timestamped_filename(
                base_name=self._file_base_name,
                extension=ext,
                prefix_date=False
            )
            dest_path = self._working_dir / system_name

            # Физическое копирование
            shutil.copy2(src, dest_path)

            # Регистрация в псевдо-таблице метаданных
            now_str = datetime.datetime.now().isoformat()
            node = FileNode(
                id=file_id,
                original_name=src.name,
                system_name=system_name,
                created_at=now_str
            )
            self._registry[file_id] = node
            self._save_registry()

            return file_id




    def delete_file(self, file_id: str) -> None:
        """Физически удаляет файл с диска и стирает его запись из таблицы."""
        with self._lock:
            if file_id in self._registry:
                node = self._registry.pop(file_id)
                file_path = self._working_dir / node.system_name
                if file_path.exists():
                    file_path.unlink(missing_ok=True)
                self._save_registry()




    def get_absolute_path(self, file_id: str) -> Path | None:
        """Возвращает полный физический путь к файлу в ОС по его ID."""
        with self._lock:
            if file_id in self._registry:
                return self._working_dir / self._registry[file_id].system_name
            return None