aircox/aircox/models/program.py

import logging
import os
import shutil

from django.conf import settings as conf
from django.contrib.auth.models import Group, Permission
from django.contrib.contenttypes.models import ContentType
from django.db import models
from django.db.models import F
from django.db.models.functions import Concat, Substr
from django.utils.translation import gettext_lazy as _

from aircox.conf import settings

from .page import Page, PageQuerySet
from .station import Station

logger = logging.getLogger("aircox")


__all__ = (
    "Program",
    "ProgramChildQuerySet",
    "ProgramQuerySet",
    "Stream",
)


class ProgramQuerySet(PageQuerySet):
    def station(self, station):
        # FIXME: reverse-lookup
        return self.filter(station=station)

    def active(self):
        return self.filter(active=True)


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 = models.ForeignKey(Group, models.CASCADE, blank=True, null=True, verbose_name=_("editors"))

    objects = ProgramQuerySet.as_manager()
    detail_url_name = "program-detail"

    @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)

    @property
    def editors_group_name(self):
        return "{self.title} editors"

    @property
    def change_permission_codename(self):
        return f"change_program_{self.slug}"

    def __init__(self, *kargs, **kwargs):
        super().__init__(*kargs, **kwargs)
        if self.slug:
            self.__initial_path = self.path
            self.__initial_cover = self.cover

    @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)

    def set_group_ownership(self):
        editors, created = Group.objects.get_or_create(name=self.editors_group_name)
        if created:
            self.editors = editors
        permission, _ = Permission.objects.get_or_create(
            name=f"change program {self.title}",
            codename=self.change_permission_codename,
            content_type=ContentType.objects.get_for_model(self),
        )
        if permission not in editors.permissions.all():
            editors.permissions.add(permission)

    class Meta:
        verbose_name = _("Program")
        verbose_name_plural = _("Programs")

    def __str__(self):
        return self.title

    def save(self, *kargs, **kwargs):
        from .sound import Sound

        super().save(*kargs, **kwargs)

        # TODO: move in signals
        path_ = getattr(self, "__initial_path", None)
        abspath = path_ and os.path.join(conf.MEDIA_ROOT, path_)
        if path_ is not None and path_ != self.path and os.path.exists(abspath) and not os.path.exists(self.abspath):
            logger.info(
                "program #%s's dir changed to %s - update it.",
                self.id,
                self.title,
            )

            shutil.move(abspath, self.abspath)
            Sound.objects.filter(path__startswith=path_).update(file=Concat("file", Substr(F("file"), len(path_))))

        self.set_group_ownership()
        super().save(*kargs, **kwargs)


class ProgramChildQuerySet(PageQuerySet):
    def station(self, station=None, id=None):
        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)


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"),
    )