models.py

#
# steering, Programme/schedule management for AURA
#
# Copyright (C) 2011-2017, 2020, Ernesto Rico Schmidt
# Copyright (C) 2017-2019, Ingo Leindecker
#
# This program is free software: you can redistribute it and/or modify it under
# the terms of the GNU Affero General Public License as published by the Free
# Software Foundation, either version 3 of the License, or (at your option) any
# later version.
#
# This program is distributed in the hope that it will be useful, but WITHOUT
# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
# FOR A PARTICULAR PURPOSE.  See the GNU Affero General Public License for more
# details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program.  If not, see <http://www.gnu.org/licenses/>.
#
import dataclasses
import datetime

import jsonschema
from rest_framework.exceptions import ValidationError
from versatileimagefield.fields import PPOIField, VersatileImageField

from django.contrib.auth.models import User
from django.core.exceptions import ValidationError as DjangoValidationError
from django.core.validators import RegexValidator
from django.db import models
from django.db.models import Max, Q
from django.utils import timezone
from django.utils.translation import gettext_lazy as _


class ScheduleConflictError(ValidationError):
    def __init__(self, *args, conflicts=None, **kwargs):
        super().__init__(*args, **kwargs)
        self.conflicts = conflicts


class Type(models.Model):
    is_active = models.BooleanField(default=True, help_text="True if type is active.")
    name = models.CharField(max_length=32, help_text="Name of the type.")
    slug = models.SlugField(max_length=32, unique=True, help_text="Slug of the type.")

    class Meta:
        ordering = ("name",)

    def __str__(self):
        return self.name


class Category(models.Model):
    description = models.TextField(blank=True, help_text="Description of the category.")
    is_active = models.BooleanField(default=True, help_text="True if category is active.")
    name = models.CharField(max_length=32, help_text="Name of the category.")
    slug = models.SlugField(max_length=32, unique=True, help_text="Slug of the category.")
    subtitle = models.CharField(blank=True, max_length=32, help_text="Subtitle of the category.")

    class Meta:
        ordering = ("name",)
        verbose_name_plural = "Categories"

    def __str__(self):
        return self.name


class Topic(models.Model):
    is_active = models.BooleanField(default=True, help_text="True if topic is active.")
    name = models.CharField(max_length=32, help_text="Name of the topic.")
    slug = models.SlugField(max_length=32, unique=True, help_text="Slug of the topic.")

    class Meta:
        ordering = ("name",)

    def __str__(self):
        return self.name


class MusicFocus(models.Model):
    is_active = models.BooleanField(default=True, help_text="True if music focus is active.")
    name = models.CharField(max_length=32, help_text="Name of the music focus.")
    slug = models.SlugField(max_length=32, unique=True, help_text="Slug of the music focus.")

    class Meta:
        ordering = ("name",)
        verbose_name_plural = "Music Focus"

    def __str__(self):
        return self.name


class FundingCategory(models.Model):
    is_active = models.BooleanField(default=True, help_text="True if funding category is active.")
    name = models.CharField(max_length=32, help_text="Name of the funding category.")
    slug = models.SlugField(max_length=32, unique=True, help_text="Slug of the funding category.")

    class Meta:
        ordering = ("name",)
        verbose_name_plural = "Funding Categories"

    def __str__(self):
        return self.name


class Language(models.Model):
    is_active = models.BooleanField(default=True, help_text="True if language is active.")
    name = models.CharField(max_length=32, help_text="Name of the language.")

    class Meta:
        ordering = ("name",)

    def __str__(self):
        return self.name


class License(models.Model):
    identifier = models.CharField(max_length=32, help_text="Identifier of the license")
    name = models.CharField(max_length=64, help_text="Name of the license")
    needs_author = models.BooleanField(default=True, help_text="True if license needs an author.")
    requires_express_permission_for_publication = models.BooleanField(
        default=True, help_text="True if express permission for publication required."
    )
    url = models.URLField(default="", blank=True, help_text="URL of the licence.")

    class Meta:
        ordering = ("name",)

    def __str__(self):
        return self.identifier


class Image(models.Model):
    alt_text = models.TextField(blank=True, default="", help_text="Alternate text for the image.")
    credits = models.TextField(blank=True, default="", help_text="Credits of the image")
    is_use_explicitly_granted_by_author = models.BooleanField(
        default=False, help_text="True if use is explicitly granted by author."
    )
    height = models.PositiveIntegerField(blank=True, null=True)
    image = VersatileImageField(
        blank=True,
        height_field="height",
        null=True,
        ppoi_field="ppoi",
        upload_to="images",
        width_field="width",
        help_text="The URI of the image.",
    )
    license = models.ForeignKey(
        License, null=True, on_delete=models.SET_NULL, related_name="images"
    )
    owner = models.CharField(max_length=150)
    ppoi = PPOIField()
    width = models.PositiveIntegerField(blank=True, null=True)

    def render(self, width: int | None = None, height: int | None = None):
        if width is None and height is None:
            return self.image.url
        elif width and height:
            return self.image.crop[f"{width}x{height}"].url
        aspect_ratio = self.width / self.height
        if width is None:
            width = int(height * aspect_ratio)
        if height is None:
            height = int(width / aspect_ratio)
        return self.image.thumbnail[f"{width}x{height}"].url


class Profile(models.Model):
    biography = models.TextField(blank=True, help_text="Biography of the profile.")
    created_at = models.DateTimeField(auto_now_add=True)
    created_by = models.CharField(max_length=150)
    email = models.EmailField(blank=True, help_text="Email address of the profile.")
    image = models.ForeignKey(Image, null=True, on_delete=models.CASCADE, related_name="profiles")
    is_active = models.BooleanField(default=True, help_text="True if the profile is active.")
    name = models.CharField(max_length=128, help_text="Display name of the profile.")
    owners = models.ManyToManyField(
        User, blank=True, related_name="profiles", help_text="User ID(s) that own this profile."
    )
    updated_at = models.DateTimeField(auto_now=True, blank=True, null=True)
    updated_by = models.CharField(blank=True, default="", max_length=150)

    class Meta:
        ordering = ("name",)
        permissions = [
            ("edit__profile__biography", "Can edit biography field"),
            ("edit__profile__email", "Can edit email field"),
            ("edit__profile__image", "Can edit image field"),
            ("edit__profile__links", "Can edit links field"),
            ("edit__profile__name", "Can edit name field"),
            ("edit__profile__owners", "Can edit owners field"),
            # overrides ownership
            ("update_profile", "Can update profile"),
        ]

    def __str__(self):
        return self.name


class LinkType(models.Model):
    is_active = models.BooleanField(default=True, help_text="True if link type is active.")
    name = models.CharField(max_length=32, help_text="Name of the link type")

    class Meta:
        ordering = ("name",)

    def __str__(self):
        return self.name


class Link(models.Model):
    type = models.ForeignKey(LinkType, default=1, on_delete=models.CASCADE)
    url = models.URLField()

    class Meta:
        abstract = True

    def __str__(self):
        return self.url


class ProfileLink(Link):
    profile = models.ForeignKey(Profile, on_delete=models.CASCADE, related_name="links")


class ShowManager(models.Manager):
    def with_max_timeslot_start(self):
        return (
            super().get_queryset().annotate(max_timeslot_start=Max("schedules__timeslots__start"))
        )


class Show(models.Model):
    category = models.ManyToManyField(Category, blank=True, related_name="shows")
    cba_series_id = models.IntegerField(blank=True, null=True)
    created_at = models.DateTimeField(auto_now_add=True)
    created_by = models.CharField(max_length=150)
    default_playlist_id = models.IntegerField(blank=True, null=True)
    description = models.TextField(blank=True, help_text="Description of this show.")
    email = models.EmailField(blank=True, null=True, help_text="Email address of this show.")
    funding_category = models.ForeignKey(
        FundingCategory, blank=True, null=True, on_delete=models.CASCADE, related_name="shows"
    )
    hosts = models.ManyToManyField(Profile, blank=True, related_name="shows")
    image = models.ForeignKey(Image, null=True, on_delete=models.CASCADE, related_name="shows")
    internal_note = models.TextField(blank=True, help_text="Internal note for this show.")
    is_active = models.BooleanField(default=True, help_text="True if this show is active.")
    is_public = models.BooleanField(default=False, help_text="True if this show is public.")
    language = models.ManyToManyField(Language, blank=True, related_name="shows")
    # TODO: is this really necessary?
    logo = models.ForeignKey(
        Image, blank=True, null=True, on_delete=models.CASCADE, related_name="logo_shows"
    )
    music_focus = models.ManyToManyField(MusicFocus, blank=True, related_name="shows")
    name = models.CharField(max_length=255, help_text="Name of this Show.")
    owners = models.ManyToManyField(User, blank=True, related_name="shows")
    predecessor = models.ForeignKey(
        "self", blank=True, null=True, on_delete=models.CASCADE, related_name="successors"
    )
    short_description = models.TextField(help_text="Short description of this show.")
    slug = models.SlugField(
        blank=True, max_length=255, unique=True, help_text="Slug of this show."
    )
    topic = models.ManyToManyField(Topic, blank=True, related_name="shows")
    type = models.ForeignKey(
        Type, blank=True, null=True, on_delete=models.CASCADE, related_name="shows"
    )
    updated_at = models.DateTimeField(auto_now=True, blank=True, null=True)
    updated_by = models.CharField(blank=True, default="", max_length=150)

    objects = ShowManager()

    class Meta:
        ordering = ("slug",)
        permissions = [
            ("display__show__internal_note", "Can display internal note field"),
            ("edit__show__categories", "Can edit category field"),
            ("edit__show__cba_series_id", "Can edit cba series id field"),
            ("edit__show__default_playlist_id", "Can edit default media source"),
            ("edit__show__description", "Can edit description field"),
            ("edit__show__email", "Can edit email field"),
            ("edit__show__funding_categories", "Can edit funding category field"),
            ("edit__show__hosts", "Can edit hosts field"),
            ("edit__show__image", "Can edit image field"),
            ("edit__show__internal_note", "Can edit internal note field"),
            ("edit__show__is_active", "Can edit is active field"),
            ("edit__show__languages", "Can edit language field"),
            ("edit__show__links", "Can edit links field"),
            ("edit__show__logo", "Can edit logo field"),
            ("edit__show__music_focuses", "Can edit music focus field"),
            ("edit__show__name", "Can edit name field"),
            ("edit__show__owners", "Can edit owners field"),
            ("edit__show__predecessor", "Can edit predecessor field"),
            ("edit__show__short_description", "Can edit short description field"),
            ("edit__show__slug", "Can edit slug field"),
            ("edit__show__topics", "Can edit topic field"),
            ("edit__show__type", "Can edit type field"),
            # overrides ownership
            ("update_show", "Can update show"),
        ]

    def __str__(self):
        return self.name

    def save(self, *args, **kwargs):
        now = timezone.datetime.now()
        today = now.date()

        if self.pk and self.is_active is False:
            # deactivating a show means:
            # - **delete all* the timeslots that belong to a schedule of this show the after now
            # - **update all** the schedules of this show have today as `last_date`
            TimeSlot.objects.filter(schedule__show=self, start__gt=now).delete()
            self.schedules.filter(Q(last_date__gt=today) | Q(last_date=None)).update(
                last_date=today
            )
        super().save(*args, **kwargs)


class ShowLink(Link):
    show = models.ForeignKey(Show, on_delete=models.CASCADE, related_name="links")


class RRule(models.Model):
    by_set_pos = models.IntegerField(
        blank=True,
        choices=[
            (1, "first"),
            (2, "second"),
            (3, "third"),
            (4, "fourth"),
            (5, "fifth"),
            (-1, "last"),
        ],
        null=True,
    )
    by_weekdays = models.CharField(
        blank=True,
        choices=[
            (None, ""),
            ("0,1,2,3,4", "business days"),
            ("5,6", "weekends"),
        ],
        null=True,
        max_length=9,
    )
    count = models.IntegerField(
        blank=True,
        null=True,
        help_text="How many occurrences should be generated.",
    )
    freq = models.IntegerField(
        choices=[
            (0, "once"),
            (1, "monthly"),
            (2, "weekly"),
            (3, "daily"),
        ]
    )
    interval = models.IntegerField(
        default=1,
        help_text="The interval between each freq iteration.",
    )
    name = models.CharField(max_length=32, unique=True)

    class Meta:
        ordering = ("pk",)
        unique_together = ("freq", "interval", "by_set_pos", "by_weekdays")
        verbose_name = _("recurrence rule")

    def __str__(self):
        return self.name


class Schedule(models.Model):
    add_business_days_only = models.BooleanField(
        default=False,
        help_text=(
            "Whether to add add_days_no but skipping the weekends. "
            "E.g. if weekday is Friday, the date returned will be the next Monday."
        ),
    )
    add_days_no = models.IntegerField(
        blank=True,
        null=True,
        help_text=(
            "Add a number of days to the generated dates. "
            "This can be useful for repetitions, like 'On the following day'."
        ),
    )
    by_weekday = models.IntegerField(
        help_text="Number of the Weekday.",
        choices=[
            (0, "Monday"),
            (1, "Tuesday"),
            (2, "Wednesday"),
            (3, "Thursday"),
            (4, "Friday"),
            (5, "Saturday"),
            (6, "Sunday"),
        ],
        null=True,
    )
    default_playlist_id = models.IntegerField(
        blank=True,
        null=True,
        help_text="A tank ID in case the timeslot's playlist_id is empty.",
    )
    end_time = models.TimeField(null=True, help_text="End time of schedule.")
    first_date = models.DateField(help_text="Start date of schedule.")
    is_repetition = models.BooleanField(
        default=False,
        help_text="Whether the schedule is a repetition.",
    )
    last_date = models.DateField(help_text="End date of schedule.", null=True)
    rrule = models.ForeignKey(
        RRule, help_text="A recurrence rule.", on_delete=models.CASCADE, related_name="schedules"
    )
    show = models.ForeignKey(
        Show,
        help_text="Show the schedule belongs to.",
        on_delete=models.CASCADE,
        related_name="schedules",
    )
    start_time = models.TimeField(help_text="Start time of schedule.")

    class Meta:
        ordering = ("first_date", "start_time")
        permissions = [
            ("edit__schedule__default_playlist_id", "Can edit default media source"),
        ]

    def __str__(self):
        WEEKDAYS = ["MO", "TU", "WE", "TH", "FR", "SA", "SU"]
        start_time = self.start_time.strftime("%H:%M")
        end_time = self.end_time.strftime("%H:%M")
        recurrence = self.rrule.name
        weekday = self.first_date.weekday()

        return f"{self.show.name} - {recurrence} {WEEKDAYS[weekday]} {start_time}-{end_time}"


class TimeSlot(models.Model):
    end = models.DateTimeField()
    memo = models.TextField(blank=True, help_text="Memo for this timeslot.")
    playlist_id = models.IntegerField(null=True, help_text="Playlist ID of this timeslot.")
    repetition_of = models.ForeignKey(
        "self", blank=True, null=True, on_delete=models.CASCADE, related_name="repetitions"
    )
    schedule = models.ForeignKey(Schedule, on_delete=models.CASCADE, related_name="timeslots")
    start = models.DateTimeField()

    class Meta:
        ordering = ("start", "end")
        permissions = [
            ("edit__timeslot__memo", "Can edit memo field"),
            ("edit__timeslot__playlist_id", "Can edit media source"),
            ("edit__timeslot__repetition_of", "Can edit repetition of field"),
        ]

    def __str__(self):
        if self.start.date() == self.end.date():
            time_span = "{0}, {1} - {2}".format(
                self.start.strftime("%x"),
                self.start.strftime("%X"),
                self.end.strftime("%X"),
            )
        else:
            time_span = "{0} - {1}".format(
                self.start.strftime("%X %x"),
                self.end.strftime("%X %x"),
            )

        return f"{str(self.schedule.show)} ({time_span})"

    @property
    def hash(self):
        string = (
            str(self.start)
            + str(self.end)
            + str(self.schedule.rrule.id)
            + str(self.schedule.by_weekday)
        )
        return str("".join(s for s in string if s.isdigit()))


class Note(models.Model):
    cba_id = models.IntegerField(blank=True, null=True, help_text="CBA entry ID.")
    content = models.TextField(help_text="Textual content of the note.")
    contributors = models.ManyToManyField(
        Profile, related_name="notes", help_text="Profile IDs that contributed to this episode."
    )
    created_at = models.DateTimeField(auto_now_add=True)
    created_by = models.CharField(max_length=150)
    image = models.ForeignKey(Image, null=True, on_delete=models.CASCADE, related_name="notes")
    language = models.ManyToManyField(Language, blank=True, related_name="episodes")
    playlist = models.TextField(blank=True)
    summary = models.TextField(blank=True, help_text="Summary of the Note.")
    tags = models.JSONField(blank=True, default=list)
    timeslot = models.OneToOneField(TimeSlot, null=True, on_delete=models.SET_NULL, unique=True)
    title = models.CharField(
        blank=True, default="", max_length=128, help_text="Title of the note."
    )
    topic = models.ManyToManyField(Topic, blank=True, related_name="episodes")
    updated_at = models.DateTimeField(auto_now=True, blank=True, null=True)
    updated_by = models.CharField(blank=True, default="", max_length=150)

    class Meta:
        ordering = ("timeslot",)
        permissions = [
            ("edit__note__cba_id", "Can edit CBA id field"),
            ("edit__note__content", "Can edit content field"),
            ("edit__note__contributors", "Can edit contributor field"),
            ("edit__note__image", "Can edit image field"),
            ("edit__note__languages", "Can edit language field"),
            ("edit__note__links", "Can edit links field"),
            ("edit__note__playlist", "Can edit playlist field"),
            ("edit__note__summary", "Can edit summary field"),
            ("edit__note__tags", "Can edit tags field"),
            ("edit__note__title", "Can edit title field"),
            ("edit__note__topics", "Can edit topics field"),
            # overrides ownership
            ("create_note", "Can create note"),
            ("update_note", "Can update note"),
        ]

    def __str__(self):
        return self.title


class NoteLink(Link):
    note = models.ForeignKey(Note, on_delete=models.CASCADE, related_name="links")


class CBA(models.Model):
    username = models.CharField("Username", blank=True, max_length=60)
    user_token = models.CharField("User Token", blank=True, max_length=255)
    created_at = models.DateTimeField(auto_now_add=True)
    created_by = models.CharField(max_length=150)
    updated_at = models.DateTimeField(auto_now=True, blank=True, null=True)
    updated_by = models.CharField(blank=True, default="", max_length=150)
    user = models.OneToOneField(User, null=True, on_delete=models.SET_NULL, related_name="cba")

    class Meta:
        permissions = [
            # overrides ownership
            ("create_cba", "Can create user CBA profile"),
            ("update_cba", "Can update user CBA profile"),
        ]

    def __str__(self):
        return self.user.username


class Playlist(models.Model):
    class Meta:
        permissions = [
            ("add__file", "Can add file media-source"),
            ("add__import", "Can add import media-source"),
            ("add__line", "Can add line media-source"),
            ("add__m3ufile", "Can add m3u media-source"),
            ("add__stream", "Can add stream media-source"),
        ]


class ImageAspectRadioField(models.CharField):
    validators = [
        RegexValidator(
            code="invalid_aspect_ratio",
            message="Enter a valid aspect ratio in the format int:int or float:float",
            regex=r"^\d+(\.\d+)?:\d+(\.\d+)?$",
        )
    ]

    def __init__(self, *args, **kwargs):
        kwargs["max_length"] = 11

        super().__init__(*args, **kwargs)


class ImageShapeField(models.CharField):
    def __init__(self, *args, **kwargs):
        kwargs["choices"] = [
            ("rect", "rect"),
            ("round", "round"),
        ]
        kwargs["max_length"] = 5

        super().__init__(*args, **kwargs)


def validate_json_value(value: list | dict, schema: dict) -> None:
    """Validates value JSON against the schema. Raises a Django `ValidationError` if invalid."""

    try:
        jsonschema.validate(instance=value, schema=schema)
    except jsonschema.exceptions.ValidationError as e:
        raise DjangoValidationError(e.args[0])


def validate_cba_domains(value):
    schema = {
        "type": "array",
        "items": {"type": "string"},
    }

    validate_json_value(value, schema)


def validate_line_in_channels(value):
    schema = {
        "type": "object",
        "patternProperties": {
            "^.*$": {"type": "string"},
        },
    }

    validate_json_value(value, schema)


def validate_fallback_pools(value):
    schema = {
        "type": "object",
        "patternProperties": {
            "^.*$": {"type": "string"},
        },
    }

    validate_json_value(value, schema)


def validate_fallback_default_pool(value):
    if value not in RadioSettings.objects.first().pools.keys():
        raise DjangoValidationError(f"Pool key '{value}' does not exist in pools.")


class RadioSettings(models.Model):
    cba_api_key = models.CharField(blank=True, max_length=64, verbose_name="CBA API key")
    cba_domains = models.JSONField(
        blank=True,
        default=list,
        help_text="JSON array of strings",
        validators=[validate_cba_domains],
        verbose_name="CBA domains",
    )
    fallback_default_pool = models.CharField(
        blank=True, max_length=32, validators=[validate_fallback_default_pool]
    )
    fallback_show = models.ForeignKey(
        Show, blank=True, null=True, on_delete=models.CASCADE, related_name="+"
    )
    line_in_channels = models.JSONField(
        blank=True,
        default=dict,
        help_text="JSON key/value pairs",
        validators=[validate_line_in_channels],
    )
    micro_show = models.ForeignKey(
        Show, blank=True, null=True, on_delete=models.CASCADE, related_name="+"
    )
    note_image_aspect_ratio = ImageAspectRadioField(default="16:9")
    note_image_shape = ImageShapeField(default="rect")
    pools = models.JSONField(
        blank=True,
        default=dict,
        help_text="JSON key/value pairs",
        validators=[validate_fallback_pools],
    )
    profile_image_aspect_ratio = ImageAspectRadioField(default="1:1")
    profile_image_shape = ImageShapeField(default="round")
    show_image_aspect_ratio = ImageAspectRadioField(default="16:9")
    show_image_shape = ImageShapeField(default="rect")
    show_logo_aspect_ratio = ImageAspectRadioField(default="1:1")
    show_logo_shape = ImageShapeField(default="rect")
    station_logo = VersatileImageField(
        blank=True,
        height_field="station_logo_height",
        null=True,
        upload_to="images",
        width_field="station_logo_width",
    )
    station_logo_height = models.PositiveIntegerField(blank=True, null=True)
    station_logo_width = models.PositiveIntegerField(blank=True, null=True)
    station_name = models.CharField(max_length=256, unique=True)
    station_website = models.URLField()

    class Meta:
        verbose_name_plural = "Radio Settings"

    def __str__(self):
        return self.station_name


@dataclasses.dataclass()
class ProgramEntry:
    id: str
    start: datetime.datetime
    end: datetime.datetime
    show: Show
    timeslot: TimeSlot | None

    def playlist_id(self) -> int | None:
        if self.timeslot and self.timeslot.playlist_id:
            return self.timeslot.playlist_id
        else:
            return self.show.default_playlist_id