249 lines
8.8 KiB
Python
249 lines
8.8 KiB
Python
"""Folder API viewset."""
|
||
|
||
import io
|
||
import posixpath
|
||
import zipfile
|
||
|
||
from django.db import transaction
|
||
from django.http import FileResponse
|
||
from filer.models import Folder
|
||
from rest_framework import status
|
||
from rest_framework.decorators import action
|
||
from rest_framework.response import Response
|
||
from rest_framework.viewsets import ModelViewSet
|
||
|
||
from documents.application import (
|
||
GuestManagedFolderError,
|
||
ProtectedFolderDocumentsError,
|
||
collect_descendant_folder_ids,
|
||
delete_folder_tree,
|
||
)
|
||
from documents.models import Document
|
||
|
||
from ..permissions import FolderPermission, filter_documents_for_user
|
||
from ..serializers import DocumentReadSerializer, FolderSerializer
|
||
from .archive import (
|
||
document_archive_filename,
|
||
safe_archive_part,
|
||
unique_archive_path,
|
||
write_document_to_archive,
|
||
)
|
||
|
||
|
||
class FolderViewSet(ModelViewSet):
|
||
"""CRUD для папок документов (filer.Folder).
|
||
|
||
Поддерживает иерархическую структуру папок с parent/children связями.
|
||
"""
|
||
|
||
queryset = Folder.objects.all()
|
||
serializer_class = FolderSerializer
|
||
permission_classes = [FolderPermission]
|
||
|
||
def get_queryset(self):
|
||
"""Возвращает папки, опционально фильтрованные по parent_id.
|
||
|
||
Параметры:
|
||
?parent_id=<id> - показать только дочерние папки указанного parent
|
||
?root=true - показать только корневые папки (parent=null)
|
||
"""
|
||
from django.db.models import Count
|
||
|
||
qs = super().get_queryset()
|
||
request = getattr(self, "request", None)
|
||
|
||
# Аннотируем количество документов в каждой папке
|
||
qs = qs.annotate(document_count=Count("documents"))
|
||
|
||
if not request:
|
||
return qs
|
||
|
||
parent_id = request.query_params.get("parent_id")
|
||
root = request.query_params.get("root", "").lower() == "true"
|
||
|
||
if root:
|
||
qs = qs.filter(parent__isnull=True)
|
||
elif parent_id:
|
||
qs = qs.filter(parent_id=parent_id)
|
||
|
||
return qs.order_by("name")
|
||
|
||
@transaction.atomic
|
||
def create(self, request, *args, **kwargs):
|
||
"""Создать папку.
|
||
|
||
Body:
|
||
{
|
||
"name": "Название папки",
|
||
"parent": <id> | null // опционально
|
||
}
|
||
"""
|
||
serializer = self.get_serializer(data=request.data)
|
||
serializer.is_valid(raise_exception=True)
|
||
|
||
# Создаем папку
|
||
name = serializer.validated_data["name"]
|
||
parent_id = request.data.get("parent")
|
||
|
||
parent = None
|
||
if parent_id:
|
||
try:
|
||
parent = Folder.objects.get(pk=parent_id)
|
||
except Folder.DoesNotExist:
|
||
return Response(
|
||
{"error": "Родительская папка не найдена"},
|
||
status=status.HTTP_404_NOT_FOUND,
|
||
)
|
||
|
||
folder = Folder.objects.create(name=name, parent=parent, owner=request.user)
|
||
|
||
result = FolderSerializer(folder).data
|
||
return Response(result, status=status.HTTP_201_CREATED)
|
||
|
||
def _collect_descendant_folder_ids(self, folder: Folder) -> set[int]:
|
||
"""Возвращает ID папки и всех её подпапок."""
|
||
return collect_descendant_folder_ids(folder.id)
|
||
|
||
def update(self, request, *args, **kwargs):
|
||
folder = self.get_object()
|
||
from guests.models import Guest
|
||
|
||
if Guest.objects.filter(document_folder_id=folder.id).exists():
|
||
return Response(
|
||
{
|
||
"detail": (
|
||
"Папка привязана к гостю. Изменяйте ее через карточку гостя."
|
||
)
|
||
},
|
||
status=status.HTTP_400_BAD_REQUEST,
|
||
)
|
||
return super().update(request, *args, **kwargs)
|
||
|
||
def partial_update(self, request, *args, **kwargs):
|
||
kwargs["partial"] = True
|
||
return self.update(request, *args, **kwargs)
|
||
|
||
def _folder_archive_paths(self, folder_ids: set[int]) -> dict[int, str]:
|
||
"""Возвращает ZIP-пути для папок внутри удаляемого дерева."""
|
||
folders = {
|
||
folder.id: folder
|
||
for folder in Folder.objects.filter(id__in=folder_ids).select_related(
|
||
"parent"
|
||
)
|
||
}
|
||
archive_paths: dict[int, str] = {}
|
||
|
||
def build_path(folder: Folder) -> str:
|
||
if folder.id in archive_paths:
|
||
return archive_paths[folder.id]
|
||
|
||
folder_name = safe_archive_part(folder.name, f"folder-{folder.id}")
|
||
if folder.parent_id in folders:
|
||
archive_path = posixpath.join(
|
||
build_path(folders[folder.parent_id]), folder_name
|
||
)
|
||
else:
|
||
archive_path = folder_name
|
||
|
||
archive_paths[folder.id] = archive_path
|
||
return archive_path
|
||
|
||
for folder in folders.values():
|
||
build_path(folder)
|
||
|
||
return archive_paths
|
||
|
||
@action(detail=True, methods=["get"], url_path="archive")
|
||
def archive(self, request, pk=None):
|
||
"""Выгрузить папку со всеми подпапками и документами в ZIP."""
|
||
folder = self.get_object()
|
||
folder_ids = self._collect_descendant_folder_ids(folder)
|
||
folder_paths = self._folder_archive_paths(folder_ids)
|
||
archive_buffer = io.BytesIO()
|
||
used_file_paths: set[str] = set()
|
||
|
||
documents = filter_documents_for_user(
|
||
Document.objects.filter(folder_id__in=folder_ids).select_related(
|
||
"file", "folder"
|
||
),
|
||
request.user,
|
||
).order_by("folder_id", "title", "id")
|
||
|
||
with zipfile.ZipFile(
|
||
archive_buffer, mode="w", compression=zipfile.ZIP_DEFLATED
|
||
) as zip_file:
|
||
for folder_id, archive_path in sorted(
|
||
folder_paths.items(), key=lambda item: item[1]
|
||
):
|
||
zip_file.writestr(f"{archive_path}/", b"")
|
||
|
||
for document in documents:
|
||
if not document.file:
|
||
continue
|
||
|
||
folder_path = folder_paths.get(document.folder_id, "")
|
||
archive_path = unique_archive_path(
|
||
posixpath.join(
|
||
folder_path,
|
||
document_archive_filename(document),
|
||
),
|
||
used_file_paths,
|
||
)
|
||
write_document_to_archive(
|
||
zip_file,
|
||
document=document,
|
||
archive_path=archive_path,
|
||
)
|
||
|
||
archive_buffer.seek(0)
|
||
archive_name = f"{safe_archive_part(folder.name, 'folder')}.zip"
|
||
return FileResponse(
|
||
archive_buffer,
|
||
as_attachment=True,
|
||
filename=archive_name,
|
||
content_type="application/zip",
|
||
)
|
||
|
||
def destroy(self, request, *args, **kwargs):
|
||
"""Удаляет папку вместе с подпапками и связанными документами."""
|
||
folder = self.get_object()
|
||
try:
|
||
delete_folder_tree(folder_id=folder.id, actor=request.user)
|
||
except GuestManagedFolderError:
|
||
return Response(
|
||
{
|
||
"detail": (
|
||
"Папка привязана к гостю. Изменяйте ее через карточку гостя."
|
||
)
|
||
},
|
||
status=status.HTTP_400_BAD_REQUEST,
|
||
)
|
||
except ProtectedFolderDocumentsError:
|
||
return Response(
|
||
{"detail": ("Папка содержит документы, которые вы не можете удалить.")},
|
||
status=status.HTTP_403_FORBIDDEN,
|
||
)
|
||
|
||
return Response(status=status.HTTP_204_NO_CONTENT)
|
||
|
||
@action(detail=True, methods=["get"])
|
||
def children(self, request, pk=None):
|
||
"""Получить дочерние папки данной папки."""
|
||
folder = self.get_object()
|
||
children = folder.children.all().order_by("name")
|
||
serializer = self.get_serializer(children, many=True)
|
||
return Response(serializer.data)
|
||
|
||
@action(detail=True, methods=["get"])
|
||
def documents(self, request, pk=None):
|
||
"""Получить документы в данной папке."""
|
||
folder = self.get_object()
|
||
documents = filter_documents_for_user(
|
||
Document.objects.filter(folder=folder),
|
||
request.user,
|
||
)
|
||
serializer = DocumentReadSerializer(
|
||
documents, many=True, context={"request": request}
|
||
)
|
||
return Response(serializer.data)
|