import logging import asyncio from pathlib import Path, PurePath from sys import platform from spotify_dl import spotify from spotify_dl import youtube as youtube_download import re import netrc import spotipy from spotipy.oauth2 import SpotifyClientCredentials from constants import NETRC_FILE, MUSIC_FOLDER, FILE_SERVICE_ADDRESS, GET_MP3, SEND_MP3 import yt_dlp import requests netrc_mod = netrc.netrc(NETRC_FILE) REMOTE_HOST_NAME = "spotipy" authTokens = netrc_mod.authenticators(REMOTE_HOST_NAME) logger = logging.getLogger("discord") spotify_ctrl = spotipy.Spotify( client_credentials_manager=SpotifyClientCredentials( client_id=authTokens[0], client_secret=authTokens[2], ) ) class MusicFileList(object): """ The `MusicFileList` class manages a list of music files, with the ability to refresh the list from a file service or local directory and update the list with new items. """ def __init__(self, uplogger) -> None: self.music_file_list = [] self.file_service_active = False self.logger = uplogger self.logger.info("Created Playlist organizer class") def refresh_file_list(self): """ The `refresh_file_list` function attempts to connect to a file service to retrieve a list of music files, and if unsuccessful, it populates the list by scanning a local directory for .mp3 files. """ try: self.logger.info("Attempt to connect to file service") response = requests.get(f"{FILE_SERVICE_ADDRESS}{GET_MP3}", timeout=360) self.music_file_list = response.json()["music_file_list"] self.file_service_active = True except requests.exceptions.RequestException as e: for mp3_item in Path.glob(Path(MUSIC_FOLDER), "**/*.mp3"): temp_music_file = mp3_item.as_posix() if platform == "win32": temp_music_file = temp_music_file.replace("/", "\\") self.music_file_list.append(temp_music_file) self.file_service_active = False self.logger.error(e.strerror) self.logger.error("Service Unavailable") def get_file_list(self): """ The `get_file_list` function returns the list of music files stored in the object. :return: The `music_file_list` attribute is being returned. """ return self.music_file_list def update_file_list(self, item): """ The `update_file_list` function appends an item to a music file list and sends a POST request with the item data to a file service address. :param item: The `item` parameter in the `update_file_list` method is the file that you want to add to the `music_file_list` attribute of the class instance. It is then sent as a JSON payload in a POST request to a file service address along with the endpoint `SEND_MP3` """ self.music_file_list.append(item) post_data = {"item": str(item)} requests.post(f"{FILE_SERVICE_ADDRESS}{SEND_MP3}", json=post_data, timeout=360) MUSIC_FILE_LIST = MusicFileList(logger) async def get_file(ctx, source, link): """ Take in three parameters: "ctx" (context), "source" (a string representing the source of the file), and "link" (a string representing the link to the file). Execute it in asynchronouse way. And get file from spotify or youtube. :param ctx: ctx stands for "context" and is a parameter commonly used in Discord.py commands. It represents the context in which a command is being executed, including information such as the message, the channel, the server, and the user who invoked the command :param source: The source parameter is likely a string that represents the source of the file that the user wants to retrieve. This could be a website, a cloud storage service, or any other location where the file is stored :param link: The "link" parameter is likely a string that represents a URL or file path to the location of the file that the function is trying to retrieve """ item_id = None item_type = None file_list = [] if source == "Spotify": dir_path = None item_type, item_id = spotify.parse_spotify_url(link) if item_type in ["album", "track", "playlist"] and item_id: logger.info(item_type) logger.info(item_id) logger.info("Spotify link provided") file_list = await asyncio.to_thread( spotify.fetch_tracks, spotify_ctrl, item_type, link ) logger.info(file_list) directory_name = await asyncio.to_thread( spotify.get_item_name, spotify_ctrl, item_type, item_id ) logger.info(directory_name) logger.info("Spotify scrape done") url_data = {"urls": []} url_dict = {} url_dict["save_path"] = Path( PurePath.joinpath(Path(MUSIC_FOLDER), Path(directory_name)) ) url_dict["save_path"].mkdir(parents=True, exist_ok=True) url_dict["songs"] = file_list url_data["urls"].append(url_dict.copy()) file_name_f = youtube_download.default_filename logger.info("YT-DL") coro = asyncio.to_thread( youtube_download.download_songs, songs=url_data, output_dir=MUSIC_FOLDER, format_str="bestaudio/best", skip_mp3=False, keep_playlist_order=False, no_overwrites=True, remove_trailing_tracks="no", use_sponsorblock="yes", file_name_f=file_name_f, multi_core=0, proxy="", ) _ = await coro logger.info("YT-DL done") dir_path = (url_dict["save_path"].resolve()).as_posix() if platform == "win32": dir_path = dir_path.replace("/", "\\") return dir_path, file_list await ctx.send( "No se jaja chyba robisz, Ty byś spotifaja nie poznawszy jakby z krzaków wyskoczył i w dupę kopnął. Plejlista, kawałek albo album prosze." ) logger.error("Wrong link provided spotify: %s", link) elif source == "Youtube": link_ok = False pat = r"^https?:\/\/(?:www\.)?[-a-zA-Z0-9@:%._\+~#=]{1,256}\.[a-zA-Z0-9()]{1,6}\b(?:[-a-zA-Z0-9()@:%_\+.~#?&\/=]*)$" if re.match(pat, link) and re.match(".*youtube.*", link): link_ok = True if link_ok: dir_path = "" file_list = [] query = link sponsorblock_postprocessor = [ { "key": "SponsorBlock", "categories": ["skip_non_music_sections"], }, { "key": "ModifyChapters", "remove_sponsor_segments": ["music_offtopic"], "force_keyframes": True, }, ] # TODO: Make sponsorblock work sponsorblock_postprocessor = [] if platform == "win32": dir_path = "G:\\Muzyka\\Youtube" else: dir_path = "/home/pi/RetroPie/mp3/Youtube" ydl_opts = { "proxy": "", "default_search": "ytsearch", "format": "bestaudio/best", "postprocessors": sponsorblock_postprocessor, "noplaylist": True, "no_color": False, "paths": {"home": dir_path}, } mp3_postprocess_opts = { "key": "FFmpegExtractAudio", "preferredcodec": "mp3", "preferredquality": "192", } ydl_opts["postprocessors"].append(mp3_postprocess_opts.copy()) with yt_dlp.YoutubeDL(ydl_opts) as ydl: try: coro = asyncio.to_thread(ydl.download, [query]) _ = await coro except Exception as exce: # pylint: disable=broad-exception-caught logger.error(exce) logger.info( "Failed to download %s, make sure yt_dlp is up to date", link ) extract = re.search("v=(...........)[&,$]*", link) ident = extract.group(1) for item in Path.glob(Path(dir_path), "**/*.mp3"): file = item.as_posix() if platform == "win32": file = file.replace("/", "\\") if re.match(".*" + ident + ".*", file): file_list.append(file) return dir_path, file_list await ctx.send( "No se jaja chyba robisz, Ty byś jutuba nie poznawszy jakby z krzaków wyskoczył i w dupę kopnął. Jutuba to nie tuba którą sobie można gdzieś wsadzić." ) logger.error("Wrong link provided youtube: %s", link) return "", None elif source == "Pornol": dir_path = "" file_list = [] query = link sponsorblock_postprocessor = [ { "key": "SponsorBlock", "categories": ["skip_non_music_sections"], }, { "key": "ModifyChapters", "remove_sponsor_segments": ["music_offtopic"], "force_keyframes": True, }, ] # TODO: Make sponsorblock work sponsorblock_postprocessor = [] dir_path = "/home/pi/RetroPie/movies/porn" ydl_opts = { "postprocessors": sponsorblock_postprocessor, "paths": {"home": dir_path}, } with yt_dlp.YoutubeDL(ydl_opts) as ydl: try: coro = asyncio.to_thread(ydl.download, [query]) _ = await coro except Exception as exce: # pylint: disable=broad-exception-caught logger.error(exce) logger.info( "Failed to download %s, make sure yt_dlp is up to date", link ) file_list.append("wiadomo co wiadomo gdzie") return dir_path, file_list