Newer
Older
#
# 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/>.
#
from typing import List, TypedDict
from drf_jsonschema_serializer import JSONSchemaField
from rest_framework.permissions import exceptions
from django.conf import settings
from django.contrib.auth.models import User
from django.core.exceptions import ObjectDoesNotExist
from django.utils import text, timezone
from program.models import (
Category,
FundingCategory,
Host,
from program.utils import delete_links
SOLUTION_CHOICES = {
"theirs": "Discard projected timeslot. Keep existing timeslot(s).",
"ours": "Create projected timeslot. Delete existing timeslot(s).",
"theirs-start": (
"Keep existing timeslot. Create projected timeslot with start time of existing end."
),
"ours-start": (
"Create projected timeslot. Change end of existing timeslot to projected start time."
),
"theirs-end": (
"Keep existing timeslot. Create projected timeslot with end of existing start time."
),
"ours-end": (
"Create projected timeslot. Change start of existing timeslot to projected end time."
),
"theirs-both": (
"Keep existing timeslot. "
"Create two projected timeslots with end of existing start and start of existing end."
),
"ours-both": (
"Create projected timeslot. Split existing timeslot into two: \n\n"
"* set existing end time to projected start,\n"
"* create another timeslot with start = projected end and end = existing end."
),
}
class ErrorSerializer(serializers.Serializer):
message = serializers.CharField()
code = serializers.CharField(allow_null=True)
class ProfileSerializer(serializers.ModelSerializer):
class Meta:
model = UserProfile
fields = (
"cba_username",
"cba_user_token",
"created_at",
"created_by",
"updated_at",
"updated_by",
)
read_only_fields = (
"created_at",
"created_by",
"updated_at",
"updated_by",
)
class UserSerializer(serializers.ModelSerializer):
is_privileged = serializers.SerializerMethodField()
permissions = serializers.SerializerMethodField()
# Add profile fields to JSON
profile = ProfileSerializer(required=False)
class Meta:
model = User
) + read_only_fields
@staticmethod
def get_permissions(obj: User) -> list[str]:
return sorted([p for p in obj.get_all_permissions() if p.startswith("program")])
@staticmethod
def get_is_privileged(obj: User) -> bool:
# return obj.groups.filter(name=settings.PRIVILEGED_GROUP).exists()
return obj.is_superuser
def create(self, validated_data):
"""
Create and return a new User instance, given the validated data.
"""
profile_data = validated_data.pop("profile") if "profile" in validated_data else None
user = super(UserSerializer, self).create(validated_data)
user.date_joined = timezone.now()
user.set_password(validated_data["password"])
if profile_data:
cba_username=profile_data.get("cba_username").strip(),
cba_user_token=profile_data.get("cba_user_token").strip(),
created_by=self.context.get("request").user.username,
user=user,
profile.save()
def update(self, instance, validated_data):
"""
Update and return an existing User instance, given the validated data.
"""
if "first_name" in validated_data:
instance.first_name = validated_data.get("first_name")
if "last_name" in validated_data:
instance.last_name = validated_data.get("last_name")
if "email" in validated_data:
instance.email = validated_data.get("email")
if "is_active" in validated_data:
instance.is_active = validated_data.get("is_active")
if "is_staff" in validated_data:
instance.is_staff = validated_data.get("is_staff")
if "is_superuser" in validated_data:
instance.is_superuser = validated_data.get("is_superuser")
profile_data = validated_data.pop("profile") if "profile" in validated_data else None
if profile_data:
# TODO: How to hook into this from ProfileSerializer without having to call it here?
try:
profile = UserProfile.objects.get(user=instance.id)
except ObjectDoesNotExist:
profile = UserProfile.objects.create(user=instance, **profile_data)
if "cba_username" in profile_data:
profile.cba_username = profile_data.get("cba_username")
if "cba_user_token" in profile_data:
profile.cba_user_token = profile_data.get("cba_user_token")
profile.updated_by = self.context.get("request").user.username
profile.save()
instance.save()
return instance
class CategorySerializer(serializers.ModelSerializer):
class Meta:
model = Category
fields = ("description", "id", "is_active", "name", "slug", "subtitle")
class LinkTypeSerializer(serializers.ModelSerializer):
class Meta:
model = LinkType
fields = ("id", "is_active", "name")
class LicenseSerializer(serializers.ModelSerializer):
fields = (
"id",
"identifier",
"name",
"needs_author",
"requires_express_permission_for_publication",
"url",
)
class HostLinkSerializer(serializers.ModelSerializer):
type_id = serializers.PrimaryKeyRelatedField(queryset=LinkType.objects.all(), source="type")
class Meta:
fields = ("type_id", "url")
model = HostLink
class PPOIField(serializers.CharField):
def validate_format(self, value: str):
if not re.match(r"\d(?:\.\d+)?x\d(?:\.\d+)?", value):
raise serializers.ValidationError("PPOI must match format: ${float}x${float}")
def __init__(self, **kwargs):
kwargs["max_length"] = 20
kwargs.setdefault("validators", [])
kwargs["validators"].append(self.validate_format)
super().__init__(**kwargs)
def to_representation(self, value: tuple[float, float]):
[left, top] = value
return f"{left}x{top}"
class Thumbnail(TypedDict):
width: float
height: float
url: str
class ImageSerializer(serializers.ModelSerializer):
license_id = serializers.PrimaryKeyRelatedField(
allow_null=True, queryset=License.objects.all(), required=False, source="license"
)
ppoi = PPOIField(required=False)
thumbnails = serializers.SerializerMethodField()
@staticmethod
def get_thumbnails(instance) -> List[Thumbnail]:
"""Returns thumbnails"""
thumbnails = []
if instance.image.name and settings.THUMBNAIL_SIZES:
for size in settings.THUMBNAIL_SIZES:
[width, height] = size.split("x")
thumbnails.append(
{
"width": int(width),
"height": int(height),
"url": instance.image.crop[size].url,
}
)
return thumbnails
class Meta:
model = Image
read_only_fields = (
"height",
"id",
"thumbnails",
"width",
)
fields = (
"alt_text",
"credits",
"image",
) + read_only_fields
def create(self, validated_data):
"""Create and return a new Image instance, given the validated data."""
image = Image.objects.create(
owner=self.context.get("request").user.username, **validated_data
)
image.save()
return image
def update(self, instance, validated_data):
"""Update and return an existing Image instance, given the validated data."""
if "alt_text" in validated_data:
instance.alt_text = validated_data.get("alt_text")
if "credits" in validated_data:
instance.credits = validated_data.get("credits")
if "license" in validated_data:
instance.license = validated_data.get("license")
if "ppoi" in validated_data:
instance.image.ppoi = validated_data.get("ppoi")
instance.save()
return instance
class HostSerializer(serializers.ModelSerializer):
image_id = serializers.PrimaryKeyRelatedField(
allow_null=True, queryset=Image.objects.all(), required=False, source="image"
links = HostLinkSerializer(many=True, required=False)
owner_ids = serializers.PrimaryKeyRelatedField(
allow_null=True, many=True, queryset=User.objects.all(), source="owners"
)
class Meta:
model = Host
read_only_fields = (
"created_at",
"created_by",
"updated_at",
"updated_by",
)
fields = (
"biography",
"email",
"id",
"image_id",
def create(self, validated_data):
"""
Create and return a new Host instance, given the validated data.
"""
links_data = validated_data.pop("links", [])
validated_data["image"] = validated_data.pop("image", None)
host = Host.objects.create(
created_by=self.context.get("request").user.username, **validated_data
)
for link_data in links_data:
HostLink.objects.create(host=host, **link_data)
host.save()
return host
def update(self, instance, validated_data):
"""
Update and return an existing Host instance, given the validated data.
A `PermissionDenied` exception will be raised if the user is not privileged or the owner of
the host and has the permissions to edit the fields.
user = self.context.get("request").user
# user_is_privileged = user.groups.filter(name=settings.PRIVILEGED_GROUP).exists()
user_is_owner = user in instance.owners.all()
user_edit_permissions = [
permission.split("__")[-1]
for permission in user.get_all_permissions()
if permission.startswith("program.edit__host")
]
# Only superusers and owners of a host with edit permissions are allowed to update it
# Being a superuser overrides the ownership
if not (user.is_superuser or (user_is_owner and len(user_edit_permissions) > 0)):
raise exceptions.PermissionDenied(detail="You are not allowed to update this host.")
# Only users with edit permissions are allowed to edit these fields
if "biography" in validated_data and "biography" not in user_edit_permissions:
raise exceptions.PermissionDenied(
detail="You are not allowed to edit the host’s biography."
)
if "name" in validated_data and "name" not in user_edit_permissions:
raise exceptions.PermissionDenied(
detail="You are not allowed to edit the host’s name."
)
if "biography" in validated_data:
instance.biography = validated_data.get("biography")
if "name" in validated_data:
instance.name = validated_data.get("name")
# Only update these fields if the user superuser, ignore otherwise
if user.is_superuser:
if "email" in validated_data:
instance.email = validated_data.get("email")
if "image" in validated_data:
instance.image = validated_data.get("image")
if "is_active" in validated_data:
instance.is_active = validated_data.get("is_active")
# optional nested objects
if links_data := validated_data.get("links"):
instance = delete_links(instance)
for link_data in links_data:
HostLink.objects.create(host=instance, **link_data)
# optional many-to-many
if "owners" in validated_data.get("owners"):
instance.owners.set(validated_data.get("owners", []))
instance.updated_by = self.context.get("request").user.username
instance.save()
return instance
class LanguageSerializer(serializers.ModelSerializer):
class Meta:
model = Language
class TopicSerializer(serializers.ModelSerializer):
class Meta:
model = Topic
class MusicFocusSerializer(serializers.ModelSerializer):
class Meta:
model = MusicFocus
class TypeSerializer(serializers.ModelSerializer):

Ingo Leindecker
committed
class Meta:
model = Type

Ingo Leindecker
committed
class FundingCategorySerializer(serializers.ModelSerializer):

Ingo Leindecker
committed
class Meta:

Ingo Leindecker
committed
class ShowLinkSerializer(serializers.ModelSerializer):
type_id = serializers.PrimaryKeyRelatedField(queryset=LinkType.objects.all(), source="type")
class Meta:
model = ShowLink
fields = ("type_id", "url")
class ShowSerializer(serializers.HyperlinkedModelSerializer):
category_ids = serializers.PrimaryKeyRelatedField(
many=True, queryset=Category.objects.all(), source="category"
)
funding_category_id = serializers.PrimaryKeyRelatedField(
queryset=FundingCategory.objects.all(), source="funding_category"
)
host_ids = serializers.PrimaryKeyRelatedField(
many=True, queryset=Host.objects.all(), source="hosts"
)
image_id = serializers.PrimaryKeyRelatedField(
allow_null=True, queryset=Image.objects.all(), required=False, source="image"
)
language_ids = serializers.PrimaryKeyRelatedField(
many=True, queryset=Language.objects.all(), source="language"
links = HostLinkSerializer(many=True, required=False)
logo_id = serializers.PrimaryKeyRelatedField(
allow_null=True, queryset=Image.objects.all(), required=False, source="logo"
)
music_focus_ids = serializers.PrimaryKeyRelatedField(
many=True, queryset=MusicFocus.objects.all(), source="music_focus"
owner_ids = serializers.PrimaryKeyRelatedField(
many=True, queryset=User.objects.all(), source="owners"
predecessor_id = serializers.PrimaryKeyRelatedField(
allow_null=True, queryset=Show.objects.all(), required=False, source="predecessor"
)
topic_ids = serializers.PrimaryKeyRelatedField(
many=True, queryset=Topic.objects.all(), source="topic"
)
type_id = serializers.PrimaryKeyRelatedField(queryset=Type.objects.all(), source="type")
class Meta:
model = Show
read_only_fields = (
"created_at",
"created_by",
"updated_at",
"updated_by",
)
"funding_category_id",
"host_ids",
"logo_id",
"music_focus_ids",
"owner_ids",
"predecessor_id",
"topic_ids",
"type_id",
def create(self, validated_data):
"""
Create and return a new Show instance, given the validated data.
"""
category = validated_data.pop("category")
hosts = validated_data.pop("hosts")
language = validated_data.pop("language")
music_focus = validated_data.pop("music_focus")
owners = validated_data.pop("owners")
topic = validated_data.pop("topic")
links_data = validated_data.pop("links", [])
validated_data["funding_category"] = validated_data.pop("funding_category")
validated_data["type"] = validated_data.pop("type")
validated_data["image"] = validated_data.pop("image", None)
validated_data["logo"] = validated_data.pop("logo", None)
validated_data["predecessor"] = validated_data.pop("predecessor", None)
validated_data["slug"] = validated_data.get(
"slug", text.slugify(validated_data.get("name"))
)
show = Show.objects.create(
created_by=self.context.get("request").user.username,
**validated_data,
# Save many-to-many relationships

jackie / Andrea Ida Malkah Klaura
committed
show.category.set(category)
show.hosts.set(hosts)
show.language.set(language)
show.music_focus.set(music_focus)
show.owners.set(owners)
show.topic.set(topic)
for link_data in links_data:
ShowLink.objects.create(show=show, **link_data)
def update(self, instance, validated_data):
"""
Update and return an existing Show instance, given the validated data.
A `PermissionDenied` exception will be raised if the user is not privileged or the owner of
a show and has the permissions to edit the fields.
"""
user = self.context.get("request").user
# user_is_privileged = user.groups.filter(name=settings.PRIVILEGED_GROUP).exists()
user_is_owner = instance in user.shows.all()
user_edit_permissions = [
permission.split("__")[-1]
for permission in user.get_all_permissions()
if permission.startswith("program.edit__show")
]
# Only superusers and owners of a show with edit permissions are allowed to update it
# Being a superuser overrides the ownership
if not (user.is_superuser or (user_is_owner and len(user_edit_permissions) > 0)):
raise exceptions.PermissionDenied(detail="You are not allowed to update this show.")
# Only users with edit permissions are allowed to update these fields
if "description" in validated_data and "description" not in user_edit_permissions:
raise exceptions.PermissionDenied(
detail="You are not allowed to edit the show’s description."
)
if "name" in validated_data and "name" not in user_edit_permissions:
raise exceptions.PermissionDenied(
detail="You are not allowed to edit the show’s name."
)
"short_description" in validated_data
and "short_description" not in user_edit_permissions
):
raise exceptions.PermissionDenied(
detail="You are not allowed to edit the show’s short description."
)
if "description" in validated_data:
instance.description = validated_data.get("description")
if "name" in validated_data:
instance.name = validated_data.get("name")
if "short_description" in validated_data:
instance.short_description = validated_data.get("short_description")
# Only update these fields if the user is superuser, ignore otherwise
if user.is_superuser:
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
if "cba_series_id" in validated_data:
instance.cba_series_id = validated_data.get("cba_series_id")
if "default_playlist_id" in validated_data:
instance.default_playlist_id = validated_data.get("default_playlist_id")
if "email" in validated_data:
instance.email = validated_data.get("email")
if "funding_category" in validated_data:
instance.funding_category = validated_data.get("funding_category")
if "image" in validated_data:
instance.image = validated_data.get("image")
if "internal_note" in validated_data:
instance.internal_note = validated_data.get("internal_note")
if "is_active" in validated_data:
instance.is_active = validated_data.get("is_active")
if "is_public" in validated_data:
instance.is_public = validated_data.get("is_public")
if "logo" in validated_data:
instance.logo = validated_data.get("logo")
if "predecessor" in validated_data:
instance.predecessor = validated_data.get("predecessor")
if "slug" in validated_data:
instance.slug = validated_data.get("slug")
if "type" in validated_data:
instance.type = validated_data.get("type")
# optional many-to-many
if "category" in validated_data:
instance.category.set(validated_data.get("category", []))
if "hosts" in validated_data:
instance.hosts.set(validated_data.get("hosts", []))
if "language" in validated_data:
instance.language.set(validated_data.get("language", []))
if "music_focus" in validated_data:
instance.music_focus.set(validated_data.get("music_focus", []))
if "owners" in validated_data:
instance.owners.set(validated_data.get("owners", []))
if "topic" in validated_data:
instance.topic.set(validated_data.get("topic", []))
# optional nested objects
if links_data := validated_data.get("links"):
instance = delete_links(instance)
for link_data in links_data:
ShowLink.objects.create(show=instance, **link_data)
instance.updated_by = self.context.get("request").user.username
instance.save()
return instance
class RRuleSerializer(serializers.ModelSerializer):
class Meta:
model = RRule
fields = (
"by_set_pos",
"by_weekdays",
"count",
"freq",
"name",
)
read_only_fields = fields
class ScheduleSerializer(serializers.ModelSerializer):
class Meta:
model = Schedule
"rrule_id",
"show_id",
"start_time",
)
class UnsavedScheduleSerializer(ScheduleSerializer):
id = serializers.IntegerField(allow_null=True)
class ScheduleInRequestSerializer(ScheduleSerializer):
dryrun = serializers.BooleanField(
write_only=True,
required=False,
help_text=(
"Whether to simulate the database changes. If true, no database changes will occur. "
"Instead a list of objects that would be created, updated and deleted if dryrun was "
"false will be returned."
),
)
"default_playlist_id",
"dryrun",
"end_time",
"rrule_id",
"show_id",
"start_time",
)
def create(self, validated_data):
"""Create and return a new Schedule instance, given the validated data."""
validated_data["rrule"] = validated_data.pop("rrule_id")
validated_data["show"] = validated_data.pop("show_id")
schedule = Schedule.objects.create(**validated_data)
schedule.save()
return schedule
def update(self, instance, validated_data):
"""Update and return an existing Schedule instance, given the validated data."""
instance.by_weekday = validated_data.get("by_weekday", instance.by_weekday)
instance.first_date = validated_data.get("first_date", instance.first_date)
instance.start_time = validated_data.get("start_time", instance.start_time)
instance.end_time = validated_data.get("end_time", instance.end_time)
instance.last_date = validated_data.get("last_date", instance.last_date)
instance.is_repetition = validated_data.get("is_repetition", instance.is_repetition)
instance.default_playlist_id = validated_data.get(
"default_playlist_id", instance.default_playlist_id
)
instance.rrule = validated_data.get("rrule_id", instance.rrule)
instance.show = validated_data.get("show_id", instance.show)
instance.add_days_no = validated_data.get("add_days_no", instance.add_days_no)
instance.add_business_days_only = validated_data.get(
"add_business_days_only", instance.add_business_days_only
)
class CollisionSerializer(serializers.Serializer):
id = serializers.IntegerField()
start = serializers.DateTimeField()
end = serializers.DateTimeField()
playlist_id = serializers.IntegerField(allow_null=True)
show_id = serializers.IntegerField()
show_name = serializers.CharField()
repetition_of_id = serializers.IntegerField(allow_null=True)
schedule_id = serializers.IntegerField()
memo = serializers.CharField()
note_id = serializers.IntegerField(allow_null=True)
class ProjectedTimeSlotSerializer(serializers.Serializer):
hash = serializers.CharField()
start = serializers.DateTimeField()
end = serializers.DateTimeField()
collisions = CollisionSerializer(many=True)
error = serializers.CharField(allow_null=True)
solution_choices = serializers.ListField(child=serializers.ChoiceField(SOLUTION_CHOICES))
class DryRunTimeSlotSerializer(serializers.Serializer):
id = serializers.PrimaryKeyRelatedField(queryset=TimeSlot.objects.all(), allow_null=True)
schedule_id = serializers.PrimaryKeyRelatedField(
queryset=Schedule.objects.all(), allow_null=True
)
playlist_id = serializers.IntegerField(allow_null=True)
start = serializers.DateField()
end = serializers.DateField()
repetition_of_id = serializers.IntegerField(allow_null=True)
memo = serializers.CharField()
class ScheduleCreateUpdateRequestSerializer(serializers.Serializer):
schedule = ScheduleInRequestSerializer()
solutions = serializers.DictField(
child=serializers.ChoiceField(SOLUTION_CHOICES), required=False
)
notes = serializers.DictField(child=serializers.IntegerField(), required=False)
playlists = serializers.DictField(child=serializers.IntegerField(), required=False)
class ScheduleResponseSerializer(serializers.Serializer):
projected = ProjectedTimeSlotSerializer(many=True)
solutions = serializers.DictField(child=serializers.ChoiceField(SOLUTION_CHOICES))
notes = serializers.DictField(child=serializers.IntegerField())
playlists = serializers.DictField(child=serializers.IntegerField())
schedule = ScheduleSerializer()
class ScheduleConflictResponseSerializer(ScheduleResponseSerializer):
schedule = UnsavedScheduleSerializer()
class ScheduleDryRunResponseSerializer(serializers.Serializer):
created = DryRunTimeSlotSerializer(many=True)
updated = DryRunTimeSlotSerializer(many=True)
deleted = DryRunTimeSlotSerializer(many=True)
class TimeSlotSerializer(serializers.ModelSerializer):
note_id = serializers.SerializerMethodField()
show_id = serializers.SerializerMethodField()
schedule_id = serializers.PrimaryKeyRelatedField(
queryset=Schedule.objects.all(), required=False, source="schedule"
)
repetition_of_id = serializers.PrimaryKeyRelatedField(
allow_null=True, queryset=TimeSlot.objects.all(), required=False, source="repetition_of"
)
class Meta:
model = TimeSlot
"id",
"note_id",
"schedule_id",
"show_id",
fields = (
"memo",
"playlist_id",
"repetition_of_id",
def get_show_id(obj) -> int:
return obj.schedule.show.id
@staticmethod
def get_note_id(obj) -> int:
return obj.note.id if hasattr(obj, "note") else None
def update(self, instance, validated_data):
"""Update and return an existing Show instance, given the validated data."""
if "memo" in validated_data:
instance.memo = validated_data.get("memo")
if "repetition_of" in validated_data:
instance.repetition_of = validated_data.get("repetition_of")
if "playlist_id" in validated_data:
instance.playlist_id = validated_data.get("playlist_id")
instance.save()
return instance
class NoteLinkSerializer(serializers.ModelSerializer):
type_id = serializers.PrimaryKeyRelatedField(queryset=LinkType.objects.all(), source="type")
class Meta:
model = NoteLink
fields = ("type_id", "url")
tags_json_schema = {"type": "array", "items": {"type": "string"}}
class NoteSerializer(serializers.ModelSerializer):
contributor_ids = serializers.PrimaryKeyRelatedField(
many=True,
queryset=Host.objects.all(),
required=False,
source="contributors",
image_id = serializers.PrimaryKeyRelatedField(
queryset=Image.objects.all(), required=False, allow_null=True, source="image"
language_ids = serializers.PrimaryKeyRelatedField(
allow_null=True,
many=True,
queryset=Language.objects.all(),
required=False,
source="language",
)
links = NoteLinkSerializer(many=True, required=False)
playlist_id = serializers.IntegerField(required=False)
tags = JSONSchemaField(tags_json_schema, required=False)
timeslot_id = serializers.PrimaryKeyRelatedField(
queryset=TimeSlot.objects.all(), required=False, source="timeslot"
topic_ids = serializers.PrimaryKeyRelatedField(
allow_null=True, many=True, queryset=Topic.objects.all(), required=False, source="topic"
)
class Meta:
model = Note
"created_at",
"created_by",
"updated_at",
"updated_by",
)
"contributor_ids",
"language_ids",