funkwhale/api/funkwhale_api/musicbrainz/serializers.py

import logging

from rest_framework import serializers

from funkwhale_api import musicbrainz
from funkwhale_api.tags import models as tags_models

from . import client

logger = logging.getLogger(__name__)


class ArtistSerializer(serializers.Serializer):
    """
    Serializer for Musicbrainz artist data.
    """

    id = serializers.CharField()
    name = serializers.CharField()

    def create(self, validated_data):
        from funkwhale_api.music.models import Artist

        data = {
            "name": validated_data["name"],
            "mbid": validated_data["id"],
        }
        artist, created = Artist.objects.get_or_create(**data)
        return artist


class ArtistCreditSerializer(serializers.Serializer):
    """
    Serializer for Musicbrainz artist data.
    """

    name = serializers.CharField()
    joinphrase = serializers.CharField(allow_blank=True)
    artist = ArtistSerializer()

    def create(self, validated_data):
        from funkwhale_api.music.models import ArtistCredit

        data = {
            "credit": validated_data["name"],
            "joinphrase": validated_data.get("joinphrase", ""),
            "artist": ArtistSerializer().create(validated_data["artist"]),
        }
        artist_credit, created = ArtistCredit.objects.get_or_create(**data)
        return artist_credit


class ReleaseForRecordingSerializer(serializers.Serializer):
    """
    Serializer for Musicbrainz release data when returned in a recording object.
    """

    id = serializers.CharField()
    title = serializers.CharField()
    artist_credit = ArtistCreditSerializer(many=True)
    tags = serializers.ListField(child=serializers.CharField(), allow_empty=True)
    date = serializers.DateField(input_formats=["%Y", "%Y/%m/%d", "%Y-%m-%d"])

    def create(self, validated_data):
        from funkwhale_api.music.models import Album

        data = {
            "title": validated_data["title"],
            "mbid": validated_data["id"],
            "release_date": validated_data.get("date", None),
        }
        album, created = Album.objects.get_or_create(**data)
        artist_credit = ArtistCreditSerializer(many=True).create(
            validated_data["artist_credit"]
        )
        album.artist_credit.set(artist_credit)
        album.save()
        tags_models.add_tags(album, *validated_data.get("tags", []))
        return album

    def update(self, instance, validated_data):
        instance.title = validated_data["title"]
        instance.release_date = validated_data.get("date")
        instance.save()
        tags_models.add_tags(instance, *validated_data.get("tags", []))

        return instance


class RecordingSerializer(serializers.Serializer):
    """
    Serializer for Musicbrainz track data.
    """

    id = serializers.CharField()
    title = serializers.CharField()
    artist_credit = ArtistCreditSerializer(many=True)
    releases = ReleaseForRecordingSerializer(many=True, required=False)
    tags = serializers.ListField(child=serializers.CharField(), allow_empty=True)
    position = serializers.IntegerField(required=False, allow_null=True)

    def create(self, validated_data):
        from funkwhale_api.music.models import Track

        data = {"mbid": validated_data["id"]}
        defaults = {
            "title": validated_data["title"],
            "mbid": validated_data["id"],
            # In mb a recording can have various releases, we take the fist one
            "album": (
                ReleaseForRecordingSerializer(many=True).create(
                    validated_data["releases"]
                )[0]
                if validated_data.get("releases")
                else None
            ),
            # this will be none if the recording is not fetched from the release endpoint
            "position": validated_data.get("position", None),
        }

        if defaults["album"] is not None and defaults["position"] is None:
            result = client.api.releases.get(
                id=validated_data["releases"][0]["id"],
                includes=["tags", "artists", "recordings"],
            )
            tracks = result["media"][0]["tracks"]
            defaults["position"] = next(
                (o for o in tracks if o["recording"]["id"] == data["mbid"]), {}
            ).get("position", None)

        track, created = Track.objects.get_or_create(**data, defaults=defaults)
        artist_credit = ArtistCreditSerializer(many=True).create(
            validated_data["artist_credit"]
        )
        track.artist_credit.set(artist_credit)
        track.save()

        tags_models.add_tags(track, *validated_data.get("tags", []))

        return track

    def update(self, instance, validated_data):
        instance.title = validated_data["title"]
        instance.save()
        tags_models.add_tags(instance, *validated_data.get("tags", []))

        return instance


class RecordingForReleaseSerializer(serializers.Serializer):
    id = serializers.CharField()
    title = serializers.CharField()
    # not in Musicbrainz recording object, but used to store the position of the track in the album
    position = serializers.IntegerField(required=False, allow_null=True)

    def create(self, validated_data):
        def replace_hyphens_in_keys(obj):
            if isinstance(obj, dict):
                return {
                    k.replace("-", "_"): replace_hyphens_in_keys(v)
                    for k, v in obj.items()
                }
            elif isinstance(obj, list):
                return [replace_hyphens_in_keys(item) for item in obj]
            else:
                return obj

        recordings_data = musicbrainz.api.recordings.get(
            id=validated_data["id"], includes=["tags", "artists"]
        )
        recordings_data = replace_hyphens_in_keys(recordings_data)
        recordings_data["position"] = validated_data.get("position", None)
        serializer = RecordingSerializer(data=recordings_data)
        serializer.is_valid(raise_exception=True)
        track = serializer.save()
        track.album = validated_data["album"]
        track.save()
        return track

    def update(self, instance, validated_data):
        instance.title = validated_data["title"]
        tags_models.add_tags(instance, *validated_data.get("tags", []))
        instance.album = validated_data["album"]
        instance.save()
        return instance


class TrackSerializer(serializers.Serializer):
    recording = RecordingForReleaseSerializer()
    position = serializers.IntegerField()


class MediaSerializer(serializers.Serializer):
    tracks = TrackSerializer(many=True)


class ReleaseSerializer(serializers.Serializer):
    """
    Serializer for Musicbrainz release data.
    """

    id = serializers.CharField()
    title = serializers.CharField()
    artist_credit = ArtistCreditSerializer(many=True)
    tags = serializers.ListField(child=serializers.CharField(), allow_empty=True)
    date = serializers.DateField(input_formats=["%Y", "%Y/%m/%d", "%Y-%m-%d"])
    media = serializers.ListField(child=MediaSerializer())

    def create(self, validated_data):
        from funkwhale_api.music.models import Album

        data = {
            "title": validated_data["title"],
            "mbid": validated_data["id"],
            "release_date": validated_data.get("date"),
        }
        album, created = Album.objects.get_or_create(**data)
        artist_credit = ArtistCreditSerializer(many=True).create(
            validated_data["artist_credit"]
        )
        album.artist_credit.set(artist_credit)
        album.save()
        tags_models.add_tags(album, *validated_data.get("tags", []))

        # an album can have various media/physical representation, we take the first one
        tracks = [t for t in validated_data["media"][0]["tracks"]]
        recordings = []
        for t in tracks:
            t["recording"]["position"] = t["position"]
            recordings.append(t["recording"])

        for r in recordings:
            r["album"] = album
            RecordingForReleaseSerializer().create(r)

        return album

    # this will never be used while FetchViewSet and third_party_fetch filter out finished fetch
    # Would be nice to have a way to manually update releases from Musicbrainz
    def update(self, instance, validated_data):
        logger.info(f"Updating release {instance} with data: {validated_data}")
        instance.title = validated_data["title"]
        instance.release_date = validated_data.get("date")
        instance.save()
        tags_models.add_tags(instance, *validated_data.get("tags", []))

        recordings = [t["recording"] for t in validated_data["media"][0]["tracks"]]
        for r in recordings:
            r["album"] = instance
            RecordingForReleaseSerializer().update(r)
        return instance