"""Space-bound navigation, folder management and trash endpoints.""" from pathlib import Path from django.conf import settings from django.core.exceptions import ValidationError from django.db import transaction from django.db.models import Sum, Count from django.utils import timezone from rest_framework.decorators import api_view from rest_framework.response import Response from .models import DriveItem, DriveSpace, ItemGrant from .spaces import ensure_space, ensure_private_space, is_admin, resolve_space, can_write_space, can_read_space from .permissions import has_space_right, item_rights from .quotas import quota_state from .services import error, item_info, get_item, chain, readable, writable, valid_name, target_parent, writable_destination, descendant_files from .storage_layout import disk_path, file_location, assert_available, prepare_parent, relocate_item, NameConflict @api_view(['GET']) def me(request): space=ensure_space(request.user) private=ensure_private_space(request.user) return Response({'id':request.user.pk,'username':request.user.username, 'is_admin':is_admin(request.user),'first_name':request.user.first_name, 'last_name':request.user.last_name,'private_space_id':private.pk, 'space_id':space.pk if space else None,'can_invite':bool(space and space.can_invite) or is_admin(request.user)}) @api_view(['GET']) def listing(request): user=request.user view=request.query_params.get('view','files') if view not in ('files','trash','shared'): return error('Vue inconnue.',400) q=request.query_params.get('q','').strip() parent_id=request.query_params.get('parent') parent=None # Trash is global across *readable* spaces, regardless of the selected space. # In particular another person's private space is never included, even for admin. if view=='trash': accessible=[space.pk for space in DriveSpace.objects.all() if can_read_space(user,space)] nodes=DriveItem.objects.filter(space_id__in=accessible,trashed_at__isnull=False).select_related('owner','parent','space').prefetch_related('user_shares','public_links') if q:nodes=nodes.filter(name__icontains=q) space=None elif view=='shared': space=None if parent_id: parent=get_item(parent_id) if not parent.is_folder or not readable(parent,user): return error('Dossier partagé inaccessible.',403) # Without inherited item permission or normal space access, browsing is forbidden. nodes=DriveItem.objects.filter(parent=parent,trashed_at__isnull=True).select_related('owner','parent','space').prefetch_related('user_shares','public_links') else: nodes=DriveItem.objects.filter(user_shares__user=user,trashed_at__isnull=True).select_related('owner','parent','space').prefetch_related('user_shares','public_links').distinct() if q:nodes=nodes.filter(name__icontains=q) else: try:space=resolve_space(user,request.query_params.get('space')) except PermissionError as ex:return error(str(ex),403) if parent_id: parent=get_item(parent_id) if not parent.is_folder or parent.space_id!=space.pk or not readable(parent,user): return error('Dossier inaccessible.',403) nodes=DriveItem.objects.filter(space=space,trashed_at__isnull=True).select_related('owner','parent','space').prefetch_related('user_shares','public_links') if q: nodes=nodes.filter(name__icontains=q) elif parent: nodes=nodes.filter(parent=parent) else: nodes=nodes.filter(parent__isnull=True) items=[] for n in nodes[:501]: if readable(n,user,allow_trashed=view=='trash'): info=item_info(n,user) if view in ('trash','shared'):info['space_name']=n.space.display_name if not n.space.private_owner_id else 'Mes fichiers' items.append(info) if len(items)>=500:break trail=[] if parent: ancestors=list(reversed(chain(parent))) if view=='shared' and not can_read_space(user,parent.space): shared_ancestors=set(ItemGrant.objects.filter(user=user,item_id__in=[a.pk for a in ancestors]).values_list('item_id',flat=True)) for index,a in enumerate(ancestors): if a.pk in shared_ancestors: ancestors=ancestors[index:];break for n in ancestors: trail.append({'id':str(n.id),'name':n.name}) return Response({'items':items,'breadcrumb':trail,'limited':len(items)>=500, 'can_write':('add' in item_rights(user,parent)) if view=='shared' and parent else (has_space_right(user,space,'add') if space else False), 'space_id':parent.space_id if view=='shared' and parent else (space.pk if space else None)}) @api_view(['GET']) def stats(request): try:space=resolve_space(request.user,request.query_params.get('space')) except PermissionError as ex:return error(str(ex),403) result=DriveItem.objects.filter(space=space,trashed_at__isnull=True,is_folder=False).aggregate(count=Count('id'),total=Sum('size')) return Response({'files':result['count'] or 0,'bytes':result['total'] or 0, **quota_state(space), 'folders':DriveItem.objects.filter(space=space,is_folder=True,trashed_at__isnull=True).count(), 'shared':0}) @api_view(['POST']) def create_folder(request): try: name=valid_name(request.data.get('name')) space,parent=writable_destination(request.user,request.data.get('space'),request.data.get('parent')) except ValueError as ex:return error(str(ex)) except PermissionError as ex:return error(str(ex),403) with transaction.atomic(): try: destination=assert_available(space,parent,name) prepare_parent(space,parent) destination.mkdir(mode=0o700) except (NameConflict,FileExistsError) as exc: return error(str(exc),409) except OSError as exc: return error(f'Création impossible : {exc}') folder=DriveItem(owner=request.user,space=space,parent=parent,name=name,is_folder=True) try: folder.full_clean();folder.save() except Exception: destination.rmdir() raise return Response(item_info(folder,request.user),status=201) @api_view(['PATCH']) def update_item(request,item_id): item=get_item(item_id) if not readable(item,request.user) or not writable(item,request.user,'modify'): return error('Modification interdite.',403) data=request.data try: new_name=valid_name(data['name']) if 'name' in data else item.name new_parent=(target_parent(data['parent'],request.user,item.space) if 'parent' in data else item.parent) except ValueError as exc:return error(str(exc)) except PermissionError as exc:return error(str(exc),403) if 'is_shared' in data: return error('Le partage global est désactivé : utilise les espaces et invitations.') if new_parent and (new_parent.pk==item.pk or item.pk in [n.pk for n in chain(new_parent)]): return error('Déplacement impossible dans un sous-dossier de cet élément.') if new_name==item.name and (new_parent.pk if new_parent else None)==item.parent_id: return Response(item_info(item,request.user)) try: relocate_item(item,new_parent,new_name) except (NameConflict,FileExistsError) as exc: return error(str(exc),409) except (OSError,ValueError,ValidationError) as exc: return error(f'Déplacement impossible : {exc}',409) return Response(item_info(item,request.user)) @api_view(['POST']) def trash_item(request,item_id): item=get_item(item_id) if not readable(item,request.user) or not writable(item,request.user,'delete'):return error('Accès refusé.',403) if item.trashed_at:return error('Élément déjà dans la corbeille.') item.trashed_at=timezone.now();item.save(update_fields=['trashed_at','updated_at']) return Response({'ok':True}) @api_view(['POST']) def restore_item(request,item_id): item=get_item(item_id) if not writable(item,request.user,'delete'):return error('Accès refusé.',403) if not item.trashed_at:return error('Cet élément n’est pas dans la corbeille.') if item.parent and any(n.trashed_at for n in chain(item.parent)): try: relocate_item(item,None,item.name) except (NameConflict,FileExistsError,OSError,ValueError,ValidationError) as exc: return error(f'Restauration impossible : {exc}',409) item.trashed_at=None;item.save() return Response(item_info(item,request.user)) @api_view(['DELETE']) def delete_item(request,item_id): item=get_item(item_id) if not writable(item,request.user,'delete'):return error('Accès refusé.',403) if not item.trashed_at:return error('Déplace d’abord cet élément dans la corbeille.') files=descendant_files(item) folder_rel=file_location(item.space,item.parent,item.name) if item.is_folder else None with transaction.atomic(): item.delete() def remove_files(): root=Path(settings.DRIVE_STORAGE_ROOT).resolve() for key in files: path=(root/key).resolve() if path.is_relative_to(root):path.unlink(missing_ok=True) if folder_rel: # Remove only empty folders. Untracked data is never deleted. base=disk_path(folder_rel) if base.is_dir(): for candidate in sorted(base.rglob('*'),key=lambda p:len(p.parts),reverse=True): if candidate.is_dir() and not candidate.is_symlink(): try:candidate.rmdir() except OSError:pass try:base.rmdir() except OSError:pass transaction.on_commit(remove_files) return Response(status=204)