musik/game/tasks.py

import google.oauth2.credentials
import googleapiclient.discovery
from celery import shared_task

from . import models


@shared_task
def generate_playlist(creds, game_pk):
    game = models.MusikGame.objects.get(pk=game_pk)
    credentials = google.oauth2.credentials.Credentials(**creds)
    yt_api = googleapiclient.discovery.build("youtube", "v3", credentials=credentials)
    pl_request = yt_api.playlists().insert(
        part="snippet,status",
        body={
            "snippet": {
                "title": f"Musik – {game.group.name} – {game.date.strftime('%x')}",
                "description": "Playlist générée par Musik",
            },
            "status": {
                "privacyStatus": "private",
            },
        },
    )
    pl_response = pl_request.execute()
    pl_id = pl_response.get("id")
    game.playlist = pl_id
    game.save()
    for music in game.musicgameorder_set.all():
        request = yt_api.playlistItems().insert(
            part="snippet",
            body={
                "snippet": {
                    "playlistId": pl_id,
                    "resourceId": {
                        "kind": "youtube#video",
                        "videoId": music.music_video.yt_id,
                    },
                }
            },
        )
        request.execute()

    game.playlist_loading = False
    game.save()


@shared_task
def delete_playlist(creds, playlist_id):
    credentials = google.oauth2.credentials.Credentials(**creds)

    yt_api = googleapiclient.discovery.build("youtube", "v3", credentials=credentials)
    pl_request = yt_api.playlists().delete(id=playlist_id)
    pl_request.execute()