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 datetime import datetime
from functools import cached_property
from zoneinfo import ZoneInfo

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.models import Q
from django.utils import text, timezone
from program.models import (
    CBA,
    Category,
    FundingCategory,
    Image,
    Language,
    License,
    LinkType,
    MusicFocus,
    Note,
    NoteLink,
    Profile,
    ProfileLink,
    ProgramEntry,
    RadioSettings,
    RRule,
    Schedule,
    Show,
    ShowLink,
    TimeSlot,
    Topic,
    Type,
    application_state_manager,
)
from program.typing import (
    Logo,
    MicroProgram,
    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

    def to_representation(self, instance):
        if not self.parent.context.get("request").user.is_authenticated:
            return None

        return super().to_representation(instance)


class UserSerializer(serializers.ModelSerializer):
    is_privileged = serializers.SerializerMethodField()
    permissions = serializers.SerializerMethodField()
    cba = CBASerializer(required=False)
    profile_ids = serializers.PrimaryKeyRelatedField(
        many=True, queryset=Profile.objects.all(), required=False, source="profiles"
    )

    class Meta:
        extra_kwargs = {
            "password": {"write_only": True},
        }
        model = User
        read_only_fields = (
            "id",
            "is_privileged",
            "permissions",
            "profile_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.split(".", 1)[0] in ["auth", "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")