Draper e31196d19f
Audio Fixes (#4492)
* handles #4491

* add typing indicators to audio playlists commands like discussed with aika.

* recheck perms upon change of token to avoid needing a reload.

* Ensure the player lock is always released... on rewrite to this as a callback to the task.

* ffs

* resolves#4495

* missed one

* aaaaaaaaa

* fix https://canary.discord.com/channels/133049272517001216/387398816317440000/766711707921678396

* some tweaks

* Clear errors to users around YouTube Quota
2020-10-20 09:57:02 -07:00

159 lines
4.7 KiB
Python

import datetime
import json
import logging
from collections import namedtuple
from dataclasses import dataclass, field
from pathlib import Path
from typing import List, MutableMapping, Optional, Union
import discord
import lavalink
from redbot.core.bot import Red
from redbot.core.i18n import Translator
from redbot.core.utils.chat_formatting import humanize_list
from ..errors import InvalidPlaylistScope, MissingAuthor, MissingGuild
from ..utils import PlaylistScope
log = logging.getLogger("red.cogs.Audio.api.utils")
_ = Translator("Audio", Path(__file__))
@dataclass
class YouTubeCacheFetchResult:
query: Optional[str]
last_updated: int
def __post_init__(self):
if isinstance(self.last_updated, int):
self.updated_on: datetime.datetime = datetime.datetime.fromtimestamp(self.last_updated)
@dataclass
class SpotifyCacheFetchResult:
query: Optional[str]
last_updated: int
def __post_init__(self):
if isinstance(self.last_updated, int):
self.updated_on: datetime.datetime = datetime.datetime.fromtimestamp(self.last_updated)
@dataclass
class LavalinkCacheFetchResult:
query: Optional[MutableMapping]
last_updated: int
def __post_init__(self):
if isinstance(self.last_updated, int):
self.updated_on: datetime.datetime = datetime.datetime.fromtimestamp(self.last_updated)
if isinstance(self.query, str):
self.query = json.loads(self.query)
@dataclass
class LavalinkCacheFetchForGlobalResult:
query: str
data: MutableMapping
def __post_init__(self):
if isinstance(self.data, str):
self.data_string = str(self.data)
self.data = json.loads(self.data)
@dataclass
class PlaylistFetchResult:
playlist_id: int
playlist_name: str
scope_id: int
author_id: int
playlist_url: Optional[str] = None
tracks: List[MutableMapping] = field(default_factory=lambda: [])
def __post_init__(self):
if isinstance(self.tracks, str):
self.tracks = json.loads(self.tracks)
@dataclass
class QueueFetchResult:
guild_id: int
room_id: int
track: dict = field(default_factory=lambda: {})
track_object: lavalink.Track = None
def __post_init__(self):
if isinstance(self.track, str):
self.track = json.loads(self.track)
if self.track:
self.track_object = lavalink.Track(self.track)
def standardize_scope(scope: str) -> str:
"""Convert any of the used scopes into one we are expecting."""
scope = scope.upper()
valid_scopes = ["GLOBAL", "GUILD", "AUTHOR", "USER", "SERVER", "MEMBER", "BOT"]
if scope in PlaylistScope.list():
return scope
elif scope not in valid_scopes:
raise InvalidPlaylistScope(
f'"{scope}" is not a valid playlist scope.'
f" Scope needs to be one of the following: {humanize_list(valid_scopes)}"
)
if scope in ["GLOBAL", "BOT"]:
scope = PlaylistScope.GLOBAL.value
elif scope in ["GUILD", "SERVER"]:
scope = PlaylistScope.GUILD.value
elif scope in ["USER", "MEMBER", "AUTHOR"]:
scope = PlaylistScope.USER.value
return scope
def prepare_config_scope(
bot: Red,
scope,
author: Union[discord.abc.User, int] = None,
guild: Union[discord.Guild, int] = None,
):
"""Return the scope used by Playlists."""
scope = standardize_scope(scope)
if scope == PlaylistScope.GLOBAL.value:
config_scope = [PlaylistScope.GLOBAL.value, bot.user.id]
elif scope == PlaylistScope.USER.value:
if author is None:
raise MissingAuthor("Invalid author for user scope.")
config_scope = [PlaylistScope.USER.value, int(getattr(author, "id", author))]
else:
if guild is None:
raise MissingGuild("Invalid guild for guild scope.")
config_scope = [PlaylistScope.GUILD.value, int(getattr(guild, "id", guild))]
return config_scope
def prepare_config_scope_for_migration23( # TODO: remove me in a future version ?
scope, author: Union[discord.abc.User, int] = None, guild: discord.Guild = None
):
"""Return the scope used by Playlists."""
scope = standardize_scope(scope)
if scope == PlaylistScope.GLOBAL.value:
config_scope = [PlaylistScope.GLOBAL.value]
elif scope == PlaylistScope.USER.value:
if author is None:
raise MissingAuthor("Invalid author for user scope.")
config_scope = [PlaylistScope.USER.value, str(getattr(author, "id", author))]
else:
if guild is None:
raise MissingGuild("Invalid guild for guild scope.")
config_scope = [PlaylistScope.GUILD.value, str(getattr(guild, "id", guild))]
return config_scope
FakePlaylist = namedtuple("Playlist", "author scope")