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 timezone
from program.models import (
Category,
FundingCategory,
Host,
from program.utils import delete_links, get_audio_url
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):
permissions = serializers.SerializerMethodField()
# Add profile fields to JSON
profile = ProfileSerializer(required=False)
class Meta:
model = User
read_only_fields = (
"id",
"permissions",
)
) + read_only_fields
@staticmethod
def get_permissions(obj: User) -> list[str]:
return sorted(
[p.split(".")[1] for p in obj.get_all_permissions() if p.startswith("program.edit")]
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.
"""
instance.first_name = validated_data.get("first_name", instance.first_name)
instance.last_name = validated_data.get("last_name", instance.last_name)
instance.email = validated_data.get("email", instance.email)
instance.is_active = validated_data.get("is_active", instance.is_active)
instance.is_staff = validated_data.get("is_staff", instance.is_staff)
instance.is_superuser = validated_data.get("is_superuser", instance.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)
profile.cba_username = profile_data.get("cba_username")
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", "name", "type")
class LicenseSerializer(serializers.ModelSerializer):
fields = (
"id",
"identifier",
"name",
"needs_author",
"requires_express_permission_for_publication",
"url",
)
class HostLinkSerializer(serializers.ModelSerializer):
class Meta:
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."""
instance.alt_text = validated_data.get("alt_text", instance.alt_text)
instance.credits = validated_data.get("credits", instance.credits)
instance.image.ppoi = validated_data.get("ppoi", instance.ppoi)
instance.license = validated_data.get("license", instance.license)
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_id", 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 privileged users and owners of a host with edit permissions are allowed to update it
# Being a privileged user overrides the ownership
if not (user_is_privileged 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 := validated_data.get("biography")
) and "biography" not in user_edit_permissions:
raise exceptions.PermissionDenied(
detail="You are not allowed to edit the host’s biography."
)
else:
instance.biography = biography
if (name := validated_data.get("name")) and "name" not in user_edit_permissions:
raise exceptions.PermissionDenied(
detail="You are not allowed to edit the host’s name."
)
else:
instance.name = name
# Only update these fields if the user is privileged, ignore otherwise
if user_is_privileged:
instance.email = validated_data.get("email", instance.email)
instance.image = validated_data.get("image_id", instance.image)
instance.is_active = validated_data.get("is_active", instance.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 := validated_data.get("owners"):
instance.owners.set(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):
class Meta:
model = ShowLink
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()
)
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
)
internal_note = serializers.SerializerMethodField()
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
)
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
)
topic_ids = serializers.PrimaryKeyRelatedField(
many=True, queryset=Topic.objects.all(), source="topic"
)
type_id = serializers.PrimaryKeyRelatedField(queryset=Type.objects.all())
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",
Loading
Loading full blame...