295 lines
11 KiB
Python
295 lines
11 KiB
Python
from django.conf import settings
|
|
from django.contrib.auth import get_user_model
|
|
from rest_framework import status, viewsets
|
|
from rest_framework.decorators import action
|
|
from rest_framework.exceptions import PermissionDenied
|
|
from rest_framework.response import Response
|
|
from drf_spectacular.utils import extend_schema, extend_schema_view
|
|
|
|
from .models import Hub, HubPermission, Tags
|
|
from .permissions import CanEditHub, CanManageHubTags, IsHubOwnerOrSuperuser
|
|
from .serializers import HubPermissionSerializer, HubSerializer, TagsSerializer, TransferInitSerializer, TransferVerifySerializer
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Hub ViewSet
|
|
# ---------------------------------------------------------------------------
|
|
|
|
@extend_schema_view(
|
|
list=extend_schema(
|
|
tags=["hubs"],
|
|
summary="List hubs",
|
|
description=(
|
|
"Returns all public hubs. Authenticated users also see hubs they are members of. "
|
|
"Admins see all hubs regardless of visibility."
|
|
),
|
|
),
|
|
retrieve=extend_schema(
|
|
tags=["hubs"],
|
|
summary="Retrieve a hub",
|
|
),
|
|
create=extend_schema(
|
|
tags=["hubs"],
|
|
summary="Create a hub",
|
|
description="Creates a new hub. The requesting user is automatically set as the owner.",
|
|
),
|
|
update=extend_schema(
|
|
tags=["hubs"],
|
|
summary="Replace a hub",
|
|
description=(
|
|
"Full update. Restricted to the owner, site admin, or moderators "
|
|
"(per-field permission flags are enforced)."
|
|
),
|
|
),
|
|
partial_update=extend_schema(
|
|
tags=["hubs"],
|
|
summary="Update a hub",
|
|
description=(
|
|
"Partial update. Restricted to the owner, site admin, or moderators "
|
|
"(per-field permission flags are enforced)."
|
|
),
|
|
),
|
|
destroy=extend_schema(
|
|
tags=["hubs"],
|
|
summary="Delete a hub",
|
|
description="Soft-deletes the hub. Owner or admin only.",
|
|
),
|
|
)
|
|
class HubViewSet(viewsets.ModelViewSet):
|
|
serializer_class = HubSerializer
|
|
permission_classes = [CanEditHub]
|
|
filterset_fields = ['is_public', 'owner']
|
|
search_fields = ['name', 'description']
|
|
ordering_fields = ['name']
|
|
ordering = ['name']
|
|
|
|
def get_queryset(self):
|
|
user = self.request.user
|
|
if user.is_superuser:
|
|
return Hub.objects.all()
|
|
return (
|
|
Hub.objects.filter(is_public=True) | Hub.objects.filter(members=user)
|
|
).distinct()
|
|
|
|
def perform_create(self, serializer):
|
|
serializer.save(owner=self.request.user)
|
|
|
|
# ------------------------------------------------------------------
|
|
# Membership actions
|
|
# ------------------------------------------------------------------
|
|
|
|
@extend_schema(
|
|
tags=["hubs"],
|
|
summary="Join a hub",
|
|
description="Adds the authenticated user as a member. Private hubs reject this request.",
|
|
request=None,
|
|
responses={200: HubSerializer},
|
|
)
|
|
@action(detail=True, methods=['post'])
|
|
def join(self, request, pk=None):
|
|
hub = self.get_object()
|
|
if not hub.is_public and not (hub.owner == request.user or request.user.is_superuser):
|
|
return Response(
|
|
{'detail': 'This hub is private.'},
|
|
status=status.HTTP_403_FORBIDDEN,
|
|
)
|
|
hub.members.add(request.user)
|
|
return Response(HubSerializer(hub, context={'request': request}).data)
|
|
|
|
@extend_schema(
|
|
tags=["hubs"],
|
|
summary="Leave a hub",
|
|
description="Removes the authenticated user from the hub's members.",
|
|
request=None,
|
|
responses={204: None},
|
|
)
|
|
@action(detail=True, methods=['post'])
|
|
def leave(self, request, pk=None):
|
|
hub = self.get_object()
|
|
hub.members.remove(request.user)
|
|
return Response(status=status.HTTP_204_NO_CONTENT)
|
|
|
|
# ------------------------------------------------------------------
|
|
# Ownership transfer actions
|
|
# ------------------------------------------------------------------
|
|
|
|
@extend_schema(
|
|
tags=["hubs"],
|
|
summary="Initiate ownership transfer",
|
|
description=(
|
|
"Generates a transfer token and records the intended new owner. "
|
|
"Only the current hub owner can initiate. The recipient must call "
|
|
"`transfer/verify` with the token to complete the transfer."
|
|
),
|
|
request=TransferInitSerializer,
|
|
responses={200: TransferInitSerializer},
|
|
)
|
|
@action(detail=True, methods=['post'], url_path='transfer/initiate')
|
|
def initiate_transfer(self, request, pk=None):
|
|
hub = self.get_object()
|
|
if hub.owner != request.user:
|
|
return Response(
|
|
{'detail': 'Only the hub owner can initiate a transfer.'},
|
|
status=status.HTTP_403_FORBIDDEN,
|
|
)
|
|
ser = TransferInitSerializer(data=request.data)
|
|
ser.is_valid(raise_exception=True)
|
|
|
|
User = get_user_model()
|
|
try:
|
|
new_owner = User.objects.get(pk=ser.validated_data['user_id'])
|
|
except User.DoesNotExist:
|
|
return Response({'detail': 'User not found.'}, status=status.HTTP_404_NOT_FOUND)
|
|
|
|
hub.create_transfer(new_owner)
|
|
transfer_url = f"{settings.FRONTEND_URL}/hubs/{hub.pk}/transfer/verify?token={hub.transfer_token}"
|
|
return Response({'detail': f'Transfer initiated to {new_owner}.', 'transfer_url': transfer_url})
|
|
|
|
@extend_schema(
|
|
tags=["hubs"],
|
|
summary="Verify ownership transfer",
|
|
description=(
|
|
"Completes the transfer when the intended new owner supplies the correct token. "
|
|
"Must be called by the transfer recipient."
|
|
),
|
|
request=TransferVerifySerializer,
|
|
responses={200: HubSerializer},
|
|
)
|
|
@action(detail=True, methods=['post'], url_path='transfer/verify')
|
|
def verify_transfer(self, request, pk=None):
|
|
hub = self.get_object()
|
|
ser = TransferVerifySerializer(data=request.data)
|
|
ser.is_valid(raise_exception=True)
|
|
try:
|
|
hub.verify_transfer(ser.validated_data['token'], request.user)
|
|
except Exception as exc:
|
|
return Response({'detail': str(exc)}, status=status.HTTP_400_BAD_REQUEST)
|
|
return Response(HubSerializer(hub, context={'request': request}).data)
|
|
|
|
@extend_schema(
|
|
tags=["hubs"],
|
|
summary="Cancel ownership transfer",
|
|
description="Cancels a pending transfer, clearing the token and recipient. Owner or admin only.",
|
|
request=None,
|
|
responses={200: None},
|
|
)
|
|
@action(detail=True, methods=['post'], url_path='transfer/cancel')
|
|
def cancel_transfer(self, request, pk=None):
|
|
hub = self.get_object()
|
|
if not (hub.owner == request.user or request.user.is_superuser):
|
|
raise PermissionDenied('Only the hub owner or superuser can cancel a transfer.')
|
|
hub.cancel_transfer()
|
|
return Response({'detail': 'Transfer cancelled.'})
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Hub Moderator (HubPermission) ViewSet
|
|
# ---------------------------------------------------------------------------
|
|
|
|
@extend_schema_view(
|
|
list=extend_schema(
|
|
tags=["hubs"],
|
|
summary="List hub moderators",
|
|
description="Returns all moderators and their permission flags for a given hub. Pass `hub` as a query param.",
|
|
),
|
|
retrieve=extend_schema(
|
|
tags=["hubs"],
|
|
summary="Retrieve a hub moderator",
|
|
),
|
|
create=extend_schema(
|
|
tags=["hubs"],
|
|
summary="Add a hub moderator",
|
|
description="Grants a user moderator permissions on the hub. Owner or admin only.",
|
|
),
|
|
partial_update=extend_schema(
|
|
tags=["hubs"],
|
|
summary="Update moderator permissions",
|
|
description="Updates one or more permission flags for a moderator. Owner or admin only.",
|
|
),
|
|
update=extend_schema(
|
|
tags=["hubs"],
|
|
summary="Replace moderator permissions",
|
|
description="Replaces all permission flags for a moderator. Owner or admin only.",
|
|
),
|
|
destroy=extend_schema(
|
|
tags=["hubs"],
|
|
summary="Remove a hub moderator",
|
|
description="Revokes all moderator permissions from a user. Owner or admin only.",
|
|
),
|
|
)
|
|
class HubPermissionViewSet(viewsets.ModelViewSet):
|
|
serializer_class = HubPermissionSerializer
|
|
permission_classes = [IsHubOwnerOrSuperuser]
|
|
filterset_fields = ['user', 'changing_name', 'changing_description', 'changing_icon', 'changing_banner', 'managing_members', 'managing_posts', 'managing_chats']
|
|
|
|
def _get_hub(self):
|
|
hub_id = self.kwargs.get('hub_pk') or self.request.query_params.get('hub')
|
|
return Hub.objects.get(pk=hub_id)
|
|
|
|
def get_queryset(self):
|
|
return HubPermission.objects.filter(hub=self._get_hub())
|
|
|
|
def perform_create(self, serializer):
|
|
hub = self._get_hub()
|
|
if not (hub.owner == self.request.user or self.request.user.is_superuser):
|
|
raise PermissionDenied('Only the hub owner or superuser can add moderators.')
|
|
serializer.save(hub=hub)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Tags ViewSet
|
|
# ---------------------------------------------------------------------------
|
|
|
|
@extend_schema_view(
|
|
list=extend_schema(
|
|
tags=["hubs"],
|
|
summary="List hub tags",
|
|
description="Returns all tags for a given hub. Pass `hub` as a query param.",
|
|
),
|
|
retrieve=extend_schema(
|
|
tags=["hubs"],
|
|
summary="Retrieve a hub tag",
|
|
),
|
|
create=extend_schema(
|
|
tags=["hubs"],
|
|
summary="Create a hub tag",
|
|
description="Adds a tag to the hub. Owner, site admin, or moderator with `managing_posts`.",
|
|
),
|
|
partial_update=extend_schema(
|
|
tags=["hubs"],
|
|
summary="Update a hub tag",
|
|
description="Updates a tag on the hub. Owner, site admin, or moderator with `managing_posts`.",
|
|
),
|
|
update=extend_schema(
|
|
tags=["hubs"],
|
|
summary="Replace a hub tag",
|
|
description="Replaces a tag on the hub. Owner, site admin, or moderator with `managing_posts`.",
|
|
),
|
|
destroy=extend_schema(
|
|
tags=["hubs"],
|
|
summary="Delete a hub tag",
|
|
description="Removes a tag from the hub. Owner, site admin, or moderator with `managing_posts`.",
|
|
),
|
|
)
|
|
class TagsViewSet(viewsets.ModelViewSet):
|
|
serializer_class = TagsSerializer
|
|
permission_classes = [CanManageHubTags]
|
|
search_fields = ['name', 'description']
|
|
ordering_fields = ['name']
|
|
ordering = ['name']
|
|
|
|
def _get_hub(self):
|
|
hub_id = self.kwargs.get('hub_pk') or self.request.query_params.get('hub')
|
|
return Hub.objects.get(pk=hub_id)
|
|
|
|
def get_queryset(self):
|
|
return Tags.objects.filter(hub=self._get_hub())
|
|
|
|
def perform_create(self, serializer):
|
|
hub = self._get_hub()
|
|
user = self.request.user
|
|
if not (user.is_superuser or hub.owner == user or hub.moderators.filter(user=user, managing_posts=True).exists()):
|
|
raise PermissionDenied('Only the hub owner, superuser, or moderator with managing_posts can create tags.')
|
|
serializer.save(hub=hub)
|
|
|