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 collections.abc import Iterable
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 import IntegrityError, transaction
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,
    Playlist,
    PlaylistEntry,
    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 UpdateFieldPermissionValidatorMixin:
    """
    Validates that fields can only be saved if the appropriate permissions have been given.

    The required field permissions are determined based the following strategy:
        1. If Serializer.Meta.custom_field_write_permission_map[{field_name}] has been set
           the permissions defined there will be used.
        2. If permission in the style of
             {app_label}.edit__{model_name}__{field_name}
           exists, that permission will be used.
        3. If none of the above are available, no permissions will be used for the field.

    The permissions for each field, for which specific write permissions have been defined,
    will be checked on update and create operations. If the user does not have any one of the
    required field permissions for the provided fields the request will be rejected with a
    403 error including a list of fields that the user is not allowed to set.
    """

    def get_validators(self):
        def validate_field_permissions(data, *args, **kwargs):
            user = self.context.get("request").user
            self._check_field_write_permissions(user, data, self.instance)

        return [*super().get_validators(), validate_field_permissions]

    def _generate_field_permission_name_candidates(self, model_name: str, field_name: str):
        prefix = f"edit__{model_name}__"
        yield prefix + field_name
        if field_name.endswith("_id"):
            yield prefix + field_name[:-3]
        if field_name.endswith("_ids"):
            yield prefix + field_name[:-4] + "s"

    def _get_field_write_permission_map(self, field_names: set[str]):
        custom_map: dict[str, str | Iterable[str]] = getattr(
            self.Meta, "custom_field_write_permission_map", {}
        )
        model = self.Meta.model
        custom_model_permissions = set(dict(model._meta.permissions).keys())
        app_label = model._meta.app_label
        model_name = model._meta.model_name
        result = {}
        for field_name in field_names:
            if perm := custom_map.get(field_name):
                if isinstance(perm, str):
                    result[field_name] = {perm}
                else:
                    result[field_name] = set(perm)
                continue
            for perm in self._generate_field_permission_name_candidates(model_name, field_name):
                if perm in custom_model_permissions:
                    result[field_name] = {f"{app_label}.{perm}"}
        return result

    def _check_field_write_permissions(self, user, validated_data, instance=None):
        field_error_map = {}
        field_write_permission_map = self._get_field_write_permission_map(validated_data.keys())
        client_name_map = {field.source: field.field_name for field in self._writable_fields}
        for field_name, required_permissions in field_write_permission_map.items():
            if not user.has_perms(required_permissions):
                # The name of the field in validated_data is not necessarily
                # the same as the field name the client sent.
                # The error mapping should reflect the name the client sent.
                client_field_name = client_name_map.get(field_name, field_name)
                field_error_map[client_field_name] = exceptions.ErrorDetail(
                    "You do not have permission to set this field.",
                    code="missing-field-permission",
                )
        if field_error_map:
            raise exceptions.PermissionDenied(field_error_map)


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