diff --git a/radiocycle/Modules/SpotifyMod.py b/radiocycle/Modules/SpotifyMod.py
index 218588b..7834a05 100644
--- a/radiocycle/Modules/SpotifyMod.py
+++ b/radiocycle/Modules/SpotifyMod.py
@@ -1,1436 +1 @@
-# █ █ ▀ █▄▀ ▄▀█ █▀█ ▀
-# █▀█ █ █ █ █▀█ █▀▄ █
-# © Copyright 2022
-#
-# https://t.me/hikariatama
-#
-# 🔒 Licensed under the GNU AGPLv3
-# 🌐 https://www.gnu.org/licenses/agpl-3.0.html
-#
-# You CANNOT edit, distribute or redistribute this file without direct permission from the author.
-#
-# ORIGINAL MODULE: https://raw.githubusercontent.com/hikariatama/ftg/master/spotify.py
-
-# =======================================
-# _ __ __ __ _
-# | |/ /___ | \/ | ___ __| |___
-# | ' // _ \ | |\/| |/ _ \ / _` / __|
-# | . \ __/ | | | | (_) | (_| \__ \
-# |_|\_\___| |_| |_|\___/ \__,_|___/
-# @ke_mods
-# =======================================
-#
-# LICENSE: CC BY-ND 4.0 (Attribution-NoDerivatives 4.0 International)
-# --------------------------------------
-# https://creativecommons.org/licenses/by-nd/4.0/legalcode
-# =======================================
-
-# meta developer: @ke_mods
-# requires: telethon spotipy pillow requests yt-dlp
-
-import asyncio
-import contextlib
-import functools
-import io
-import logging
-import re
-import textwrap
-import time
-import traceback
-import os
-from types import FunctionType
-
-import requests
-import spotipy
-from PIL import Image, ImageDraw, ImageEnhance, ImageFilter, ImageFont, ImageOps
-from telethon.errors import FloodWaitError
-from telethon.tl.functions.account import UpdateProfileRequest
-from telethon.tl.types import Message
-
-from .. import loader, utils
-
-logger = logging.getLogger(__name__)
-logging.getLogger("spotipy").setLevel(logging.CRITICAL)
-
-class Banners:
- def __init__(
- self,
- title: str,
- artists: list,
- duration: int,
- progress: int,
- track_cover: bytes,
- font
- ):
- self.title = title
- self.artists = ", ".join(artists) if isinstance(artists, list) else artists
- self.duration = duration
- self.progress = progress
- self.track_cover = track_cover
- self.font_url = font
-
- def _get_font(self, size, font_bytes):
- return ImageFont.truetype(io.BytesIO(font_bytes), size)
-
- def _prepare_cover(self, size, radius):
- cover = Image.open(io.BytesIO(self.track_cover)).convert("RGBA")
- cover = cover.resize((size, size), Image.Resampling.LANCZOS)
-
- mask = Image.new("L", (size, size), 0)
- draw = ImageDraw.Draw(mask)
- draw.rounded_rectangle((0, 0, size, size), radius=radius, fill=255)
-
- output = Image.new("RGBA", (size, size), (0, 0, 0, 0))
- output.paste(cover, (0, 0), mask=mask)
- return output
-
- def _prepare_background(self, w, h):
- bg = Image.open(io.BytesIO(self.track_cover)).convert("RGBA")
- bg = bg.resize((w, h), Image.Resampling.BICUBIC)
- bg = bg.filter(ImageFilter.GaussianBlur(radius=40))
- bg = ImageEnhance.Brightness(bg).enhance(0.4)
- return bg
-
- def _draw_progress_bar(self, draw, x, y, w, h, progress_pct, color="white", bg_color="#5e5e5e"):
- draw.rounded_rectangle((x, y, x + w, y + h), radius=h/2, fill=bg_color)
-
- fill_w = int(w * progress_pct)
- if fill_w > 0:
- draw.rounded_rectangle((x, y, x + fill_w, y + h), radius=h/2, fill=color)
-
- dot_radius = h * 1.2
- dot_x = x + fill_w
- dot_y = y + (h / 2)
-
- draw.ellipse(
- (dot_x - dot_radius, dot_y - dot_radius, dot_x + dot_radius, dot_y + dot_radius),
- fill=color
- )
-
- def horizontal(self):
- W, H = 1500, 600
- padding = 60
- cover_size = 480
-
- font_bytes = requests.get(self.font_url).content
- title_font = self._get_font(55, font_bytes)
- artist_font = self._get_font(45, font_bytes)
- time_font = self._get_font(25, font_bytes)
-
- img = self._prepare_background(W, H)
- draw = ImageDraw.Draw(img)
-
- cover = self._prepare_cover(cover_size, 30)
- img.paste(cover, (padding, (H - cover_size) // 2), cover)
-
- text_x = padding + cover_size + 60
- text_y_start = 100
- text_width_limit = W - text_x - padding
-
- display_title = self.title
- while title_font.getlength(display_title) > text_width_limit and len(display_title) > 0:
- display_title = display_title[:-1]
- if len(display_title) < len(self.title): display_title += "…"
-
- display_artist = self.artists
- while artist_font.getlength(display_artist) > text_width_limit and len(display_artist) > 0:
- display_artist = display_artist[:-1]
- if len(display_artist) < len(self.artists): display_artist += "…"
-
- draw.text((text_x, text_y_start), display_title, font=title_font, fill="white")
- draw.text((text_x, text_y_start + 70), display_artist, font=artist_font, fill="#B3B3B3")
-
- cur_time = f"{(self.progress//1000//60):02}:{(self.progress//1000%60):02}"
- dur_time = f"{(self.duration//1000//60):02}:{(self.duration//1000%60):02}"
-
- cur_w = time_font.getlength(cur_time)
- dur_w = time_font.getlength(dur_time)
-
- bar_y = 480
- bar_h = 8
- gap = 25
-
- draw.text((text_x, bar_y - 12), cur_time, font=time_font, fill="white")
-
- bar_start_x = text_x + cur_w + gap
- bar_end_x = text_x + text_width_limit - dur_w - gap
- bar_w = bar_end_x - bar_start_x
-
- prog_pct = self.progress / self.duration if self.duration > 0 else 0
- self._draw_progress_bar(draw, bar_start_x, bar_y, bar_w, bar_h, prog_pct)
-
- draw.text((bar_end_x + gap, bar_y - 12), dur_time, font=time_font, fill="white")
-
- by = io.BytesIO()
- img.save(by, format="PNG")
- by.seek(0)
- by.name = "banner.png"
- return by
-
- def vertical(self):
- W, H = 1000, 1500
- padding = 80
- cover_size = 800
-
- font_bytes = requests.get(self.font_url).content
- title_font = self._get_font(60, font_bytes)
- artist_font = self._get_font(45, font_bytes)
- time_font = self._get_font(35, font_bytes)
-
- img = self._prepare_background(W, H)
- draw = ImageDraw.Draw(img)
-
- cover = self._prepare_cover(cover_size, 40)
- cover_x = (W - cover_size) // 2
- cover_y = 120
- img.paste(cover, (cover_x, cover_y), cover)
-
- text_area_y = cover_y + cover_size + 120
- text_width_limit = W - (padding * 2)
-
- display_title = self.title
- while title_font.getlength(display_title) > text_width_limit and len(display_title) > 0:
- display_title = display_title[:-1]
- if len(display_title) < len(self.title): display_title += "…"
-
- display_artist = self.artists
- while artist_font.getlength(display_artist) > text_width_limit and len(display_artist) > 0:
- display_artist = display_artist[:-1]
- if len(display_artist) < len(self.artists): display_artist += "…"
-
- title_w = title_font.getlength(display_title)
- draw.text(((W - title_w) / 2, text_area_y), display_title, font=title_font, fill="white")
-
- artist_w = artist_font.getlength(display_artist)
- draw.text(((W - artist_w) / 2, text_area_y + 75), display_artist, font=artist_font, fill="#B3B3B3")
-
- bar_y = text_area_y + 260
- bar_h = 8
- bar_w = W - (padding * 2)
- prog_pct = self.progress / self.duration if self.duration > 0 else 0
-
- self._draw_progress_bar(draw, padding, bar_y, bar_w, bar_h, prog_pct, color="white", bg_color="#5e5e5e")
-
- cur_time = f"{(self.progress//1000//60):02}:{(self.progress//1000%60):02}"
- dur_time = f"{(self.duration//1000//60):02}:{(self.duration//1000%60):02}"
-
- draw.text((padding, bar_y + 40), cur_time, font=time_font, fill="#B3B3B3")
-
- dur_w = time_font.getlength(dur_time)
- draw.text((W - padding - dur_w, bar_y + 40), dur_time, font=time_font, fill="#B3B3B3")
-
- by = io.BytesIO()
- img.save(by, format="PNG")
- by.seek(0)
- by.name = "banner.png"
- return by
-
-@loader.tds
-class SpotifyMod(loader.Module):
- """Card with the currently playing track on Spotify."""
-
- strings = {
- "name": "SpotifyMod",
- "need_auth": (
- "❌ Please execute"
- " .sauth before performing this action."
- ),
- "on-repeat": (
- "🔄 Set on-repeat."
- ),
- "off-repeat": (
- "🔄 Stopped track"
- " repeat."
- ),
- "skipped": (
- "➡️ Skipped track."
- ),
- "playing": "▶️ Playing...",
- "back": (
- "⬅️ Switched to previous"
- " track"
- ),
- "paused": "❌ Pause",
- "restarted": (
- "✅️ Playing track"
- " from the"
- " beginning"
- ),
- "liked": (
- "❤️ Liked current"
- " playback"
- ),
- "unlike": (
- "❌"
- " Unliked current playback"
- ),
- "err": (
- "❌ An error occurred."
- "\n{}"
- ),
- "already_authed": (
- "❌ Already authorized"
- ),
- "authed": (
- "✅ Authentication"
- " successful"
- ),
- "deauth": (
- "🚪 Successfully logged out"
- " of account"
- ),
- "auth": (
- '🔗 Follow this'
- " link, allow access, then enter .scode https://... with"
- " the link you received."
- ),
- "no_music": (
- "❌ No music is playing!"
- ),
- "dl_err": (
- "❌ Failed to download"
- " track."
- ),
- "volume_changed": (
- "🔊"
- " Volume changed to {}%."
- ),
- "volume_invalid": (
- "❌ Volume level must be"
- " a number between 0 and 100."
- ),
- "volume_err": (
- "❌ An error occurred while"
- " changing volume."
- ),
- "no_volume_arg": (
- "❌ Please specify a"
- " volume level between 0 and 100."
- ),
- "searching_tracks": (
- "🕔 Searching for tracks"
- " matching {}..."
- ),
- "no_search_query": (
- "❌ Please specify a"
- " search query."
- ),
- "no_tracks_found": (
- "❌ No tracks found for"
- " {}."
- ),
- "search_results": (
- "✅ Search results for"
- " {}:\n\n{}"
- ),
- "downloading_search_track": (
- "🕔 Downloading {}..."
- ),
- "download_success": (
- "✅ Successfully downloaded {} - {}"
- ),
- "invalid_track_number": (
- "❌ Invalid track number."
- " Please search first or provide a valid number from the list."
- ),
- "device_list": (
- "📄 Available devices:\n{}"
- ),
- "no_devices_found": (
- "❌ No devices found."
- ),
- "device_changed": (
- "✅ Playback transferred to"
- " {}."
- ),
- "invalid_device_id": (
- "❌ Invalid device ID."
- " Use .sdevice to see available devices."
- ),
- "search_results_cleared": "✅ Search results cleared",
- "autobio": (
- "🎧 Spotify autobio {}"
- ),
- "no_ytdlp": "❌ yt-dlp not found... Check config or install yt-dlp ({}terminal pip install yt-dlp)",
- "snowt_failed": "\n\n❌ Download failed",
- "uploading_banner": "\n\n🕔 Uploading banner...",
- "downloading_track": "\n\n🕔 Downloading track...",
- "no_playlists": "❌ No playlists found.",
- "playlists_list": "📄 Your playlists:\n\n{}",
- "added_to_playlist": "✅ Added {} to {}",
- "removed_from_playlist": "✅ Removed {} from {}",
- "invalid_playlist_index": "❌ Invalid playlist number.",
- "no_cached_playlists": "❌ Use .splaylists first.",
- "playlist_created": "✅ Playlist {} created.",
- "playlist_deleted": "✅ Playlist {} deleted.",
- "no_playlist_name": "❌ Please specify a playlist name.",
- }
-
- strings_ru = {
- "_cls_doc": "Карточка с играющим треком в Spotify.",
- "need_auth": (
- "❌ Выполни"
- " .sauth перед выполнением этого действия."
- ),
- "err": (
- "❌ Произошла ошибка."
- "\n{}"
- ),
- "on-repeat": (
- "🔄 Включен повтор трека."
- ),
- "off-repeat": (
- "🔄 Повтор трека отключён."
- ),
- "skipped": (
- "➡️ Трек пропущен."
- ),
- "playing": "▶️ Играет...",
- "back": (
- "⬅️ Переключено на предыдущий трек"
- ),
- "paused": "❌ Пауза",
- "restarted": (
- "✅️ Воспроизведение трека с начала..."
- ),
- "liked": (
- "❤️ Текущий трек добавлен в избранное"
- ),
- "unlike": (
- "❌ Убрал лайк с текущего трека"
- ),
- "already_authed": (
- "❌ Уже авторизован"
- ),
- "authed": (
- "✅ Успешная аутентификация"
- ),
- "deauth": (
- "🚪 Успешный выход из аккаунта"
- ),
- "auth": (
- '🔗 Пройдите по этой ссылке, разрешите вход, затем введите .scode https://... с ссылкой которую вы получили.'
- ),
- "no_music": (
- "❌ Музыка не играет!"
- ),
- "dl_err": (
- "❌ Не удалось скачать трек."
- ),
- "volume_changed": (
- "🔊"
- " Громкость изменена на {}%."
- ),
- "volume_invalid": (
- "❌ Уровень громкости должен"
- " быть числом от 0 до 100."
- ),
- "volume_err": (
- "❌ Произошла ошибка при"
- " изменении громкости."
- ),
- "no_volume_arg": (
- "❌ Пожалуйста, укажите"
- " уровень громкости от 0 до 100."
- ),
- "searching_tracks": (
- "🕔 Идет поиск треков"
- " по запросу {}..."
- ),
- "no_search_query": (
- "❌ Пожалуйста, укажите"
- " поисковый запрос."
- ),
- "no_tracks_found": (
- "❌ По запросу '{}'"
- " ничего не найдено."
- ),
- "search_results": (
- "✅ Результаты поиска"
- " по запросу {}:\n\n{}"
- ),
- "downloading_search_track": (
- "🕔 Скачиваю {}..."
- ),
- "download_success": (
- "✅ Трек {} - {} успешно скачан."
- ),
- "invalid_track_number": (
- "❌ Некорректный номер трека."
- " Сначала выполните поиск или укажите правильный номер из списка."
- ),
- "device_list": (
- "📄 Доступные устройства:\n{}"
- ),
- "no_devices_found": (
- "❌ Устройства не найдены."
- ),
- "device_changed": (
- "✅ Воспроизведение переключено на"
- " {}."
- ),
- "invalid_device_id": (
- "❌ Некорректный ID устройства."
- " Используйте .sdevice , чтобы увидеть доступные устройства."
- ),
- "search_results_cleared": "✅ Результаты поиска очищены",
- "autobio": (
- "🎧 Обновление био"
- " включено {}"
- ),
- "no_ytdlp": "❌ yt-dlp не найден... Проверьте конфиг или установите yt-dlp ({}terminal pip install yt-dlp)",
- "snowt_failed": "\n\n❌ Ошибка скачивания.",
- "uploading_banner": "\n\n🕔 Загрузка баннера...",
- "downloading_track": "\n\n🕔 Скачивание трека...",
- "no_playlists": "❌ Плейлисты не найдены.",
- "playlists_list": "📄 Ваши плейлисты:\n\n{}",
- "added_to_playlist": "✅ Трек {} добавлен в {}",
- "removed_from_playlist": "✅ Трек {} удален из {}",
- "invalid_playlist_index": "❌ Неверный номер плейлиста.",
- "no_cached_playlists": "❌ Сначала используйте .splaylists.",
- "playlist_created": "✅ Плейлист {} создан.",
- "playlist_deleted": "✅ Плейлист {} удален.",
- "no_playlist_name": "❌ Пожалуйста, укажите название плейлиста.",
- }
- strings_jp = {
- "_cls_doc": "Spotify からのメッセージ",
- "need_auth": (
- "❌ この操作を行う前に "
- ".sauth を実行してください。"
- ),
- "on-repeat": (
- "🔄 リピート再生を設定しました。"
- ),
- "off-repeat": (
- "🔄 リピート再生を解除しました。"
- ),
- "skipped": (
- "➡️ スキップしました。"
- ),
- "playing": "▶️ 再生中...",
- "back": (
- "⬅️ 前のトラックに戻りました。"
- ),
- "paused": "❌ 一時停止",
- "restarted": (
- "✅️ 最初から再生します。"
- ),
- "liked": (
- "❤️ お気に入りに追加しました。"
- ),
- "unlike": (
- "❌"
- " お気に入りから削除しました。"
- ),
- "err": (
- "❌ エラーが発生しました。"
- "\n{}"
- ),
- "already_authed": (
- "❌ 既に認証されています。"
- ),
- "authed": (
- "✅ 認証に成功しました。"
- ),
- "deauth": (
- "🚪 ログアウトしました。"
- ),
- "auth": (
- '🔗 リンクをクリックしてアクセスを許可し、取得したURLを使って .scode https://... を入力してください。'
- ),
- "no_music": (
- "❌ 音楽は再生されていません！"
- ),
- "dl_err": (
- "❌ トラックのダウンロードに失敗しました。"
- ),
- "volume_changed": (
- "🔊"
- " 音量を {}% に変更しました。"
- ),
- "volume_invalid": (
- "❌ 音量は0から100の数字で指定してください。"
- ),
- "volume_err": (
- "❌ 音量の変更中にエラーが発生しました。"
- ),
- "no_volume_arg": (
- "❌ 0から100の間で音量を指定してください。"
- ),
- "searching_tracks": (
- "🕔 {} を検索中..."
- ),
- "no_search_query": (
- "❌ 検索キーワードを指定してください。"
- ),
- "no_tracks_found": (
- "❌ {} は見つかりませんでした。"
- ),
- "search_results": (
- "✅ {} の検索結果:\n\n{}"
- ),
- "downloading_search_track": (
- "🕔 {} をダウンロード中..."
- ),
- "download_success": (
- "✅ {} - {} のダウンロードに成功しました。"
- ),
- "invalid_track_number": (
- "❌ トラック番号が無効です。"
- " 先に検索するか、リストから有効な番号を指定してください。"
- ),
- "device_list": (
- "📄 利用可能なデバイス:\n{}"
- ),
- "no_devices_found": (
- "❌ デバイスが見つかりません。"
- ),
- "device_changed": (
- "✅ 再生デバイスを"
- " {} に切り替えました。"
- ),
- "invalid_device_id": (
- "❌ デバイスIDが無効です。"
- " .sdevice で利用可能なデバイスを確認してください。"
- ),
- "search_results_cleared": "✅ 検索結果をクリアしました。",
- "autobio": (
- "🎧 Spotify AutoBio: {}"
- ),
- "no_ytdlp": "❌ yt-dlpが見つかりません... 設定を確認するか、インストールしてください ({}terminal pip install yt-dlp)",
- "snowt_failed": "\n\n❌ ダウンロードに失敗しました。",
- "uploading_banner": "\n\n🕔 バナーをアップロード中...",
- "downloading_track": "\n\n🕔 トラックをダウンロード中...",
- "no_playlists": "❌ プレイリストが見つかりません。",
- "playlists_list": "📄 あなたのプレイリスト:\n\n{}",
- "added_to_playlist": "✅ {} を {} に追加しました。",
- "removed_from_playlist": "✅ {} を {} から削除しました。",
- "invalid_playlist_index": "❌ プレイリスト番号が無効です。",
- "no_cached_playlists": "❌ 先に .splaylists を使用してください。",
- "playlist_created": "✅ プレイリスト {} を作成しました。",
- "playlist_deleted": "✅ プレイリスト {} を削除しました。",
- "no_playlist_name": "❌ プレイリスト名を指定してください。",
- }
-
- def __init__(self):
- self._client_id = "e0708753ab60499c89ce263de9b4f57a"
- self._client_secret = "80c927166c664ee98a43a2c0e2981b4a"
- self.scope = (
- "user-read-playback-state playlist-read-private playlist-read-collaborative"
- " user-modify-playback-state user-library-modify"
- " playlist-modify-public playlist-modify-private"
- )
- self.sp_auth = spotipy.oauth2.SpotifyOAuth(
- client_id=self._client_id,
- client_secret=self._client_secret,
- redirect_uri="https://thefsch.github.io/spotify/",
- scope=self.scope,
- )
- self.config = loader.ModuleConfig(
- loader.ConfigValue(
- "show_banner",
- True,
- "Show banner with track info",
- validator=loader.validators.Boolean(),
- ),
- loader.ConfigValue(
- "custom_text",
- (
- "🎧 Now playing: {track} — {artists}\n"
- "🔗 song.link"
- ),
- """Custom text, supports {track}, {artists}, {album}, {playlist}, {playlist_owner}, {spotify_url}, {songlink}, {progress}, {duration}, {device} placeholders""",
- validator=loader.validators.String(),
- ),
- loader.ConfigValue(
- "font",
- "https://raw.githubusercontent.com/kamekuro/assets/master/fonts/Onest-Bold.ttf",
- "Custom font. Specify URL to .ttf file",
- validator=loader.validators.String(),
- ),
- loader.ConfigValue(
- "auto_bio_template",
- "🎧 {}",
- lambda: "Template for Spotify AutoBio",
- ),
- loader.ConfigValue(
- "ytdlp_path",
- "",
- "Path to ytdlp binary",
- validator=loader.validators.String(),
- ),
- loader.ConfigValue(
- "banner_version",
- "horizontal",
- lambda: "Banner version",
- validator=loader.validators.Choice(["horizontal", "vertical"]),
- ),
- )
-
- async def client_ready(self, client, db):
- self.font_ready = asyncio.Event()
-
- self._premium = getattr(await client.get_me(), "premium", False)
- try:
- self.sp = spotipy.Spotify(auth=self.get("acs_tkn")["access_token"])
- except Exception:
- self.set("acs_tkn", None)
- self.sp = None
-
- if self.get("autobio", False):
- self.autobio.start()
-
- def tokenized(func) -> FunctionType:
- @functools.wraps(func)
- async def wrapped(*args, **kwargs):
- if not args[0].get("acs_tkn", False) or not args[0].sp:
- await utils.answer(args[1], args[0].strings("need_auth"))
- return
-
- return await func(*args, **kwargs)
-
- wrapped.__doc__ = func.__doc__
- wrapped.__module__ = func.__module__
-
- return wrapped
-
- def error_handler(func) -> FunctionType:
- @functools.wraps(func)
- async def wrapped(*args, **kwargs):
- try:
- return await func(*args, **kwargs)
- except Exception:
- logger.exception(traceback.format_exc())
- with contextlib.suppress(Exception):
- await utils.answer(
- args[1],
- args[0].strings("err").format(traceback.format_exc()),
- )
-
- wrapped.__doc__ = func.__doc__
- wrapped.__module__ = func.__module__
-
- return wrapped
-
-
- @loader.loop(interval=90)
- async def autobio(self):
- try:
- current_playback = self.sp.current_playback()
- track = current_playback["item"]["name"]
- track = re.sub(r"([(].*?[)])", "", track).strip()
- except Exception:
- return
-
- bio = self.config["auto_bio_template"].format(f"{track}")
-
- try:
- await self._client(
- UpdateProfileRequest(about=bio[: 140 if self._premium else 70])
- )
- except FloodWaitError as e:
- logger.info(f"Sleeping {max(e.seconds, 60)} bc of floodwait")
- await asyncio.sleep(max(e.seconds, 60))
- return
-
- async def _download_track(self, message, query: str, caption: str = ""):
- dl_dir = os.path.join(os.getcwd(), "spotifymod")
- if not os.path.exists(dl_dir):
- os.makedirs(dl_dir, exist_ok=True)
-
- for f in os.listdir(dl_dir):
- try:
- os.remove(os.path.join(dl_dir, f))
- except:
- pass
-
- try:
- squery = query.replace('"', '').replace("'", "")
-
- cmd = (
- f'{self.config["ytdlp_path"]} -x --audio-format mp3 --add-metadata '
- f'-o "{dl_dir}/%(title)s [%(id)s].%(ext)s" '
- f'"ytsearch1:{squery}"'
- )
-
- proc = await asyncio.create_subprocess_shell(
- cmd,
- stdout=asyncio.subprocess.PIPE,
- stderr=asyncio.subprocess.PIPE
- )
- await proc.communicate()
-
- files = [f for f in os.listdir(dl_dir) if f.endswith(".mp3")]
-
- if files:
- target_file = os.path.join(dl_dir, files[0])
- await utils.answer(message, caption, file=target_file)
- else:
- await utils.answer(message, self.strings("snowt_failed"))
-
- except Exception as e:
- logger.error(e)
- await utils.answer(message, self.strings("dl_err"))
-
- finally:
- if os.path.exists(dl_dir):
- for f in os.listdir(dl_dir):
- try:
- os.remove(os.path.join(dl_dir, f))
- except:
- pass
-
-
- @error_handler
- @tokenized
- @loader.command(
- ru_doc="- ➕ Добавить текущий трек в плейлист (используйте номер из .splaylists)"
- )
- async def splaylistadd(self, message: Message):
- """- ➕ Add current track to playlist (use number from .splaylists)"""
- args = utils.get_args_raw(message)
- if not args or not args.isdigit():
- await utils.answer(message, self.strings("invalid_playlist_index"))
- return
-
- index = int(args) - 1
- playlists = self.get("last_playlists", [])
-
- if not playlists:
- await utils.answer(message, self.strings("no_cached_playlists"))
- return
-
- if index < 0 or index >= len(playlists):
- await utils.answer(message, self.strings("invalid_playlist_index"))
- return
-
- current = self.sp.current_playback()
- if not current or not current.get("item"):
- await utils.answer(message, self.strings("no_music"))
- return
-
- track_uri = current["item"]["uri"]
- track_name = current["item"]["name"]
- artists = ", ".join([a["name"] for a in current["item"]["artists"]])
- full_track_name = f"{artists} - {track_name}"
-
- playlist_id = playlists[index]["id"]
- playlist_name = playlists[index]["name"]
-
- try:
- self.sp.playlist_add_items(playlist_id, [track_uri])
- except spotipy.exceptions.SpotifyException as e:
- if e.http_status == 403 and "Insufficient client scope" in str(e):
- await utils.answer(message, self.strings("need_auth"))
- return
- raise e
-
- await utils.answer(message, self.strings("added_to_playlist").format(utils.escape_html(full_track_name), utils.escape_html(playlist_name)))
-
- @error_handler
- @tokenized
- @loader.command(
- ru_doc="- ➖ Удалить текущий трек из плейлиста (используйте номер из .splaylists)"
- )
- async def splaylistrem(self, message: Message):
- """- ➖ Remove current track from playlist (use number from .splaylists)"""
- args = utils.get_args_raw(message)
- if not args or not args.isdigit():
- await utils.answer(message, self.strings("invalid_playlist_index"))
- return
-
- index = int(args) - 1
- playlists = self.get("last_playlists", [])
-
- if not playlists:
- await utils.answer(message, self.strings("no_cached_playlists"))
- return
-
- if index < 0 or index >= len(playlists):
- await utils.answer(message, self.strings("invalid_playlist_index"))
- return
-
- current = self.sp.current_playback()
- if not current or not current.get("item"):
- await utils.answer(message, self.strings("no_music"))
- return
-
- track_uri = current["item"]["uri"]
- track_name = current["item"]["name"]
- artists = ", ".join([a["name"] for a in current["item"]["artists"]])
- full_track_name = f"{artists} - {track_name}"
-
- playlist_id = playlists[index]["id"]
- playlist_name = playlists[index]["name"]
-
- try:
- self.sp.playlist_remove_all_occurrences_of_items(playlist_id, [track_uri])
- except spotipy.exceptions.SpotifyException as e:
- if e.http_status == 403 and "Insufficient client scope" in str(e):
- await utils.answer(message, self.strings("need_auth"))
- return
- raise e
-
- await utils.answer(message, self.strings("removed_from_playlist").format(utils.escape_html(full_track_name), utils.escape_html(playlist_name)))
-
- @error_handler
- @tokenized
- @loader.command(
- ru_doc="- 🆕 Создать новый плейлист"
- )
- async def splaylistcreate(self, message: Message):
- """- 🆕 Create a new playlist"""
- name = utils.get_args_raw(message)
- if not name:
- await utils.answer(message, self.strings("no_playlist_name"))
- return
-
- user_id = self.sp.me()["id"]
- self.sp.user_playlist_create(user_id, name)
- await utils.answer(message, self.strings("playlist_created").format(utils.escape_html(name)))
-
- @error_handler
- @tokenized
- @loader.command(
- ru_doc="- 🗑 Удалить плейлист (используйте номер из .splaylists)"
- )
- async def splaylistdelete(self, message: Message):
- """- 🗑 Delete playlist (use number from .splaylists)"""
- args = utils.get_args_raw(message)
- if not args or not args.isdigit():
- await utils.answer(message, self.strings("invalid_playlist_index"))
- return
-
- index = int(args) - 1
- playlists = self.get("last_playlists", [])
-
- if not playlists:
- await utils.answer(message, self.strings("no_cached_playlists"))
- return
-
- if index < 0 or index >= len(playlists):
- await utils.answer(message, self.strings("invalid_playlist_index"))
- return
-
- playlist_id = playlists[index]["id"]
- playlist_name = playlists[index]["name"]
-
- self.sp.current_user_unfollow_playlist(playlist_id)
- await utils.answer(message, self.strings("playlist_deleted").format(utils.escape_html(playlist_name)))
-
- @error_handler
- @tokenized
- @loader.command(
- ru_doc="- 📃 Получить все плейлисты"
- )
- async def splaylists(self, message: Message):
- """- 📃 Get all playlists"""
- user_id = self.sp.me()["id"]
- playlists = self.sp.current_user_playlists()
-
- editable_playlists = []
- for playlist in playlists["items"]:
- if playlist["owner"]["id"] == user_id or playlist["collaborative"]:
- editable_playlists.append(playlist)
-
- self.set("last_playlists", editable_playlists)
-
- playlist_list_text = ""
- for i, playlist in enumerate(editable_playlists):
- name = utils.escape_html(playlist["name"])
- url = playlist["external_urls"]["spotify"]
- count = playlist["tracks"]["total"]
- playlist_list_text += f"{i + 1}. {name} ({count} tracks)\n"
-
- if not playlist_list_text:
- await utils.answer(message, self.strings("no_playlists"))
- else:
- await utils.answer(message, self.strings("playlists_list").format(playlist_list_text))
-
- @error_handler
- @tokenized
- @loader.command(
- ru_doc="- ℹ️ Переключить стриминг воспроизведения в био"
- )
- async def sbiocmd(self, message: Message):
- """- ℹ️ Toggle bio playback streaming"""
- current = self.get("autobio", False)
- new = not current
- self.set("autobio", new)
- await utils.answer(
- message,
- self.strings("autobio").format("enabled" if new else "disabled"),
- )
-
- if new:
- self.autobio.start()
- else:
- self.autobio.stop()
-
- @error_handler
- @tokenized
- @loader.command(
- ru_doc="- 🔊 Изменить громкость. .svolume <0-100>"
- )
- async def svolume(self, message: Message):
- """- 🔊 Change playback volume. .svolume <0-100>"""
- try:
- args = utils.get_args_raw(message)
- if not args:
- await utils.answer(message, self.strings("no_volume_arg"))
- return
-
- volume_percent = int(args)
- if 0 <= volume_percent <= 100:
- self.sp.volume(volume_percent)
- await utils.answer(message, self.strings("volume_changed").format(volume_percent))
- else:
- await utils.answer(message, self.strings("volume_invalid"))
- except ValueError:
- await utils.answer(message, self.strings("volume_invalid"))
- except Exception:
- await utils.answer(message, self.strings("volume_err"))
-
- @error_handler
- @tokenized
- @loader.command(
- ru_doc=(
- "- 🎵 Выбрать устройство для воспроизведения. Например: .sdevice \n"
- "- 📝 Показать список устройств: .sdevice"
- )
- )
- async def sdevicecmd(self, message: Message):
- """- 🎵 Set preferred playback device. Usage: .sdevice or .sdevice to list devices"""
- args = utils.get_args_raw(message)
- devices = self.sp.devices()["devices"]
-
- if not args:
- if not devices:
- await utils.answer(message, self.strings("no_devices_found"))
- return
-
- device_list_text = ""
- for i, device in enumerate(devices):
- is_active = "(active)" if device["is_active"] else ""
- device_list_text += (
- f"{i+1}. {device['name']}"
- f" ({device['type']}) {is_active}\n"
- )
-
- await utils.answer(message, self.strings("device_list").format(device_list_text.strip()))
- return
-
- device_id = None
- try:
- device_number = int(args)
- if 0 < device_number <= len(devices):
- device_id = devices[device_number - 1]["id"]
- device_name = devices[device_number - 1]["name"]
- else:
- await utils.answer(message, self.strings("invalid_device_id"))
- return
- except ValueError:
- found_device = next((d for d in devices if d["id"] == args.strip()), None)
- if found_device:
- device_id = found_device["id"]
- device_name = found_device["name"]
- else:
- await utils.answer(message, self.strings("invalid_device_id"))
- return
-
- self.sp.transfer_playback(device_id=device_id)
- await utils.answer(message, self.strings("device_changed").format(device_name))
-
- @error_handler
- @tokenized
- @loader.command(
- ru_doc="- 💫 Включить повтор трека"
- )
- async def srepeatcmd(self, message: Message):
- """- 💫 Repeat"""
- self.sp.repeat("track")
- await utils.answer(message, self.strings("on-repeat"))
-
- @error_handler
- @tokenized
- @loader.command(
- ru_doc="- ✋ Остановить повтор"
- )
- async def sderepeatcmd(self, message: Message):
- """- ✋ Stop repeat"""
- self.sp.repeat("context")
- await utils.answer(message, self.strings("off-repeat"))
-
- @error_handler
- @tokenized
- @loader.command(
- ru_doc="- 👉 Следующий трек"
- )
- async def snextcmd(self, message: Message):
- """- 👉 Next track"""
- self.sp.next_track()
- await utils.answer(message, self.strings("skipped"))
-
- @error_handler
- @tokenized
- @loader.command(
- ru_doc="- 🤚 Продолжить воспроизведение"
- )
- async def sresumecmd(self, message: Message):
- """- 🤚 Resume"""
- self.sp.start_playback()
- await utils.answer(message, self.strings("playing"))
-
- @error_handler
- @tokenized
- @loader.command(
- ru_doc="- 🤚 Пауза"
- )
- async def spausecmd(self, message: Message):
- """- 🤚 Pause"""
- self.sp.pause_playback()
- await utils.answer(message, self.strings("paused"))
-
- @error_handler
- @tokenized
- @loader.command(
- ru_doc="- ⏮ Предыдущий трек"
- )
- async def sbackcmd(self, message: Message):
- """- ⏮ Previous track"""
- self.sp.previous_track()
- await utils.answer(message, self.strings("back"))
-
- @error_handler
- @tokenized
- @loader.command(
- ru_doc="- ⏪ Перезапустить трек"
- )
- async def sbegincmd(self, message: Message):
- """- ⏪ Restart track"""
- self.sp.seek_track(0)
- await utils.answer(message, self.strings("restarted"))
-
- @error_handler
- @tokenized
- @loader.command(
- ru_doc="- ❤️ Лайкнуть играющий трек"
- )
- async def slikecmd(self, message: Message):
- """- ❤️ Like current track"""
- cupl = self.sp.current_playback()
- self.sp.current_user_saved_tracks_add([cupl["item"]["id"]])
- await utils.answer(message, self.strings("liked"))
-
- @error_handler
- @tokenized
- @loader.command(
- ru_doc="- 💔 Убрать лайк с играющего трека"
- )
- async def sunlikecmd(self, message: Message):
- """- 💔 Unlike current track"""
- cupl = self.sp.current_playback()
- self.sp.current_user_saved_tracks_delete([cupl["item"]["id"]])
- await utils.answer(message, self.strings("unlike"))
-
- @error_handler
- @loader.command(
- ru_doc="- Получить ссылку для авторизации"
- )
- async def sauthcmd(self, message: Message):
- """- Get authorization link"""
- if self.get("acs_tkn", False) and not self.sp:
- await utils.answer(message, self.strings("already_authed"))
- else:
- self.sp_auth.get_authorize_url()
- await utils.answer(
- message,
- self.strings("auth").format(self.sp_auth.get_authorize_url()),
- )
-
- @error_handler
- @loader.command(
- ru_doc="- Вставить код авторизации"
- )
- async def scodecmd(self, message: Message):
- """- Paste authorization code"""
- url = message.message.split(" ")[1]
- code = self.sp_auth.parse_auth_response_url(url)
- self.set("acs_tkn", self.sp_auth.get_access_token(code, True, False))
- self.sp = spotipy.Spotify(auth=self.get("acs_tkn")["access_token"])
- await utils.answer(message, self.strings("authed"))
-
- @error_handler
- @loader.command(
- ru_doc="- Выйти из аккаунта"
- )
- async def unauthcmd(self, message: Message):
- """- Log out of account"""
- self.set("acs_tkn", None)
- del self.sp
- await utils.answer(message, self.strings("deauth"))
-
- @error_handler
- @tokenized
- @loader.command(
- ru_doc="- Обновить токен авторизации"
- )
- async def stokrefreshcmd(self, message: Message):
- """- Refresh authorization token"""
- self.set(
- "acs_tkn",
- self.sp_auth.refresh_access_token(self.get("acs_tkn")["refresh_token"]),
- )
- self.set("NextRefresh", time.time() + 45 * 60)
- self.sp = spotipy.Spotify(auth=self.get("acs_tkn")["access_token"])
- await utils.answer(message, self.strings("authed"))
-
- @error_handler
- @tokenized
- @loader.command(
- ru_doc="- 🎧 Показать карточку играющего трека"
- )
- async def snowcmd(self, message: Message):
- """- 🎧 View current track card."""
- current_playback = self.sp.current_playback()
- if not current_playback or not current_playback.get("is_playing", False):
- await utils.answer(message, self.strings("no_music"))
- return
-
- track = current_playback["item"]["name"]
- track_id = current_playback["item"]["id"]
- artists = ", ".join([a["name"] for a in current_playback["item"]["artists"]])
- album_name = current_playback["item"]["album"].get("name", "Unknown Album")
- duration_ms = current_playback["item"].get("duration_ms", 0)
- progress_ms = current_playback.get("progress_ms", 0)
-
- duration = f"{duration_ms//1000//60}:{duration_ms//1000%60:02}"
- progress = f"{progress_ms//1000//60}:{progress_ms//1000%60:02}"
-
- spotify_url = f"https://open.spotify.com/track/{track_id}"
- songlink = f"https://song.link/s/{track_id}"
-
- try:
- device_raw = (
- current_playback["device"]["name"]
- + " "
- + current_playback["device"]["type"].lower()
- )
- device = device_raw.replace("computer", "").replace("smartphone", "").strip()
- except Exception:
- device = None
-
- try:
- playlist_id = current_playback["context"]["uri"].split(":")[-1]
- playlist = self.sp.playlist(playlist_id)
- playlist_name = playlist.get("name", None)
- try:
- playlist_owner = (
- f''
- f'{playlist["owner"]["display_name"]}'
- )
- except KeyError:
- playlist_owner = playlist.get("owner", {}).get("display_name", "")
- except Exception:
- playlist_name = ""
- playlist_owner = ""
-
- text = self.config["custom_text"].format(
- track=utils.escape_html(track),
- artists=utils.escape_html(artists),
- album=utils.escape_html(album_name),
- duration=duration,
- progress=progress,
- device=device,
- spotify_url=spotify_url,
- songlink=songlink,
- playlist=utils.escape_html(playlist_name) if playlist_name else "",
- playlist_owner=playlist_owner or "",
- )
-
- if self.config["show_banner"]:
- cover_url = current_playback["item"]["album"]["images"][0]["url"]
-
- tmp_msg = await utils.answer(message, text + self.strings("uploading_banner"))
-
- banners = Banners(
- title=track,
- artists=artists,
- duration=duration_ms,
- progress=progress_ms,
- track_cover=requests.get(cover_url).content,
- font=self.config["font"],
- )
- file = getattr(banners, self.config["banner_version"], banners.horizontal)()
-
- await utils.answer(tmp_msg, text, file=file)
- else:
- await utils.answer(message, text)
-
- @error_handler
- @tokenized
- @loader.command(
- ru_doc="- 🎧 Скачать играющий трек"
- )
- async def snowtcmd(self, message: Message):
- """- 🎧 Download current track."""
- current_playback = self.sp.current_playback()
- if not current_playback or not current_playback.get("is_playing", False):
- await utils.answer(message, self.strings("no_music"))
- return
-
- track = current_playback["item"]["name"]
- artists = ", ".join([a["name"] for a in current_playback["item"]["artists"]])
- album_name = current_playback["item"]["album"].get("name", "Unknown Album")
- duration_ms = current_playback["item"].get("duration_ms", 0)
- progress_ms = current_playback.get("progress_ms", 0)
-
- duration = f"{duration_ms//1000//60}:{duration_ms//1000%60:02}"
- progress = f"{progress_ms//1000//60}:{progress_ms//1000%60:02}"
-
- spotify_url = f"https://open.spotify.com/track/{current_playback['item']['id']}"
- songlink = f"https://song.link/s/{current_playback['item']['id']}"
-
- try:
- device_raw = (
- current_playback["device"]["name"]
- + " "
- + current_playback["device"]["type"].lower()
- )
- device = device_raw.replace("computer", "").replace("smartphone", "").strip()
- except Exception:
- device = None
-
- try:
- playlist_id = current_playback["context"]["uri"].split(":")[-1]
- playlist = self.sp.playlist(playlist_id)
- playlist_name = playlist.get("name", None)
- try:
- playlist_owner = (
- f''
- f'{playlist["owner"]["display_name"]}'
- )
- except KeyError:
- playlist_owner = playlist.get("owner", {}).get("display_name", "")
- except Exception:
- playlist_name = ""
- playlist_owner = ""
-
- text = self.config["custom_text"].format(
- track=utils.escape_html(track),
- artists=utils.escape_html(artists),
- album=utils.escape_html(album_name),
- duration=duration,
- progress=progress,
- device=device,
- spotify_url=spotify_url,
- songlink=songlink,
- playlist=utils.escape_html(playlist_name) if playlist_name else "",
- playlist_owner=playlist_owner or "",
- )
-
- msg = await utils.answer(message, text + self.strings("downloading_track"))
-
- await self._download_track(msg, f"{artists} {track}", caption=text)
-
- @error_handler
- @tokenized
- @loader.command(
- ru_doc=(
- "- 🔍 Поиск треков. Например: .ssearch Imagine Dragons Believer\n"
- "- 🎧 Скачать трек: .ssearch 1 (где 1 — номер трека из списка)"
- )
- )
- async def ssearchcmd(self, message: Message):
- """🔍 Search for tracks. Usage: .ssearch or .ssearch to download"""
- args = utils.get_args_raw(message)
- if not args:
- await utils.answer(message, self.strings("no_search_query"))
- return
-
- try:
- track_number = int(args)
- search_results = self.get("last_search_results", [])
-
- if not search_results:
- await utils.answer(message, self.strings("no_tracks_found"))
- return
-
- if track_number <= 0 or track_number > len(search_results):
- raise ValueError
-
- msg = await utils.answer(message, self.strings("downloading_track"))
-
- track_info = search_results[track_number - 1]
- track_name = track_info["name"]
- artists = ", ".join([a["name"] for a in track_info["artists"]])
-
- caption_text = self.strings("download_success").format(
- utils.escape_html(track_name),
- utils.escape_html(artists)
- )
-
- await self._download_track(msg, f"{artists} {track_name}", caption=caption_text)
- return
-
- except ValueError:
- await utils.answer(message, self.strings("searching_tracks").format(args))
-
- results = self.sp.search(q=args, limit=5, type="track")
-
- if not results or not results["tracks"]["items"]:
- await utils.answer(message, self.strings("no_tracks_found").format(args))
- return
-
- self.set("last_search_results", results["tracks"]["items"])
-
- tracks_list = []
- for i, track in enumerate(results["tracks"]["items"]):
- track_name = track["name"]
- artists = ", ".join([artist["name"] for artist in track["artists"]])
- track_url = track["external_urls"]["spotify"]
- tracks_list.append(
- "{number}. {track_name} — {artists}\n🔗 Spotify".format(
- number=i + 1,
- track_name=utils.escape_html(track_name),
- artists=utils.escape_html(artists),
- track_url=track_url,
- )
- )
-
- text = "\n".join(tracks_list)
- await utils.answer(message, self.strings("search_results").format(args, text))
-
-
- @loader.command(
- ru_doc="- 🔄 Сброс результатов поиска по трекам"
- )
- async def ssearchresetcmd(self, message: Message):
- """- 🔄 Reset track search results"""
- self.set("last_search_results", [])
- await utils.answer(message, self.strings["search_results_cleared"])
-
- async def watcher(self, message: Message):
- """Watcher is used to update token"""
- if not self.sp:
- return
-
- if self.get("NextRefresh", False):
- ttc = self.get("NextRefresh", 0)
- crnt = time.time()
- if ttc < crnt:
- self.set(
- "acs_tkn",
- self.sp_auth.refresh_access_token(
- self.get("acs_tkn")["refresh_token"]
- ),
- )
- self.set("NextRefresh", time.time() + 45 * 60)
- self.sp = spotipy.Spotify(auth=self.get("acs_tkn")["access_token"])
- else:
- self.set(
- "acs_tkn",
- self.sp_auth.refresh_access_token(self.get("acs_tkn")["refresh_token"]),
- )
- self.set("NextRefresh", time.time() + 45 * 60)
- self.sp = spotipy.Spotify(auth=self.get("acs_tkn")["access_token"])
+# License Violation. Original module has GPL-v3 but author of nowhere tried to change it to CC BY-ND 4.0