refactor: format code

Author: WorldObservationLog

src/api.py

@@ -74,14 +74,16 @@ def init(self):
 
     @retry(retry=retry_if_exception_type((httpx.HTTPError, SSLError, FileNotFoundError)),
            wait=wait_random_exponential(multiplier=1, max=it(Config).download.maxWaitTime),
-           stop=stop_after_attempt(it(Config).download.retryTime), before_sleep=before_sleep_log(it(GlobalLogger).logger, "WARNING"))
+           stop=stop_after_attempt(it(Config).download.retryTime),
+           before_sleep=before_sleep_log(it(GlobalLogger).logger, "WARNING"))
     async def _request(self, *args, **kwargs):
         async with self.request_lock:
             return await self.client.request(*args, **kwargs)
 
     @retry(retry=retry_if_exception_type((httpx.HTTPError, SSLError, FileNotFoundError)),
            wait=wait_random_exponential(multiplier=1, max=it(Config).download.maxWaitTime),
-           stop=stop_after_attempt(it(Config).download.retryTime), before_sleep=before_sleep_log(it(GlobalLogger).logger, "WARNING"))
+           stop=stop_after_attempt(it(Config).download.retryTime),
+           before_sleep=before_sleep_log(it(GlobalLogger).logger, "WARNING"))
     async def download_song(self, url: str) -> bytes:
         async with self.download_lock:
             result = BytesIO()
@@ -110,7 +112,8 @@ async def get_album_info(self, album_id: str, storefront: str, lang: str):
         return album_info_obj
 
     async def get_album_tracks(self, album_id: str, storefront: str, lang: str, offset: int = 0):
-        req = await self._request("GET", f"https://amp-api.music.apple.com/v1/catalog/{storefront}/albums/{album_id}/tracks?offset={offset}")
+        req = await self._request("GET",
+                                  f"https://amp-api.music.apple.com/v1/catalog/{storefront}/albums/{album_id}/tracks?offset={offset}")
         album_info_obj = AlbumTracks.model_validate(req.json())
         tracks = album_info_obj.data
         if album_info_obj.next:

src/cmd.py

@@ -1,8 +1,8 @@
 import argparse
 import asyncio
 import copy
-import sys
 import os
+import sys
 
 import grpc.aio
 from creart import it
@@ -17,12 +17,10 @@
 from src.logger import GlobalLogger
 from src.measurer import Measurer
 from src.qemu import QemuInstance
-from src.measurer import Measurer
-from src.qemu import QemuInstance
+from src.quality import print_song_quality, print_album_quality, print_playlist_quality, key_to_Headers
 from src.rip import Ripper
 from src.url import AppleMusicURL, URLType
 from src.utils import check_dep, run_sync, safely_create_task, config_outdated
-from src.quality import print_song_quality, print_album_quality, print_playlist_quality, key_to_Headers
 
 
 class InteractiveShell:
@@ -54,21 +52,23 @@ def __init__(self, loop: asyncio.AbstractEventLoop):
                 loop.run_until_complete(asyncio.sleep(3))
         else:
             loop.run_until_complete(it(WrapperManager).init(it(Config).instance.url, it(Config).instance.secure))
-        safely_create_task(it(WrapperManager).decrypt_init(on_success=self.ripper.on_decrypt_success, on_failure=self.ripper.on_decrypt_failed))
+        safely_create_task(it(WrapperManager).decrypt_init(on_success=self.ripper.on_decrypt_success,
+                                                           on_failure=self.ripper.on_decrypt_failed))
         try:
             loop.run_until_complete(self.show_status())
         except grpc.aio._call.AioRpcError:
             it(GlobalLogger).logger.error("Unable to connect to the wrapper-manager")
             sys.exit()
 
         if config_outdated():
-            it(GlobalLogger).logger.warning("The configuration file is out of date. Please refer to config.example.toml to update it")
+            it(GlobalLogger).logger.warning(
+                "The configuration file is out of date. Please refer to config.example.toml to update it")
 
         self.parser = argparse.ArgumentParser(exit_on_error=False)
         subparser = self.parser.add_subparsers()
         download_parser = subparser.add_parser("download", aliases=["dl"])
         quality_parser = subparser.add_parser("quality", aliases=["qa"])
-        download_parser.add_argument("url",  nargs='*', type=str)
+        download_parser.add_argument("url", nargs='*', type=str)
         download_parser.add_argument("-c", "--codec",
                                      choices=["alac", "ec3", "aac", "aac-binaural", "aac-downmix", "aac-legacy", "ac3"],
                                      default="alac")
@@ -77,8 +77,8 @@ def __init__(self, loop: asyncio.AbstractEventLoop):
         download_parser.add_argument("-l", "--language", default=it(Config).region.language, action="store")
         download_parser.add_argument("--include-participate-songs", default=False, dest="include", action="store_true")
 
-        quality_parser.add_argument("url",  nargs='*', type=str)
-        quality_parser.add_argument("-i","--invert", default=False, action="store_true")
+        quality_parser.add_argument("url", nargs='*', type=str)
+        quality_parser.add_argument("-i", "--invert", default=False, action="store_true")
         quality_parser.add_argument("--codec-id", default=True, action="store_false")
         quality_parser.add_argument("--codec", default=True, action="store_false")
         quality_parser.add_argument("--bitrate", default=True, action="store_false")
@@ -99,7 +99,8 @@ async def show_status(self):
         it(WrapperManager).status.cache_invalidate()
         st_resp = await it(WrapperManager).status()
         if not st_resp.regions:
-            it(GlobalLogger).logger.error("The currently used wrapper-manager instance has no available account. Please execute login command to log in.")
+            it(GlobalLogger).logger.error(
+                "The currently used wrapper-manager instance has no available account. Please execute login command to log in.")
         it(GlobalLogger).logger.info(f"Regions available on wrapper-manager instance: {', '.join(st_resp.regions)}")
 
     async def handle_batch_mode(self, args, cmds):
@@ -108,16 +109,17 @@ async def handle_batch_mode(self, args, cmds):
                 self.batch_mode = True
                 self.batch_args = args
                 self.batch_command = cmds[0]
-                it(GlobalLogger).logger.info("Entering batch mode. Enter one or more URLs per line (space-separated), type 'exit' to quit")
+                it(GlobalLogger).logger.info(
+                    "Entering batch mode. Enter one or more URLs per line (space-separated), type 'exit' to quit")
         except:
             pass
-    
+
     async def batch_mode_parser(self, cmds: str):
-        args=self.batch_args
-        args.url=copy.deepcopy(cmds)
-        if cmds[0]!="exit":
+        args = self.batch_args
+        args.url = copy.deepcopy(cmds)
+        if cmds[0] != "exit":
             cmds[0] = self.batch_command
-        return cmds,args
+        return cmds, args
 
     async def command_parser(self, cmd: str):
         if not cmd.strip():
@@ -139,15 +141,15 @@ async def command_parser(self, cmd: str):
                 await self.show_status()
             case "exit":
                 if self.batch_mode:
-                    self.batch_mode=False
+                    self.batch_mode = False
                     it(GlobalLogger).logger.info("Batch mode exited. Returning to normal command mode.")
                 else:
                     self.handle_exit()
             case "quality" | "qa":
-                safely_create_task(self.do_quality(args.url, args)) 
-
+                safely_create_task(self.do_quality(args.url, args))
 
-    async def do_download(self, raw_urls: list[str], codec: str, force_download: bool, language: str, include: bool = False):
+    async def do_download(self, raw_urls: list[str], codec: str, force_download: bool, language: str,
+                          include: bool = False):
         for raw_url in raw_urls:
             url = AppleMusicURL.parse_url(raw_url)
             if not url:
@@ -158,18 +160,22 @@ async def do_download(self, raw_urls: list[str], codec: str, force_download: boo
                     continue
             match url.type:
                 case URLType.Song:
-                    safely_create_task(self.ripper.rip_song(url, codec, Flags(force_save=force_download, language=language)))
+                    safely_create_task(
+                        self.ripper.rip_song(url, codec, Flags(force_save=force_download, language=language)))
                 case URLType.Album:
-                    safely_create_task(self.ripper.rip_album(url, codec, Flags(force_save=force_download, language=language)))
+                    safely_create_task(
+                        self.ripper.rip_album(url, codec, Flags(force_save=force_download, language=language)))
                 case URLType.Artist:
-                    safely_create_task(self.ripper.rip_artist(url, codec, Flags(force_save=force_download, language=language,
-                                                                    include_participate_in_works=include)))
+                    safely_create_task(
+                        self.ripper.rip_artist(url, codec, Flags(force_save=force_download, language=language,
+                                                                 include_participate_in_works=include)))
                 case URLType.Playlist:
-                    safely_create_task(self.ripper.rip_playlist(url, codec, Flags(force_save=force_download, language=language)))
+                    safely_create_task(
+                        self.ripper.rip_playlist(url, codec, Flags(force_save=force_download, language=language)))
                 case _:
                     it(GlobalLogger).logger.error(f"Unsupported URLType - {raw_url}")
-                    continue        
-    
+                    continue
+
     async def do_quality(self, raw_urls: list[str], args):
         all_fields = list(key_to_Headers.keys())
         show_fields = []
@@ -246,7 +252,8 @@ def completer(self):
 
     def handle_exit(self):
         if it(Measurer).tasks_count() > 0:
-            it(GlobalLogger).logger.info("There is still {} tasks, do you really want to exit? (y/N)".format(it(Measurer).tasks_count()))
+            it(GlobalLogger).logger.info(
+                "There is still {} tasks, do you really want to exit? (y/N)".format(it(Measurer).tasks_count()))
             response = input().strip().lower()
             if response != 'y':
                 return
@@ -255,7 +262,8 @@ def handle_exit(self):
         os._exit(0)
 
     async def handle_command(self):
-        session = PromptSession("> ", bottom_toolbar=self.bottom_toolbar, completer=self.completer(), refresh_interval=1)
+        session = PromptSession("> ", bottom_toolbar=self.bottom_toolbar, completer=self.completer(),
+                                refresh_interval=1)
 
         while True:
             try:
@@ -301,11 +309,10 @@ async def logout_flow(self):
         it(GlobalLogger).logger.info("Logout Success!")
         it(WrapperManager).status.cache_invalidate()
 
-
     async def start(self):
         with patch_stdout():
             try:
                 await self.handle_command()
             finally:
                 if it(Config).localInstance.enable:
-                    self.localInstance.terminate()
+                    self.localInstance.terminate()

src/logger.py

@@ -74,7 +74,8 @@ def not_exist(self):
             f"Unable to download {self.item_type}. This {self.item_type} does not exist in all available storefronts")
 
     def language_not_exist(self, region: str, current_language: str, default_language: str):
-        self.logger.warning(f"Selected language {current_language} does not exist in region {region.upper()}, falling back to {default_language}")
+        self.logger.warning(
+            f"Selected language {current_language} does not exist in region {region.upper()}, falling back to {default_language}")
 
     def already_exist(self):
         self.logger.info(f"Song already exists")
@@ -113,4 +114,4 @@ def selected_codec(self, selected_codec):
         self.logger.info(f"Selected codec: {selected_codec}")
 
     def codec_alternative(self):
-        self.logger.warning("Unable to find the specified audio, switched to another audio codec")
+        self.logger.warning("Unable to find the specified audio, switched to another audio codec")

src/metadata.py

@@ -1,10 +1,8 @@
 from typing import Optional, Dict, List
 
-from creart import it
 from mutagen.mp4 import MP4Cover
 from pydantic import BaseModel
 
-
 from src.models import AlbumMeta
 from src.models.song_data import Datum
 from src.utils import ttml_convent, count_total_track_and_disc
@@ -167,7 +165,6 @@ def _rating(content_rating: Optional[str]) -> int:
     def set_lyrics(self, lyrics: str):
         self.lyrics = lyrics
 
-
     def set_playlist_index(self, index: int):
         self.playlist_index = index
 

src/mp4.py

@@ -203,7 +203,7 @@ def encapsulate(song_info: SongInfo, decrypted_media: bytes, atmos_convent: bool
 
 
 def write_metadata(song: bytes, metadata: SongMetadata, embed_metadata: list[str],
-                         cover_format: str, params: dict[str, Any]) -> bytes:
+                   cover_format: str, params: dict[str, Any]) -> bytes:
     tmp_dir = TemporaryDirectory()
     name = uuid.uuid4().hex
     song_name = Path(tmp_dir.name) / Path(f"{name}.m4a")

src/qemu.py

@@ -40,7 +40,7 @@ class QGAClient:
     reader: asyncio.StreamReader
     writer: asyncio.StreamWriter
 
-    @retry(retry=retry_if_exception_type((asyncio.TimeoutError)),
+    @retry(retry=retry_if_exception_type(asyncio.TimeoutError),
            wait=wait_random_exponential(multiplier=1, max=it(Config).download.maxWaitTime),
            stop=stop_after_attempt(8), before_sleep=before_sleep_log(it(GlobalLogger).logger, "WARNING"))
     async def init(self):
@@ -66,7 +66,8 @@ async def read_file(self, path: str):
 
     async def write_file(self, path: str, content: str):
         fp = await self.send_cmd("guest-file-open", {"path": path, "mode": "w"})
-        await self.send_cmd("guest-file-write", {"handle": fp, "buf-b64": base64.standard_b64encode(content.encode()).decode()})
+        await self.send_cmd("guest-file-write",
+                            {"handle": fp, "buf-b64": base64.standard_b64encode(content.encode()).decode()})
         await self.send_cmd("guest-file-close", {"handle": fp})
 
     async def execute(self, path: str, args: list[str]):

src/quality.py

@@ -14,12 +14,12 @@
 from src.grpc.manager import WrapperManager
 
 Headers = [
-    "Codec ID", 
-    "Codec", 
-    "Bitrate", 
-    "Average Bitrate", 
-    "Channels", 
-    "Sample Rate", 
+    "Codec ID",
+    "Codec",
+    "Bitrate",
+    "Average Bitrate",
+    "Channels",
+    "Sample Rate",
     "Bit Depth"
 ]
 
@@ -33,6 +33,7 @@
     "bit_depth": "Bit Depth"
 }
 
+
 async def get_available_audio_quality(m3u8_url: str):
     parsed_m3u8 = m3u8.loads(await it(WebAPI).download_m3u8(m3u8_url), uri=m3u8_url)
     result = []
@@ -47,6 +48,7 @@ async def get_available_audio_quality(m3u8_url: str):
                                        bit_depth=playlist.media[0].extras.get("bit_depth", None)))
     return result
 
+
 async def print_song_quality(url: Song, show_fields: list[str]):
     raw_metadata = await it(WebAPI).get_song_info(url.id, url.storefront, it(Config).region.language)
     metadata = SongMetadata.parse_from_song_data(raw_metadata)
@@ -62,13 +64,15 @@ async def print_song_quality(url: Song, show_fields: list[str]):
     print_formatted_text(f"Available audio qualities for song: {metadata.artist} - {metadata.title}")
     print_formatted_text(tabulate(filtered_data, headers=filtered_headers, tablefmt="grid"))
 
+
 async def print_playlist_quality(url: Playlist, show_fields: list[str]):
     playlist_info = await it(WebAPI).get_playlist_info_and_tracks(url.id, url.storefront, it(Config).region.language)
     playlist_info = playlist_write_song_index(playlist_info)
     for track in playlist_info.data[0].relationships.tracks.data:
         song = Song(id=track.id, storefront=url.storefront, url="", type=URLType.Song)
         safely_create_task(print_song_quality(song, show_fields))
 
+
 async def print_album_quality(url: Album, show_fields: list[str]):
     album_info = await it(WebAPI).get_album_info(url.id, url.storefront, it(Config).region.language)
     for track in album_info.data[0].relationships.tracks.data: