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 typing import Literal, NotRequired, TypedDict

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 (
    Category,
    FundingCategory,
    Host,
    HostLink,
    Image,
    Language,
    License,
    LinkType,
    MusicFocus,
    Note,
    NoteLink,
    RadioSettings,
    RRule,
    Schedule,
    Show,
    ShowLink,
    TimeSlot,
    Topic,
    Type,
    UserProfile,
)
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)
    host_ids = serializers.PrimaryKeyRelatedField(
        many=True, queryset=Host.objects.all(), source="hosts"
    )

    class Meta:
        model = User
        read_only_fields = (
            "id",
            "is_privileged",
            "permissions",
            "host_ids",
        )
        fields = (
            "email",
            "first_name",
            "is_active",
            "is_staff",
            "is_superuser",
            "last_name",
            "password",
            "profile",
            "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.
        """

        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"])
        user.save()

        if profile_data:
            profile = UserProfile(
                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()

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

        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: