import os from django.conf import settings as conf from django.contrib.auth.models import Group from django.db import models from django.utils.translation import gettext_lazy as _ from aircox.conf import settings from .page import Page, PageQuerySet from .station import Station __all__ = ( "ProgramQuerySet", "Program", "ProgramChildQuerySet", "Stream", ) class ProgramQuerySet(PageQuerySet): def station(self, station): # FIXME: reverse-lookup return self.filter(station=station) def active(self): return self.filter(active=True) def editor(self, user): """Return programs for which user is an editor. Superuser is considered as editor of all groups. """ if user.is_superuser: return self groups = user.groups.all() return self.filter(editors_group__in=groups) class Program(Page): """A Program can either be a Streamed or a Scheduled program. A Streamed program is used to generate non-stop random playlists when there is not scheduled diffusion. In such a case, a Stream is used to describe diffusion informations. A Scheduled program has a schedule and is the one with a normal use case. Renaming a Program rename the corresponding directory to matches the new name if it does not exists. """ # explicit foreign key in order to avoid related name clashes station = models.ForeignKey(Station, models.CASCADE, verbose_name=_("station")) active = models.BooleanField( _("active"), default=True, help_text=_("if not checked this program is no longer active"), ) sync = models.BooleanField( _("syncronise"), default=True, help_text=_("update later diffusions according to schedule changes"), ) editors_group = models.ForeignKey(Group, models.CASCADE, verbose_name=_("editors")) objects = ProgramQuerySet.as_manager() detail_url_name = "program-detail" list_url_name = "program-list" edit_url_name = "program-edit" @property def path(self): """Return program's directory path.""" return os.path.join(settings.PROGRAMS_DIR, self.slug.replace("-", "_")) @property def abspath(self): """Return absolute path to program's dir.""" return os.path.join(conf.MEDIA_ROOT, self.path) @property def archives_path(self): return os.path.join(self.path, settings.SOUND_ARCHIVES_SUBDIR) @property def excerpts_path(self): return os.path.join(self.path, settings.SOUND_ARCHIVES_SUBDIR) def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) if self.slug: self.__initial_path = self.path @classmethod def get_from_path(cl, path): """Return a Program from the given path. We assume the path has been given in a previous time by this model (Program.path getter). """ if path.startswith(settings.PROGRAMS_DIR_ABS): path = path.replace(settings.PROGRAMS_DIR_ABS, "") while path[0] == "/": path = path[1:] path = path[: path.index("/")] return cl.objects.filter(slug=path.replace("_", "-")).first() def ensure_dir(self, subdir=None): """Make sur the program's dir exists (and optionally subdir). Return True if the dir (or subdir) exists. """ path = os.path.join(self.abspath, subdir) if subdir else self.abspath os.makedirs(path, exist_ok=True) return os.path.exists(path) class Meta: verbose_name = _("Program") verbose_name_plural = _("Programs") def __str__(self): return self.title def save(self, *args, **kwargs): if not self.editors_group_id: from aircox import permissions saved = permissions.program.init(self) if saved: return super().save() class ProgramChildQuerySet(PageQuerySet): def station(self, station=None, id=None): # lookup `__program` is due to parent being a page subclass (page is # concrete). return ( self.filter(parent__program__station=station) if id is None else self.filter(parent__program__station__id=id) ) def program(self, program=None, id=None): return self.parent(program, id) def editor(self, user): programs = Program.objects.editor(user) return self.filter(parent__program__in=programs) class Stream(models.Model): """When there are no program scheduled, it is possible to play sounds in order to avoid blanks. A Stream is a Program that plays this role, and whose linked to a Stream. All sounds that are marked as good and that are under the related program's archive dir are elligible for the sound's selection. """ program = models.ForeignKey( Program, models.CASCADE, verbose_name=_("related program"), ) delay = models.TimeField( _("delay"), blank=True, null=True, help_text=_("minimal delay between two sound plays"), ) begin = models.TimeField( _("begin"), blank=True, null=True, help_text=_("used to define a time range this stream is " "played"), ) end = models.TimeField( _("end"), blank=True, null=True, help_text=_("used to define a time range this stream is " "played"), )