PYRSS-Website/apps/home/views.py

# -*- encoding: utf-8 -*-

import json
import logging

import httpx
from django.conf import settings
from django.utils import timezone
from asgiref.sync import sync_to_async
from django.shortcuts import redirect
from django.http import JsonResponse, HttpResponse
from django.views.generic import TemplateView, View
from django.core.exceptions import PermissionDenied

from apps.home.models import Server, DiscordChannel
from apps.authentication.models import DiscordUser, ServerMember

log = logging.getLogger(__name__)


class IndexView(TemplateView):
    """
    View for the Index page.
    """

    template_name = "home/index.html"


@sync_to_async
def get_user_access_token(user: DiscordUser) -> str:
    return user.access_token

@sync_to_async
def is_user_authenticated(user: DiscordUser) -> bool:
    return user.is_authenticated


class GuildsView(View):

    async def get(self, request, *args, **kwargs):
        if not await is_user_authenticated(request.user):
            return redirect("/oauth2/login")

        access_token = await get_user_access_token(request.user)

        async with httpx.AsyncClient() as client:
            response = await client.get(
                url=f"{settings.DISCORD_API_URL}/users/@me/guilds",
                headers={"Authorization": f"Bearer {access_token}"}
            )
            status = response.status_code
            guild_data = response.json()

        if status != 200:
            return JsonResponse(guild_data, status=status, safe=False)

        # guilds where the user either administrates or owns
        cleaned_guild_data = []

        for item in guild_data:
            cleaned_data = await self.setup_server(request.user, item)
            if cleaned_data:
                cleaned_guild_data.append(cleaned_data)

        return JsonResponse(cleaned_guild_data, safe=False)

    async def setup_server(self, user: DiscordUser, data: dict):
        is_owner = data["owner"]
        permissions = data["permissions"]
        admin_perm = 1 << 3

        # Ignore servers where the user isn't an administrator or owner
        if not ((int(permissions) & admin_perm) == admin_perm or is_owner):
            await self.delete_member(user, data["id"])
            return

        server = await Server.objects.aupdate_or_create(
            id=data["id"],
            defaults={
                "name": data["name"],
                "icon_hash": data["icon"]
            }
        )
    
        await ServerMember.objects.aupdate_or_create(
            user=user,
            server=server[0],
            defaults={
                "permissions": permissions,
                "is_owner": is_owner
            }
        )

        return data

    @staticmethod
    async def delete_member(user: DiscordUser, server_id: int):
        try:
            member = await ServerMember.objects.aget(user=user, server_id=server_id)
            await member.adelete()
        except ServerMember.DoesNotExist:
            pass


class ChannelsView(View):

    async def get(self, request, *args, **kwargs):
        if not await is_user_authenticated(request.user):
            return redirect("/oauth2/login")

        guild_id = request.GET.get("guild")
        server = await Server.objects.aget(pk=guild_id)

        if not ServerMember.objects.filter(server=server, user=request.user).aexists():
            raise PermissionDenied("You aren't a member of this server.")

        channels_data, status = await self._get_channel_data(guild_id)
        if status != 200:
            return JsonResponse(channels_data, status=status, safe=False)

        cleaned_channels_data = await self._clean_channels_data(server, channels_data)
        await self._cleanup_dead_channels(server, cleaned_channels_data)

        return JsonResponse(cleaned_channels_data, safe=False)

    async def _get_channel_data(self, guild_id: int) -> tuple[list[dict], int]:
        """
        Returns the raw channel data and a response status code
        from the Discord API.
        """

        async with httpx.AsyncClient() as client:
            response = await client.get(
                url=f"{settings.DISCORD_API_URL}/guilds/{guild_id}/channels",
                headers={"Authorization": f"Bot {settings.BOT_TOKEN}"}
            )
            return response.json(), response.status_code

    async def _clean_channels_data(self, server: Server, channels_data) -> list[dict]:
        """
        Returns a sorted & cleaned list of channel data, also performs a
        database setup for each channel to be stored.
        """

        cleaned_channels_data = []

        for item in channels_data:
            cleaned_data = await self._setup_channel(server, item)
            if cleaned_data:
                cleaned_channels_data.append(cleaned_data)

        cleaned_channels_data.sort(key=lambda ch: ch.get("position"))
        return cleaned_channels_data

    async def _setup_channel(self, server: Server, data: dict) -> dict:
        """
        Create or update an instance of DiscordChannel representing the given
        `data` dictionary, returns the data or `NoneType` if `data['type'] != 0`.
        """

        if data.get("type") != 0:
            return

        await DiscordChannel.objects.aupdate_or_create(
            id=data["id"],
            server=server,
            defaults={
                "name": data["name"],
                "is_nsfw": data["nsfw"]
            }
        )
        return data

    async def _cleanup_dead_channels(self, server: Server, channel_data: list[dict]):
        """
        Deletes any unused instances of `DiscordChannel` against the given server.
        """

        channel_ids = [item["id"] for item in channel_data]
        count, _ = await DiscordChannel.objects.filter(server=server).exclude(id__in=channel_ids).adelete()
        log.info("Deleted %s dead DiscordChannel object(s)", count)