# -*- encoding: utf-8 -*- import logging from django.db.models import Subquery from django.db.utils import IntegrityError from django_filters import rest_framework as rest_filters from rest_framework import status, permissions, filters, generics from rest_framework.response import Response from rest_framework.pagination import PageNumberPagination from rest_framework.authentication import SessionAuthentication, TokenAuthentication from rest_framework.parsers import MultiPartParser, FormParser from apps.home.models import SubChannel, Filter, Subscription, SavedGuilds, TrackedContent, ArticleMutator, GuildSettings from apps.authentication.models import DiscordUser from .metadata import ExpandedMetadata from .serializers import ( SubChannelSerializer, FilterSerializer, SubscriptionSerializer_GET, SubscriptionSerializer_POST, SavedGuildSerializer, TrackedContentSerializer_GET, TrackedContentSerializer_POST, ArticleMutatorSerializer, GuildSettingsSerializer ) log = logging.getLogger(__name__) class DefaultPagination(PageNumberPagination): """Default class for pagination in API views.""" page_size = 10 page_size_query_param = "page_size" max_page_size = 25 def is_automated_admin(user): return user.user_type == DiscordUser.USER_TYPES.AUTOMATED_USER and user.is_superuser # ================================================================================================= # SubChannel Views class SubChannel_ListView(generics.ListCreateAPIView): """ View to provide a list of SubChannel model instances. Can also be used to create a new instance. Supports: GET, POST """ authentication_classes = [SessionAuthentication, TokenAuthentication] permission_classes = [permissions.IsAuthenticated] pagination_class = DefaultPagination serializer_class = SubChannelSerializer queryset = SubChannel.objects.all().order_by("id") filter_backends = [filters.SearchFilter, rest_filters.DjangoFilterBackend, filters.OrderingFilter] filterset_fields = ["id", "channel_id", "channel_name", "subscription"] search_fields = ["channel_name"] def get_queryset(self): if self.request.user.is_superuser: return SubChannel.objects.all() saved_guilds = SavedGuilds.objects.filter(added_by=self.request.user) guild_ids = [guild.guild_id for guild in saved_guilds] return SubChannel.objects.filter(subscription__guild_id__in=guild_ids) def post(self, request): serializer = self.get_serializer(data=request.data) serializer.is_valid(raise_exception=True) try: self.perform_create(serializer) except IntegrityError as err: return Response( {"detail": str(err)}, status=status.HTTP_500_INTERNAL_SERVER_ERROR, exception=True ) headers = self.get_success_headers(serializer.data) return Response(serializer.data, status=status.HTTP_201_CREATED, headers=headers) class SubChannel_DetailView(generics.RetrieveUpdateDestroyAPIView): """ View to provide details on a particular SubChannel model instances. Supports: GET, PUT, PATCH, DELETE """ authentication_classes = [SessionAuthentication, TokenAuthentication] permission_classes = [permissions.IsAuthenticated] parser_classes = [MultiPartParser, FormParser] serializer_class = SubChannelSerializer queryset = SubChannel.objects.all().order_by("id") def get_queryset(self): if self.request.user.is_superuser: return SubChannel.objects.all() saved_guilds = SavedGuilds.objects.filter(added_by=self.request.user) guild_ids = [guild.guild_id for guild in saved_guilds] return SubChannel.objects.filter(subscription__guild_id__in=guild_ids) # ================================================================================================= # Filter Views class Filter_ListView(generics.ListCreateAPIView): """ View to provide a list of Filter model instances. Can also be used to create a new instance. Supports: GET, POST """ authentication_classes = [SessionAuthentication, TokenAuthentication] permission_classes = [permissions.IsAuthenticated] pagination_class = DefaultPagination serializer_class = FilterSerializer metadata_class = ExpandedMetadata filter_backends = [filters.SearchFilter, rest_filters.DjangoFilterBackend, filters.OrderingFilter] filterset_fields = ["id", "name", "matching_algorithm", "match", "is_insensitive", "is_whitelist", "guild_id"] search_fields = ["name", "match"] def get_queryset(self): if self.request.user.is_superuser: return Filter.objects.all().order_by("id") saved_guild_ids = SavedGuilds.objects \ .filter(added_by=self.request.user.id) \ .values("guild_id") return Filter.objects \ .filter(guild_id__in=Subquery(saved_guild_ids)) \ .order_by("id") def post(self, request): serializer = self.get_serializer(data=request.data) serializer.is_valid(raise_exception=True) try: self.perform_create(serializer) except IntegrityError as err: return Response( {"detail": str(err)}, status=status.HTTP_500_INTERNAL_SERVER_ERROR, exception=True ) headers = self.get_success_headers(serializer.data) return Response(serializer.data, status=status.HTTP_201_CREATED, headers=headers) class Filter_DetailView(generics.RetrieveUpdateDestroyAPIView): """ View to provide details on a particular Filter model instances. Supports: GET, PUT, PATCH, DELETE """ authentication_classes = [SessionAuthentication, TokenAuthentication] permission_classes = [permissions.IsAuthenticated] parser_classes = [MultiPartParser, FormParser] serializer_class = FilterSerializer def get_queryset(self): if self.request.user.is_superuser: return Filter.objects.all().order_by("id") saved_guild_ids = SavedGuilds.objects \ .filter(added_by=self.request.user.id) \ .values("guild_id") return Filter.objects \ .filter(guild_id__in=Subquery(saved_guild_ids)) \ .order_by("id") # ================================================================================================= # Subscription Views class Subscription_ListView(generics.ListCreateAPIView): """ View to provide a list of Subscription model instances. Can also be used to create a new instance. Supports: GET, POST """ authentication_classes = [SessionAuthentication, TokenAuthentication] permission_classes = [permissions.IsAuthenticated] pagination_class = DefaultPagination metadata_class = ExpandedMetadata filter_backends = [filters.SearchFilter, rest_filters.DjangoFilterBackend, filters.OrderingFilter] filterset_fields = [ "id", "name", "url", "guild_id", "creation_datetime", "extra_notes", "filters", "article_title_mutators", "article_desc_mutators", "embed_colour", "published_threshold", "active" ] search_fields = ["name", "url", "extra_notes"] ordering_fields = ["name", "creation_datetime", "active"] read_serializer_class = SubscriptionSerializer_GET write_serializer_class = SubscriptionSerializer_POST def get_serializer_class(self): if self.request.method == "POST": return self.write_serializer_class return self.read_serializer_class def get_queryset(self): if self.request.user.is_superuser: return Subscription.objects.all().order_by("-creation_datetime") saved_guild_ids = SavedGuilds.objects \ .filter(added_by=self.request.user.id) \ .values("guild_id") return Subscription.objects \ .filter(guild_id__in=Subquery(saved_guild_ids)) \ .order_by("-creation_datetime") def post(self, request): serializer = self.get_serializer(data=request.data) serializer.is_valid(raise_exception=True) try: self.perform_create(serializer) except IntegrityError as err: return Response( {"detail": str(err)}, status=status.HTTP_500_INTERNAL_SERVER_ERROR, exception=True ) headers = self.get_success_headers(serializer.data) return Response(serializer.data, status=status.HTTP_201_CREATED, headers=headers) class Subscription_DetailView(generics.RetrieveUpdateDestroyAPIView): """ View to provide details on a particular Subscription model instances. Supports: GET, PUT, PATCH, DELETE """ authentication_classes = [SessionAuthentication, TokenAuthentication] permission_classes = [permissions.IsAuthenticated] parser_classes = [MultiPartParser, FormParser] serializer_class = SubscriptionSerializer_POST def get_queryset(self): if self.request.user.is_superuser: return Subscription.objects.all().order_by("-creation_datetime") saved_guild_ids = SavedGuilds.objects \ .filter(added_by=self.request.user.id) \ .values("guild_id") return Subscription.objects \ .filter(guild_id__in=Subquery(saved_guild_ids)) \ .order_by("-creation_datetime") class Subscription_SubChannelView(generics.DestroyAPIView): """ View to erase all subscription channels quickly. """ authentication_classes = [SessionAuthentication, TokenAuthentication] permission_classes = [permissions.IsAuthenticated] parser_classes = [MultiPartParser, FormParser] def delete(self, *args, **kwargs): if self.request.user.is_superuser: subscriptions = Subscription.objects.all() else: saved_guild_ids = SavedGuilds.objects \ .filter(added_by=self.request.user.id) \ .values("guild_id") subscriptions = Subscription.objects \ .filter(guild_id__in=Subquery(saved_guild_ids)) if not subscriptions: return Response( {"detail": "You are forbidden from viewing this subscription"}, status=status.HTTP_403_FORBIDDEN ) subscription = subscriptions.filter(id=kwargs["pk"]).first() if not subscription: return Response( {"detail": "not found"}, status=status.HTTP_404_NOT_FOUND ) channels = SubChannel.objects.filter(subscription=subscription) channels.delete(); return Response(status=status.HTTP_204_NO_CONTENT) # ================================================================================================= # SavedGuild Views class SavedGuild_ListView(generics.ListCreateAPIView): """ View to provide a list of SavedGuild model instances. Can also be used to create a new instance. Supports: GET, POST """ authentication_classes = [SessionAuthentication, TokenAuthentication] permission_classes = [permissions.IsAuthenticated] pagination_class = None serializer_class = SavedGuildSerializer metadata_class = ExpandedMetadata filter_backends = [filters.SearchFilter, rest_filters.DjangoFilterBackend, filters.OrderingFilter] filterset_fields = ["id", "guild_id", "name", "icon", "added_by", "permissions", "owner"] search_fields = ["name"] def get_queryset(self): if self.request.user.is_superuser: return SavedGuilds.objects.all() return SavedGuilds.objects.filter(added_by=self.request.user) def post(self, request): # TODO: # the data used for admin/owner verification is provided # from the client, this is a potential attack vector, and # should be rewritten. is_owner = request.data["owner"].lower() == "true" # Check user is admin in server if not (self.is_server_admin(request.data["permissions"]) or is_owner): return Response( {"detail": "You must be a server administrator"}, status=status.HTTP_403_FORBIDDEN, exception=False ) serializer = self.get_serializer(data=request.data) serializer.is_valid(raise_exception=True) try: self.perform_create(serializer) except IntegrityError as err: return Response( {"detail": str(err)}, status=status.HTTP_500_INTERNAL_SERVER_ERROR, exception=True ) headers = self.get_success_headers(serializer.data) return Response(serializer.data, status=status.HTTP_201_CREATED, headers=headers) def is_server_admin(self, permissions) -> bool: return (int(permissions) & 1 << 3) == 1 << 3 class SavedGuild_DetailView(generics.RetrieveDestroyAPIView): """ View to provide details on a particular SavedGuild model instances. Supports: GET, DELETE """ authentication_classes = [SessionAuthentication, TokenAuthentication] permission_classes = [permissions.IsAuthenticated] parser_classes = [MultiPartParser, FormParser] serializer_class = SavedGuildSerializer def get_queryset(self): if self.request.user.is_superuser: return SavedGuilds.objects.all() return SavedGuilds.objects.filter(added_by=self.request.user) # ================================================================================================= # GuildSettings Views class GuildSettings_ListView(generics.ListCreateAPIView): """ View to provide a list of GuildSettings model instances. Can also be used to create a new instance. Supports: GET, POST """ authentication_classes = [SessionAuthentication, TokenAuthentication] permission_classes = [permissions.IsAuthenticated] pagination_class = DefaultPagination serializer_class = GuildSettingsSerializer metadata_class = ExpandedMetadata filter_backends = [filters.SearchFilter, rest_filters.DjangoFilterBackend, filters.OrderingFilter] filterset_fields = ["id", "guild_id", "default_embed_colour", "active"] def get_queryset(self): if self.request.user.is_superuser: return GuildSettings.objects.all() saved_guilds = SavedGuilds.objects.filter(added_by=self.request.user) guild_ids = [guild.guild_id for guild in saved_guilds] return GuildSettings.objects.filter(guild_id__in=guild_ids) def post(self, request): saved_guilds = SavedGuilds.objects.filter(added_by=request.user) if not saved_guilds: return Response( {"detail": "You must have an instance of saved guild with this guild_id"}, status=status.HTTP_403_FORBIDDEN, exception=False ) serializer = self.get_serializer(data=request.data) serializer.is_valid(raise_exception=True) try: self.perform_create(serializer) except IntegrityError as err: return Response( {"detail": str(err)}, status=status.HTTP_500_INTERNAL_SERVER_ERROR, exception=True ) headers = self.get_success_headers(serializer.data) return Response(serializer.data, status=status.HTTP_201_CREATED, headers=headers) def is_server_admin(self, permissions) -> bool: return (int(permissions) & 1 << 3) == 1 << 3 class GuildSettings_DetailView(generics.RetrieveUpdateDestroyAPIView): """ View to provide details on a particular GuildSettings model instances. Supports: GET, DELETE """ authentication_classes = [SessionAuthentication, TokenAuthentication] permission_classes = [permissions.IsAuthenticated] parser_classes = [MultiPartParser, FormParser] serializer_class = GuildSettingsSerializer def get_queryset(self): if self.request.user.is_superuser: return GuildSettings.objects.all() saved_guilds = SavedGuilds.objects.filter(added_by=self.request.user) guild_ids = [guild.guild_id for guild in saved_guilds] return GuildSettings.objects.filter(guild_id__in=guild_ids) # ================================================================================================= # TrackedContent Views class TrackedContent_ListView(generics.ListCreateAPIView): """ View to provide a list of TrackedContent model instances. Can also be used to create a new instance. Supports: GET, POST """ authentication_classes = [SessionAuthentication, TokenAuthentication] permission_classes = [permissions.IsAuthenticated] pagination_class = DefaultPagination metadata_class = ExpandedMetadata queryset = TrackedContent.objects.all().order_by("-creation_datetime") filter_backends = [filters.SearchFilter, rest_filters.DjangoFilterBackend, filters.OrderingFilter] filterset_fields = ["guid", "title", "url", "subscription", "subscription__guild_id", "channel_id", "blocked", "creation_datetime"] search_fields = ["guid", "title", "url", "subscription__name", "subscription__url"] ordering_fields = ["title", "subscription", "creation_datetime", "blocked"] read_serializer_class = TrackedContentSerializer_GET write_serializer_class = TrackedContentSerializer_POST def get_serializer_class(self): if self.request.method == "POST": return self.write_serializer_class return self.read_serializer_class def get_queryset(self): if self.request.user.is_superuser: return TrackedContent.objects.all() saved_guilds = SavedGuilds.objects.filter(added_by=self.request.user) guild_ids = [guild.guild_id for guild in saved_guilds] return TrackedContent.objects.filter(subscription__guild_id__in=guild_ids) def post(self, request): serializer = self.get_serializer(data=request.data) serializer.is_valid(raise_exception=True) try: self.perform_create(serializer) except IntegrityError as err: return Response( {"detail": str(err)}, status=status.HTTP_500_INTERNAL_SERVER_ERROR, exception=True ) headers = self.get_success_headers(serializer.data) return Response(serializer.data, status=status.HTTP_201_CREATED, headers=headers) class TrackedContent_DetailView(generics.RetrieveUpdateDestroyAPIView): """ View to provide details on a particular TrackedContent model instances. Supports: GET, PUT, PATCH, DELETE """ authentication_classes = [SessionAuthentication, TokenAuthentication] permission_classes = [permissions.IsAuthenticated] parser_classes = [MultiPartParser, FormParser] serializer_class = TrackedContentSerializer_POST queryset = TrackedContent.objects.all().order_by("-creation_datetime") def get_queryset(self): if self.request.user.is_superuser: return TrackedContent.objects.all() saved_guilds = SavedGuilds.objects.filter(added_by=self.request.user) guild_ids = [guild.guild_id for guild in saved_guilds] return TrackedContent.objects.filter(subscription__guild_id__in=guild_ids) class ArticleMutator_ListView(generics.ListCreateAPIView): """ View to provide a list of ArticleMutator model instances. Can also be used to create a new instance. Supports: GET, POST """ authentication_classes = [SessionAuthentication, TokenAuthentication] permission_classes = [permissions.IsAuthenticated] metadata_class = ExpandedMetadata queryset = ArticleMutator.objects.all().order_by("id") filter_backends = [filters.SearchFilter, rest_filters.DjangoFilterBackend, filters.OrderingFilter] filterset_fields = ["id", "name", "value"] serializer_class = ArticleMutatorSerializer def post(self, request): serializer = self.get_serializer(data=request.data) serializer.is_valid(raise_exception=True) try: self.perform_create(serializer) except IntegrityError as err: return Response( {"detail": str(err)}, status=status.HTTP_500_INTERNAL_SERVER_ERROR, exception=True ) headers = self.get_success_headers(serializer.data) return Response(serializer.data, status=status.HTTP_201_CREATED, headers=headers) class ArticleMutator_DetailView(generics.RetrieveUpdateDestroyAPIView): """ View to provide details on a particular ArticleMutator model instances. Supports: GET, PUT, PATCH, DELETE """ authentication_classes = [SessionAuthentication, TokenAuthentication] permission_classes = [permissions.IsAuthenticated] parser_classes = [MultiPartParser, FormParser] serializer_class = ArticleMutatorSerializer queryset = ArticleMutator.objects.all().order_by("id")