serializers.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 re

from drf_jsonschema_serializer import JSONSchemaField
from rest_framework import serializers
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.db import IntegrityError
from django.utils import text, timezone
from program.models import (
    CBA,
    Category,
    FundingCategory,
    Host,
    HostLink,
    Image,
    Language,
    License,
    LinkType,
    MusicFocus,
    Note,
    NoteLink,
    RadioSettings,
    RRule,
    Schedule,
    Show,
    ShowLink,
    TimeSlot,
    Topic,
    Type,
)
from program.typing import (
    Logo,
    MicroProgram,
    NestedEpisode,
    NestedSchedule,
    NestedShow,
    ProgramFallback,
    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):
    class Meta:
        model = CBA
        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 = (
            "id",
            "is_privileged",
            "permissions",
            "host_ids",
        )
        fields = (
            "cba",
            "email",
            "first_name",
            "is_active",
            "is_staff",
            "is_superuser",
            "last_name",
            "password",
            "username",
        ) + 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"])
        user.save()

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

        return 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