Corban-Lee Jones
b6d3efa014
Some checks are pending
Build and Push Docker Image / build (push) Waiting to run
305 lines
10 KiB
Python
305 lines
10 KiB
Python
# -*- encoding: utf-8 -*-
|
|
|
|
import logging
|
|
|
|
from django.db.models import Q
|
|
from django_filters import rest_framework as rest_filters
|
|
from rest_framework import permissions, filters, generics, status
|
|
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 rest_framework.filters import BaseFilterBackend
|
|
|
|
from apps.home.models import (
|
|
Server,
|
|
ContentFilter,
|
|
MessageMutator,
|
|
MessageStyle,
|
|
Subscription,
|
|
Content
|
|
)
|
|
from apps.authentication.models import DiscordUser, ServerMember
|
|
from .metadata import ExpandedMetadata
|
|
from .serializers import (
|
|
ServerSerializer,
|
|
ContentFilterSerializer,
|
|
MessageMutatorSerializer,
|
|
MessageStyleSerializer,
|
|
SubscriptionSerializer,
|
|
ContentSerializer
|
|
)
|
|
from .permissions import HasServerAccess
|
|
from .errors import NotAMemberError
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
|
|
class DefaultPagination(PageNumberPagination):
|
|
"""Default class for pagination in API views."""
|
|
|
|
page_size = 10
|
|
page_query_param = "page"
|
|
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
|
|
|
|
|
|
class ListView(generics.ListAPIView):
|
|
authentication_classes = [SessionAuthentication, TokenAuthentication]
|
|
permission_classes = [permissions.IsAuthenticated]
|
|
pagination_class = DefaultPagination
|
|
metadata_class = ExpandedMetadata
|
|
filter_backends = [filters.SearchFilter, rest_filters.DjangoFilterBackend, filters.OrderingFilter]
|
|
|
|
|
|
class ListCreateView(generics.ListCreateAPIView):
|
|
authentication_classes = [SessionAuthentication, TokenAuthentication]
|
|
permission_classes = [permissions.IsAuthenticated]
|
|
pagination_class = DefaultPagination
|
|
metadata_class = ExpandedMetadata
|
|
filter_backends = [filters.SearchFilter, rest_filters.DjangoFilterBackend, filters.OrderingFilter]
|
|
|
|
|
|
class DetailView(generics.RetrieveAPIView):
|
|
authentication_classes = [SessionAuthentication, TokenAuthentication]
|
|
permission_classes = [permissions.IsAuthenticated]
|
|
parser_classes = [MultiPartParser, FormParser]
|
|
|
|
|
|
class ChangableDetailView(generics.RetrieveUpdateDestroyAPIView):
|
|
authentication_classes = [SessionAuthentication, TokenAuthentication]
|
|
permission_classes = [permissions.IsAuthenticated]
|
|
parser_classes = [MultiPartParser, FormParser]
|
|
|
|
|
|
class DeletableDetailView(generics.RetrieveDestroyAPIView):
|
|
authentication_classes = [SessionAuthentication, TokenAuthentication]
|
|
permission_classes = [permissions.IsAuthenticated]
|
|
parser_classes = [MultiPartParser, FormParser]
|
|
|
|
|
|
# region Servers
|
|
|
|
class Server_ListView(ListView):
|
|
filterset_fields = ("id", "name", "icon_hash", "active")
|
|
search_fields = ("name")
|
|
ordering_fields = ("id", "name", "active")
|
|
serializer_class = ServerSerializer
|
|
|
|
def get_queryset(self):
|
|
if self.request.user.is_superuser:
|
|
return Server.objects.all().order_by("id")
|
|
|
|
servers = ServerMember.objects.filter(user=self.request.user).values_list("server", flat=True)
|
|
return Server.objects.filter(id__in=servers).order_by("id")
|
|
|
|
|
|
class Server_DetailView(DeletableDetailView):
|
|
serializer_class = ServerSerializer
|
|
|
|
def get_queryset(self):
|
|
if self.request.user.is_superuser:
|
|
return Server.objects.all()
|
|
|
|
servers = ServerMember.objects.filter(user=self.request.user).values_list("server", flat=True)
|
|
return Server.objects.filter(id__in=servers)
|
|
|
|
def destroy(self, request, *args, **kwargs):
|
|
server = self.get_object()
|
|
|
|
if not self.request.user.is_superuser:
|
|
member = ServerMember.objects.get(server=server, user=request.user)
|
|
if not member.is_owner:
|
|
return Response(
|
|
{"detail": "Only the owner can destroy server data."},
|
|
status=status.HTTP_403_FORBIDDEN
|
|
)
|
|
|
|
return super().destroy(request, *args, **kwargs)
|
|
|
|
|
|
# region Filters
|
|
|
|
class ContentFilter_ListView(ListCreateView):
|
|
filterset_fields = ("id", "server", "name", "match", "matching_algorithm", "is_insensitive", "is_whitelist")
|
|
search_fields = ("name", "match")
|
|
ordering_fields = ("id", "server", "name", "match", "matching_algorithm", "is_insensitive", "is_whitelist")
|
|
serializer_class = ContentFilterSerializer
|
|
|
|
def get_queryset(self):
|
|
if self.request.user.is_superuser:
|
|
return ContentFilter.objects.all().order_by("name")
|
|
|
|
servers = ServerMember.objects.filter(user=self.request.user).values_list("server", flat=True)
|
|
return ContentFilter.objects.filter(server__in=servers).order_by("name")
|
|
|
|
|
|
class ContentFilter_DetailView(ChangableDetailView):
|
|
serializer_class = ContentFilterSerializer
|
|
|
|
def get_queryset(self):
|
|
if self.request.user.is_superuser:
|
|
return ContentFilter.objects.all()
|
|
|
|
servers = ServerMember.objects.filter(user=self.request.user).values_list("server", flat=True)
|
|
return ContentFilter.objects.filter(server__in=servers)
|
|
|
|
|
|
# region Mutators
|
|
|
|
class MessageMutator_ListView(ListView): # instances of this one are pre-defined ONLY
|
|
filterset_fields = ("id", "name", "value")
|
|
search_fields = ("name", "value")
|
|
ordering_fields = ("id", "name", "value")
|
|
serializer_class = MessageMutatorSerializer
|
|
|
|
def get_queryset(self):
|
|
return MessageMutator.objects.all()
|
|
|
|
|
|
class MessageMutator_DetailView(DetailView):
|
|
serializer_class = MessageMutatorSerializer
|
|
|
|
def get_queryset(self):
|
|
return MessageMutator.objects.all()
|
|
|
|
|
|
# Message Styles
|
|
|
|
class MessageStyle_ListView(ListCreateView):
|
|
filterset_fields = ("id", "server", "name", "is_embed", "is_hyperlinked", "show_author", "show_timestamp", "show_images", "fetch_images", "title_mutator", "description_mutator")
|
|
search_fields = ("name",)
|
|
ordering_fields = ("id", "server", "name", "is_embed", "is_hyperlinked", "show_author", "show_timestamp", "show_images", "fetch_images")
|
|
serializer_class = MessageStyleSerializer
|
|
|
|
def get_queryset(self):
|
|
if self.request.user.is_superuser:
|
|
return MessageStyle.objects.all().order_by("name")
|
|
|
|
servers = ServerMember.objects.filter(user=self.request.user).values_list("server", flat=True)
|
|
return MessageStyle.objects.filter(server__in=servers).order_by("name")
|
|
|
|
|
|
class MessageStyle_DetailView(ChangableDetailView):
|
|
serializer_class = MessageStyleSerializer
|
|
|
|
def get_queryset(self):
|
|
if self.request.user.is_superuser:
|
|
return MessageStyle.objects.all()
|
|
|
|
servers = ServerMember.objects.filter(user=self.request.user).values_list("server", flat=True)
|
|
return MessageStyle.objects.filter(server__in=servers)
|
|
|
|
|
|
# region Subscriptions
|
|
|
|
class Subscription_ListView(ListCreateView):
|
|
filterset_fields = ("id", "server", "name", "url", "created_at", "updated_at", "extra_notes", "active", "publish_threshold", "filters", "message_style")
|
|
search_fields = ("name", "url", "extra_notes")
|
|
ordering_fields = ("id", "server", "name", "url", "created_at", "updated_at", "extra_notes", "active", "message_style")
|
|
serializer_class = SubscriptionSerializer
|
|
|
|
def get_queryset(self):
|
|
if self.request.user.is_superuser:
|
|
return Subscription.objects.all().order_by("name")
|
|
|
|
servers = ServerMember.objects.filter(user=self.request.user).values_list("server", flat=True)
|
|
return Subscription.objects.filter(server__in=servers).order_by("name")
|
|
|
|
|
|
class Subscription_DetailView(ChangableDetailView):
|
|
serializer_class = SubscriptionSerializer
|
|
|
|
def get_queryset(self):
|
|
if self.request.user.is_superuser:
|
|
return Subscription.objects.all()
|
|
|
|
servers = ServerMember.objects.filter(user=self.request.user).values_list("server", flat=True)
|
|
return Subscription.objects.filter(server__in=servers)
|
|
|
|
|
|
# region Content
|
|
|
|
|
|
class ContentFilterBackend(BaseFilterBackend):
|
|
"""
|
|
If `match_any=true` in querystring, matches 'any' params instead
|
|
of 'all' params.
|
|
"""
|
|
|
|
_MATCH_ANY_PARAM = "match_any"
|
|
_IGNORE_PARAMS = [
|
|
DefaultPagination.page_query_param,
|
|
DefaultPagination.page_size_query_param,
|
|
"search",
|
|
"subscription"
|
|
]
|
|
|
|
def filter_queryset(self, request, queryset, view):
|
|
filters = Q()
|
|
|
|
for param, value in request.query_params.items():
|
|
if param in self._IGNORE_PARAMS or param == self._MATCH_ANY_PARAM:
|
|
continue
|
|
|
|
query = Q(**{param: value})
|
|
|
|
if param not in self._IGNORE_PARAMS:
|
|
filters |= query
|
|
else:
|
|
filters &= query
|
|
|
|
return queryset.filter(filters)
|
|
|
|
|
|
class Content_ListView(ListCreateView):
|
|
search_fields = (
|
|
"item_id",
|
|
"item_guid",
|
|
"item_url",
|
|
"item_title",
|
|
"item_description",
|
|
"item_content_hash",
|
|
"item_image_url",
|
|
"item_thumbnail_url",
|
|
"item_published",
|
|
"item_author",
|
|
"item_author_url",
|
|
"item_feed_title",
|
|
"item_feed_url"
|
|
)
|
|
filterset_fields = search_fields + ("id", "subscription", "subscription__server")
|
|
ordering_fields = filterset_fields
|
|
serializer_class = ContentSerializer
|
|
filter_backends = [
|
|
ContentFilterBackend,
|
|
filters.SearchFilter,
|
|
rest_filters.DjangoFilterBackend,
|
|
filters.OrderingFilter
|
|
]
|
|
# filter_backends = (FlexibleFilterBackend,)
|
|
|
|
def get_queryset(self):
|
|
if self.request.user.is_superuser:
|
|
return Content.objects.all().order_by("-subscription__created_at", "id")
|
|
|
|
servers = ServerMember.objects.filter(user=self.request.user).values_list("server", flat=True)
|
|
subscriptions = Subscription.objects.filter(server__in=servers).values_list("id", flat=True)
|
|
return Content.objects.filter(subscription__in=subscriptions).order_by("-subscription__created_at", "id")
|
|
|
|
|
|
class Content_DetailView(ChangableDetailView):
|
|
serializer_class = ContentSerializer
|
|
|
|
def get_queryset(self):
|
|
if self.request.user.is_superuser:
|
|
return Content.objects.all()
|
|
|
|
servers = ServerMember.objects.filter(user=self.request.user).values_list("server", flat=True)
|
|
subscriptions = Subscription.objects.filter(server__in=servers).values_list("id", flat=True)
|
|
return Content.objects.filter(subscription__in=subscriptions)
|