WIP

#35

ala.py

@@ -0,0 +1,6 @@
+# from searching_utils.spotify_to_youtube import get_url_without_api
+# get_url_without_api("https://open.spotify.com/intl-de/track/3ElbrKKlIkCHTB4hfFXzpb?si=c98ed9eeb77a4140")
+
+from searching_utils import query_to_youtube
+
+print(query_to_youtube.get_url("aposjd"))

ddl_retrievers/spotify_ddl_retriever.py

@@ -4,9 +4,6 @@
 import yt_dlp as youtube_dl
 from models.music_information import MusicInformation
 
-load_dotenv()
-
-spotdl = Spotdl(client_id=os.getenv('SPOTIFY_CLIENT_ID'), client_secret=os.getenv('SPOTIFY_CLIENT_SECRET'))
 
 async def get_streaming_url(spotify_url) -> MusicInformation:    
     song = spotdl.search([spotify_url])

ddl_retrievers/universal_ddl_retriever.py

@@ -10,7 +10,6 @@ async def get_streaming_url(url) -> MusicInformation:
 
     with yt_dlp.YoutubeDL(ydl_opts) as ydl:
         info_dict = ydl.extract_info(url, download=False)
-
         track_link = info_dict['url']
         track_name = info_dict['title']
         track_author = info_dict['uploader']

docker-compose.yml

@@ -10,4 +10,4 @@ services:
       - DISCORD_TOKEN=${DISCORD_TOKEN}
       - SPOTIFY_CLIENT_ID=${SPOTIFY_CLIENT_ID}
       - SPOTIFY_CLIENT_SECRET=${SPOTIFY_CLIENT_SECRET}
-    restart: unless-stopped
+    restart: unless-stopped

enums/platform.py → enums/source.py

@@ -1,9 +1,9 @@
 from enum import Enum
 
-class Platform(Enum):
+class Source(Enum):
     NO_URL = 0
     TIK_TOK = 1
     SPOTIFY = 2
     YOUTUBE = 3
     SOUND_CLOUD = 4
-    ANYTHING_ELSE = 5
+    UNKNOWN_SOURCE = 5

exceptions/discord_exceptions.py

@@ -0,0 +1,2 @@
+class InvalidDiscordTokenError(Exception):
+    pass

exceptions/spotify_exceptions.py

@@ -0,0 +1,5 @@
+class SpotifyRequestError(Exception):
+    pass
+
+class InvalidSpotifyCredentialsError(Exception):
+    pass

main.py

@@ -17,7 +17,10 @@
 import db_utils.db_utils as db_utils
 from discord_utils import embed_generator, player
 from ai_server_utils import rvc_server_checker
+from exceptions.spotify_exceptions import InvalidSpotifyCredentialsError
 from platform_handlers import music_url_getter
+from setup.apply_config import apply_config
+from setup.check_credentials import check_discord_credentials, check_spotify_credentials
 load_dotenv()
 
 model_choices = []
@@ -26,37 +29,20 @@
 # if(isServerRunning):
 #     model_choices, index_choices = rvc_server_checker.fetch_choices()
 
+check_discord_credentials()
+
 intents = discord.Intents.default()
 intents.message_content = True
 
 bot = commands.Bot(command_prefix=[".", "!", "$"], intents=intents, help_command=None)
 
 @bot.event
 async def on_ready():
+    await check_spotify_credentials()
     await setup_db()
+    await apply_config(bot)
     await bot.change_presence(status=discord.Status.do_not_disturb, activity=discord.Activity(type=discord.ActivityType.listening, name="to da kuhle songs"))
     print(f"Bot is ready and logged in as {bot.user.name}")
-
-    config = configparser.ConfigParser()
-    config.read('config.ini')
-
-    ask_in_dms = config.getboolean('Bot', 'AskInDMs', fallback=False)
-    admin_userid = config.getint('Admin', 'UserID', fallback=0)
-
-    if ask_in_dms and admin_userid:
-        user = await bot.fetch_user(admin_userid)
-
-        dm_channel = await user.create_dm()
-
-        messages = await dm_channel.history().flatten()
-
-        for msg in messages:
-            try:
-                await msg.delete()
-            except:
-                print("skiped message")
-
-        await user.send(f"Bot is ready and logged in as {bot.user.name}")
 
 @bot.command(aliases=['next', 'advance', 'skip_song', 'move_on', 'play_next'])
 async def skip(ctx):

platform_handlers/audio_content_type_finder.py

@@ -1,12 +1,12 @@
 from enums.audio_content_type import AudioContentType
-from enums.platform import Platform
+from enums.source import Source
 import requests
 
 from urllib.parse import urlparse
 from urllib.parse import parse_qs
 
-async def get_audio_content_type(query_url: str, platform: Platform) -> AudioContentType:
-    if platform is Platform.YOUTUBE:
+async def get_audio_content_type(query_url: str, platform: Source) -> AudioContentType:
+    if platform is Source.YOUTUBE:
         parse_result = urlparse(query_url)
         query_params = parse_qs(parse_result.query)
         result = query_params.get("list", [None])[0]
@@ -19,7 +19,7 @@ async def get_audio_content_type(query_url: str, platform: Platform) -> AudioCon
         else:
             return AudioContentType.SINGLE_SONG
 
-    elif platform is Platform.SOUND_CLOUD:
+    elif platform is Source.SOUND_CLOUD:
         parse_result = urlparse(query_url)
         path = parse_result.path
         path_segments = path.strip("/").split("/")
@@ -29,7 +29,7 @@ async def get_audio_content_type(query_url: str, platform: Platform) -> AudioCon
         else:
             return AudioContentType.SINGLE_SONG
 
-    elif platform is Platform.SPOTIFY:
+    elif platform is Source.SPOTIFY:
         parse_result = urlparse(query_url)
         path = parse_result.path
         path_segments = path.strip("/").split("/")
@@ -43,10 +43,10 @@ async def get_audio_content_type(query_url: str, platform: Platform) -> AudioCon
         else:
             return AudioContentType.NOT_SUPPORTED
 
-    elif platform is Platform.NO_URL:
+    elif platform is Source.NO_URL:
         return AudioContentType.QUERY
 
-    elif platform is Platform.ANYTHING_ELSE:
+    elif platform is Source.UNKNOWN_SOURCE:
         response = requests.get(query_url)
         contentType = response.headers['content-type']
 
@@ -55,7 +55,7 @@ async def get_audio_content_type(query_url: str, platform: Platform) -> AudioCon
         else:
             return AudioContentType.YT_DLP
 
-    elif platform is Platform.TIK_TOK:
+    elif platform is Source.TIK_TOK:
         return AudioContentType.SINGLE_SONG
 
     else:

platform_handlers/music_platform_finder.py → platform_handlers/music_source_finder.py

@@ -1,24 +1,24 @@
 from urllib.parse import urlsplit
-from enums.platform import Platform
+from enums.source import Source
 
-async def find_platform(query_url: str) -> Platform:
+async def find_music_source(query_url: str) -> Source:
     split_url = urlsplit(query_url)
     hostname = split_url.hostname
 
     if not hostname:
-        return Platform.NO_URL
+        return Source.NO_URL
 
     elif "spotify" in hostname:
-        return Platform.SPOTIFY
+        return Source.SPOTIFY
 
     elif "tiktok" in hostname:
-        return Platform.TIK_TOK
+        return Source.TIK_TOK
 
     elif "youtube" in hostname or "youtu" in hostname:
-        return Platform.YOUTUBE
+        return Source.YOUTUBE
 
     elif "soundcloud" in hostname:
-        return Platform.SOUND_CLOUD
+        return Source.SOUND_CLOUD
 
     else:
-        return Platform.ANYTHING_ELSE
+        return Source.UNKNOWN_SOURCE

platform_handlers/music_url_getter.py

@@ -9,9 +9,9 @@
 from models.music_information import MusicInformation
 import ddl_retrievers
 from platform_handlers.audio_content_type_finder import get_audio_content_type
-from platform_handlers.music_platform_finder import find_platform
+from platform_handlers.music_source_finder import find_music_source
 from enums.audio_content_type import AudioContentType
-from enums.platform import Platform
+from enums.source import Source
 import yt_dlp
 import ytmusicapi
 
@@ -20,19 +20,20 @@
 import os
 
 async def get_streaming_url(query_url: str) -> MusicInformation:
-    platform = await find_platform(query_url)
+    platform = await find_music_source(query_url)
 
-    if platform is Platform.SPOTIFY:
+    if platform is Source.SPOTIFY:
         return await ddl_retrievers.spotify_ddl_retriever.get_streaming_url(query_url)
 
-    elif platform is Platform.TIK_TOK:
+    elif platform is Source.TIK_TOK:
         return await ddl_retrievers.tiktok_ddl_retriever.get_streaming_url(query_url)
 
-    elif platform is Platform.SOUND_CLOUD:
+    elif platform is Source.SOUND_CLOUD:
         parsed_url = urlparse(query_url)
-
         subdomain = parsed_url.hostname.split('.')[0]
 
+        sc_url = query_url
+
         if "api" in subdomain:
             response = requests.get(f"https://w.soundcloud.com/player/?url={query_url}")
             html_content = response.text
@@ -42,13 +43,12 @@ async def get_streaming_url(query_url: str) -> MusicInformation:
             canonical_link = soup.find('link', rel='canonical')
             href = canonical_link.get('href')
 
-            return await ddl_retrievers.universal_ddl_retriever.get_streaming_url(href)
+            sc_url = href
 
-        else:
-            return await ddl_retrievers.universal_ddl_retriever.get_streaming_url(query_url)
+        return await ddl_retrievers.universal_ddl_retriever.get_streaming_url(sc_url)
 
 
-    elif platform is Platform.ANYTHING_ELSE:
+    elif platform is Source.UNKNOWN_SOURCE:
         audio_content_type = await get_audio_content_type(query_url, platform)
 
         if audio_content_type is AudioContentType.YT_DLP:
@@ -63,14 +63,14 @@ async def get_streaming_url(query_url: str) -> MusicInformation:
         return await ddl_retrievers.universal_ddl_retriever.get_streaming_url(query_url)
 
 async def get_urls(query: str) -> List[str]:
-    platform = await find_platform(query)
+    platform = await find_music_source(query)
     audio_content_type = await get_audio_content_type(query, platform)
 
     if audio_content_type is AudioContentType.NOT_SUPPORTED:
         return []
 
     # TikTok
-    elif audio_content_type is Platform.TIK_TOK:
+    elif audio_content_type is Source.TIK_TOK:
         return [query]
 
     elif audio_content_type is AudioContentType.QUERY:
@@ -83,7 +83,7 @@ async def get_urls(query: str) -> List[str]:
         return [query]
 
     # Spotify
-    elif (audio_content_type is AudioContentType.PLAYLIST or audio_content_type is AudioContentType.ALBUM) and platform is Platform.SPOTIFY:
+    elif (audio_content_type is AudioContentType.PLAYLIST or audio_content_type is AudioContentType.ALBUM) and platform is Source.SPOTIFY:
         client_credentials_manager = SpotifyClientCredentials(client_id=os.getenv('SPOTIFY_CLIENT_ID'), client_secret=os.getenv('SPOTIFY_CLIENT_SECRET'))
         sp = spotipy.Spotify(client_credentials_manager=client_credentials_manager)
 
@@ -104,7 +104,7 @@ async def get_urls(query: str) -> List[str]:
             raise NotImplementedError("This type of Spotify content is not implemented.")
 
     # Soundcloud and Youtube
-    elif (audio_content_type is AudioContentType.PLAYLIST or audio_content_type is AudioContentType.RADIO) and platform != Platform.SPOTIFY:
+    elif (audio_content_type is AudioContentType.PLAYLIST or audio_content_type is AudioContentType.RADIO) and platform != Source.SPOTIFY:
         ydl_opts = {
             'extract_flat': True,
             'quiet': True,

requirements.txt

@@ -9,3 +9,4 @@ python-dotenv
 ytmusicapi
 peewee
 spotapi
+requests

searching_utils/query_to_youtube.py

@@ -0,0 +1,18 @@
+import ytmusicapi
+from yt_dlp import YoutubeDL
+
+def get_url(query: str) -> str:
+    yt = ytmusicapi.YTMusic()
+    results = yt.search(query)
+
+    if results:
+        video_id = results[0]["videoId"]
+        return f"https://music.youtube.com/watch?v={video_id}"
+    else:
+        with YoutubeDL({"quiet": True}) as ydl:
+            search_results = ydl.extract_info(f"ytsearch:{query}", download=False)
+            if "entries" in search_results and search_results["entries"]:
+                first_result = search_results["entries"][0]
+                return f"https://www.youtube.com/watch?v={first_result['id']}"
+
+    raise Exception("There was an error getting the url")

searching_utils/spotify_to_youtube.py

@@ -0,0 +1,39 @@
+import os
+from bs4 import BeautifulSoup
+import requests
+from spotdl import Spotdl
+from exceptions.spotify_exceptions import MissingSpotifyCredentialsError
+from searching_utils.query_to_youtube import get_url
+
+def get_url(spotify_url: str) -> str:
+    client_id = os.getenv('SPOTIFY_CLIENT_ID')
+    client_secret = os.getenv('SPOTIFY_CLIENT_SECRET')
+
+    if client_id and client_secret:
+        return get_url_with_credentials(spotify_url, client_id, client_secret)
+    elif client_id or client_secret:
+        raise MissingSpotifyCredentialsError("Both client ID and client secret are required.")
+    else:
+        return get_url_without_credentials(spotify_url)
+
+def get_url_without_credentials(spotify_url: str) -> str:
+    response = requests.get(spotify_url)
+
+    if response.status_code != 200:
+        raise ValueError(f"Request failed with status code {response.status_code}")
+
+    soup = BeautifulSoup(response.text, 'html.parser')
+
+    meta_tag = soup.find('meta', {'property': 'og:description'})
+
+    if not meta_tag or 'content' not in meta_tag.attrs:
+        raise ValueError("Could not find the 'og:description' meta tag")
+
+    search_query = meta_tag['content']
+
+    return get_url(search_query)
+
+def get_url_with_credentials(spotify_url: str, client_id: str, client_secret: str) -> str:
+    spotdl = Spotdl(client_id=client_id, client_secret=client_secret)
+    song = spotdl.search([spotify_url])
+    return spotdl.get_download_urls(song)[0]

setup/apply_config.py

@@ -0,0 +1,24 @@
+import configparser
+from discord import Bot
+
+async def apply_config(bot: Bot):
+    config = configparser.ConfigParser()
+    config.read('config.ini')
+
+    ask_in_dms = config.getboolean('Bot', 'AskInDMs', fallback=False)
+    admin_userid = config.getint('Admin', 'UserID', fallback=0)
+
+    if ask_in_dms and admin_userid:
+        user = await bot.fetch_user(admin_userid)
+
+        dm_channel = await user.create_dm()
+
+        messages = await dm_channel.history().flatten()
+
+        for msg in messages:
+            try:
+                await msg.delete()
+            except Exception as e:
+                print(f"Skipped message due to error: {str(e)}")
+
+        await user.send(f"Bot is ready and logged in as {bot.user.name}")