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"