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

Ernesto Rico Schmidt
committed
from django.db import IntegrityError
from django.utils import text, timezone
from program.typing import (
Logo,
NestedEpisode,
NestedSchedule,
NestedShow,
RadioCBASettings,
RadioImageRequirementsSettings,
RadioPlayoutSettings,
RadioProgramSettings,
RadioStationSettings,
)
from program.utils import update_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 CBASerializer(serializers.ModelSerializer):
read_only_fields = (
"created_at",
"created_by",
"updated_at",
"updated_by",
)
fields = (
"username",
"user_token",
) + read_only_fields
class UserSerializer(serializers.ModelSerializer):
is_privileged = serializers.SerializerMethodField()
permissions = serializers.SerializerMethodField()
cba = CBASerializer(required=False)
host_ids = serializers.PrimaryKeyRelatedField(
many=True, queryset=Host.objects.all(), required=False, source="hosts"
class Meta:
extra_kwargs = {
"password": {"write_only": True},
}
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.is_superuser
def create(self, validated_data):
"""
Create and return a new User instance, given the validated data.
"""
cba_data = validated_data.pop("cba") if "cba" in validated_data else None
user = super(UserSerializer, self).create(validated_data)
user.date_joined = timezone.now()
user.set_password(validated_data["password"])
if cba_data:
CBA.objects.create(
username=cba_data.get("username").strip(),
user_token=cba_data.get("user_token").strip(),
created_by=self.context.get("request").user.username,
user=user,
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")
cba_data = validated_data.pop("cba") if "cba" in validated_data else None
except ObjectDoesNotExist:
user = self.context.get("request").user
if cba:
# having the update_cba permission overrides being the user
if not (user.has_perm("update_cba") or user.id == instance.id):
raise exceptions.PermissionDenied(
detail="You do not have permission to update this user CBA profile."
if "username" in cba_data:
cba.username = cba_data.get("username")
if "user_token" in cba_data:
cba.user_token = cba_data.get("user_token")
cba.updated_by = self.context.get("request").user.username
cba.save()
else:
# having the create_cba permission overrides being the user
if not (user.has_perm("create_cba") or user.id == instance.id):
raise exceptions.PermissionDenied(
detail="You do not have permission to create this user CBA profile."
created_by=self.context.get("request").user.username,
user=instance,
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 ImageSerializer(serializers.ModelSerializer):
license_id = serializers.PrimaryKeyRelatedField(
allow_null=True,
queryset=License.objects.all(),
required=False,
source="license",
help_text="`License` ID of this image.",
ppoi = PPOIField(required=False, help_text="PPOI specifies the crop centerpoint of the image.")

Ernesto Rico Schmidt
committed
url = serializers.SerializerMethodField()

Ernesto Rico Schmidt
committed
@staticmethod
def get_url(instance: Image) -> str:
"""Returns the image URL, using settings.SITE_URL to include the protocol and avoid mixed
media warnings."""
return f"{settings.SITE_URL}{instance.image.url}"
extra_kwargs = {"image": {"write_only": True}}
model = Image
read_only_fields = (
"height",
"id",

Ernesto Rico Schmidt
committed
"url",
"width",
)
fields = (
"alt_text",
"credits",
"image",

Konrad Mohrfeldt
committed
"is_use_explicitly_granted_by_author",
) + 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 ImageRenderSerializer(serializers.Serializer):
width = serializers.IntegerField(required=False, min_value=1)
height = serializers.IntegerField(required=False, min_value=1)
class HostSerializer(serializers.ModelSerializer):
image_id = serializers.PrimaryKeyRelatedField(
allow_null=True,
queryset=Image.objects.all(),
required=False,
source="image",
help_text="`Image` id of the host.",
)
links = HostLinkSerializer(
many=True, required=False, help_text="Array of `HostLink` objects. Can be empty."
owner_ids = serializers.PrimaryKeyRelatedField(
many=True,
queryset=User.objects.all(),
source="owners",
help_text="`User` id(s) identifying this host.",
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."""
user = self.context.get("request").user
user_is_owner = user in instance.owners.all()
user_permissions = set(
permission.split("__")[-1]
for permission in user.get_all_permissions()
if permission.startswith("program.edit__host")
)
update_fields = set(validated_data.keys())
# having the update_host permission overrides the ownership
if not (user.has_perm("program.update_host") or (user_is_owner and user_permissions)):
raise exceptions.PermissionDenied(detail="You are not allowed to update this host.")
# without the update_host permission, fields without edit permission are not allowed
if not user.has_perm("program.update_host") and (
not_allowed := update_fields.difference(user_permissions)
):
detail = {field: "You are not allowed to edit this field" for field in not_allowed}
raise exceptions.PermissionDenied(detail=detail)
if "biography" in validated_data:
instance.biography = validated_data.get("biography")
if "name" in validated_data:
instance.name = validated_data.get("name")
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 many-to-many
if "owners" in validated_data:
instance.owners.set(validated_data.get("owners"))
# optional nested objects
if "links" in validated_data:
instance = update_links(instance, validated_data.get("links"))
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",
help_text="Array of `Category` IDs.",
)
cba_series_id = serializers.IntegerField(
allow_null=True, required=False, help_text="CBA series ID."
)
default_playlist_id = serializers.IntegerField(
allow_null=True, required=False, help_text="Default `Playlist` ID for this show."
)
funding_category_id = serializers.PrimaryKeyRelatedField(
queryset=FundingCategory.objects.all(),
source="funding_category",
help_text="`FundingCategory` ID.",
)
host_ids = serializers.PrimaryKeyRelatedField(
many=True,
queryset=Host.objects.all(),
source="hosts",
help_text="`Host` IDs of this show.",
)
image_id = serializers.PrimaryKeyRelatedField(
allow_null=True,
queryset=Image.objects.all(),
required=False,
source="image",
help_text="`Image` ID of this show.",
)
language_ids = serializers.PrimaryKeyRelatedField(
many=True,
queryset=Language.objects.all(),
source="language",
help_text="`Language` IDs of this show.",
links = HostLinkSerializer(many=True, required=False, help_text="Array of `Link` objects.")
logo_id = serializers.PrimaryKeyRelatedField(
allow_null=True,
queryset=Image.objects.all(),
required=False,
source="logo",
help_text="`Image` ID of the logo of this show.",
)
music_focus_ids = serializers.PrimaryKeyRelatedField(
many=True,
queryset=MusicFocus.objects.all(),
source="music_focus",
help_text="Array of `MusicFocus` IDs.",
owner_ids = serializers.PrimaryKeyRelatedField(
many=True,
queryset=User.objects.all(),
source="owners",
help_text="Array of `User` IDs owning this Show.",
predecessor_id = serializers.PrimaryKeyRelatedField(
allow_null=True,
queryset=Show.objects.all(),
required=False,
source="predecessor",
help_text="`Show` ID that predeceeded this one.",
)
topic_ids = serializers.PrimaryKeyRelatedField(
many=True, queryset=Topic.objects.all(), source="topic", help_text="Array of `Topic` IDs."
)
type_id = serializers.PrimaryKeyRelatedField(
queryset=Type.objects.all(), source="type", help_text="Array of `Type` IDs."
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 to_internal_value(self, data):
if data.get("cba_series_id") == "":
data["cba_series_id"] = None
if data.get("default_playlist_id") == "":
data["default_playlist_id"] = None
return super().to_internal_value(data)
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."""
user = self.context.get("request").user
user_is_owner = user in instance.owners.all()
user_permissions = set(
permission.split("__")[-1]
for permission in user.get_all_permissions()
if permission.startswith("program.edit__show")
)
update_fields = set(validated_data.keys())
# having update_show permission overrides the ownership
if not (user.has_perm("program.update_show") or (user_is_owner and user_permissions)):
raise exceptions.PermissionDenied(detail="You are not allowed to update this show.")
# without the update_show permission, fields without edit permission are not allowed
if not user.has_perm("program.update_show") and (
not_allowed := update_fields.difference(user_permissions)
detail = {field: "You are not allowed to edit this field" for field in not_allowed}
raise exceptions.PermissionDenied(detail=detail)
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")
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" in validated_data:
instance = update_links(instance, validated_data.get("links"))
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):
start = serializers.DateTimeField()
end = serializers.DateTimeField()
playlist_id = serializers.IntegerField(allow_null=True)
show_id = serializers.IntegerField()
show_name = serializers.CharField()
schedule_id = serializers.IntegerField()
memo = serializers.CharField()
timeslot_id = serializers.IntegerField()
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(help_text="`Schedule` object.")
solutions = serializers.DictField(
child=serializers.ChoiceField(SOLUTION_CHOICES),
required=False,
help_text="Array of solution choices.",
)
notes = serializers.DictField(
child=serializers.IntegerField(), required=False, help_text="Array of `Note` objects."
)
playlists = serializers.DictField(
child=serializers.IntegerField(), required=False, help_text="Array of `Playlist` IDs."
)
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",
help_text="`Schedule` ID of this timeslot.",
)
repetition_of_id = serializers.PrimaryKeyRelatedField(
allow_null=True,
queryset=TimeSlot.objects.all(),
required=False,
source="repetition_of",
help_text="This timeslot is a repetition of `Timeslot` ID.",
)
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