conjurer/music_functions.py

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,
)
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="",
            )
            _ = 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 = {
                "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"]