serializers.py

import logging
import urllib.parse
import uuid

from django.core.exceptions import ObjectDoesNotExist
from django.core.paginator import Paginator
from django.db import transaction
from django.urls import reverse
from django.utils import timezone
from rest_framework import serializers

from funkwhale_api.common import utils as common_utils
from funkwhale_api.common import models as common_models
from funkwhale_api.moderation import models as moderation_models
from funkwhale_api.moderation import serializers as moderation_serializers
from funkwhale_api.moderation import signals as moderation_signals
from funkwhale_api.music import licenses
from funkwhale_api.music import models as music_models
from funkwhale_api.music import tasks as music_tasks
from funkwhale_api.tags import models as tags_models

from . import activity, actors, contexts, jsonld, models, tasks, utils

logger = logging.getLogger(__name__)


def include_if_not_none(data, value, field):
    if value is not None:
        data[field] = value


class MultipleSerializer(serializers.Serializer):
    """
    A serializer that will try multiple serializers in turn
    """

    def __init__(self, *args, **kwargs):
        self.allowed = kwargs.pop("allowed")
        super().__init__(*args, **kwargs)

    def to_internal_value(self, v):
        last_exception = None
        for serializer_class in self.allowed:
            s = serializer_class(data=v)
            try:
                s.is_valid(raise_exception=True)
            except serializers.ValidationError as e:
                last_exception = e
            else:
                return s.validated_data

        raise last_exception


class TruncatedCharField(serializers.CharField):
    def __init__(self, *args, **kwargs):
        self.truncate_length = kwargs.pop("truncate_length")
        super().__init__(*args, **kwargs)

    def to_internal_value(self, v):
        v = super().to_internal_value(v)
        if v:
            v = v[: self.truncate_length]
        return v


class TagSerializer(jsonld.JsonLdSerializer):
    type = serializers.ChoiceField(choices=[contexts.AS.Hashtag])
    name = serializers.CharField(max_length=100)

    class Meta:
        jsonld_mapping = {"name": jsonld.first_val(contexts.AS.name)}

    def validate_name(self, value):
        if value.startswith("#"):
            # remove trailing #
            value = value[1:]
        return value


def tag_list(tagged_items):
    return [
        repr_tag(item.tag.name)
        for item in sorted(set(tagged_items.all()), key=lambda i: i.tag.name)
    ]


def is_mimetype(mt, allowed_mimetypes):
    for allowed in allowed_mimetypes:
        if allowed.endswith("/*"):
            if mt.startswith(allowed.replace("*", "")):
                return True
        else:
            if mt == allowed:
                return True
    return False


class MediaSerializer(jsonld.JsonLdSerializer):
    mediaType = serializers.CharField()

    def __init__(self, *args, **kwargs):
        self.allowed_mimetypes = kwargs.pop("allowed_mimetypes", [])
        self.allow_empty_mimetype = kwargs.pop("allow_empty_mimetype", False)
        super().__init__(*args, **kwargs)
        self.fields["mediaType"].required = not self.allow_empty_mimetype
        self.fields["mediaType"].allow_null = self.allow_empty_mimetype

    def validate_mediaType(self, v):
        if not self.allowed_mimetypes:
            # no restrictions
            return v
        if self.allow_empty_mimetype and not v:
            return None

        if not is_mimetype(v, self.allowed_mimetypes):
            raise serializers.ValidationError(
                "Invalid mimetype {}. Allowed: {}".format(v, self.allowed_mimetypes)
            )
        return v


class LinkSerializer(MediaSerializer):
    type = serializers.ChoiceField(choices=[contexts.AS.Link])
    href = serializers.URLField(max_length=500)
    bitrate = serializers.IntegerField(min_value=0, required=False)
    size = serializers.IntegerField(min_value=0, required=False)

    class Meta:
        jsonld_mapping = {
            "href": jsonld.first_id(contexts.AS.href),
            "mediaType": jsonld.first_val(contexts.AS.mediaType),
            "bitrate": jsonld.first_val(contexts.FW.bitrate),
            "size": jsonld.first_val(contexts.FW.size),
        }


class LinkListSerializer(serializers.ListField):
    def __init__(self, *args, **kwargs):
        kwargs.setdefault("child", LinkSerializer(jsonld_expand=False))
        self.keep_mediatype = kwargs.pop("keep_mediatype", [])
        super().__init__(*args, **kwargs)

    def to_internal_value(self, v):
        links = super().to_internal_value(v)
        if not self.keep_mediatype:
            # no further filtering required
            return links
        links = [
            link
            for link in links
            if link.get("mediaType")
            and is_mimetype(link["mediaType"], self.keep_mediatype)
        ]
        if not self.allow_empty and len(links) == 0:
            self.fail("empty")

        return links


class ImageSerializer(MediaSerializer):
    type = serializers.ChoiceField(choices=[contexts.AS.Image, contexts.AS.Link])
    href = serializers.URLField(max_length=500, required=False)
    url = serializers.URLField(max_length=500, required=False)

    class Meta:
        jsonld_mapping = {
            "url": jsonld.first_id(contexts.AS.url),
            "href": jsonld.first_id(contexts.AS.href),
            "mediaType": jsonld.first_val(contexts.AS.mediaType),
        }

    def validate(self, data):
        validated_data = super().validate(data)
        if "url" not in validated_data:
            try:
                validated_data["url"] = validated_data["href"]
            except KeyError:
                if self.required:
                    raise serializers.ValidationError(
                        "You need to provide a url or href"
                    )

        return validated_data


class URLSerializer(jsonld.JsonLdSerializer):
    href = serializers.URLField(max_length=500)
    mediaType = serializers.CharField(required=False)

    class Meta:
        jsonld_mapping = {
            "href": jsonld.first_id(contexts.AS.href, aliases=[jsonld.raw("@id")]),
            "mediaType": jsonld.first_val(contexts.AS.mediaType),
        }


class EndpointsSerializer(jsonld.JsonLdSerializer):
    sharedInbox = serializers.URLField(max_length=500, required=False)