import asyncio import logging import re import uuid from pathlib import Path, PurePath from sys import platform import discord import requests import yt_dlp from constants import ( FILE_SERVICE_ADDRESS, GET_MP3, GET_PLAYLIST, MUSIC_FOLDER, SEND_MP3, SPOTIFY_CTRL, YOUTUBE_AUTH, ) from spotify_dl import spotify from spotify_dl import youtube as youtube_download 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, logger_name) -> None: self.music_file_list = [] self.file_service_active = False self.logger = logging.getLogger("discord") self.logger.info("Created Playlist organizer class") async def refresh_file_list(self, bot): """ 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") finally: if self.file_service_active: self.logger.info("Radio Status: Probably Active") status = discord.Status.online # radio_hardkor = discord.Activity( # name = "Radio Hammer/Radio Conjurer/Wolne Bałuity Kurwa", # url = "http://95.175.16.246:666/mp3-stream", # type = discord.ActivityType.streaming, # platform = "Liquidsoap+RaspberyPi+RolandSeratoDJ202", # state = "Where the f*** is the DJ booth?", # details= "This Is Radio Hardkor based in Wolne Księstwo Bałuty, the best pirate radiostation on both sides of Łódka river", # buttons= [{"label":"RADIO", "url":"http://95.175.16.246:666/mp3-stream"}], # assets= {"large_image": "fuckery.jpg", "large_text":"Fuckewry", "small_image":"willowisp.png", "small_text":"Hi!"} # ) radio_hardkor = discord.Streaming( name="Radio Hardkor", url="http://95.175.16.246:666/mp3-stream" ) await bot.change_presence(status=status, activity=radio_hardkor) else: self.logger.info("Radio Status: Rather Unknown") await bot.change_presence( activity=discord.Game(name="Axe Throwing Darts") ) 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("discord") 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 """ logger = logging.getLogger("discord") 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="", YT_AUTH = YOUTUBE_AUTH ) _ = 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) or re.match(".*youtu.be.*", 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 = { "username": YOUTUBE_AUTH[0], "password": YOUTUBE_AUTH[1], "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 async def search_music(ctx, how_many=0, slowa_kluczowe=None): """ Take in a context object and an optional integer parameter "how_many" and perform search operation on music library. :param ctx: ctx stands for "context" and is a parameter commonly used in Discord.py commands. It represents the context in which the command was invoked, including information such as the message, the channel, the server, and the user who invoked the command :param how_many: how_many is a parameter that specifies the number of search results to be returned. It is an optional parameter with a default value of 0, which means that if no value is provided for how_many, the function will return all search results, defaults to 0 (optional) """ # i sieć neuronową z możliwością wyrażenia opinii o dopasowaniu. logger = logging.getLogger("discord") if slowa_kluczowe: word_list = slowa_kluczowe else: word_list = ctx.message.content.split() logger.info("Wyszukuje") jrequest = { "lista_slow": word_list, "dlugosc_plejlisty": how_many, "UUID": str(uuid.uuid4()), } coroutine = asyncio.to_thread( requests.post, f"{FILE_SERVICE_ADDRESS}{GET_PLAYLIST}", json=jrequest, timeout=360, ) return_data = await coroutine if not return_data.status_code == 200: await ctx.send("Wołaj szefa - coś się wyjebało") return logger.info(return_data.json()["data"]) return return_data.json()["data"]