Migrate to Pycord (#9)

.gitignore

@@ -150,3 +150,7 @@ cython_debug/
 #  and can be added to the global gitignore or merged into this file.  For a more nuclear
 #  option (not recommended) you can uncomment the following to ignore the entire idea folder.
 .idea/
+
+# JSON files are used only for testing. If this behaviour will change in the future specific
+# files will be ignored.
+*.json

.vscode/launch.json

@@ -5,10 +5,10 @@
     "version": "0.2.0",
     "configurations": [
         {
-            "name": "Python: Spoyt Discord",
+            "name": "Python: Spoyt",
             "type": "python",
             "request": "launch",
-            "module": "Spoyt.Discord",
+            "module": "Spoyt",
             "console": "integratedTerminal",
             "justMyCode": true
         }

README.md

@@ -1,29 +1,18 @@
 # Spoyt
 
-Spotify to YouTube; Discord and Guilded link converter.
+Discord bot that allows you to "convert" Spotify links to YouTube videos.
 
 ## Usage
 
-Just send a message with share link from Spotify. Bot will automatically find
-the track in Spotify database, and search its name and artists in YouTube.
-If possible, it will try to delete your message, but you can disable it
-by permitting permissions.
+1. Invite the bot with this link: <https://discord.com/api/oauth2/authorize?client_id=948274806325903410&permissions=2147485696&scope=bot%20applications.commands>.
 
-Invite the bot by one of following links:
-- Discord: https://discord.com/api/oauth2/authorize?client_id=948274806325903410&permissions=3072&scope=bot
-- Guilded: https://www.guilded.gg/b/93177486-3a1d-4464-a202-1ddd6354844b
+1. Use `/track` or `/playlist` command to search. Bot will try to find the track or playlist (respectively) using Spotify API, and search it in YouTube (also using API).
 
-## Support
-
-You can join one of my servers (or both):
+### Note
 
-- Discord: [discord.gg/SRdmrPpf2z](https://discord.gg/SRdmrPpf2z)
-- Guilded: [guilded.gg/Anonymous-Canteen](https://guilded.gg/Anonymous-Canteen)
+YouTube searching currently applies only to `/track`. I'm currently inspecting YouTube API limiations.
 
-## How to run
+## Support
 
-Make sure you have Python `>=3.8` installed.
-```
-[py|python|python3] -(O|OO)m Spoyt.[Discord|Guilded]
-```
+You can join my server: <https://discord.gg/SRdmrPpf2z>.
 

Spoyt/__main__.py

@@ -1,9 +1,19 @@
 # -*- coding: utf-8 -*-
 from logging import INFO, basicConfig
 
+from discord import ApplicationContext, Bot, DiscordException, Option
+from discord.ext.commands import BucketType, cooldown, CommandOnCooldown
 from rich.logging import RichHandler
 
-from Spoyt.logging import log
+from Spoyt.api.spotify import search_track, search_playlist, url_to_id
+from Spoyt.api.youtube import search_video
+from Spoyt.embeds import CommandOnCooldownEmbed, ErrorEmbed, IncorrectInputEmbed, SpotifyPlaylistkNotFoundEmbed, SpotifyTrackEmbed, \
+    SpotifyPlaylistEmbed, SpotifyTrackNotFoundEmbed, SpotifyUnreachableEmbed, YouTubeVideoEmbed, \
+        UnderCunstructionEmbed
+from Spoyt.exceptions import SpotifyNotFoundException, SpotifyUnreachableException, YouTubeException
+from Spoyt.logger import log
+from Spoyt.settings import BOT_TOKEN
+from Spoyt.utils import check_env
 
 if __name__ == '__main__':
     basicConfig(
@@ -12,6 +22,104 @@
         datefmt='[%x]',
         handlers=[RichHandler(rich_tracebacks=True)]
     )
-    log.info('This is general Spoyt module.')
-    log.info('To run specific bot please run "Discord" or "Guilded" module.')
-    log.info('Remember to set "BOT_TOKEN" environment variables.')
+    if not check_env():
+        log.critical('Aborting start')
+        exit()
+
+    log.info('Starting Discord bot')
+
+    bot = Bot()
+
+    @bot.event
+    async def on_ready() -> None:
+        log.info(f'Logged in as "{bot.user}"')
+
+    @bot.event
+    async def on_application_command_error(
+        ctx: ApplicationContext,
+        exception: DiscordException
+    ) -> None:
+        if isinstance(exception, CommandOnCooldown):
+            await ctx.respond(embed=CommandOnCooldownEmbed(
+                description=f'Retry in {int(exception.retry_after)} second(s).'
+            ))
+        else:
+            raise exception
+
+    @bot.slash_command(
+        name='track',
+        description='Search for a track'
+    )
+    @cooldown(1, 5.0, BucketType.guild)
+    async def track(
+        ctx: ApplicationContext,
+        url: Option(
+            input_type=str,
+            name='url',
+            description='Starts with "https://open.spotify.com/track/..."',
+            required=True
+    )) -> None:
+        if not url.startswith('https://open.spotify.com/track/'):
+            await ctx.respond(embed=IncorrectInputEmbed())
+            return
+
+        track_id = url_to_id(url)
+        try:
+            track = search_track(track_id)
+        except SpotifyNotFoundException:
+            await ctx.respond(embed=SpotifyTrackNotFoundEmbed())
+            return
+        except SpotifyUnreachableException:
+            await ctx.respond(embed=SpotifyUnreachableEmbed())
+            return
+
+        await ctx.respond(embed=SpotifyTrackEmbed(track))
+
+        youtube_query = '{} {}'.format(track.name, ' '.join(track.artists))
+        try:
+            youtube_result = search_video(youtube_query)
+        except YouTubeException as e:
+            await ctx.channel.send(embed=ErrorEmbed(
+                description=f'```diff\n- {e}\n```'
+            ))
+            return
+        await ctx.channel.send(embed=YouTubeVideoEmbed(youtube_result))
+
+        log.info(f'Successfully converted "{track.name}" track')
+
+    @bot.slash_command(
+        name='playlist',
+        description='Search for a playlist'
+    )
+    @cooldown(1, 30.0, BucketType.guild)
+    async def playlist(
+        ctx: ApplicationContext,
+        url: Option(
+            input_type=str,
+            name='url',
+            description='Starts with "https://open.spotify.com/playlist/..."',
+            required=True
+        )
+    ) -> None:
+        if not url.startswith('https://open.spotify.com/playlist/'):
+            await ctx.respond(embed=IncorrectInputEmbed())
+            return
+
+        playlist_id = url_to_id(url)
+        try:
+            playlist = search_playlist(playlist_id)
+        except SpotifyNotFoundException:
+            await ctx.respond(embed=SpotifyPlaylistkNotFoundEmbed())
+            return
+        except SpotifyUnreachableException:
+            await ctx.respond(embed=SpotifyUnreachableEmbed())
+            return
+
+        await ctx.respond(embed=SpotifyPlaylistEmbed(playlist))
+
+        await ctx.channel.send(embed=UnderCunstructionEmbed(
+            description='YouTube searching is currently available only for `/track`.'
+        ))
+        log.info('Playlist conversion issued.')
+
+    bot.run(BOT_TOKEN)

Spoyt/api/spotify.py

@@ -0,0 +1,114 @@
+# -*- coding: utf-8 -*-
+from spotipy import Spotify, SpotifyException, SpotifyClientCredentials
+
+from Spoyt.exceptions import SpotifyNotFoundException, SpotifyUnreachableException
+from Spoyt.logger import log
+from Spoyt.settings import SPOTIFY_CLIENT_ID, SPOTIFY_CLIENT_SECRET
+
+
+class Track:
+    def __init__(self, payload: dict) -> None:
+        self.name: str = payload.get('name')
+        self.track_id: str = payload.get('id')
+        self.artists: list[str] = list(map(
+            lambda a: a.get('name'),
+            payload.get('artists', {})
+        ))
+        self.release_date: str = payload.get('album', {}).get('release_date')
+        self.cover_url: str = payload.get('album', {}).get('images', [{}])[0].get('url')
+
+    @property
+    def is_single_artist(self) -> bool:
+        return len(self.artists) == 1
+
+    @property
+    def track_url(self) -> str:
+        return f'https://open.spotify.com/track/{self.track_id}'
+
+
+class User:
+    def __init__(self, payload: dict) -> None:
+        self.name: str = payload.get('display_name')
+        self.id: str = payload.get('id')
+        self.user_url: str = payload.get('external_urls', {}).get('spotify')
+        self.avatar_url: str = payload.get('images', [{}])[-1].get('url')
+
+class Playlist:
+    def __init__(self, payload: dict) -> None:
+        self.name: str = payload.get('name')
+        self.description: str = payload.get('description')
+        self.playlist_id: str = payload.get('id')
+        self.cover_url: str = payload.get('images', [{}])[0].get('url')
+        self.tracks: list[Track] = list(map(
+            lambda a: Track(a.get('track', {})),
+            payload.get('tracks', {}).get('items')
+        ))
+        self.total_tracks: int = payload.get('tracks', {}).get('total')
+        self.query_limit: int = payload.get('tracks', {}).get('limit')
+
+        self.owner: User = search_user(payload.get('owner', {}).get('id'))
+
+    @property
+    def url(self) -> str:
+        return f'https://open.spotify.com/playlist/{self.playlist_id}'
+
+    @property
+    def is_query_limited(self) -> bool:
+        return len(self.tracks) == self.query_limit
+
+
+def url_to_id(url: str) -> str:
+    """
+    Removes trailing parameters like share source, then extractd ID.
+
+    For example this: "https://open.spotify.com/track/4cOdK2wGLETKBW3PvgPWqT?si=8a1b522f00744ee1",
+    becomes: "4cOdK2wGLETKBW3PvgPWqT".
+    """
+    return url.split('?')[0].split('&')[0].split('/')[-1]
+
+
+def spotify_connect() -> Spotify:
+    return Spotify(
+        auth_manager=SpotifyClientCredentials(
+            client_id=SPOTIFY_CLIENT_ID,
+            client_secret=SPOTIFY_CLIENT_SECRET
+        )
+    )
+
+# Search functions should not return `class Track` or `class Playlist`
+# because of checks if connections was successful during runtime.
+
+def search_track(track_id: str) -> Track:
+    log.info(f'Searching track by ID "{track_id}"')
+    try:
+        track: dict | None = spotify_connect().track(track_id=track_id)
+    except SpotifyException:
+        raise SpotifyNotFoundException
+    if not track:
+        log.error('Spotify unreachable')
+        raise SpotifyUnreachableException
+    return Track(track)
+
+
+def search_playlist(playlist_id: str) -> Playlist:
+    log.info(f'Searching playlist by ID "{playlist_id}"')
+    try:
+        playlist: dict | None = spotify_connect().playlist(playlist_id=playlist_id)
+    except SpotifyException:
+        raise SpotifyNotFoundException
+    if not playlist:
+        log.error('Spotify unreachable')
+        raise SpotifyUnreachableException
+    return Playlist(playlist)
+
+
+def search_user(user_id: str) -> User:
+    log.info(f'Searching user by ID "{user_id}"')
+    try:
+        user: dict | None = spotify_connect().user(user=user_id)
+    except SpotifyException:
+        raise SpotifyNotFoundException
+    if not user:
+        log.error('Spotify unreachable')
+        raise SpotifyUnreachableException
+    return User(user)

Spoyt/api/youtube.py

@@ -0,0 +1,48 @@
+# -*- coding: utf-8 -*-
+from json import loads as json_loads
+
+from requests import get as requests_get
+
+from Spoyt.exceptions import YouTubeException, YouTubeForbiddenException
+from Spoyt.logger import log
+from Spoyt.settings import YOUTUBE_API_KEY
+
+
+class YouTubeVideo:
+    def __init__(self, payload: dict) -> None:
+        item: dict = payload.get('items', [{}])[0]
+        snippet: dict = item.get('snippet', {})
+        self.video_id: str = item.get('id', {}).get('videoId')
+        self.title: str = snippet.get('title')
+        self.description: str = snippet.get('description')
+        self.published_date: str = snippet.get('publishTime', '')[:10]
+
+    @property
+    def video_link(self) -> str:
+        return f'https://www.youtube.com/watch?v={self.video_id}'
+
+    @property
+    def video_thumbnail(self) -> str:
+        return f'https://i.ytimg.com/vi/{self.video_id}/default.jpg'
+
+
+def search_video(query: str) -> YouTubeVideo:
+    log.info(f'Searching YouTube: "{query}"')
+    yt_r = requests_get(
+        'https://www.googleapis.com/youtube/v3/search'
+        '?key={}'
+        '&part=snippet'
+        '&maxResults=1'
+        '&q={}'.format(YOUTUBE_API_KEY, query)
+    )
+    content = json_loads(yt_r.content)
+    if (error_code := yt_r.status_code) == 200:
+        video = YouTubeVideo(content)
+        log.info(f'Found YouTube video "{video.title}" ({video.video_link})')
+    elif error_code == 403:
+        log.critical(content['error']['message'])
+        raise YouTubeForbiddenException
+    else:
+        log.error(content['error']['message'])
+        raise YouTubeException
+    return video