mirror of
https://github.com/migatu/conjurer.git
synced 2026-07-14 21:38:38 +00:00
248 lines
10 KiB
Python
248 lines
10 KiB
Python
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
|