Merge pull request #101 from uright008/feature/quality

WorldObservationLog · web-flow · commit 2d0bc19a010e · 2026-02-13T16:42:01.000+08:00
Feature: add quality command and retrytime &amp; Refactor: move batch download into a separate function
diff --git a/README.md b/README.md
@@ -29,6 +29,14 @@ https://music.apple.com/jp/playlist/bocchi-the-rock/pl.u-Ympg5s39LRqp
 https://music.apple.com/jp/album/nameless-name-single/1688539265
 # Download multiple songs in one line
 dl https://music.apple.com/jp/playlist/bocchi-the-rock/pl.u-Ympg5s39LRqp https://music.apple.com/jp/album/nameless-name-single/1688539265
+# Check the available quality of the song
+quality https://music.apple.com/jp/playlist/bocchi-the-rock/pl.u-Ympg5s39LRqp
+# Or a shorter command
+qa https://music.apple.com/jp/playlist/bocchi-the-rock/pl.u-Ympg5s39LRqp
+# You can hide a column by enabling it in the options.
+qa --codec-id https://music.apple.com/jp/playlist/bocchi-the-rock/pl.u-Ympg5s39LRqp
+# When you add --invert, it works the opposite way, showing the column for whichever option you enable.
+qa --invert --codec-id https://music.apple.com/jp/playlist/bocchi-the-rock/pl.u-Ympg5s39LRqp
 ```
 
 # Support Codec
diff --git a/config.example.toml b/config.example.toml
@@ -1,5 +1,5 @@
 # DO NOT EDIT IT
-version = "0.0.9"
+version = "0.0.10"
 
 [instance]
 url = "127.0.0.1:8080"
@@ -106,6 +106,8 @@ maxBitDepth = 24
 # Example: "\"C:\\Program Files\\DAUM\\PotPlayer\\PotPlayerMini64.exe\" \"{filename}\""
 # Pay attention to escaping issues
 afterDownloaded = ""
+retryTime = 8
+maxWaitTime = 30
 
 [metadata]
 # Metadata to be written to the song
diff --git a/pyproject.toml b/pyproject.toml
@@ -27,6 +27,7 @@ creart = "^0.3.0"
 mutagen = "^1.47.0"
 hishel = "^0.1.5"
 async-lru = "^2.0.5"
+tabulate = "^0.9.0"
 
 [build-system]
 requires = ["poetry-core"]
diff --git a/src/api.py b/src/api.py
@@ -59,8 +59,8 @@ def __init__(self, proxy: str, parallel_num: int):
         self.request_lock = asyncio.Semaphore(256)
 
     @retry(retry=retry_if_exception_type((httpx.HTTPError, SSLError, FileNotFoundError)),
-           wait=wait_random_exponential(multiplier=1, max=60),
-           stop=stop_after_attempt(32))
+           wait=wait_random_exponential(multiplier=1, max=it(Config).download.maxWaitTime),
+           stop=stop_after_attempt(it(Config).download.retryTime))
     def _set_token(self):
         with httpx.Client() as client:
             resp = client.get("https://music.apple.com", follow_redirects=True)
@@ -73,15 +73,15 @@ def init(self):
         pass
 
     @retry(retry=retry_if_exception_type((httpx.HTTPError, SSLError, FileNotFoundError)),
-           wait=wait_random_exponential(multiplier=1, max=60),
-           stop=stop_after_attempt(32), before_sleep=before_sleep_log(it(GlobalLogger).logger, "WARNING"))
+           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"))
     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=60),
-           stop=stop_after_attempt(32), before_sleep=before_sleep_log(it(GlobalLogger).logger, "WARNING"))
+           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"))
     async def download_song(self, url: str) -> bytes:
         async with self.download_lock:
             result = BytesIO()
diff --git a/src/cmd.py b/src/cmd.py
@@ -1,5 +1,6 @@
 import argparse
 import asyncio
+import copy
 import sys
 import os
 
@@ -19,6 +20,7 @@
 from src.rip import on_decrypt_success, on_decrypt_failed, rip_song, rip_album, rip_artist, rip_playlist
 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:
@@ -60,7 +62,8 @@ def __init__(self, loop: asyncio.AbstractEventLoop):
         self.parser = argparse.ArgumentParser(exit_on_error=False)
         subparser = self.parser.add_subparsers()
         download_parser = subparser.add_parser("download", aliases=["dl"])
-        download_parser.add_argument("url",  nargs='*' ,type=str)
+        quality_parser = subparser.add_parser("quality", aliases=["qa"])
+        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")
@@ -69,12 +72,23 @@ 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("--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")
+        quality_parser.add_argument("--average-bitrate", default=True, action="store_false")
+        quality_parser.add_argument("--channels", default=True, action="store_false")
+        quality_parser.add_argument("--sample-rate", default=True, action="store_false")
+        quality_parser.add_argument("--bit-depth", default=True, action="store_false")
+        quality_parser.add_argument("-b", "--batch", default=False, action="store_true")
+
         subparser.add_parser("status")
         subparser.add_parser("login")
         subparser.add_parser("logout")
         subparser.add_parser("exit")
 
-        self.batch_download_mode = False
+        self.batch_mode = False
 
     async def show_status(self):
         it(WrapperManager).status.cache_invalidate()
@@ -83,36 +97,50 @@ async def show_status(self):
             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):
+        try:
+            if args.batch:
+                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")
+        except:
+            pass
+    
+    async def batch_mode_parser(self, cmds: str):
+        args=self.batch_args
+        args.url=copy.deepcopy(cmds)
+        if cmds[0]!="exit":
+            cmds[0] = self.batch_command
+        return cmds,args
+
     async def command_parser(self, cmd: str):
         if not cmd.strip():
             return
         cmds = cmd.split(" ")
-        if self.batch_download_mode:
-            if cmds[0]=="exit":
-                self.batch_download_mode=False
-                it(GlobalLogger).logger.info("Batch mode exited. Returning to normal command mode.")
+        if self.batch_mode:
+            cmds, args = await self.batch_mode_parser(cmds)
+        else:
+            try:
+                args = self.parser.parse_args(cmds)
+            except (argparse.ArgumentError, argparse.ArgumentTypeError, SystemExit):
+                it(GlobalLogger).logger.warning(f"Unknown command: {cmd}")
                 return
-            args=self.batch_download_args
-            await self.do_download(cmds, args.codec, args.force, args.language, args.include)
-            return
-        try:
-            args = self.parser.parse_args(cmds)
-        except (argparse.ArgumentError, argparse.ArgumentTypeError, SystemExit):
-            it(GlobalLogger).logger.warning(f"Unknown command: {cmd}")
-            return
+            await self.handle_batch_mode(args, cmds)
         match cmds[0]:
             case "download" | "dl":
-                if args.batch:
-                    self.batch_download_mode = True
-                    self.batch_download_args = args
-                    it(GlobalLogger).logger.info("Entering batch mode. Enter one or more URLs per line (space-separated), type 'exit' to quit")
-                    return
-                
-                await self.do_download(args.url, args.codec, args.force, args.language, args.include)
+                safely_create_task(self.do_download(args.url, args.codec, args.force, args.language, args.include))
             case "status":
                 await self.show_status()
             case "exit":
-                self.handle_exit()
+                if self.batch_mode:
+                    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)) 
+
 
     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:
@@ -133,6 +161,34 @@ async def do_download(self, raw_urls: list[str], codec: str, force_download: boo
                                                                     include_participate_in_works=include)))
                 case URLType.Playlist:
                     safely_create_task(rip_playlist(url, codec, Flags(force_save=force_download, language=language)))
+                case _:
+                    it(GlobalLogger).logger.error(f"Unsupported URLType - {raw_url}")
+                    continue        
+    
+    async def do_quality(self, raw_urls: list[str], args):
+        all_fields = list(key_to_Headers.keys())
+        show_fields = []
+        for field in all_fields:
+            if args.invert:
+                show_fields = [f for f in key_to_Headers if not getattr(args, f)]
+            else:
+                show_fields = [f for f in key_to_Headers if getattr(args, f)]
+
+        for raw_url in raw_urls:
+            url = AppleMusicURL.parse_url(raw_url)
+            if not url:
+                real_url = await it(WebAPI).get_real_url(raw_url)
+                url = AppleMusicURL.parse_url(real_url)
+                if not url:
+                    it(GlobalLogger).logger.error(f"Illegal URL! - {raw_url}")
+                    continue
+            match url.type:
+                case URLType.Song:
+                    safely_create_task(print_song_quality(url, show_fields))
+                case URLType.Album:
+                    safely_create_task(print_album_quality(url, show_fields))
+                case URLType.Playlist:
+                    safely_create_task(print_playlist_quality(url, show_fields))
                 case _:
                     it(GlobalLogger).logger.error(f"Unsupported URLType - {raw_url}")
                     continue
@@ -143,6 +199,7 @@ def bottom_toolbar(self):
     def completer(self):
         mycompleter = {
             "dl": {
+                "--batch": None,
                 "--codec": {
                     "ec3": None,
                     "aac": None,
@@ -164,6 +221,17 @@ def completer(self):
                 },
                 "--include-participate-songs": None
             },
+            "qa": {
+                "--invert": None,
+                "--codec-id": None,
+                "--codec": None,
+                "--bitrate": None,
+                "--average-bitrate": None,
+                "--channels": None,
+                "--sample-rate": None,
+                "--bit-depth": None,
+                "--batch": None
+            },
             "status": None,
             "login": None,
             "logout": None,
diff --git a/src/config.py b/src/config.py
@@ -5,7 +5,7 @@
 from creart.creator import AbstractCreator, CreateTargetInfo
 from pydantic import BaseModel
 
-CONFIG_VERSION = "0.0.9"
+CONFIG_VERSION = "0.0.10"
 
 
 class Instance(BaseModel):
@@ -51,6 +51,8 @@ class Download(BaseModel):
     maxSampleRate: int = 192000
     maxBitDepth: int = 24
     afterDownloaded: str = ""
+    retryTime: int = 8
+    maxWaitTime: int = 30
 
 
 class Metadata(BaseModel):
diff --git a/src/grpc/manager.py b/src/grpc/manager.py
@@ -13,6 +13,7 @@
 from src.grpc.manager_pb2 import *
 from src.grpc.manager_pb2_grpc import WrapperManagerServiceStub, google_dot_protobuf_dot_empty__pb2
 from src.logger import GlobalLogger
+from src.config import Config
 from src.utils import safely_create_task
 
 
@@ -129,8 +130,8 @@ async def _decrypt_keepalive(self):
 
     @retry(retry=((retry_if_exception_type(WrapperManagerException)) & (
             retry_if_not_exception_message('no available instance'))),
-           wait=wait_random_exponential(multiplier=1, max=60),
-           stop=stop_after_attempt(32), before_sleep=before_sleep_log(it(GlobalLogger).logger, "WARNING"))
+           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"))
     async def m3u8(self, adam_id: str) -> str:
         resp: M3U8Reply = await self._stub.M3U8(M3U8Request(data=M3U8DataRequest(adam_id=adam_id)))
         if resp.header.code != 0:
@@ -139,8 +140,8 @@ async def m3u8(self, adam_id: str) -> str:
 
     @retry(retry=((retry_if_exception_type(WrapperManagerException)) & (
             retry_if_not_exception_message('no such account'))),
-           wait=wait_random_exponential(multiplier=1, max=60),
-           stop=stop_after_attempt(32), before_sleep=before_sleep_log(it(GlobalLogger).logger, "WARNING"))
+           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"))
     async def logout(self, username: str):
         resp: LogoutReply = await self._stub.Logout(LogoutRequest(data=LogoutData(username=username)))
         if resp.header.code != 0:
@@ -149,8 +150,8 @@ async def logout(self, username: str):
 
     @retry(retry=((retry_if_exception_type(WrapperManagerException)) & (
             retry_if_not_exception_message('no available instance'))),
-           wait=wait_random_exponential(multiplier=1, max=60),
-           stop=stop_after_attempt(32), before_sleep=before_sleep_log(it(GlobalLogger).logger, "WARNING"))
+           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"))
     async def lyrics(self, adam_id: str, language: str, region: str) -> str:
         resp: LyricsReply = await self._stub.Lyrics(LyricsRequest(
             data=LyricsDataRequest(adam_id=adam_id, language=language, region=region)))
@@ -160,8 +161,8 @@ async def lyrics(self, adam_id: str, language: str, region: str) -> str:
 
     @retry(retry=((retry_if_exception_type(WrapperManagerException)) & (
             retry_if_not_exception_message('no available instance'))),
-           wait=wait_random_exponential(multiplier=1, max=60),
-           stop=stop_after_attempt(32), before_sleep=before_sleep_log(it(GlobalLogger).logger, "WARNING"))
+           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"))
     async def webPlayback(self, adam_id: str) -> str:
         resp: WebPlaybackReply = await self._stub.WebPlayback(WebPlaybackRequest(
             data=WebPlaybackDataRequest(adam_id=adam_id)
@@ -172,8 +173,8 @@ async def webPlayback(self, adam_id: str) -> str:
 
     @retry(retry=((retry_if_exception_type(WrapperManagerException)) & (
             retry_if_not_exception_message('no available instance'))),
-           wait=wait_random_exponential(multiplier=1, max=60),
-           stop=stop_after_attempt(32), before_sleep=before_sleep_log(it(GlobalLogger).logger, "WARNING"))
+           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"))
     async def license(self, adam_id: str, challenge: str, kid: str) -> str:
         resp: LicenseReply = await self._stub.License(LicenseRequest(
             data=LicenseDataRequest(adam_id=adam_id, challenge=challenge, uri=kid)
diff --git a/src/qemu.py b/src/qemu.py
@@ -41,7 +41,7 @@ class QGAClient:
     writer: asyncio.StreamWriter
 
     @retry(retry=retry_if_exception_type((asyncio.TimeoutError)),
-           wait=wait_random_exponential(multiplier=1, max=60),
+           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):
         self.reader, self.writer = await asyncio.open_connection("127.0.0.1", 32766)
diff --git a/src/quality.py b/src/quality.py
