Cleanup with a bit of black, flake8 and mypy.

.flake8

@@ -1,5 +1,9 @@
 [flake8]
 ignore = E203, E266, E501, W503, C901
+
+# This would be ideal
+#ignore = E203, E501, W503
+
 max-line-length = 99
 max-complexity = 18
 select = B,C,E,F,W,T4,B9,TC,TC1

musicbot/bot.py

@@ -104,9 +104,7 @@ def __init__(self, config_file=None, perms_file=None, aliases_file=None):
 
         self.aiolocks = defaultdict(asyncio.Lock)
         self.filecache = AudioFileCache(self)
-        self.downloader = downloader.Downloader(
-            self, download_folder=self.config.audio_cache_path
-        )
+        self.downloader = downloader.Downloader(self)
 
         log.info("Starting MusicBot {}".format(BOTVERSION))
 
@@ -160,7 +158,7 @@ async def _doBotInit(self, use_certifi: bool = False):
                 headers={"User-Agent": self.http.user_agent}
             )
 
-        self.spotify = None
+        self.spotify: Spotify
         if self.config._spotify:
             try:
                 self.spotify = Spotify(
@@ -2138,7 +2136,7 @@ async def cmd_shuffleplay(
             leftover_args,
             song_url,
             head=False,
-            shuffle_entries=True
+            shuffle_entries=True,
         )
 
         return Response(
@@ -2306,7 +2304,7 @@ async def cmd_move(self, channel, command, leftover_args):
 
         Swaps the location of a song within the playlist.
         """
-        # TODO: this command needs some tlc. args renamed, better checks.
+        # TODO: move command needs some tlc. args renamed, better checks.
         player = self.get_player_in(channel.guild)
         if not player:
             raise exceptions.CommandError(
@@ -2434,7 +2432,7 @@ async def _cmd_play(
         leftover_args,
         song_url,
         head,
-        shuffle_entries: bool = False
+        shuffle_entries: bool = False,
     ):
         """
         This function handles actually playing any given URL or song subject.
@@ -2575,7 +2573,7 @@ async def _cmd_play(
                     expire_in=30,
                 )
 
-            # If the result has usable entries, we assume it is _type = playlist
+            # If the result has usable entries, we assume it is a playlist
             if "entries" in info and (info.playlist_count or info.entry_count):
                 await self._do_playlist_checks(player, author, info)
 

musicbot/downloader.py

@@ -2,20 +2,24 @@
 import copy
 import logging
 import functools
-import yt_dlp as youtube_dl
+import yt_dlp as youtube_dl  # type: ignore
 
 from collections import UserDict
 from concurrent.futures import ThreadPoolExecutor
 from types import MappingProxyType
-from typing import NoReturn, Optional, Any, List, Dict
+from typing import TYPE_CHECKING, Optional, Any, List, Dict
 from pprint import pformat
-from yt_dlp.networking.exceptions import NoSupportingHandlers
-from yt_dlp.utils import UnsupportedError, DownloadError
+from yt_dlp.networking.exceptions import NoSupportingHandlers  # type: ignore
+from yt_dlp.utils import UnsupportedError, DownloadError  # type: ignore
 
 from .exceptions import ExtractionError
 from .spotify import Spotify
 from .utils import get_header
 
+if TYPE_CHECKING:
+    from .bot import MusicBot
+
+
 log = logging.getLogger(__name__)
 
 # Immutable dict is needed, because something is modifying the 'outtmpl' value. I suspect it being ytdl, but I'm not sure.
@@ -53,18 +57,20 @@
 
 
 class Downloader:
-    def __init__(self, bot, download_folder=None):
-        self.bot = bot
+    def __init__(self, bot: "MusicBot"):
+        self.bot: "MusicBot" = bot
+        self.download_folder: str = bot.config.audio_cache_path
         self.thread_pool = ThreadPoolExecutor(max_workers=2)
-        self.download_folder = download_folder
 
         # Copy immutable dict and use the mutable copy for everything else.
         ytdl_format_options = ytdl_format_options_immutable.copy()
 
-        if download_folder:
+        if self.download_folder:
             # print("setting template to " + os.path.join(download_folder, otmpl))
             otmpl = ytdl_format_options["outtmpl"]
-            ytdl_format_options["outtmpl"] = os.path.join(download_folder, otmpl)
+            ytdl_format_options["outtmpl"] = os.path.join(
+                self.download_folder, str(otmpl)
+            )
 
         self.unsafe_ytdl = youtube_dl.YoutubeDL(ytdl_format_options)
         self.safe_ytdl = youtube_dl.YoutubeDL(
@@ -136,8 +142,8 @@ def _sanitize_and_log(
             if field in data:
                 data[field] = redacted_str
 
-        if log.getEffectiveLevel() <= logging.NOISY:
-            log.noise(f"Sanitized YTDL Extraction Info:\n{pformat(data)}")
+        if log.getEffectiveLevel() <= logging.NOISY:  # type: ignore[attr-defined]
+            log.noise(f"Sanitized YTDL Extraction Info:\n{pformat(data)}")  # type: ignore[attr-defined]
         else:
             log.debug(f"Sanitized YTDL Extraction Info:  {data}")
 
@@ -194,7 +200,7 @@ async def _filtered_extract_info(self, song_subject: str, *args, **kwargs):
         :returns: YtdlpResponseDict containing sanitized extraction data.
         :raises: YoutubeDLError as base exception.
         """
-        log.noise(f"Called extract_info with:  '{song_subject}', {args}, {kwargs}")
+        log.noise(f"Called extract_info with:  '{song_subject}', {args}, {kwargs}")  # type: ignore[attr-defined]
         as_stream_url = kwargs.pop("as_stream", False)
 
         # handle extracting spotify links before ytdl get ahold of them.
@@ -251,7 +257,7 @@ async def _filtered_extract_info(self, song_subject: str, *args, **kwargs):
         except NoSupportingHandlers:
             # due to how we allow search service strings we can't just encode this by default.
             # on the other hand, this method prevents cmd_stream from taking search strings.
-            log.noise(
+            log.noise(  # type: ignore[attr-defined]
                 "Caught NoSupportingHandlers, trying again after replacing colon with space."
             )
             song_subject = song_subject.replace(":", " ")
@@ -274,7 +280,7 @@ async def _filtered_extract_info(self, song_subject: str, *args, **kwargs):
             and type(data.get("entries", None)) is list
             and data.get("playlist_count", 0) == 1
         ):
-            log.noise(
+            log.noise(  # type: ignore[attr-defined]
                 "Extractor youtube:search returned single-entry result, replacing base info with entry info."
             )
             entry_info = copy.deepcopy(data["entries"][0])
@@ -285,7 +291,7 @@ async def _filtered_extract_info(self, song_subject: str, *args, **kwargs):
         return data
 
     async def safe_extract_info(self, *args, **kwargs):
-        log.noise(f"Called safe_extract_info with:  {args}, {kwargs}")
+        log.noise(f"Called safe_extract_info with:  {args}, {kwargs}")  # type: ignore[attr-defined]
         return await self.bot.loop.run_in_executor(
             self.thread_pool,
             functools.partial(self.safe_ytdl.extract_info, *args, **kwargs),
@@ -304,7 +310,7 @@ def __init__(self, data: Dict) -> None:
         super().__init__(data)
         self._propagate_entry_data()
 
-    def _propagate_entry_data(self) -> NoReturn:
+    def _propagate_entry_data(self) -> None:
         """ensure the `__input_subject` key is set on all child entries."""
         subject = self.get("__input_subject", None)
         if not subject:
@@ -460,9 +466,9 @@ def url(self) -> str:
         return self.data.get("url", "")
 
     @property
-    def webpage_url(self) -> Optional[str]:
+    def webpage_url(self) -> str:
         """returns value for data key 'webpage_url' or None"""
-        return self.data.get("webpage_url", None)
+        return self.data.get("webpage_url", "")
 
     @property
     def webpage_basename(self) -> Optional[str]:
@@ -479,6 +485,11 @@ def original_url(self) -> Optional[str]:
         """returns value for data key 'original_url' or None"""
         return self.data.get("original_url", None)
 
+    @property
+    def video_id(self) -> str:
+        """returns the id if it exists or empty string."""
+        return self.data.get("id", "")
+
     @property
     def title(self) -> str:
         """returns title value if it exists, empty string otherwise."""

musicbot/entry.py

@@ -134,12 +134,11 @@ class URLPlaylistEntry(BasePlaylistEntry):
     def __init__(
         self,
         playlist: "Playlist",
-        info: "YtdlpResponseDict",
+        info: YtdlpResponseDict,
         **meta,
     ) -> None:
         super().__init__()
 
-        # TODO: perhaps remove arguments here that are inside info dict, and just pass info dict.
         self.playlist = playlist
         self.downloader = playlist.bot.downloader
 
@@ -548,14 +547,13 @@ async def _really_download(self, *, hash=False):
         self.downloaded_bytes = os.path.getsize(self.filename)
 
 
-# TODO: make this use info dict instead.
 class StreamPlaylistEntry(BasePlaylistEntry):
     SERIAL_VERSION = 2
 
     def __init__(
         self,
         playlist: "Playlist",
-        info: "YtdlpResponseDict",
+        info: YtdlpResponseDict,
         **meta,
     ):
         super().__init__()

musicbot/playlist.py

@@ -3,13 +3,17 @@
 from collections import deque
 from itertools import islice
 from random import shuffle
+from typing import TYPE_CHECKING, Union
 
 from .constructs import Serializable
 from .exceptions import ExtractionError, WrongEntryTypeError, InvalidDataError
 from .lib.event_emitter import EventEmitter
 
 from .entry import URLPlaylistEntry, StreamPlaylistEntry
 
+if TYPE_CHECKING:
+    from .bot import MusicBot
+
 log = logging.getLogger(__name__)
 
 
@@ -18,7 +22,7 @@ class Playlist(EventEmitter, Serializable):
     A playlist that manages the list of songs that will be played.
     """
 
-    def __init__(self, bot):
+    def __init__(self, bot: "MusicBot") -> None:
         super().__init__()
         self.bot = bot
         self.loop = bot.loop
@@ -28,7 +32,7 @@ def __init__(self, bot):
     def __iter__(self):
         return iter(self.entries)
 
-    def __len__(self):
+    def __len__(self) -> int:
         return len(self.entries)
 
     def shuffle(self):