-
Konrad Mohrfeldt authoredThis new endpoint allows clients to easily purge steering’s application state. It is primarily designed for use in integration tests as outlined in aura#387. The endpoint is available in debug mode so that developers can inspect the schema. However, any request to it is denied unless the OPERATING_MODE environment variable is set to "tests". This is a precaution for administrators that use the debug mode to analyze application issues in production.
Konrad Mohrfeldt authoredThis new endpoint allows clients to easily purge steering’s application state. It is primarily designed for use in integration tests as outlined in aura#387. The endpoint is available in debug mode so that developers can inspect the schema. However, any request to it is denied unless the OPERATING_MODE environment variable is set to "tests". This is a precaution for administrators that use the debug mode to analyze application issues in production.
models.py 27.62 KiB
#
# 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 collections
import dataclasses
import datetime
import typing
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 topic 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
class ApplicationStateManager:
categorized_models: typing.Mapping[str, list[typing.Type[models.Model]]] = {
"classifications": [
Type,
Category,
Topic,
MusicFocus,
FundingCategory,
Language,
License,
LinkType,
RRule,
],
"settings": [RadioSettings],
"auth": [CBA, User],
"media": [Image],
"program": [Note, TimeSlot, Schedule, Show],
}
@property
def _models(self):
result = []
for _models in self.categorized_models.values():
result.extend(_models)
return result
@property
def _model_map(self):
return {model._meta.label: model for model in self._models}
@property
def model_category_choices(self):
return sorted(self.categorized_models.keys())
@property
def model_choices(self):
return sorted(self._model_map.keys())
def purge(
self,
model_category_names: typing.Iterable[str] | None = None,
model_names: typing.Iterable[str] | None = None,
invert_selection: bool = False,
):
model_map = self._model_map
model_category_names = set(model_category_names or [])
model_names = set(model_names or [])
selected_models: set[typing.Type[models.Model]] = set()
for category_name in model_category_names:
selected_models.update(self.categorized_models[category_name])
for model_name in model_names:
selected_models.add(model_map[model_name])
if invert_selection:
selected_models = set(self._models).difference(selected_models)
# Some models may have dependent state and therefore need to be deleted in order.
ordered_selected_models = [model for model in self._models if model in selected_models]
deleted_model_count_map = collections.defaultdict(int)
for model in ordered_selected_models:
for model_path, count in model.objects.all().delete()[1].items():
deleted_model_count_map[model_path] += count
return deleted_model_count_map
application_state_manager = ApplicationStateManager()