deadcxap/YandexMusicDiscordBot
1
0
Fork 0
mirror of https://github.com/deadcxap/YandexMusicDiscordBot.git synced 2026-01-09 07:41:53 +03:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #8 from Lemon4ksan/dev

Browse Source 
Обновление бота #4
This commit is contained in:
Bananchiki
2025-03-20 18:38:10 +03:00
committed by GitHub
parent 32f9e6e2c8 43de5636d8
commit 45b91d53b4
18 changed files with 1187 additions and 869 deletions
Download Patch File Download Diff File Expand all files Collapse all files

20
.dockerignore Normal file
View File

@@ -0,0 +1,20 @@
.git
.gitignore
venv/
env/
ENV/
__pycache__/
*.py[cod]
*$py.class
build/
dist/
.vscode/
.idea/
.DS_Store
.env

52
.github/workflows/docker-image.yml vendored Normal file
View File

@@ -0,0 +1,52 @@
name: Docker Image CI
on:
push:
branches:
- '**'
pull_request:
branches:
- '**'
jobs:
build:
runs-on: ubuntu-latest
env:
IMAGE_NAME: ${{ secrets.DOCKERHUB_USERNAME }}/yandexmusicdiscordbot
steps:
- name: Checkout repository
uses: actions/checkout@v4
- name: Set up Docker Buildx
uses: docker/setup-buildx-action@v2
- name: Log in to Docker Hub
uses: docker/login-action@v2
with:
username: ${{ secrets.DOCKERHUB_USERNAME }}
password: ${{ secrets.DOCKERHUB_TOKEN }}
- name: Get short SHA
id: vars
run: echo "short_sha=${GITHUB_SHA:0:7}" >> $GITHUB_OUTPUT
- name: Build and push Docker image
uses: docker/build-push-action@v5
with:
context: .
push: true
tags: ${{ env.IMAGE_NAME }}:${{ github.ref_name }}-${{ steps.vars.outputs.short_sha }}
- name: Set the latest tag for the main branch
if: github.ref == 'refs/heads/main'
run: |
docker pull $IMAGE_NAME:${{ github.ref_name }}-${{ steps.vars.outputs.short_sha }}
docker tag $IMAGE_NAME:${{ github.ref_name }}-${{ steps.vars.outputs.short_sha }} $IMAGE_NAME:latest
docker push $IMAGE_NAME:latest
- name: Set the latest tag for the dev branch
if: github.ref == 'refs/heads/dev'
run: |
docker pull $IMAGE_NAME:${{ github.ref_name }}-${{ steps.vars.outputs.short_sha }}
docker tag $IMAGE_NAME:${{ github.ref_name }}-${{ steps.vars.outputs.short_sha }} $IMAGE_NAME:latest-dev
docker push $IMAGE_NAME:latest-dev

14
Dockerfile Normal file
View File

@@ -0,0 +1,14 @@
FROM python:3.13-slim
RUN apt-get update && apt-get install -y ffmpeg && apt-get clean && rm -rf /var/lib/apt/lists/*
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY MusicBot /app/MusicBot
ENV PYTHONPATH=/app
CMD ["python", "./MusicBot/main.py"]

136
MusicBot/cogs/general.py
View File

@@ -9,8 +9,8 @@ from yandex_music.exceptions import UnauthorizedError
from yandex_music import ClientAsync as YMClient
from MusicBot.ui import ListenView
from MusicBot.database import BaseUsersDatabase, BaseGuildsDatabase
from MusicBot.cogs.utils import generate_item_embed
from MusicBot.database import BaseUsersDatabase
from MusicBot.cogs.utils import BaseBot, generate_item_embed
users_db = BaseUsersDatabase()
@@ -22,8 +22,7 @@ async def get_search_suggestions(ctx: discord.AutocompleteContext) -> list[str]:
return []
uid = ctx.interaction.user.id
token = await users_db.get_ym_token(uid)
if not token:
if not (token := await users_db.get_ym_token(uid)):
logging.info(f"[GENERAL] User {uid} has no token")
return []
@@ -33,15 +32,13 @@ async def get_search_suggestions(ctx: discord.AutocompleteContext) -> list[str]:
logging.info(f"[GENERAL] User {uid} provided invalid token")
return []
content_type = ctx.options['тип']
search = await client.search(ctx.value)
if not search:
if not (search := await client.search(ctx.value)):
logging.warning(f"[GENERAL] Failed to search for '{ctx.value}' for user {uid}")
return []
logging.debug(f"[GENERAL] Searching for '{ctx.value}' for user {uid}")
if content_type not in ('Трек', 'Альбом', 'Артист', 'Плейлист'):
if (content_type := ctx.options['тип']) not in ('Трек', 'Альбом', 'Артист', 'Плейлист'):
logging.error(f"[GENERAL] Invalid content type '{content_type}' for user {uid}")
return []
@@ -64,8 +61,7 @@ async def get_user_playlists_suggestions(ctx: discord.AutocompleteContext) -> li
return []
uid = ctx.interaction.user.id
token = await users_db.get_ym_token(uid)
if not token:
if not (token := await users_db.get_ym_token(uid)):
logging.info(f"[GENERAL] User {uid} has no token")
return []
@@ -84,12 +80,10 @@ async def get_user_playlists_suggestions(ctx: discord.AutocompleteContext) -> li
return [playlist.title for playlist in playlists_list if playlist.title and ctx.value in playlist.title][:100]
class General(Cog):
class General(Cog, BaseBot):
def __init__(self, bot: discord.Bot):
self.bot = bot
self.db = BaseGuildsDatabase()
self.users_db = users_db
BaseBot.__init__(self, bot)
account = discord.SlashCommandGroup("account", "Команды, связанные с аккаунтом.")
@@ -168,75 +162,81 @@ class General(Cog):
"Запустить станцию. Без уточнения станции, запускает Мою Волну.\n```/voice vibe <название станции>```"
)
else:
await ctx.respond('❌ Неизвестная команда.', delete_after=15, ephemeral=True)
await self.respond(ctx, "error", "Неизвестная команда.", delete_after=15, ephemeral=True)
return
await ctx.respond(embed=embed, ephemeral=True)
@account.command(description="Ввести токен Яндекс Музыки.")
@discord.option("token", type=discord.SlashCommandOptionType.string, description="Токен.")
@discord.option("token", type=discord.SlashCommandOptionType.string, description="Токен для доступа к API Яндекс Музыки.")
async def login(self, ctx: discord.ApplicationContext, token: str) -> None:
logging.info(f"[GENERAL] Login command invoked by user {ctx.author.id} in guild {ctx.guild_id}")
try:
client = await YMClient(token).init()
except UnauthorizedError:
logging.info(f"[GENERAL] Invalid token provided by user {ctx.author.id}")
await ctx.respond('❌ Недействительный токен.', delete_after=15, ephemeral=True)
await self.respond(ctx, "error", "Недействительный токен.", delete_after=15, ephemeral=True)
return
if not client.me or not client.me.account:
logging.warning(f"[GENERAL] Failed to get user info for user {ctx.author.id}")
await ctx.respond('❌ Не удалось получить информацию о пользователе.', delete_after=15, ephemeral=True)
await self.respond(ctx, "error", "Не удалось получить информацию о пользователе.", delete_after=15, ephemeral=True)
return
await self.users_db.update(ctx.author.id, {'ym_token': token})
await ctx.respond(f'✅ Привет, {client.me.account.first_name}!', delete_after=15, ephemeral=True)
await self.respond(ctx, "success", f"Привет, {client.me.account.first_name}!", delete_after=15, ephemeral=True)
self._ym_clients[token] = client
logging.info(f"[GENERAL] User {ctx.author.id} logged in successfully")
@account.command(description="Удалить токен из базы данных бота.")
async def remove(self, ctx: discord.ApplicationContext) -> None:
logging.info(f"[GENERAL] Remove command invoked by user {ctx.author.id} in guild {ctx.guild_id}")
if not await self.users_db.get_ym_token(ctx.user.id):
if not (token := await self.users_db.get_ym_token(ctx.user.id)):
logging.info(f"[GENERAL] No token found for user {ctx.author.id}")
await ctx.respond('❌ Токен не указан.', delete_after=15, ephemeral=True)
await self.respond(ctx, "error", "Токен не указан.", delete_after=15, ephemeral=True)
return
if token in self._ym_clients:
del self._ym_clients[token]
await self.users_db.update(ctx.user.id, {'ym_token': None})
await ctx.respond(f'✅ Токен был удалён.', delete_after=15, ephemeral=True)
logging.info(f"[GENERAL] Token removed for user {ctx.author.id}")
await self.respond(ctx, "success", "Токен был удалён.", delete_after=15, ephemeral=True)
@account.command(description="Получить плейлист «Мне нравится»")
async def likes(self, ctx: discord.ApplicationContext) -> None:
logging.info(f"[GENERAL] Likes command invoked by user {ctx.author.id} in guild {ctx.guild_id}")
token = await self.users_db.get_ym_token(ctx.user.id)
if not token:
logging.info(f"[GENERAL] No token found for user {ctx.user.id}")
await ctx.respond("❌ Укажите токен через /account login.", delete_after=15, ephemeral=True)
guild = await self.db.get_guild(ctx.guild_id, projection={'single_token_uid'})
if guild['single_token_uid'] and ctx.author.id != guild['single_token_uid']:
await self.respond(ctx, "error", "Только владелец токена может делиться личными плейлистами.", delete_after=15, ephemeral=True)
return
try:
client = await YMClient(token).init()
except UnauthorizedError:
logging.info(f"[GENERAL] Invalid token for user {ctx.user.id}")
await ctx.respond('❌ Неверный токен. Укажите новый через /account login.', delete_after=15, ephemeral=True)
if not (client := await self.init_ym_client(ctx)):
return
try:
likes = await client.users_likes_tracks()
except UnauthorizedError:
logging.warning(f"[GENERAL] Unknown token error for user {ctx.user.id}")
await ctx.respond("❌ Произошла неизвестная ошибка при попытке получения лайков. Пожалуйста, сообщите об этом разработчику.", delete_after=15, ephemeral=True)
await self.respond(
ctx, "error",
"Произошла неизвестная ошибка при попытке получения лайков. Пожалуйста, сообщите об этом разработчику.",
delete_after=15, ephemeral=True
)
return
if likes is None:
logging.info(f"[GENERAL] Failed to fetch likes for user {ctx.user.id}")
await ctx.respond('❌ Что-то пошло не так. Повторите попытку позже.', delete_after=15, ephemeral=True)
await self.respond(ctx, "error", "Что-то пошло не так. Повторите попытку позже.", delete_after=15, ephemeral=True)
return
elif not likes:
logging.info(f"[GENERAL] Empty likes for user {ctx.user.id}")
await ctx.respond('❌ У вас нет треков в плейлисте «Мне нравится».', delete_after=15, ephemeral=True)
await self.respond(ctx, "error", "У вас нет треков в плейлисте «Мне нравится».", delete_after=15, ephemeral=True)
return
await ctx.defer() # Sometimes it takes a while to fetch all tracks, so we defer the response
@@ -262,33 +262,27 @@ class General(Cog):
# NOTE: Recommendations can be accessed by using /find, but it's more convenient to have it in separate command.
logging.debug(f"[GENERAL] Recommendations command invoked by user {ctx.user.id} in guild {ctx.guild_id} for type '{content_type}'")
token = await self.users_db.get_ym_token(ctx.user.id)
if not token:
await ctx.respond("❌ Укажите токен через /account login.", delete_after=15, ephemeral=True)
guild = await self.db.get_guild(ctx.guild_id, projection={'single_token_uid'})
if guild['single_token_uid'] and ctx.author.id != guild['single_token_uid']:
await self.respond(ctx, "error", "Только владелец токена может делиться личными плейлистами.", delete_after=15, ephemeral=True)
return
try:
client = await YMClient(token).init()
except UnauthorizedError:
logging.info(f"[GENERAL] User {ctx.user.id} provided invalid token")
await ctx.respond('❌ Неверный токен. Укажите новый через /account login.', delete_after=15, ephemeral=True)
if not (client := await self.init_ym_client(ctx)):
return
search = await client.search(content_type, type_='playlist')
if not search or not search.playlists:
logging.info(f"[GENERAL] Failed to fetch recommendations for user {ctx.user.id}")
await ctx.respond('❌ Что-то пошло не так. Повторите попытку позже.', delete_after=15, ephemeral=True)
await self.respond(ctx, "error", "Что-то пошло не так. Повторите попытку позже.", delete_after=15, ephemeral=True)
return
playlist = search.playlists.results[0]
if playlist is None:
if (playlist := search.playlists.results[0]) is None:
logging.info(f"[GENERAL] Failed to fetch recommendations for user {ctx.user.id}")
await ctx.respond('❌ Что-то пошло не так. Повторите попытку позже.', delete_after=15, ephemeral=True)
await self.respond(ctx, "error", "Что-то пошло не так. Повторите попытку позже.", delete_after=15, ephemeral=True)
tracks = await playlist.fetch_tracks_async()
if not tracks:
if not await playlist.fetch_tracks_async():
logging.info(f"[GENERAL] User {ctx.user.id} search for '{content_type}' returned no tracks")
await ctx.respond("❌ Пустой плейлист.", delete_after=15, ephemeral=True)
await self.respond(ctx, "error", "Пустой плейлист.", delete_after=15, ephemeral=True)
return
await ctx.respond(embed=await generate_item_embed(playlist), view=ListenView(playlist))
@@ -304,36 +298,29 @@ class General(Cog):
async def playlist(self, ctx: discord.ApplicationContext, name: str) -> None:
logging.info(f"[GENERAL] Playlist command invoked by user {ctx.user.id} in guild {ctx.guild_id}")
token = await self.users_db.get_ym_token(ctx.user.id)
if not token:
logging.info(f"[GENERAL] No token found for user {ctx.user.id}")
await ctx.respond("❌ Укажите токен через /account login.", delete_after=15, ephemeral=True)
guild = await self.db.get_guild(ctx.guild_id, projection={'single_token_uid'})
if guild['single_token_uid'] and ctx.author.id != guild['single_token_uid']:
await self.respond(ctx, "error", "олько владелец токена может делиться личными плейлистами.", delete_after=15, ephemeral=True)
return
try:
client = await YMClient(token).init()
except UnauthorizedError:
logging.info(f"[GENERAL] User {ctx.user.id} provided invalid token")
await ctx.respond('❌ Неверный токен. Укажите новый через /account login.', delete_after=15, ephemeral=True)
if not (client := await self.init_ym_client(ctx)):
return
try:
playlists = await client.users_playlists_list()
except UnauthorizedError:
logging.warning(f"[GENERAL] Unknown token error for user {ctx.user.id}")
await ctx.respond("❌ Произошла неизвестная ошибка при попытке получения плейлистов. Пожалуйста, сообщите об этом разработчику.", delete_after=15, ephemeral=True)
await self.respond(ctx, "error", "Произошла неизвестная ошибка при попытке получения плейлистов. Пожалуйста, сообщите об этом разработчику.", delete_after=15, ephemeral=True)
return
playlist = next((playlist for playlist in playlists if playlist.title == name), None)
if not playlist:
if not (playlist := next((playlist for playlist in playlists if playlist.title == name), None)):
logging.info(f"[GENERAL] User {ctx.user.id} playlist '{name}' not found")
await ctx.respond("❌ Плейлист не найден.", delete_after=15, ephemeral=True)
await self.respond(ctx, "error", "Плейлист не найден.", delete_after=15, ephemeral=True)
return
tracks = await playlist.fetch_tracks_async()
if not tracks:
if not await playlist.fetch_tracks_async():
logging.info(f"[GENERAL] User {ctx.user.id} playlist '{name}' is empty")
await ctx.respond("❌ Плейлист пуст.", delete_after=15, ephemeral=True)
await self.respond(ctx, "error", "Плейлист пуст.", delete_after=15, ephemeral=True)
return
await ctx.respond(embed=await generate_item_embed(playlist), view=ListenView(playlist))
@@ -361,23 +348,12 @@ class General(Cog):
) -> None:
logging.info(f"[GENERAL] Find command invoked by user {ctx.user.id} in guild {ctx.guild_id} for '{content_type}' with name '{name}'")
token = await self.users_db.get_ym_token(ctx.user.id)
if not token:
logging.info(f"[GENERAL] No token found for user {ctx.user.id}")
await ctx.respond("❌ Укажите токен через /account login.", delete_after=15, ephemeral=True)
if not (client := await self.init_ym_client(ctx)):
return
try:
client = await YMClient(token).init()
except UnauthorizedError:
logging.info(f"[GENERAL] User {ctx.user.id} provided invalid token")
await ctx.respond("❌ Недействительный токен. Если это не так, попробуйте ещё раз.", delete_after=15, ephemeral=True)
return
search_result = await client.search(name, nocorrect=True)
if not search_result:
if not (search_result := await client.search(name, nocorrect=True)):
logging.warning(f"Failed to search for '{name}' for user {ctx.user.id}")
await ctx.respond("❌ Что-то пошло не так. Повторите попытку позже.", delete_after=15, ephemeral=True)
await self.respond(ctx, "error", "Что-то пошло не так. Повторите попытку позже.", delete_after=15, ephemeral=True)
return
if content_type == 'Трек':
@@ -391,7 +367,7 @@ class General(Cog):
if not content:
logging.info(f"[GENERAL] User {ctx.user.id} search for '{name}' returned no results")
await ctx.respond("❌ По запросу ничего не найдено.", delete_after=15, ephemeral=True)
await self.respond(ctx, "error", "По запросу ничего не найдено.", delete_after=15, ephemeral=True)
return
result = content.results[0]

65
MusicBot/cogs/settings.py
View File

@@ -5,11 +5,12 @@ import discord
from discord.ext.commands import Cog
from MusicBot.database import BaseUsersDatabase, BaseGuildsDatabase
from MusicBot.cogs.utils import BaseBot
def setup(bot):
bot.add_cog(Settings(bot))
class Settings(Cog):
class Settings(Cog, BaseBot):
settings = discord.SlashCommandGroup("settings", "Команды для изменения настроек бота.")
@@ -21,62 +22,84 @@ class Settings(Cog):
@settings.command(name="show", description="Показать текущие настройки бота.")
async def show(self, ctx: discord.ApplicationContext) -> None:
if not ctx.guild_id:
logging.warning("[SETTINGS] Show command invoked without guild_id")
await ctx.respond("❌ Эта команда может быть использована только на сервере.", ephemeral=True)
logging.info("[SETTINGS] Show command invoked without guild_id")
await self.respond(ctx, "error", "Эта команда может быть использована только на сервере.", ephemeral=True)
return
guild = await self.db.get_guild(ctx.guild_id, projection={'allow_change_connect': 1, 'vote_switch_track': 1, 'vote_add': 1})
guild = await self.db.get_guild(ctx.guild_id, projection={
'allow_change_connect': 1, 'vote_switch_track': 1, 'vote_add': 1, 'use_single_token': 1
})
vote = "✅ - Переключение" if guild['vote_switch_track'] else "❌ - Переключение"
vote += "\n✅ - Добавление в очередь" if guild['vote_add'] else "\n❌ - Добавление в очередь"
connect = "\n✅ - Разрешено всем" if guild['allow_change_connect'] else "\n❌ - Только для участникам с правами управления каналом"
token = "🔐 - Используется токен пользователя, запустившего бота" if guild['use_single_token'] else "🔒 - Используется личный токен пользователя"
embed = discord.Embed(title="Настройки бота", color=0xfed42b)
embed.set_author(name='YandexMusic', icon_url="https://github.com/Lemon4ksan/YandexMusicDiscordBot/blob/main/assets/Logo.png?raw=true")
embed.add_field(name="__Голосование__", value=vote, inline=False)
embed.add_field(name="__Подключение/Отключение бота__", value=connect, inline=False)
embed.add_field(name="__Подключение/Отключение__", value=connect, inline=False)
embed.add_field(name="__Токен__", value=token, inline=False)
await ctx.respond(embed=embed, ephemeral=True)
@settings.command(name="toggle", description="Переключить параметр настроек.")
@settings.command(name="toggle", description="Переключить параметры основных настроек.")
@discord.option(
"параметр",
parameter_name="vote_type",
description="Тип голосования.",
type=discord.SlashCommandOptionType.string,
choices=['Переключение', 'Добавление в очередь', 'Добавление/Отключение бота']
choices=[
'Переключение треков без голосования для всех',
'Добавление в очередь без голосования для всех',
'Добавление/Отключение бота от канала для всех',
'Использовать токен запустившего пользователя для всех'
]
)
async def toggle(
self,
ctx: discord.ApplicationContext,
vote_type: Literal['Переключение', 'Добавление в очередь', 'Добавление/Отключение бота']
vote_type: Literal[
'Переключение треков без голосования для всех',
'Добавление в очередь без голосования для всех',
'Добавление/Отключение бота от канала для всех',
'Использовать токен запустившего пользователя для всех'
]
) -> None:
member = cast(discord.Member, ctx.author)
if not member.guild_permissions.manage_channels:
await ctx.respond("❌ У вас нет прав для выполнения этой команды.", delete_after=15, ephemeral=True)
return
if not ctx.guild_id:
logging.warning("[SETTINGS] Toggle command invoked without guild_id")
await ctx.respond("❌ Эта команда может быть использована только на сервере.", ephemeral=True)
logging.info("[SETTINGS] Toggle command invoked without guild_id")
await self.respond(ctx, "error", "Эта команда может быть использована только на сервере.", delete_after=15, ephemeral=True)
return
member = cast(discord.Member, ctx.user)
if not member.guild_permissions.manage_channels:
await self.respond(ctx, "error", "У вас нет прав для выполнения этой команды.", delete_after=15, ephemeral=True)
return
guild = await self.db.get_guild(ctx.guild_id, projection={
'vote_switch_track': 1, 'vote_add': 1, 'allow_change_connect': 1})
'vote_switch_track': 1, 'vote_add': 1, 'allow_change_connect': 1, 'use_single_token': 1
})
if vote_type == 'Переключение':
if vote_type == 'Переключение треков без голосования для всех':
await self.db.update(ctx.guild_id, {'vote_switch_track': not guild['vote_switch_track']})
response_message = "Голосование за переключение трека " + ("❌ выключено." if guild['vote_switch_track'] else "✅ включено.")
elif vote_type == 'Добавление в очередь':
elif vote_type == 'Добавление в очередь без голосования для всех':
await self.db.update(ctx.guild_id, {'vote_add': not guild['vote_add']})
response_message = "Голосование за добавление в очередь " + ("❌ выключено." if guild['vote_add'] else "✅ включено.")
elif vote_type == 'Добавление/Отключение бота':
elif vote_type == 'Добавление/Отключение бота от канала для всех':
await self.db.update(ctx.guild_id, {'allow_change_connect': not guild['allow_change_connect']})
response_message = f"Добавление/Отключение бота от канала теперь {'✅ разрешено' if not guild['allow_change_connect'] else '❌ запрещено'} участникам без прав управления каналом."
elif vote_type == 'Использовать токен запустившего пользователя для всех':
await self.db.update(ctx.guild_id, {'use_single_token': not guild['use_single_token']})
response_message = f"Использование единого токена для прослушивания теперь {'✅ включено' if not guild['use_single_token'] else '❌ выключено'}."
else:
response_message = "❌ Неизвестный тип голосования."
response_message = "Неизвестный тип настроек."
await ctx.respond(response_message, delete_after=15, ephemeral=True)
await self.respond(ctx, 'info', response_message, delete_after=30, ephemeral=True)

5
MusicBot/cogs/utils/__init__.py
View File

@@ -1,8 +1,9 @@
from .embeds import generate_item_embed
from .voice_extension import VoiceExtension, menu_views
from .voice_extension import VoiceExtension
from .base_bot import BaseBot
__all__ = [
"generate_item_embed",
"VoiceExtension",
"menu_views"
"BaseBot"
]

243
MusicBot/cogs/utils/base_bot.py Normal file
View File

@@ -0,0 +1,243 @@
import asyncio
import logging
from typing import Any, Literal, cast
import yandex_music.exceptions
from yandex_music import ClientAsync as YMClient
import discord
from discord import Interaction, ApplicationContext, RawReactionActionEvent, MISSING
from MusicBot.database import VoiceGuildsDatabase, BaseUsersDatabase
class BaseBot:
menu_views: dict[int, Any] = {} # Store menu views and delete them when needed to prevent memory leaks for after callbacks.
_ym_clients: dict[str, YMClient] = {} # Store YM clients to prevent creating new ones for each command.
def __init__(self, bot: discord.Bot | None) -> None:
self.bot = bot
self.db = VoiceGuildsDatabase()
self.users_db = BaseUsersDatabase()
async def init_ym_client(
self,
ctx: ApplicationContext | Interaction | RawReactionActionEvent,
token: str | None = None
) -> YMClient | None:
"""Initialize Yandex Music client. Return client on success. Return None if no token found and respond to the context.
Args:
ctx (ApplicationContext | Interaction | RawReactionActionEvent): Context.
token (str | None, optional): Token. Fetched from database if not provided. Defaults to None.
Returns:
(YMClient | None): Client or None.
"""
logging.debug("[BASE_BOT] Initializing Yandex Music client")
if not (token := await self.get_ym_token(ctx)):
logging.debug("[BASE_BOT] No token found")
await self.respond(ctx, "error", "Укажите токен через /account login.", delete_after=15, ephemeral=True)
return None
try:
if token in self._ym_clients:
client = self._ym_clients[token]
await client.account_status()
return client
client = await YMClient(token).init()
except yandex_music.exceptions.UnauthorizedError:
del self._ym_clients[token]
await self.respond(ctx, "error", "Недействительный токен Yandex Music.", ephemeral=True, delete_after=15)
return None
self._ym_clients[token] = client
return client
async def get_ym_token(self, ctx: ApplicationContext | Interaction | RawReactionActionEvent) -> str | None:
"""Get Yandex Music token from context. It's either individual or single."""
uid = ctx.user_id if isinstance(ctx, discord.RawReactionActionEvent) else ctx.user.id if ctx.user else None
if not ctx.guild_id or not uid:
logging.info("[VC_EXT] No guild id or user id found")
return None
guild = await self.db.get_guild(ctx.guild_id, projection={'single_token_uid': 1})
if guild['single_token_uid']:
return await self.users_db.get_ym_token(guild['single_token_uid'])
else:
return await self.users_db.get_ym_token(uid)
async def respond(
self,
ctx: ApplicationContext | Interaction | RawReactionActionEvent,
response_type: Literal['info', 'success', 'error'] | None = None,
content: str | None = None,
*,
delete_after: float | None = None,
ephemeral: bool = False,
embed: discord.Embed | None = None,
view: discord.ui.View | None = None,
**kwargs: Any
) -> discord.Interaction | discord.WebhookMessage | discord.Message | None:
"""Send response message based on context type. `self.bot` must be set in order to use RawReactionActionEvent context type.
RawReactionActionEvent can't be ephemeral.
Args:
ctx (ApplicationContext | Interaction | RawReactionActionEvent): Context.
content (str): Message content to send. If embed is not set, used as description.
response_type (Literal['info', 'success', 'error'] | None, optional): Response type. Applies if embed is not specified.
delete_after (float, optional): Time after which the message will be deleted. Defaults to None.
ephemeral (bool, optional): Whether the message is ephemeral. Defaults to False.
embed (discord.Embed, optional): Discord embed. Defaults to None.
view (discord.ui.View, optional): Discord view. Defaults to None.
kwargs: Additional arguments for embed generation. Applies if embed is not specified.
Returns:
(discord.InteractionMessage | discord.WebhookMessage | discord.Message | None): Message or None. Type depends on the context type.
"""
if not embed and response_type:
if content:
kwargs['description'] = content
embed = self.generate_response_embed(ctx, response_type, **kwargs)
content = None
if not isinstance(ctx, RawReactionActionEvent) and not view and ctx.response.is_done():
view = MISSING
if not isinstance(ctx, RawReactionActionEvent):
return await ctx.respond(content, delete_after=delete_after, ephemeral=ephemeral, view=view, embed=embed)
elif self.bot:
channel = self.bot.get_channel(ctx.channel_id)
if isinstance(channel, (discord.abc.Messageable)):
return await channel.send(content, delete_after=delete_after, view=view, embed=embed) # type: ignore
return None
async def get_message_by_id(
self,
ctx: ApplicationContext | Interaction | RawReactionActionEvent,
message_id: int
) -> discord.Message | None:
"""Get message by id based on context type. self.bot must be set in order to use RawReactionActionEvent context type.
Args:
ctx (ApplicationContext | Interaction | RawReactionActionEvent): Context.
message_id (int): Message id.
Returns:
(discord.Message | None): Message or None.
Raises:
ValueError: Bot instance is not set.
discord.DiscordException: Failed to get message.
"""
try:
if isinstance(ctx, ApplicationContext):
return await ctx.fetch_message(message_id)
elif isinstance(ctx, Interaction):
return ctx.client.get_message(message_id)
elif not self.bot:
raise ValueError("Bot instance is not set.")
else:
return self.bot.get_message(message_id)
except discord.DiscordException as e:
logging.debug(f"[BASE_BOT] Failed to get message: {e}")
raise
async def get_discord_user_by_id(self, ctx: ApplicationContext | Interaction | RawReactionActionEvent, user_id: int) -> discord.User | None:
if isinstance(ctx, ApplicationContext) and ctx.user:
logging.debug(f"[BASE_BOT] Getting user {user_id} from ApplicationContext")
return await ctx.bot.fetch_user(user_id)
elif isinstance(ctx, Interaction):
logging.debug(f"[BASE_BOT] Getting user {user_id} from Interaction")
return await ctx.client.fetch_user(user_id)
elif not self.bot:
raise ValueError("Bot instance is not available")
else:
logging.debug(f"[BASE_BOT] Getting user {user_id} from bot instance")
return await self.bot.fetch_user(user_id)
async def get_viber_id_from_ctx(self, ctx: ApplicationContext | Interaction | RawReactionActionEvent) -> int | None:
if not ctx.guild_id:
logging.warning("[BASE_BOT] Guild not found")
return None
guild = await self.db.get_guild(ctx.guild_id, projection={'current_viber_id': 1})
if guild['current_viber_id']:
return guild['current_viber_id']
return ctx.user_id if isinstance(ctx, discord.RawReactionActionEvent) else ctx.user.id if ctx.user else None
async def init_menu_view(self, ctx: ApplicationContext | Interaction | RawReactionActionEvent, gid: int, *, disable: bool = False) -> None:
from MusicBot.ui import MenuView
self.menu_views[gid] = await MenuView(ctx).init(disable=disable)
def generate_response_embed(
self,
ctx: ApplicationContext | Interaction | RawReactionActionEvent,
embed_type: Literal['info', 'success', 'error'] = 'info',
**kwargs: Any
) -> discord.Embed:
if isinstance(ctx, Interaction):
name = ctx.client.user.name if ctx.client.user else None
icon_url = ctx.client.user.avatar.url if ctx.client.user and ctx.client.user.avatar else None
elif isinstance(ctx, ApplicationContext):
name = ctx.bot.user.name if ctx.bot.user else None
icon_url = ctx.bot.user.avatar.url if ctx.bot.user and ctx.bot.user.avatar else None
elif self.bot:
name = self.bot.user.name if self.bot.user else None
icon_url = self.bot.user.avatar.url if self.bot.user and self.bot.user.avatar else None
else:
name = icon_url = None
if not name:
name = 'YandexMusic'
if not icon_url:
icon_url="https://github.com/Lemon4ksan/YandexMusicDiscordBot/blob/main/assets/Logo.png?raw=true"
embed = discord.Embed(**kwargs)
embed.set_author(name=name, icon_url=icon_url)
if embed_type == 'info':
embed.color = 0xfed42b
elif embed_type == 'success':
embed.set_author(name = "✅ Успех")
embed.color = discord.Color.green()
else:
embed.set_author(name = "❌ Ошибка")
embed.color = discord.Color.red()
return embed
def get_current_event_loop(self, ctx: ApplicationContext | Interaction | RawReactionActionEvent) -> asyncio.AbstractEventLoop:
"""Get the current event loop. If the context is a RawReactionActionEvent, get the loop from the self.bot instance.
Args:
ctx (ApplicationContext | Interaction | RawReactionActionEvent): Context.
Raises:
TypeError: If the context is not a RawReactionActionEvent, ApplicationContext or Interaction.
ValueError: If the context is a RawReactionActionEvent and the bot is not set.
Returns:
asyncio.AbstractEventLoop: Current event loop.
"""
if isinstance(ctx, Interaction):
return ctx.client.loop
elif isinstance(ctx, ApplicationContext):
return ctx.bot.loop
elif isinstance(ctx, RawReactionActionEvent):
if not self.bot:
raise ValueError("Bot is not set.")
return self.bot.loop
else:
raise TypeError(f"Invalid context type: '{type(ctx).__name__}'.")

672
MusicBot/cogs/utils/voice_extension.py
View File

File diff suppressed because it is too large Load Diff

272
MusicBot/cogs/voice.py
View File

@@ -7,8 +7,8 @@ from discord.ext.commands import Cog
from yandex_music import ClientAsync as YMClient
from yandex_music.exceptions import UnauthorizedError
from MusicBot.cogs.utils import VoiceExtension
from MusicBot.database import BaseUsersDatabase
from MusicBot.cogs.utils import VoiceExtension, menu_views
from MusicBot.ui import QueueView, generate_queue_embed
def setup(bot: discord.Bot):
@@ -20,8 +20,7 @@ async def get_vibe_stations_suggestions(ctx: discord.AutocompleteContext) -> lis
if not ctx.interaction.user or not ctx.value or len(ctx.value) < 2:
return []
token = await users_db.get_ym_token(ctx.interaction.user.id)
if not token:
if not (token := await users_db.get_ym_token(ctx.interaction.user.id)):
logging.info(f"[GENERAL] User {ctx.interaction.user.id} has no token")
return []
@@ -46,63 +45,71 @@ class Voice(Cog, VoiceExtension):
@Cog.listener()
async def on_voice_state_update(self, member: discord.Member, before: discord.VoiceState, after: discord.VoiceState) -> None:
gid = member.guild.id
guild = await self.db.get_guild(gid, projection={'current_menu': 1})
guild = await self.db.get_guild(member.guild.id, projection={'current_menu': 1})
channel = after.channel or before.channel
if not channel:
logging.warning(f"[VOICE] No channel found for member {member.id}")
if not after.channel or not before.channel:
logging.debug(f"[VOICE] No channel found for member {member.id}")
return
vc = cast(discord.VoiceClient | None, discord.utils.get(self.typed_bot.voice_clients, guild=await self.typed_bot.fetch_guild(gid)))
vc = cast(
discord.VoiceClient | None,
discord.utils.get(
self.typed_bot.voice_clients,
guild=await self.typed_bot.fetch_guild(member.guild.id)
)
)
for member in channel.members:
if not vc:
logging.info(f"[VOICE] No voice client found for guild {member.guild.id}")
return
for member in set(before.channel.members + after.channel.members):
if member.id == self.typed_bot.user.id: # type: ignore # should be logged in
logging.info(f"[VOICE] Voice state update for member {member.id} in guild {member.guild.id}")
break
else:
logging.debug(f"[VOICE] Bot is not in the channel {channel.id}")
logging.debug(f"[VOICE] Bot is not in the channel {after.channel.id}")
return
if not vc:
logging.info(f"[VOICE] No voice client found for guild {gid}")
return
if len(after.channel.members) == 1:
logging.info(f"[VOICE] Clearing history and stopping playback for guild {member.guild.id}")
if len(channel.members) == 1:
logging.info(f"[VOICE] Clearing history and stopping playback for guild {gid}")
if member.guild.id in menu_views:
menu_views[member.guild.id].stop()
del menu_views[member.guild.id]
if member.guild.id in self.menu_views:
self.menu_views[member.guild.id].stop()
del self.menu_views[member.guild.id]
if guild['current_menu']:
message = self.typed_bot.get_message(guild['current_menu'])
if message:
if (message := self.typed_bot.get_message(guild['current_menu'])):
await message.delete()
await self.db.update(gid, {
await self.db.update(member.guild.id, {
'previous_tracks': [], 'next_tracks': [], 'votes': {},
'current_track': None, 'current_menu': None, 'vibing': False,
'repeat': False, 'shuffle': False, 'is_stopped': True
})
vc.stop()
if member.guild.id in menu_views:
menu_views[member.guild.id].stop()
del menu_views[member.guild.id]
if member.guild.id in self.menu_views:
self.menu_views[member.guild.id].stop()
del self.menu_views[member.guild.id]
@Cog.listener()
async def on_raw_reaction_add(self, payload: discord.RawReactionActionEvent) -> None:
logging.debug(f"[VOICE] Reaction added by user {payload.user_id} in channel {payload.channel_id}")
if not self.typed_bot.user or not payload.member:
return
bot_id = self.typed_bot.user.id
if payload.user_id == bot_id:
if not payload.guild_id:
logging.info(f"[VOICE] No guild id in reaction payload")
return
channel = cast(discord.VoiceChannel, self.typed_bot.get_channel(payload.channel_id))
if not channel:
if payload.user_id == self.typed_bot.user.id:
return
channel = self.typed_bot.get_channel(payload.channel_id)
if not isinstance(channel, discord.VoiceChannel):
logging.info(f"[VOICE] Channel {payload.channel_id} is not a voice channel")
return
try:
@@ -114,19 +121,17 @@ class Voice(Cog, VoiceExtension):
logging.info(f"[VOICE] Message {payload.message_id} not found in channel {payload.channel_id}")
return
if not message or message.author.id != bot_id:
if not message or message.author.id != self.typed_bot.user.id:
logging.info(f"[VOICE] Message {payload.message_id} is not a bot message")
return
if not await self.users_db.get_ym_token(payload.user_id):
guild = await self.db.get_guild(payload.guild_id)
if not guild['use_single_token'] and not (guild['single_token_uid'] or await self.users_db.get_ym_token(payload.user_id)):
await message.remove_reaction(payload.emoji, payload.member)
await channel.send("Для участия в голосовании необходимо авторизоваться через /account login.", delete_after=15)
await channel.send("❌ Для участия в голосовании необходимо авторизоваться через /account login.", delete_after=15)
return
guild_id = payload.guild_id
if not guild_id:
return
guild = await self.db.get_guild(guild_id)
votes = guild['votes']
if str(payload.message_id) not in votes:
@@ -147,7 +152,7 @@ class Voice(Cog, VoiceExtension):
if len(vote_data['positive_votes']) >= required_votes:
logging.info(f"[VOICE] Enough positive votes for message {payload.message_id}")
await message.delete()
await self.proccess_vote(payload, guild, channel, vote_data)
await self.proccess_vote(payload, guild, vote_data)
del votes[str(payload.message_id)]
elif len(vote_data['negative_votes']) >= required_votes:
@@ -156,29 +161,30 @@ class Voice(Cog, VoiceExtension):
await message.edit(content='Запрос был отклонён.', delete_after=15)
del votes[str(payload.message_id)]
await self.db.update(guild_id, {'votes': votes})
await self.db.update(payload.guild_id, {'votes': votes})
@Cog.listener()
async def on_raw_reaction_remove(self, payload: discord.RawReactionActionEvent) -> None:
logging.debug(f"[VOICE] Reaction removed by user {payload.user_id} in channel {payload.channel_id}")
if not self.typed_bot.user:
if not self.typed_bot.user or not payload.member:
return
guild_id = payload.guild_id
if not guild_id:
if not payload.guild_id:
return
guild = await self.db.get_guild(guild_id, projection={'votes': 1})
channel = self.typed_bot.get_channel(payload.channel_id)
if not isinstance(channel, discord.VoiceChannel):
logging.info(f"[VOICE] Channel {payload.channel_id} is not a voice channel")
return
guild = await self.db.get_guild(payload.guild_id, projection={'votes': 1})
votes = guild['votes']
if str(payload.message_id) not in votes:
logging.info(f"[VOICE] Message {payload.message_id} not found in votes")
return
channel = cast(discord.VoiceChannel, self.typed_bot.get_channel(payload.channel_id))
if not channel:
return
try:
message = await channel.fetch_message(payload.message_id)
except discord.Forbidden:
@@ -199,13 +205,13 @@ class Voice(Cog, VoiceExtension):
logging.info(f"[VOICE] User {payload.user_id} removed negative vote for message {payload.message_id}")
del vote_data['negative_votes'][payload.user_id]
await self.db.update(guild_id, {'votes': votes})
await self.db.update(payload.guild_id, {'votes': votes})
@voice.command(name="menu", description="Создать или обновить меню проигрывателя.")
async def menu(self, ctx: discord.ApplicationContext) -> None:
logging.info(f"[VOICE] Menu command invoked by user {ctx.author.id} in guild {ctx.guild_id}")
if await self.voice_check(ctx) and not await self.send_menu_message(ctx):
await ctx.respond("❌ Не удалось создать меню.", ephemeral=True)
await self.respond(ctx, "error", "Не удалось создать меню.", ephemeral=True)
@voice.command(name="join", description="Подключиться к голосовому каналу, в котором вы сейчас находитесь.")
async def join(self, ctx: discord.ApplicationContext) -> None:
@@ -213,37 +219,59 @@ class Voice(Cog, VoiceExtension):
if not ctx.guild_id:
logging.warning("[VOICE] Join command invoked without guild_id")
await ctx.respond("❌ Эта команда может быть использована только на сервере.", ephemeral=True)
await self.respond(ctx, "error", "Эта команда может быть использована только на сервере.", ephemeral=True)
return
if ctx.author.id not in ctx.channel.voice_states:
logging.debug("[VC_EXT] User is not connected to the voice channel")
await self.respond(ctx, "error", "Вы должны находиться в голосовом канале.", delete_after=15, ephemeral=True)
return
member = cast(discord.Member, ctx.author)
guild = await self.db.get_guild(ctx.guild_id, projection={'allow_change_connect': 1})
guild = await self.db.get_guild(ctx.guild_id, projection={'allow_change_connect': 1, 'use_single_token': 1})
if guild['use_single_token'] and not await self.users_db.get_ym_token(ctx.author.id):
await self.respond(
ctx, "error",
"У вас нет токена Яндекс Музыки. Используйте команду /account login для установки токена, " \
"попросите участника с токеном запустить бота или отключите использование общего токена в настройках сервера.",
delete_after=15, ephemeral=True
)
return
await ctx.defer(ephemeral=True)
if not member.guild_permissions.manage_channels and not guild['allow_change_connect']:
response_message = "❌ У вас нет прав для выполнения этой команды."
response_message = ("error", "У вас нет прав для выполнения этой команды.")
elif isinstance(ctx.channel, discord.VoiceChannel):
try:
await ctx.channel.connect()
except TimeoutError:
response_message = "❌ Не удалось подключиться к голосовому каналу."
response_message = ("error", "Не удалось подключиться к голосовому каналу.")
except discord.ClientException:
response_message = "❌ Бот уже находится в голосовом канале. Выключите его с помощью команды /voice leave."
response_message = ("error", "Бот уже находится в голосовом канале.\nВыключите его с помощью команды /voice leave.")
except discord.DiscordException as e:
logging.error(f"[VOICE] DiscordException: {e}")
response_message = ("error", "Произошла неизвестная ошибка при подключении к голосовому каналу.")
else:
response_message = "✅ Подключение успешно!"
response_message = ("success", "Подключение успешно!")
if guild['use_single_token']:
response_message = ("success", "Подключение успешно! Ваш токен будет использован для всех операций с музыкой на этом сервере.")
await self.db.update(ctx.guild_id, {'single_token_uid': ctx.author.id})
else:
response_message = "❌ Вы должны отправить команду в чате голосового канала."
response_message = ("error", "Вы должны отправить команду в чате голосового канала.")
logging.info(f"[VOICE] Join command response: {response_message}")
await ctx.respond(response_message, delete_after=15, ephemeral=True)
await self.respond(ctx, *response_message, delete_after=15, ephemeral=True)
@voice.command(description="Заставить бота покинуть голосовой канал.")
async def leave(self, ctx: discord.ApplicationContext) -> None:
logging.info(f"[VOICE] Leave command invoked by user {ctx.author.id} in guild {ctx.guild_id}")
if not ctx.guild_id:
logging.warning("[VOICE] Leave command invoked without guild_id")
await ctx.respond("❌ Эта команда может быть использована только на сервере.", ephemeral=True)
logging.info("[VOICE] Leave command invoked without guild_id")
await self.respond(ctx, "error", "Эта команда может быть использована только на сервере.", ephemeral=True)
return
member = cast(discord.Member, ctx.author)
@@ -251,29 +279,30 @@ class Voice(Cog, VoiceExtension):
if not member.guild_permissions.manage_channels and not guild['allow_change_connect']:
logging.info(f"[VOICE] User {ctx.author.id} does not have permissions to execute leave command in guild {ctx.guild_id}")
await ctx.respond("❌ У вас нет прав для выполнения этой команды.", delete_after=15, ephemeral=True)
await self.respond(ctx, "error", "У вас нет прав для выполнения этой команды.", delete_after=15, ephemeral=True)
return
if not await self.voice_check(ctx):
return
if (vc := await self.get_voice_client(ctx)) and await self.voice_check(ctx) and vc.is_connected:
res = await self.stop_playing(ctx, vc=vc, full=True)
if not res:
await ctx.respond("❌ Не удалось отключиться.", delete_after=15, ephemeral=True)
return
if not (vc := await self.get_voice_client(ctx)) or not vc.is_connected:
logging.info(f"[VOICE] Voice client is not connected in guild {ctx.guild_id}")
await self.respond(ctx, "error", "Бот не подключен к голосовому каналу.", delete_after=15, ephemeral=True)
return
await vc.disconnect(force=True)
await ctx.respond("✅ Отключение успешно!", delete_after=15, ephemeral=True)
logging.info(f"[VOICE] Successfully disconnected from voice channel in guild {ctx.guild_id}")
else:
await ctx.respond("❌ Бот не подключен к голосовому каналу.", delete_after=15, ephemeral=True)
if not await self.stop_playing(ctx, vc=vc, full=True):
await self.respond(ctx, "error", "Не удалось отключиться.", delete_after=15, ephemeral=True)
return
await vc.disconnect(force=True)
logging.info(f"[VOICE] Successfully disconnected from voice channel in guild {ctx.guild_id}")
await self.db.update(ctx.guild_id, {'single_token_uid': None})
await self.respond(ctx, "success", "Отключение успешно!", delete_after=15, ephemeral=True)
@queue.command(description="Очистить очередь треков и историю прослушивания.")
async def clear(self, ctx: discord.ApplicationContext) -> None:
logging.info(f"[VOICE] Clear queue command invoked by user {ctx.author.id} in guild {ctx.guild_id}")
if not ctx.guild_id:
logging.warning("[VOICE] Clear command invoked without guild_id")
await ctx.respond("❌ Эта команда может быть использована только на сервере.", ephemeral=True)
return
if not await self.voice_check(ctx):
return
@@ -283,14 +312,14 @@ class Voice(Cog, VoiceExtension):
if len(channel.members) > 2 and not member.guild_permissions.manage_channels:
logging.info(f"Starting vote for clearing queue in guild {ctx.guild_id}")
response_message = f"{member.mention} хочет очистить историю прослушивания и очередь треков.\n\n Выполнить действие?."
message = cast(discord.Interaction, await ctx.respond(response_message, delete_after=60))
message = cast(discord.Interaction, await self.respond(ctx, "info", response_message, delete_after=60))
response = await message.original_response()
await response.add_reaction('✅')
await response.add_reaction('❌')
await self.db.update_vote(
ctx.guild_id,
response.id,
@@ -305,40 +334,28 @@ class Voice(Cog, VoiceExtension):
return
await self.db.update(ctx.guild_id, {'previous_tracks': [], 'next_tracks': []})
await ctx.respond("✅ Очередь и история сброшены.", delete_after=15, ephemeral=True)
await self.respond(ctx, "success", "Очередь и история сброшены.", delete_after=15, ephemeral=True)
logging.info(f"[VOICE] Queue and history cleared in guild {ctx.guild_id}")
@queue.command(description="Получить очередь треков.")
async def get(self, ctx: discord.ApplicationContext) -> None:
logging.info(f"[VOICE] Get queue command invoked by user {ctx.author.id} in guild {ctx.guild_id}")
if not ctx.guild_id:
logging.warning("[VOICE] Get command invoked without guild_id")
await ctx.respond("❌ Эта команда может быть использована только на сервере.", ephemeral=True)
return
if not await self.voice_check(ctx):
return
await self.users_db.update(ctx.user.id, {'queue_page': 0})
tracks = await self.db.get_tracks_list(ctx.guild_id, 'next')
if len(tracks) == 0:
await ctx.respond("❌ Очередь пуста.", ephemeral=True)
await self.respond(ctx, "error", "Очередь прослушивания пуста.", delete_after=15, ephemeral=True)
return
embed = generate_queue_embed(0, tracks)
await ctx.respond(embed=embed, view=await QueueView(ctx).init(), ephemeral=True)
await ctx.respond(embed=generate_queue_embed(0, tracks), view=QueueView(ctx, tracks), ephemeral=True)
logging.info(f"[VOICE] Queue embed sent to user {ctx.author.id} in guild {ctx.guild_id}")
@voice.command(description="Прервать проигрывание, удалить историю, очередь и текущий плеер.")
async def stop(self, ctx: discord.ApplicationContext) -> None:
logging.info(f"[VOICE] Stop command invoked by user {ctx.author.id} in guild {ctx.guild_id}")
if not ctx.guild_id:
logging.warning("[VOICE] Stop command invoked without guild_id")
await ctx.respond("❌ Эта команда может быть использована только на сервере.", ephemeral=True)
return
if not await self.voice_check(ctx):
return
@@ -350,7 +367,7 @@ class Voice(Cog, VoiceExtension):
logging.info(f"Starting vote for stopping playback in guild {ctx.guild_id}")
response_message = f"{member.mention} хочет полностью остановить проигрывание.\n\n Выполнить действие?."
message = cast(discord.Interaction, await ctx.respond(response_message, delete_after=60))
message = cast(discord.Interaction, await self.respond(ctx, "info", response_message, delete_after=60))
response = await message.original_response()
await response.add_reaction('✅')
@@ -372,9 +389,9 @@ class Voice(Cog, VoiceExtension):
await ctx.defer(ephemeral=True)
res = await self.stop_playing(ctx, full=True)
if res:
await ctx.respond("✅ Воспроизведение остановлено.", delete_after=15, ephemeral=True)
await self.respond(ctx, "success", "Воспроизведение остановлено.", delete_after=15, ephemeral=True)
else:
await ctx.respond("❌ Произошла ошибка при остановке воспроизведения.", delete_after=15, ephemeral=True)
await self.respond(ctx, "error", "Произошла ошибка при остановке воспроизведения.", delete_after=15, ephemeral=True)
@voice.command(description="Запустить Мою Волну.")
@discord.option(
@@ -387,36 +404,24 @@ class Voice(Cog, VoiceExtension):
)
async def vibe(self, ctx: discord.ApplicationContext, name: str | None = None) -> None:
logging.info(f"[VOICE] Vibe (user) command invoked by user {ctx.user.id} in guild {ctx.guild_id}")
if not await self.voice_check(ctx):
return
if not ctx.guild_id:
logging.warning("[VOICE] Vibe command invoked without guild_id")
await ctx.respond("❌ Эта команда может быть использована только на сервере.", ephemeral=True)
return
guild = await self.db.get_guild(ctx.guild_id, projection={'current_menu': 1, 'vibing': 1})
if guild['vibing']:
logging.info(f"[VOICE] Action declined: vibing is already enabled in guild {ctx.guild_id}")
await ctx.respond("❌ Моя Волна уже включена. Используйте /voice stop, чтобы остановить воспроизведение.", delete_after=15, ephemeral=True)
await self.respond(ctx, "error", "Моя Волна уже включена. Используйте /voice stop, чтобы остановить воспроизведение.", delete_after=15, ephemeral=True)
return
await ctx.defer(invisible=False)
if name:
token = await users_db.get_ym_token(ctx.user.id)
if not token:
logging.info(f"[GENERAL] User {ctx.user.id} has no token")
if not (client := await self.init_ym_client(ctx)):
return
try:
client = await YMClient(token).init()
except UnauthorizedError:
logging.info(f"[GENERAL] User {ctx.user.id} provided invalid token")
return
stations = await client.rotor_stations_list()
for content in stations:
for content in (await client.rotor_stations_list()):
if content.station and content.station.name == name and content.ad_params:
break
else:
@@ -424,40 +429,42 @@ class Voice(Cog, VoiceExtension):
if not content:
logging.debug(f"[VOICE] Station {name} not found")
await ctx.respond("❌ Станция не найдена.", delete_after=15, ephemeral=True)
await self.respond(ctx, "error", "Станция не найдена.", delete_after=15, ephemeral=True)
return
_type, _id = content.ad_params.other_params.split(':') if content.ad_params else (None, None)
vibe_type, vibe_id = content.ad_params.other_params.split(':') if content.ad_params else (None, None)
if not _type or not _id:
if not vibe_type or not vibe_id:
logging.debug(f"[VOICE] Station {name} has no ad params")
await ctx.respond("❌ Станция не найдена.", delete_after=15, ephemeral=True)
await self.respond(ctx, "error", "Станция не найдена.", delete_after=15, ephemeral=True)
return
else:
_type, _id = 'user', 'onyourwave'
vibe_type, vibe_id = 'user', 'onyourwave'
content = None
member = cast(discord.Member, ctx.author)
channel = cast(discord.VoiceChannel, ctx.channel)
await self.users_db.reset_vibe_settings(ctx.user.id)
if len(channel.members) > 2 and not member.guild_permissions.manage_channels:
logging.info(f"Starting vote for starting vibe in guild {ctx.guild_id}")
if _type == 'user' and _id == 'onyourwave':
if vibe_type == 'user' and vibe_id == 'onyourwave':
station = "Моя Волна"
elif content and content.station:
station = content.station.name
else:
logging.warning(f"[VOICE] Station {name} not found")
await ctx.respond("❌ Станция не найдена.", delete_after=15, ephemeral=True)
await self.respond(ctx, "error", "Станция не найдена.", delete_after=15, ephemeral=True)
return
response_message = f"{member.mention} хочет запустить станцию **{station}**.\n\n Выполнить действие?"
message = cast(discord.WebhookMessage, await ctx.respond(response_message))
message = cast(discord.WebhookMessage, await self.respond(ctx, "info", response_message, delete_after=60))
await message.add_reaction('✅')
await message.add_reaction('❌')
await self.db.update_vote(
ctx.guild_id,
message.id,
@@ -466,20 +473,19 @@ class Voice(Cog, VoiceExtension):
'negative_votes': list(),
'total_members': len(channel.members),
'action': 'vibe_station',
'vote_content': [_type, _id, ctx.user.id]
'vote_content': [vibe_type, vibe_id, ctx.user.id]
}
)
return
if not await self.update_vibe(ctx, _type, _id):
await ctx.respond("❌ Операция не удалась. Возможно, у вес нет подписки на Яндекс Музыку.", delete_after=15, ephemeral=True)
if not await self.update_vibe(ctx, vibe_type, vibe_id):
await self.respond(ctx, "error", "Операция не удалась. Возможно, у вес нет подписки на Яндекс Музыку.", delete_after=15, ephemeral=True)
return
if guild['current_menu']:
await ctx.respond("✅ Моя Волна включена.", delete_after=15, ephemeral=True)
await self.respond(ctx, "success", "Моя Волна включена.", delete_after=15, ephemeral=True)
elif not await self.send_menu_message(ctx, disable=True):
await ctx.respond("❌ Не удалось отправить меню. Попробуйте позже.", delete_after=15, ephemeral=True)
await self.respond(ctx, "error", "Не удалось отправить меню. Попробуйте позже.", delete_after=15, ephemeral=True)
next_track = await self.db.get_track(ctx.guild_id, 'next')
if next_track:
if (next_track := await self.db.get_track(ctx.guild_id, 'next')):
await self.play_track(ctx, next_track)

31
MusicBot/database/base.py
View File

@@ -1,3 +1,4 @@
import os
from typing import Iterable, Any, cast
from pymongo import AsyncMongoClient, ReturnDocument, UpdateOne
from pymongo.asynchronous.collection import AsyncCollection
@@ -6,7 +7,11 @@ from pymongo.results import UpdateResult
from .user import User, ExplicitUser
from .guild import Guild, ExplicitGuild, MessageVotes
client: AsyncMongoClient = AsyncMongoClient("mongodb://localhost:27017/")
mongo_server = os.getenv('MONGO_URI')
if not mongo_server:
raise ValueError('MONGO_URI environment variable is not set')
client: AsyncMongoClient = AsyncMongoClient(mongo_server)
db = client.YandexMusicBot
users: AsyncCollection[ExplicitUser] = db.users
@@ -15,9 +20,6 @@ guilds: AsyncCollection[ExplicitGuild] = db.guilds
class BaseUsersDatabase:
DEFAULT_USER = User(
ym_token=None,
playlists=[],
playlists_page=0,
queue_page=0,
vibe_batch_id=None,
vibe_type=None,
vibe_id=None,
@@ -65,10 +67,14 @@ class BaseUsersDatabase:
)
return cast(str | None, user.get('ym_token') if user else None)
async def add_playlist(self, uid: int, playlist_data: dict) -> UpdateResult:
return await users.update_one(
async def reset_vibe_settings(self, uid: int) -> None:
await users.update_one(
{'_id': uid},
{'$push': {'playlists': playlist_data}}
{'$set': {'vibe_settings': {
'mood': 'all',
'diversity': 'default',
'lang': 'any'
}}}
)
@@ -79,7 +85,6 @@ class BaseGuildsDatabase:
current_track=None,
current_menu=None,
is_stopped=True,
always_allow_menu=False,
allow_change_connect=True,
vote_switch_track=True,
vote_add=True,
@@ -87,7 +92,9 @@ class BaseGuildsDatabase:
repeat=False,
votes={},
vibing=False,
current_viber_id=None
current_viber_id=None,
use_single_token=False,
single_token_uid=None
)
async def update(self, gid: int, data: Guild | dict[str, Any]) -> UpdateResult:
@@ -125,9 +132,3 @@ class BaseGuildsDatabase:
{'_id': gid},
{'$set': {f'votes.{mid}': data}}
)
async def clear_queue(self, gid: int) -> UpdateResult:
return await guilds.update_one(
{'_id': gid},
{'$set': {'next_tracks': []}}
)

12
MusicBot/database/guild.py
View File

@@ -10,13 +10,12 @@ class MessageVotes(TypedDict):
]
vote_content: Any | None
class Guild(TypedDict, total=False):
class Guild(TypedDict, total=False): # Don't forget to change base.py if you add a new field
next_tracks: list[dict[str, Any]]
previous_tracks: list[dict[str, Any]]
current_track: dict[str, Any] | None
current_menu: int | None
is_stopped: bool
always_allow_menu: bool
is_stopped: bool # Prevents the `after` callback of play_track
allow_change_connect: bool
vote_switch_track: bool
vote_add: bool
@@ -25,6 +24,8 @@ class Guild(TypedDict, total=False):
votes: dict[str, MessageVotes]
vibing: bool
current_viber_id: int | None
use_single_token: bool
single_token_uid: int | None
class ExplicitGuild(TypedDict):
_id: int
@@ -32,8 +33,7 @@ class ExplicitGuild(TypedDict):
previous_tracks: list[dict[str, Any]]
current_track: dict[str, Any] | None
current_menu: int | None
is_stopped: bool # Prevents the `after` callback of play_track
always_allow_menu: bool
is_stopped: bool
allow_change_connect: bool
vote_switch_track: bool
vote_add: bool
@@ -42,3 +42,5 @@ class ExplicitGuild(TypedDict):
votes: dict[str, MessageVotes]
vibing: bool
current_viber_id: int | None
use_single_token: bool
single_token_uid: int | None

8
MusicBot/database/user.py
View File

@@ -6,11 +6,8 @@ VibeSettingsOptions: TypeAlias = Literal[
'russian', 'not-russian', 'without-words', 'any',
]
class User(TypedDict, total=False):
class User(TypedDict, total=False): # Don't forget to change base.py if you add a new field
ym_token: str | None
playlists: list[tuple[str, int]]
playlists_page: int
queue_page: int
vibe_batch_id: str | None
vibe_type: Literal['track', 'album', 'artist', 'playlist', 'user'] | None
vibe_id: str | int | None
@@ -19,9 +16,6 @@ class User(TypedDict, total=False):
class ExplicitUser(TypedDict):
_id: int
ym_token: str | None
playlists: list[tuple[str, int]] # name / tracks count
playlists_page: int
queue_page: int
vibe_batch_id: str | None
vibe_type: Literal['track', 'album', 'artist', 'playlist', 'user'] | None
vibe_id: str | int | None

42
MusicBot/ui/find.py
View File

@@ -20,7 +20,7 @@ class PlayButton(Button, VoiceExtension):
if not interaction.guild_id:
logging.info("[FIND] No guild found in PlayButton callback")
await interaction.respond("❌ Эта команда доступна только на серверах.", ephemeral=True, delete_after=15)
await self.respond(interaction, "error", "Эта команда доступна только на серверах.", ephemeral=True, delete_after=15)
return
if not await self.voice_check(interaction):
@@ -28,7 +28,7 @@ class PlayButton(Button, VoiceExtension):
guild = await self.db.get_guild(interaction.guild_id, projection={'current_track': 1, 'current_menu': 1, 'vote_add': 1, 'vibing': 1})
if guild['vibing']:
await interaction.respond("❌ Нельзя добавлять треки в очередь, пока запущена волна.", ephemeral=True, delete_after=15)
await self.respond(interaction, "error", "Нельзя добавлять треки в очередь, пока запущена волна.", ephemeral=True, delete_after=15)
return
channel = cast(discord.VoiceChannel, interaction.channel)
@@ -38,54 +38,54 @@ class PlayButton(Button, VoiceExtension):
tracks = [self.item]
action = 'add_track'
vote_message = f"{member.mention} хочет добавить трек **{self.item.title}** в очередь.\n\n Голосуйте за добавление."
response_message = f"✅ Трек **{self.item.title}** был добавлен в очередь."
response_message = f"Трек **{self.item.title}** был добавлен в очередь."
elif isinstance(self.item, Album):
album = await self.item.with_tracks_async()
if not album or not album.volumes:
logging.debug("[FIND] Failed to fetch album tracks in PlayButton callback")
await interaction.respond("❌ Не удалось получить треки альбома.", ephemeral=True, delete_after=15)
await self.respond(interaction, "error", "Не удалось получить треки альбома.", ephemeral=True, delete_after=15)
return
tracks = [track for volume in album.volumes for track in volume]
action = 'add_album'
vote_message = f"{member.mention} хочет добавить альбом **{self.item.title}** в очередь.\n\n Голосуйте за добавление."
response_message = f"✅ Альбом **{self.item.title}** был добавлен в очередь."
response_message = f"Альбом **{self.item.title}** был добавлен в очередь."
elif isinstance(self.item, Artist):
artist_tracks = await self.item.get_tracks_async()
if not artist_tracks:
logging.debug("[FIND] Failed to fetch artist tracks in PlayButton callback")
await interaction.respond("❌ Не удалось получить треки артиста.", ephemeral=True, delete_after=15)
await self.respond(interaction, "error", "Не удалось получить треки артиста.", ephemeral=True, delete_after=15)
return
tracks = artist_tracks.tracks.copy()
action = 'add_artist'
vote_message = f"{member.mention} хочет добавить треки от **{self.item.name}** в очередь.\n\n Голосуйте за добавление."
response_message = f"✅ Песни артиста **{self.item.name}** были добавлены в очередь."
response_message = f"Песни артиста **{self.item.name}** были добавлены в очередь."
elif isinstance(self.item, Playlist):
short_tracks = await self.item.fetch_tracks_async()
if not short_tracks:
logging.debug("[FIND] Failed to fetch playlist tracks in PlayButton callback")
await interaction.respond("❌ Не удалось получить треки из плейлиста.", ephemeral=True, delete_after=15)
await self.respond(interaction, "error", "Не удалось получить треки из плейлиста.", ephemeral=True, delete_after=15)
return
tracks = [cast(Track, short_track.track) for short_track in short_tracks]
action = 'add_playlist'
vote_message = f"{member.mention} хочет добавить плейлист **{self.item.title}** в очередь.\n\n Голосуйте за добавление."
response_message = f"✅ Плейлист **{self.item.title}** был добавлен в очередь."
response_message = f"Плейлист **{self.item.title}** был добавлен в очередь."
elif isinstance(self.item, list):
tracks = self.item.copy()
if not tracks:
logging.debug("[FIND] Empty tracks list in PlayButton callback")
await interaction.respond("❌ Не удалось получить треки.", ephemeral=True, delete_after=15)
await self.respond(interaction, "error", "Не удалось получить треки.", ephemeral=True, delete_after=15)
return
action = 'add_playlist'
vote_message = f"{member.mention} хочет добавить плейлист **Мне Нравится** в очередь.\n\n Голосуйте за добавление."
response_message = f"✅ Плейлист **«Мне нравится»** был добавлен в очередь."
response_message = f"Плейлист **«Мне нравится»** был добавлен в очередь."
else:
raise ValueError(f"Unknown item type: '{type(self.item).__name__}'")
@@ -93,7 +93,7 @@ class PlayButton(Button, VoiceExtension):
if guild['vote_add'] and len(channel.members) > 2 and not member.guild_permissions.manage_channels:
logging.info(f"Starting vote for '{action}' (from PlayButton callback)")
message = cast(discord.Interaction, await interaction.respond(vote_message, delete_after=60))
message = cast(discord.Interaction, await self.respond(interaction, "info", vote_message, delete_after=60))
response = await message.original_response()
await response.add_reaction('✅')
@@ -113,9 +113,9 @@ class PlayButton(Button, VoiceExtension):
return
if guild['current_menu']:
await interaction.respond(response_message, delete_after=15)
await self.respond(interaction, "success", response_message, delete_after=15)
elif not await self.send_menu_message(interaction, disable=True):
await interaction.respond('❌ Не удалось отправить сообщение.', ephemeral=True, delete_after=15)
await self.respond(interaction, "error", "Не удалось отправить сообщение.", ephemeral=True, delete_after=15)
if guild['current_track']:
logging.debug(f"[FIND] Adding tracks to queue")
@@ -125,7 +125,7 @@ class PlayButton(Button, VoiceExtension):
track = tracks.pop(0)
await self.db.modify_track(interaction.guild_id, tracks, 'next', 'extend')
if not await self.play_track(interaction, track):
await interaction.respond('❌ Не удалось воспроизвести трек.', ephemeral=True, delete_after=15)
await self.respond(interaction, "error", "Не удалось воспроизвести трек.", ephemeral=True, delete_after=15)
if interaction.message:
await interaction.message.delete()
@@ -150,7 +150,7 @@ class MyVibeButton(Button, VoiceExtension):
guild = await self.db.get_guild(interaction.guild_id, projection={'current_menu': 1, 'vibing': 1})
if guild['vibing']:
await interaction.respond('❌ Волна уже запущена. Остановите её с помощью команды /voice stop.', ephemeral=True, delete_after=15)
await self.respond(interaction, "error", "Волна уже запущена. Остановите её с помощью команды /voice stop.", ephemeral=True, delete_after=15)
return
track_type_map = {
@@ -160,7 +160,7 @@ class MyVibeButton(Button, VoiceExtension):
if isinstance(self.item, Playlist):
if not self.item.owner:
logging.warning(f"[VIBE] Playlist owner is None")
await interaction.respond("❌ Не удалось получить информацию о плейлисте. Отсутствует владелец.", ephemeral=True, delete_after=15)
await self.respond(interaction, "error", "Не удалось получить информацию о плейлисте. Отсутствует владелец.", ephemeral=True, delete_after=15)
return
_id = self.item.owner.login + '_' + str(self.item.kind)
@@ -187,7 +187,7 @@ class MyVibeButton(Button, VoiceExtension):
case list():
response_message = f"{member.mention} хочет запустить станцию **Моя Волна**.\n\n Выполнить действие?"
message = cast(discord.Interaction, await interaction.respond(response_message))
message = cast(discord.Interaction, await self.respond(interaction, "info", response_message))
response = await message.original_response()
await response.add_reaction('✅')
@@ -207,7 +207,7 @@ class MyVibeButton(Button, VoiceExtension):
return
if not guild['current_menu'] and not await self.send_menu_message(interaction, disable=True):
await interaction.respond('❌ Не удалось отправить сообщение.', ephemeral=True, delete_after=15)
await self.respond(interaction, "error", "Не удалось отправить сообщение.", ephemeral=True, delete_after=15)
await self.update_vibe(interaction, track_type_map[type(self.item)], _id)
@@ -229,8 +229,8 @@ class ListenView(View):
link_app = f"yandexmusic://artist/{item.id}"
link_web = f"https://music.yandex.ru/artist/{item.id}"
elif isinstance(item, Playlist):
link_app = f"yandexmusic://playlist/{item.playlist_uuid}"
link_web = f"https://music.yandex.ru/playlist/{item.playlist_uuid}"
link_app = f"yandexmusic://playlists/{item.playlist_uuid}"
link_web = f"https://music.yandex.ru/playlists/{item.playlist_uuid}"
elif isinstance(item, list): # Can't open other person's likes
self.add_item(PlayButton(item, label="Слушать в голосовом канале", style=ButtonStyle.gray))
self.add_item(MyVibeButton(item, label="Моя Волна", style=ButtonStyle.gray, emoji="🌊", row=1))

316
MusicBot/ui/menu.py
View File

@@ -1,5 +1,6 @@
import logging
from typing import Self, cast
from time import monotonic
from typing import Self, Literal, cast
from discord.ui import View, Button, Item, Select
from discord import (
@@ -9,26 +10,28 @@ from discord import (
import yandex_music.exceptions
from yandex_music import TrackLyrics, Playlist, ClientAsync as YMClient
from MusicBot.cogs.utils.voice_extension import VoiceExtension, menu_views
from MusicBot.cogs.utils import VoiceExtension
class ToggleButton(Button, VoiceExtension):
def __init__(self, *args, **kwargs):
def __init__(self, root: 'MenuView', *args, **kwargs):
super().__init__(*args, **kwargs)
VoiceExtension.__init__(self, None)
self.root = root
async def callback(self, interaction: Interaction) -> None:
callback_type = interaction.custom_id
if callback_type not in ('repeat', 'shuffle'):
if (callback_type := interaction.custom_id) not in ('repeat', 'shuffle'):
raise ValueError(f"Invalid callback type: '{callback_type}'")
logging.info(f'[MENU] {callback_type.capitalize()} button callback')
if not (gid := interaction.guild_id) or not interaction.user:
logging.warning('[MENU] Failed to get guild ID.')
await interaction.respond("❌ Что-то пошло не так. Попробуйте снова.", delete_after=15, ephemeral=True)
await self.respond(interaction, "error", "Что-то пошло не так. Попробуйте снова.", delete_after=15, ephemeral=True)
return
if not await self.voice_check(interaction, check_vibe_privilage=True):
if not await self.voice_check(interaction):
return
guild = await self.db.get_guild(gid)
@@ -40,7 +43,7 @@ class ToggleButton(Button, VoiceExtension):
action = "выключить" if guild[callback_type] else "включить"
task = "перемешивание треков" if callback_type == 'shuffle' else "повтор трека"
message = cast(Interaction, await interaction.respond(f"{member.mention} хочет {action} {task}.\n\nВыполнить действие?", delete_after=60))
message = cast(Interaction, await self.respond(interaction, "info", f"{member.mention} хочет {action} {task}.\n\nВыполнить действие?", delete_after=60))
response = await message.original_response()
await response.add_reaction('✅')
@@ -61,8 +64,10 @@ class ToggleButton(Button, VoiceExtension):
await self.db.update(gid, {callback_type: not guild[callback_type]})
if not await self.update_menu_view(interaction, button_callback=True):
await interaction.respond("❌ Что-то пошло не так. Попробуйте снова.", delete_after=15, ephemeral=True)
button = self.root.repeat_button if callback_type == 'repeat' else self.root.shuffle_button
button.style = ButtonStyle.secondary if guild[callback_type] else ButtonStyle.success
await interaction.edit(view=await self.root.update())
class PlayPauseButton(Button, VoiceExtension):
def __init__(self, **kwargs):
@@ -71,7 +76,8 @@ class PlayPauseButton(Button, VoiceExtension):
async def callback(self, interaction: Interaction) -> None:
logging.info('[MENU] Play/Pause button callback...')
if not await self.voice_check(interaction, check_vibe_privilage=True):
if not await self.voice_check(interaction):
return
if not (gid := interaction.guild_id) or not interaction.user:
@@ -88,7 +94,7 @@ class PlayPauseButton(Button, VoiceExtension):
logging.info(f"[MENU] User {interaction.user.id} started vote to pause/resume track in guild {gid}")
task = "приостановить" if vc.is_playing() else "возобновить"
message = cast(Interaction, await interaction.respond(f"{member.mention} хочет {task} проигрывание.\n\nВыполнить действие?", delete_after=60))
message = cast(Interaction, await self.respond(interaction, "info", f"{member.mention} хочет {task} проигрывание.\n\nВыполнить действие?", delete_after=60))
response = await message.original_response()
await response.add_reaction('✅')
@@ -107,18 +113,31 @@ class PlayPauseButton(Button, VoiceExtension):
)
return
if vc.is_paused():
vc.resume()
else:
vc.pause()
try:
embed = interaction.message.embeds[0]
except IndexError:
await interaction.respond("❌ Нет воспроизводимого трека.", delete_after=15, ephemeral=True)
await self.respond(interaction, "error", "Нет воспроизводимого трека.", delete_after=15, ephemeral=True)
return
if vc.is_paused():
vc.resume()
embed.remove_footer()
else:
vc.pause()
guild = await self.db.get_guild(interaction.guild_id, projection={'single_token_uid': 1})
if not vc.is_paused() and guild['single_token_uid']:
user = await self.get_discord_user_by_id(interaction, guild['single_token_uid'])
if guild['single_token_uid'] and user:
embed.set_footer(text=f"Используется токен {user.display_name}", icon_url=user.display_avatar.url)
else:
embed.set_footer(text='Используется токен (неизвестный пользователь)')
elif vc.is_paused():
embed.set_footer(text='Приостановлено')
else:
embed.remove_footer()
await interaction.edit(embed=embed)
@@ -128,8 +147,8 @@ class SwitchTrackButton(Button, VoiceExtension):
VoiceExtension.__init__(self, None)
async def callback(self, interaction: Interaction) -> None:
callback_type = interaction.custom_id
if callback_type not in ('next', 'previous'):
if (callback_type := interaction.custom_id) not in ('next', 'previous'):
raise ValueError(f"Invalid callback type: '{callback_type}'")
if not (gid := interaction.guild_id) or not interaction.user:
@@ -138,15 +157,15 @@ class SwitchTrackButton(Button, VoiceExtension):
logging.info(f'[MENU] {callback_type.capitalize()} track button callback')
if not await self.voice_check(interaction, check_vibe_privilage=True):
if not await self.voice_check(interaction):
return
tracks_type = callback_type + '_tracks'
guild = await self.db.get_guild(gid, projection={tracks_type: 1, 'vote_switch_track': 1})
guild = await self.db.get_guild(gid, projection={tracks_type: 1, 'vote_switch_track': 1, 'vibing': 1})
if not guild[tracks_type]:
if not guild[tracks_type] and not guild['vibing']:
logging.info(f"[MENU] No tracks in '{tracks_type}' list in guild {gid}")
await interaction.respond(f"❌ Нет треков в {'очереди' if callback_type == 'next' else 'истории'}.", delete_after=15, ephemeral=True)
await self.respond(interaction, "error", f"Нет треков в {'очереди' if callback_type == 'next' else 'истории'}.", delete_after=15, ephemeral=True)
return
member = cast(Member, interaction.user)
@@ -156,7 +175,7 @@ class SwitchTrackButton(Button, VoiceExtension):
logging.info(f"[MENU] User {interaction.user.id} started vote to skip track in guild {gid}")
task = "пропустить текущий трек" if callback_type == 'next' else "вернуться к предыдущему треку"
message = cast(Interaction, await interaction.respond(f"{member.mention} хочет {task}.\n\nВыполнить переход?", delete_after=60))
message = cast(Interaction, await self.respond(interaction, "info", f"{member.mention} хочет {task}.\n\nВыполнить переход?", delete_after=60))
response = await message.original_response()
await response.add_reaction('✅')
@@ -176,17 +195,18 @@ class SwitchTrackButton(Button, VoiceExtension):
return
if callback_type == 'next':
title = await self.next_track(interaction, button_callback=True)
title = await self.play_next_track(interaction, button_callback=True)
else:
title = await self.previous_track(interaction, button_callback=True)
title = await self.play_previous_track(interaction, button_callback=True)
if not title:
await interaction.respond(f"❌ Что-то пошло не так. Попробуйте позже.", delete_after=15, ephemeral=True)
await self.respond(interaction, "error", "Что-то пошло не так. Попробуйте позже.", delete_after=15, ephemeral=True)
class ReactionButton(Button, VoiceExtension):
def __init__(self, *args, **kwargs):
def __init__(self, root: 'MenuView', *args, **kwargs):
super().__init__(*args, **kwargs)
VoiceExtension.__init__(self, None)
self.root = root
async def callback(self, interaction: Interaction):
callback_type = interaction.custom_id
@@ -199,34 +219,81 @@ class ReactionButton(Button, VoiceExtension):
return
if not (vc := await self.get_voice_client(interaction)) or not vc.is_playing:
await interaction.respond("❌ Нет воспроизводимого трека.", delete_after=15, ephemeral=True)
await self.respond(interaction, "error", "Нет воспроизводимого трека.", delete_after=15, ephemeral=True)
channel = cast(VoiceChannel, interaction.channel)
res = await self.react_track(interaction, callback_type)
if callback_type == 'like' and res[0]:
await self._update_menu_views_dict(interaction)
await interaction.edit(view=menu_views[gid])
await interaction.respond(
f"✅ Трек был {'добавлен в понравившиеся.' if res[1] == 'added' else 'удалён из понравившихся.'}",
delete_after=15, ephemeral=True
)
button = self.root.like_button
response_message = f"Трек был {'добавлен в понравившиеся.' if res[1] == 'added' else 'удалён из понравившихся.'}"
elif callback_type == 'dislike' and res[0]:
if len(channel.members) == 2 and not await self.next_track(interaction, vc=vc, button_callback=True):
await interaction.respond("✅ Воспроизведение приостановлено. Нет треков в очереди.", delete_after=15)
if len(channel.members) == 2:
await self.play_next_track(interaction, vc=vc, button_callback=True)
return
await self._update_menu_views_dict(interaction)
await interaction.edit(view=menu_views[gid])
await interaction.respond(
f"✅ Трек был {'добавлен в дизлайки.' if res[1] == 'added' else 'удалён из дизлайков.'}",
delete_after=15, ephemeral=True
)
button = self.root.dislike_button
response_message =f"Трек был {'добавлен в дизлайки.' if res[1] == 'added' else 'удалён из дизлайков.'}"
else:
logging.debug(f"[VC_EXT] Failed to get {callback_type} tracks")
await interaction.respond("❌ Операция не удалась. Попробуйте позже.", delete_after=15, ephemeral=True)
await self.respond(interaction, "error", "Операция не удалась. Попробуйте позже.", delete_after=15, ephemeral=True)
return
if len(channel.members) == 2:
button.style = ButtonStyle.success if res[1] == 'added' else ButtonStyle.secondary
await interaction.edit(view=await self.root.update())
else:
await self.respond(interaction, "success", response_message, delete_after=15, ephemeral=True)
async def react_track(
self,
ctx: ApplicationContext | Interaction,
action: Literal['like', 'dislike']
) -> tuple[bool, Literal['added', 'removed'] | None]:
"""Like or dislike current track. Return track title on success.
Args:
ctx (ApplicationContext | Interaction): Context.
action (Literal['like', 'dislike']): Action to perform.
Returns:
(tuple[bool, Literal['added', 'removed'] | None]): Tuple with success status and action.
"""
if not (gid := ctx.guild_id) or not ctx.user:
logging.warning("[VC_EXT] Guild or User not found")
return (False, None)
if not (current_track := await self.db.get_track(gid, 'current')):
logging.debug("[VC_EXT] Current track not found")
return (False, None)
if not (client := await self.init_ym_client(ctx)):
return (False, None)
if action == 'like':
tracks = await client.users_likes_tracks()
add_func = client.users_likes_tracks_add
remove_func = client.users_likes_tracks_remove
else:
tracks = await client.users_dislikes_tracks()
add_func = client.users_dislikes_tracks_add
remove_func = client.users_dislikes_tracks_remove
if tracks is None:
logging.debug(f"[VC_EXT] No {action}s found")
return (False, None)
if str(current_track['id']) not in [str(track.id) for track in tracks]:
logging.debug(f"[VC_EXT] Track not found in {action}s. Adding...")
await add_func(current_track['id'])
return (True, 'added')
else:
logging.debug(f"[VC_EXT] Track found in {action}s. Removing...")
await remove_func(current_track['id'])
return (True, 'removed')
class LyricsButton(Button, VoiceExtension):
def __init__(self, **kwargs):
@@ -239,12 +306,10 @@ class LyricsButton(Button, VoiceExtension):
if not await self.voice_check(interaction) or not interaction.guild_id or not interaction.user:
return
client = await self.init_ym_client(interaction)
if not client:
if not (client := await self.init_ym_client(interaction)):
return
current_track = await self.db.get_track(interaction.guild_id, 'current')
if not current_track:
if not (current_track := await self.db.get_track(interaction.guild_id, 'current')):
logging.debug('[MENU] No current track found')
return
@@ -252,7 +317,7 @@ class LyricsButton(Button, VoiceExtension):
lyrics = cast(TrackLyrics, await client.tracks_lyrics(current_track['id']))
except yandex_music.exceptions.NotFoundError:
logging.debug('[MENU] Lyrics not found')
await interaction.respond("❌ Текст песни не найден. Яндекс нам соврал (опять)!", delete_after=15, ephemeral=True)
await self.respond(interaction, "error", "Текст песни не найден. Яндекс нам соврал (опять)!", delete_after=15, ephemeral=True)
return
embed = Embed(
@@ -285,20 +350,20 @@ class MyVibeButton(Button, VoiceExtension):
member = cast(Member, interaction.user)
channel = cast(VoiceChannel, interaction.channel)
track = await self.db.get_track(interaction.guild_id, 'current')
if len(channel.members) > 2 and not member.guild_permissions.manage_channels:
logging.info(f"Starting vote for starting vibe in guild {interaction.guild_id}")
if track:
response_message = f"{member.mention} хочет запустить волну по треку **{track['title']}**.\n\n Выполнить действие?"
_type = 'track'
_id = track['id']
vibe_type = 'track'
vibe_id = track['id']
else:
response_message = f"{member.mention} хочет запустить станцию **Моя Волна**.\n\n Выполнить действие?"
_type = 'user'
_id = 'onyourwave'
vibe_type = 'user'
vibe_id = 'onyourwave'
message = cast(Interaction, await interaction.respond(response_message))
message = cast(Interaction, await self.respond(interaction, "info", response_message))
response = await message.original_response()
await response.add_reaction('✅')
@@ -312,7 +377,7 @@ class MyVibeButton(Button, VoiceExtension):
'negative_votes': list(),
'total_members': len(channel.members),
'action': 'vibe_station',
'vote_content': [_type, _id, interaction.user.id]
'vote_content': [vibe_type, vibe_id, interaction.user.id]
}
)
return
@@ -334,10 +399,9 @@ class MyVibeButton(Button, VoiceExtension):
if not res:
logging.info('[MENU] Failed to start the vibe')
await interaction.respond('❌ Не удалось запустить "Мою Волну". Возможно, у вас нет подписки на Яндекс Музыку.', ephemeral=True)
await self.respond(interaction, "error", "Не удалось запустить **Мою Волну**. Возможно, у вас нет подписки на Яндекс Музыку.", ephemeral=True)
next_track = await self.db.get_track(interaction.guild_id, 'next')
if next_track:
if (next_track := await self.db.get_track(interaction.guild_id, 'next')):
await self.play_track(interaction, next_track, button_callback=True)
class MyVibeSelect(Select, VoiceExtension):
@@ -354,7 +418,7 @@ class MyVibeSelect(Select, VoiceExtension):
if not interaction.user:
logging.warning('[MENU] No user in select callback')
return
custom_id = interaction.custom_id
if custom_id not in ('diversity', 'mood', 'lang'):
logging.error(f'[MENU] Unknown custom_id: {custom_id}')
@@ -465,7 +529,7 @@ class MyVibeSettingsButton(Button, VoiceExtension):
if not await self.voice_check(interaction, check_vibe_privilage=True):
return
await interaction.respond('Настройки "Моей Волны"', view=await MyVibeSettingsView(interaction).init(), ephemeral=True)
await self.respond(interaction, "info", "Настройки **Волны**", view=await MyVibeSettingsView(interaction).init(), ephemeral=True)
class AddToPlaylistSelect(Select, VoiceExtension):
def __init__(self, ym_client: YMClient, *args, **kwargs):
@@ -517,11 +581,11 @@ class AddToPlaylistSelect(Select, VoiceExtension):
)
if not res:
await interaction.respond('❌ Что-то пошло не так. Попробуйте позже.', delete_after=15, ephemeral=True)
await self.respond(interaction, "error", "Что-то пошло не так. Попробуйте позже.", delete_after=15, ephemeral=True)
elif track_in_playlist:
await interaction.respond('🗑 Трек был удалён из плейлиста.', delete_after=15, ephemeral=True)
await self.respond(interaction, "success", "🗑 Трек был удалён из плейлиста.", delete_after=15, ephemeral=True)
else:
await interaction.respond('📩 Трек был добавлен в плейлист.', delete_after=15, ephemeral=True)
await self.respond(interaction, "success", "📩 Трек был добавлен в плейлист.", delete_after=15, ephemeral=True)
class AddToPlaylistButton(Button, VoiceExtension):
@@ -533,23 +597,20 @@ class AddToPlaylistButton(Button, VoiceExtension):
if not await self.voice_check(interaction) or not interaction.guild_id:
return
current_track = await self.db.get_track(interaction.guild_id, 'current')
if not current_track:
await interaction.respond('❌ Нет воспроизводимого трека.', delete_after=15, ephemeral=True)
if not await self.db.get_track(interaction.guild_id, 'current'):
await self.respond(interaction, "error", "Нет воспроизводимого трека.", delete_after=15, ephemeral=True)
return
client = await self.init_ym_client(interaction)
if not client:
await interaction.respond('❌ Что-то пошло не так. Попробуйте позже.', delete_after=15, ephemeral=True)
if not (client := await self.init_ym_client(interaction)):
await self.respond(interaction, "error", "Что-то пошло не так. Попробуйте позже.", delete_after=15, ephemeral=True)
return
if not (vc := await self.get_voice_client(interaction)) or not vc.is_playing:
await interaction.respond("❌ Нет воспроизводимого трека.", delete_after=15, ephemeral=True)
await self.respond(interaction, "error", "Нет воспроизводимого трека.", delete_after=15, ephemeral=True)
return
playlists = await client.users_playlists_list()
if not playlists:
await interaction.respond('❌ У вас нет плейлистов.', delete_after=15, ephemeral=True)
if not (playlists := await client.users_playlists_list()):
await self.respond(interaction, "error", "У вас нет плейлистов.", delete_after=15, ephemeral=True)
return
view = View(
@@ -576,42 +637,76 @@ class MenuView(View, VoiceExtension):
VoiceExtension.__init__(self, None)
self.ctx = ctx
self.repeat_button = ToggleButton(style=ButtonStyle.secondary, emoji='🔂', row=0, custom_id='repeat')
self.shuffle_button = ToggleButton(style=ButtonStyle.secondary, emoji='🔀', row=0, custom_id='shuffle')
self.repeat_button = ToggleButton(self, style=ButtonStyle.secondary, emoji='🔂', row=0, custom_id='repeat')
self.shuffle_button = ToggleButton(self, style=ButtonStyle.secondary, emoji='🔀', row=0, custom_id='shuffle')
self.play_pause_button = PlayPauseButton(style=ButtonStyle.primary, emoji='⏯', row=0)
self.next_button = SwitchTrackButton(style=ButtonStyle.primary, emoji='⏭', row=0, custom_id='next')
self.prev_button = SwitchTrackButton(style=ButtonStyle.primary, emoji='⏮', row=0, custom_id='previous')
self.like_button = ReactionButton(style=ButtonStyle.secondary, emoji='❤️', row=1, custom_id='like')
self.dislike_button = ReactionButton(style=ButtonStyle.secondary, emoji='💔', row=1, custom_id='dislike')
self.like_button = ReactionButton(self, style=ButtonStyle.secondary, emoji='❤️', row=1, custom_id='like')
self.dislike_button = ReactionButton(self, style=ButtonStyle.secondary, emoji='💔', row=1, custom_id='dislike')
self.lyrics_button = LyricsButton(style=ButtonStyle.secondary, emoji='📋', row=1)
self.add_to_playlist_button = AddToPlaylistButton(style=ButtonStyle.secondary, emoji='📁', row=1)
self.vibe_button = MyVibeButton(style=ButtonStyle.secondary, emoji='🌊', row=1)
self.vibe_settings_button = MyVibeSettingsButton(style=ButtonStyle.success, emoji='🛠', row=1)
self.current_vibe_button: MyVibeButton | MyVibeSettingsButton = self.vibe_button
async def init(self, *, disable: bool = False) -> Self:
if not self.ctx.guild_id:
return self
self.guild = await self.db.get_guild(self.ctx.guild_id, projection={'repeat': 1, 'shuffle': 1, 'current_track': 1, 'current_menu': 1, 'vibing': 1})
if self.guild['repeat']:
self.repeat_button.style = ButtonStyle.success
if self.guild['shuffle']:
self.shuffle_button.style = ButtonStyle.success
current_track = self.guild['current_track']
likes = await self.get_likes(self.ctx)
await self.update(disable=disable)
self.add_item(self.repeat_button)
self.add_item(self.prev_button)
self.add_item(self.play_pause_button)
self.add_item(self.next_button)
self.add_item(self.shuffle_button)
self.add_item(self.like_button)
self.add_item(self.dislike_button)
self.add_item(self.lyrics_button)
self.add_item(self.add_to_playlist_button)
self.add_item(self.current_vibe_button)
return self
async def update(self, *, disable: bool = False) -> Self:
if not self.ctx.guild_id:
return self
if not isinstance(self.ctx, RawReactionActionEvent) and len(cast(VoiceChannel, self.ctx.channel).members) == 2:
if likes and current_track and str(current_track['id']) in [str(like.id) for like in likes]:
self.enable_all_items()
self.guild = await self.db.get_guild(self.ctx.guild_id, projection={
'repeat': 1, 'shuffle': 1, 'current_track': 1, 'current_viber_id': 1, 'vibing': 1, 'single_token_uid': 1
})
if self.guild['repeat']:
self.repeat_button.style = ButtonStyle.success
else:
self.repeat_button.style = ButtonStyle.secondary
if self.guild['shuffle']:
self.shuffle_button.style = ButtonStyle.success
else:
self.shuffle_button.style = ButtonStyle.secondary
current_track = self.guild['current_track']
if not isinstance(self.ctx, RawReactionActionEvent) \
and len(cast(VoiceChannel, self.ctx.channel).members) == 2 \
and not self.guild['single_token_uid']:
if current_track and str(current_track['id']) in [str(like.id) for like in await self.get_reacted_tracks(self.ctx, 'like')]:
self.like_button.style = ButtonStyle.success
else:
self.like_button.style = ButtonStyle.secondary
if current_track and str(current_track['id']) in [str(dislike.id) for dislike in await self.get_reacted_tracks(self.ctx, 'dislike')]:
self.dislike_button.style = ButtonStyle.success
else:
self.dislike_button.style = ButtonStyle.secondary
else:
self.like_button.style = ButtonStyle.secondary
self.dislike_button.style = ButtonStyle.secondary
if not current_track:
self.lyrics_button.disabled = True
@@ -621,35 +716,40 @@ class MenuView(View, VoiceExtension):
elif not current_track['lyrics_available']:
self.lyrics_button.disabled = True
self.add_item(self.like_button)
self.add_item(self.dislike_button)
self.add_item(self.lyrics_button)
self.add_item(self.add_to_playlist_button)
if self.guild['single_token_uid']:
self.like_button.disabled = True
self.dislike_button.disabled = True
self.add_to_playlist_button.disabled = True
if self.guild['vibing']:
self.add_item(self.vibe_settings_button)
self.current_vibe_button = self.vibe_settings_button
else:
self.add_item(self.vibe_button)
self.current_vibe_button = self.vibe_button
if disable:
self.disable_all_items()
if self.timeout:
self.__timeout_expiry = monotonic() + self.timeout
return self
async def on_timeout(self) -> None:
logging.debug('[MENU] Menu timed out. Deleting menu message')
if not self.ctx.guild_id:
return
if self.guild['current_menu']:
await self.stop_playing(self.ctx)
await self.db.update(self.ctx.guild_id, {'current_menu': None, 'previous_tracks': [], 'vibing': False})
message = await self.get_menu_message(self.ctx, self.guild['current_menu'])
if message:
if self.guild['current_menu']:
await self.db.update(self.ctx.guild_id, {
'current_menu': None, 'repeat': False, 'shuffle': False,
'previous_tracks': [], 'next_tracks': [], 'votes': {},
'vibing': False, 'current_viber_id': None
})
if (message := await self.get_menu_message(self.ctx, self.guild['current_menu'])):
await message.delete()
logging.debug('[MENU] Successfully deleted menu message')
else:
logging.debug('[MENU] No menu message found')
self.stop()
self.stop()

96
MusicBot/ui/other.py
View File

@@ -1,5 +1,5 @@
from math import ceil
from typing import Self, Any
from typing import Any
from discord.ui import View, Button, Item
from discord import ApplicationContext, ButtonStyle, Interaction, Embed, HTTPException
@@ -9,83 +9,87 @@ from MusicBot.cogs.utils.voice_extension import VoiceExtension
def generate_queue_embed(page: int, tracks_list: list[dict[str, Any]]) -> Embed:
count = 15 * page
length = len(tracks_list)
embed = Embed(
title=f"Всего: {length}",
color=0xfed42b,
)
embed.set_author(name="Очередь треков")
embed.set_footer(text=f"Страница {page + 1} из {ceil(length / 15)}")
for i, track in enumerate(tracks_list[count:count + 15], start=1 + count):
duration = track['duration_ms']
if duration:
duration_m = duration // 60000
duration_s = ceil(duration / 1000) - duration_m * 60
if track['duration_ms']:
duration_m = track['duration_ms'] // 60000
duration_s = ceil(track['duration_ms'] / 1000) - duration_m * 60
embed.add_field(name=f"{i} - {track['title']} - {duration_m}:{duration_s:02d}", value="", inline=False)
return embed
class QueueNextButton(Button, VoiceExtension):
def __init__(self, **kwargs):
class QueueNextButton(Button):
def __init__(self, root:' QueueView', **kwargs):
Button.__init__(self, **kwargs)
VoiceExtension.__init__(self, None)
self.root = root
async def callback(self, interaction: Interaction) -> None:
if not interaction.user or not interaction.guild:
return
self.root.page += 1
self.root.update()
embed = generate_queue_embed(self.root.page, self.root.tracks)
await interaction.edit(embed=embed, view=self.root)
user = await self.users_db.get_user(interaction.user.id)
page = user['queue_page'] + 1
await self.users_db.update(interaction.user.id, {'queue_page': page})
tracks = await self.db.get_tracks_list(interaction.guild.id, 'next')
embed = generate_queue_embed(page, tracks)
await interaction.edit(embed=embed, view=await QueueView(interaction).init())
class QueuePrevButton(Button, VoiceExtension):
def __init__(self, **kwargs):
class QueuePrevButton(Button):
def __init__(self, root: 'QueueView', **kwargs):
Button.__init__(self, **kwargs)
VoiceExtension.__init__(self, None)
self.root = root
async def callback(self, interaction: Interaction) -> None:
if not interaction.user or not interaction.guild:
return
user = await self.users_db.get_user(interaction.user.id)
page = user['queue_page'] - 1
await self.users_db.update(interaction.user.id, {'queue_page': page})
tracks = await self.db.get_tracks_list(interaction.guild.id, 'next')
embed = generate_queue_embed(page, tracks)
await interaction.edit(embed=embed, view=await QueueView(interaction).init())
self.root.page -= 1
self.root.update()
embed = generate_queue_embed(self.root.page, self.root.tracks)
await interaction.edit(embed=embed, view=self.root)
class QueueView(View, VoiceExtension):
def __init__(self, ctx: ApplicationContext | Interaction, *items: Item, timeout: float | None = 360, disable_on_timeout: bool = False):
def __init__(
self,
ctx: ApplicationContext | Interaction,
tracks: list[dict[str, Any]],
*items: Item,
timeout: float | None = 360,
disable_on_timeout: bool = False
):
View.__init__(self, *items, timeout=timeout, disable_on_timeout=disable_on_timeout)
VoiceExtension.__init__(self, None)
self.ctx = ctx
self.next_button = QueueNextButton(style=ButtonStyle.primary, emoji='▶️')
self.prev_button = QueuePrevButton(style=ButtonStyle.primary, emoji='◀️')
self.tracks = tracks
self.page = 0
async def init(self) -> Self:
if not self.ctx.user or not self.ctx.guild:
return self
tracks = await self.db.get_tracks_list(self.ctx.guild.id, 'next')
user = await self.users_db.get_user(self.ctx.user.id)
count = 15 * user['queue_page']
if not tracks[count + 15:]:
self.next_button = QueueNextButton(self, style=ButtonStyle.primary, emoji='▶️')
self.prev_button = QueuePrevButton(self, style=ButtonStyle.primary, emoji='◀️', disabled=True)
if not self.tracks[15:]:
self.next_button.disabled = True
if not tracks[:count]:
self.prev_button.disabled = True
self.prev_button.disabled = True
self.add_item(self.prev_button)
self.add_item(self.next_button)
return self
def update(self):
count = 15 * self.page
if self.tracks[15:]:
self.next_button.disabled = False
else:
self.next_button.disabled = True
if self.tracks[:count]:
self.prev_button.disabled = False
else:
self.prev_button.disabled = True
async def on_timeout(self) -> None:
try:
await super().on_timeout()
except HTTPException:
pass
self.stop()
self.stop()

36
README.md
View File

@@ -68,15 +68,45 @@ pip install -r requirements.txt
Создайте файл `.env` и добавьте в него переменные окружения. Пример:
```env
TOKEN='XXXXXX' # Токен бота
EXPLICIT_EID='1325879701117472869' # ID эмодзи explicit
DEBUG='False' # Включение DEBUG логов (True/False)
TOKEN='XXXXXX' # Токен бота
DEBUG='False' # Включение DEBUG логов (True/False)
EXPLICIT_EID='1325879701117472869' # ID эмодзи explicit
MONGO_URI='mongodb://localhost:27017/' # Адрес сервера MongoDB
```
Запустите сервер MongoDB (настройки по умолчанию) и создайте базу данных YandexMusicBot с коллекциями guilds и users (через Compass или mongosh).
Запустите бота (`python ./MusicBot/main.py`).
## Запуск в Docker ![Main Build](https://img.shields.io/github/actions/workflow/status/lemon4ksan/YandexMusicDiscordBot/docker-image.yml?branch=main&label=main) ![Dev Build](https://img.shields.io/github/actions/workflow/status/lemon4ksan/YandexMusicDiscordBot/docker-image.yml?branch=dev&label=dev)
Возможен запуск как из командной строки, так и с помощью docker-compose.
### docker cli
> [!NOTE]
> При этом методе запуска вам необходимо самостоятельно установить MongoDB и указать адрес сервера в команде запуска.
```bash
docker run -d \
--name yandex-music-discord-bot \
--restart unless-stopped \
-e TOKEN=XXXXXX \
-e EXPLICIT_EID=1325879701117472869 \
-e DEBUG=False \
-e MONGO_URI="mongodb://mongodb:27017/" \
lemon4ksan/yandexmusicdiscordbot:latest
```
### docker-compose (рекомендованный)
> [!NOTE]
> При первом запуске БД и коллекции будут созданы автоматически.
```bash
docker-compose up -d
```
## Настройка бота
Так должны выглядить настройки бота:

33
docker-compose.yml Normal file
View File

@@ -0,0 +1,33 @@
services:
app:
container_name: yandex-music-discord-bot
image: lemon4ksan/yandexmusicdiscordbot:latest
restart: unless-stopped
depends_on:
- mongodb
env_file:
- .env
environment:
MONGO_URI: "mongodb://ymdb-mongodb:27017"
networks:
- ymdb_network
mongodb:
container_name: ymdb-mongodb
image: mongo:latest
restart: unless-stopped
volumes:
- mongodb_data:/data/db
- ./init-mongodb.js:/docker-entrypoint-initdb.d/init-mongodb.js:ro
networks:
- ymdb_network
healthcheck:
test: echo 'db.runCommand("ping").ok' | mongo localhost:27017 --quiet
interval: 30s
timeout: 10s
retries: 5
volumes:
mongodb_data:
networks:
ymdb_network:

3
init-mongodb.js Normal file
View File

@@ -0,0 +1,3 @@
db = db.getSiblingDB('YandexMusicBot');
db.createCollection('guilds');
db.createCollection('users');