MarietjeDjango/marietje/queues/models.py

from django.contrib.auth import get_user_model
from django.db import models
from django.db.models import Q
from django.conf import settings
from django.utils import timezone

from queues.exceptions import RequestException
from songs.models import Song

User = get_user_model()


class Playlist(models.Model):
    def __str__(self):
        return "Playlist #" + str(self.id)


class PlaylistSong(models.Model):
    playlist = models.ForeignKey(
        Playlist,
        on_delete=models.SET_NULL,
        blank=True,
        null=True,
        related_name="songs",
    )
    song = models.ForeignKey(
        Song,
        on_delete=models.SET_NULL,
        blank=True,
        null=True,
    )
    user = models.ForeignKey(
        settings.AUTH_USER_MODEL,
        on_delete=models.SET_NULL,
        blank=True,
        null=True,
        db_index=True,
    )
    played_at = models.DateTimeField(blank=True, null=True)

    # 0: Queued.
    # 1: Playing.
    # 2: Played.
    # 3: Cancelled.
    STATECHOICE = (
        (0, "Queued"),
        (1, "Playing"),
        (2, "Played"),
        (3, "Cancelled"),
    )
    state = models.IntegerField(default=0, db_index=True, choices=STATECHOICE)

    def move_down(self):
        other_song = PlaylistSong.objects.filter(playlist=self.playlist, id__gt=self.id).first()
        old_id = self.id
        self.id = other_song.id
        other_song.id = old_id
        self.save()
        other_song.save()

    def __str__(self):
        return "Playlist #" + str(self.playlist_id) + ": " + str(self.song)


class Queue(models.Model):
    class Meta:
        permissions = (
            ("can_skip", "Can skip the currently playing song"),
            ("can_move", "Can move all songs in the queue"),
            ("can_cancel", "Can cancel all songs in the queue"),
            ("can_control_volume", "Can control the volume of Marietje"),
            ("unlimited_queue_length", "Is unlimited by maximum queue length"),
        )

    name = models.TextField()
    playlist = models.ForeignKey(
        Playlist,
        on_delete=models.SET_NULL,
        blank=True,
        null=True,
    )
    random_playlist = models.ForeignKey(
        Playlist, on_delete=models.SET_NULL, blank=True, null=True, related_name="random_playlist_set"
    )
    started_at = models.DateTimeField(blank=True, null=True)
    player_token = models.TextField(blank=True, null=True)

    def get_songs(self):
        self.fill_random_queue()
        return (
            PlaylistSong.objects.filter(
                Q(playlist=self.playlist_id) | Q(playlist_id=self.random_playlist_id), Q(state=0) | Q(state=1)
            )
            .order_by("-state", "playlist_id", "id")
            .select_related("song", "user")
        )

    def current_song(self):
        songs = self.get_songs()
        if not songs:
            return None
        return songs[0]

    def queue(self):
        songs = self.get_songs()
        if len(songs) < 2:
            return []
        return songs[1:]

    def request(self, song, user):
        if user is not None and not user.has_perm("queues.unlimited_queue_length"):
            playlist_songs = PlaylistSong.objects.filter(playlist=self.playlist, state=0).order_by("id")

            seconds_in_a_row = sum(ps.song.duration for ps in playlist_songs if ps.user == user)
            msg = "You cannot request more than " + str(settings.MAX_MINUTES_IN_A_ROW) + " minutes in a row."
            if settings.LIMIT_ALWAYS:
                if seconds_in_a_row > settings.MAX_MINUTES_IN_A_ROW * 60:
                    raise RequestException(msg)
            else:
                now = timezone.now()
                if (
                    seconds_in_a_row > 0
                    and seconds_in_a_row + song.duration > settings.MAX_MINUTES_IN_A_ROW * 60
                    and settings.LIMIT_HOURS[0] <= now.hour < settings.LIMIT_HOURS[1]
                ):
                    raise RequestException(msg)

            if {ps for ps in playlist_songs if ps.song == song}:
                raise RequestException("This song is already in the queue.")

        playlist_song = PlaylistSong.objects.create(playlist=self.playlist, song=song, user=user)

        # If the song was auto-queue'd, then remove it from the auto-queue
        autolist_songs = PlaylistSong.objects.filter(playlist=self.random_playlist, state=0, song=song)
        autolist_songs.delete()

        return playlist_song

    def fill_random_queue(self):
        song_count = PlaylistSong.objects.filter(playlist_id=self.random_playlist_id, state=0).count()
        while song_count < 5:
            song = Song.objects.filter(deleted=False).order_by("?").first()
            if song is None:
                return
            playlist_song = PlaylistSong(playlist=self.random_playlist, song=song, user=None)
            playlist_song.save()
            song_count += 1

    def log_action(self, user: User, action: str, description: str) -> "QueueLogEntry":
        """
        Log a queue action.

        :param user: The user performing the action.
        :param action: An identifier of the action performed.
        :param description: An optional description for the action.
        :return: The created QueueLogEntry object.
        """
        return QueueLogEntry.objects.create(
            queue=self,
            user=user,
            action=action,
            description=description,
        )

    def __str__(self):
        return str(self.name)


class UserQueue(models.Model):
    """
    UserQueue model.

    This model connects a user to its queue.
    """

    user = models.OneToOneField(User, on_delete=models.CASCADE, related_name="queue")
    queue = models.ForeignKey(Queue, on_delete=models.SET_NULL, null=True, blank=True, related_name="users")

    def __str__(self):
        """Convert this object to string."""
        return "Queue for user {}".format(self.user)


class QueueCommand(models.Model):
    queue = models.ForeignKey(
        Queue,
        on_delete=models.CASCADE,
        db_index=True,
    )
    command = models.TextField()

    def __str__(self):
        return str(self.command)


class QueueLogEntry(models.Model):
    """Model for logging queue events."""

    queue = models.ForeignKey(Queue, on_delete=models.CASCADE, related_name="logs")
    user = models.ForeignKey(User, on_delete=models.CASCADE, null=True, blank=True)
    action = models.CharField(max_length=255)
    timestamp = models.DateTimeField(auto_now_add=True)
    description = models.CharField(max_length=255)

    def __str__(self):
        return f"{self.queue} {self.action} by {self.user} at {self.timestamp}"

    class Meta:
        verbose_name = "player log entry"
        verbose_name_plural = "player log entries"