112 lines
4.2 KiB
Python
112 lines
4.2 KiB
Python
"""Shared file-tree access and serialization helpers."""
|
|
import uuid
|
|
from django.core.exceptions import ValidationError
|
|
from django.http import Http404
|
|
from rest_framework.response import Response
|
|
from .models import DriveItem, ItemGrant
|
|
from .spaces import can_read_space, can_write_space
|
|
from .permissions import has_space_right, item_rights
|
|
|
|
|
|
def error(message, code=400):
|
|
return Response({'detail':message}, status=code)
|
|
|
|
|
|
def item_info(item, user=None):
|
|
data = {
|
|
'id':str(item.id), 'name':item.name, 'is_folder':item.is_folder,
|
|
'size':item.size,'mime_type':item.mime_type,'owner_id':item.owner_id,
|
|
'owner':item.owner.username,'parent':str(item.parent_id) if item.parent_id else None,
|
|
'space_id':item.space_id, 'is_shared':item.user_shares.exists() or any(link.revoked_at is None for link in item.public_links.all()),
|
|
'trashed_at':item.trashed_at.isoformat() if item.trashed_at else None,
|
|
'created_at':item.created_at.isoformat(),'updated_at':item.updated_at.isoformat(),
|
|
}
|
|
if user is not None:
|
|
data['can_write'] = writable(item,user)
|
|
data['permissions'] = sorted(item_rights(user, item))
|
|
return data
|
|
|
|
|
|
def get_item(pk):
|
|
try:
|
|
return DriveItem.objects.select_related('owner','parent','space').get(pk=pk)
|
|
except (DriveItem.DoesNotExist,ValueError,ValidationError):
|
|
raise Http404
|
|
|
|
|
|
def chain(item):
|
|
ancestors=[]
|
|
visited=set()
|
|
while item is not None:
|
|
if item.pk in visited:
|
|
raise Http404
|
|
visited.add(item.pk)
|
|
ancestors.append(item)
|
|
item=item.parent
|
|
return ancestors
|
|
|
|
|
|
def readable(item, user, allow_trashed=False):
|
|
if 'view' not in item_rights(user, item):
|
|
return False
|
|
return allow_trashed or not any(n.trashed_at is not None for n in chain(item))
|
|
|
|
|
|
def writable(item, user, action='modify'):
|
|
return action in item_rights(user, item)
|
|
|
|
|
|
def valid_name(name):
|
|
if not isinstance(name,str) or not name.strip() or len(name)>255:
|
|
raise ValueError('Nom invalide (1 à 255 caractères).')
|
|
name=name.strip()
|
|
if name in ('.','..') or any(c in name for c in ('/','\\','\x00','\r','\n')):
|
|
raise ValueError('Ce nom contient des caractères interdits.')
|
|
return name
|
|
|
|
|
|
def target_parent(parent_id, user, space):
|
|
if not parent_id:
|
|
return None
|
|
try:
|
|
parent=DriveItem.objects.select_related('parent','space').get(pk=uuid.UUID(str(parent_id)))
|
|
except (ValueError,DriveItem.DoesNotExist,AttributeError):
|
|
raise ValueError('Dossier de destination introuvable.')
|
|
if parent.space_id != space.id:
|
|
raise PermissionError('Impossible de déplacer un élément entre deux espaces.')
|
|
if not parent.is_folder or not readable(parent,user) or not writable(parent,user,'add'):
|
|
raise PermissionError('Dossier de destination inaccessible en écriture.')
|
|
return parent
|
|
|
|
|
|
def descendant_files(node):
|
|
files=[]
|
|
pending=[node.pk]
|
|
while pending:
|
|
batch=pending[:500];pending=pending[500:]
|
|
children=list(DriveItem.objects.filter(parent_id__in=batch).values_list('pk','is_folder','storage_key','storage_location'))
|
|
pending.extend(row[0] for row in children)
|
|
for _,is_folder,key,location in children:
|
|
if not is_folder and key: files.append(location or f'{key[:2]}/{key}')
|
|
if not node.is_folder and node.storage_key:
|
|
key=node.storage_key
|
|
files.append(node.storage_location or f'{key[:2]}/{key}')
|
|
return files
|
|
|
|
|
|
def writable_destination(user, space_id, parent_id):
|
|
"""Uploads may target a folder shared with this user, without space membership."""
|
|
from .spaces import resolve_space
|
|
from .permissions import has_space_right
|
|
if parent_id:
|
|
folder=get_item(parent_id)
|
|
if space_id is not None and str(folder.space_id)!=str(space_id):
|
|
raise PermissionError('Le dossier ne correspond pas à cet espace.')
|
|
if not folder.is_folder or not readable(folder,user) or not writable(folder,user,'add'):
|
|
raise PermissionError('Dossier inaccessible en ajout.')
|
|
return folder.space, folder
|
|
space=resolve_space(user,space_id)
|
|
if not has_space_right(user,space,'add'):
|
|
raise PermissionError('Ajout interdit dans cet espace.')
|
|
return space,None
|