Merge pull request #29 from Diaoxiaozhang/develop

feat: change default download format to mp3
Diaoxiaozhang · Dec 5, 2023 · 26f5c84 · 26f5c84
2 parents c3a0dbb + 4ea9d8f
commit 26f5c84
Show file tree

Hide file tree

Showing 2 changed files with 26 additions and 28 deletions.
diff --git a/cli.py b/cli.py
@@ -22,14 +22,11 @@
     else:
         username = ximalaya.judge_cookie(cookie)
     if os.path.isdir(path):
-        print(
-            f"检测到已设置下载路径为{path}，如果想要修改下载路径，请修改config.json文件中的path字段，将path字段删除或者将值设置为空字符串可恢复默认下载路径")
+        print(f"检测到已设置下载路径为{path}，如果想要修改下载路径，请修改config.json文件中的path字段，将path字段删除或者将值设置为空字符串可恢复默认下载路径")
     else:
-        print(
-            '在config文件中未检测到有效的下载路径，将使用默认下载路径，如果想要修改下载路径，请修改config.json文件中的path字段为你想要的下载路径')
+        print('在config文件中未检测到有效的下载路径，将使用默认下载路径，如果想要修改下载路径，请修改config.json文件中的path字段为你想要的下载路径')
         path = './download'
-    response = requests.get(f"https://www.ximalaya.com/mobile-playpage/track/v3/baseInfo/{int(time.time() * 1000)}?device=web&trackId=188017958&trackQualityLevel=1",
-                            headers=ximalaya.default_headers)
+    response = requests.get(f"https://www.ximalaya.com/mobile-playpage/track/v3/baseInfo/{int(time.time() * 1000)}?device=web&trackId=188017958&trackQualityLevel=1",headers=ximalaya.default_headers)
     if response.json()["ret"] == 927 and not username:
         print("检测到当前ip不在中国大陆，由于喜马拉雅官方限制，必须登录才能继续使用，将自动跳转到登录流程")
         ximalaya.login()
@@ -78,8 +75,7 @@
                 sound_id = int(_)
             except ValueError:
                 try:
-                    sound_id = re.search(
-                        r"ximalaya.com/sound/(?P<sound_id>\d+)", _).group('sound_id')
+                    sound_id = re.search(r"ximalaya.com/sound/(?P<sound_id>\d+)", _).group('sound_id')
                 except Exception:
                     print("输入有误，请重新输入！")
                     continue
@@ -102,8 +98,7 @@
                 if choice == "":
                     choice = "1"
                 if choice == "0" or choice == "1":
-                    ximalaya.get_sound(
-                        sound_info["name"], sound_info[int(choice)], path)
+                    ximalaya.get_sound(sound_info["name"], sound_info[int(choice)], path)
                     break
                 elif choice == "2" and sound_info[2] != "":
                     ximalaya.get_sound(sound_info["name"], sound_info[2], path)
@@ -117,8 +112,7 @@
                 album_id = int(input_album)
             except ValueError:
                 try:
-                    album_id = re.search(
-                        r"ximalaya.com/album/(?P<album_id>\d+)", input_album).group('album_id')
+                    album_id = re.search(r"ximalaya.com/album/(?P<album_id>\d+)", input_album).group('album_id')
                 except Exception:
                     print("输入有误，请重新输入！")
                     continue
@@ -129,15 +123,12 @@
             if album_type == 0:
                 print(f"成功解析免费专辑{album_id}，专辑名{album_name}，共{len(sounds)}个声音")
             elif album_type == 1:
-                print(
-                    f"成功解析已购付费专辑{album_id}，专辑名{album_name}，共{len(sounds)}个声音")
+                print(f"成功解析已购付费专辑{album_id}，专辑名{album_name}，共{len(sounds)}个声音")
             elif album_type == 2:
                 if logined is True:
-                    print(
-                        f"成功解析付费专辑{album_id}，专辑名{album_name}，但是当前登陆账号未购买此专辑或未开通vip")
+                    print(f"成功解析付费专辑{album_id}，专辑名{album_name}，但是当前登陆账号未购买此专辑或未开通vip")
                 else:
-                    print(
-                        f"成功解析付费专辑{album_id}，专辑名{album_name}，但是当前未登陆账号，请登录再尝试下载")
+                    print(f"成功解析付费专辑{album_id}，专辑名{album_name}，但是当前未登陆账号，请登录再尝试下载")
                 continue
             else:
                 continue
@@ -189,8 +180,7 @@
                             if choice == "":
                                 choice = "1"
                             if choice == "0" or choice == "1" or choice == "2":
-                                loop.run_until_complete(ximalaya.get_selected_sounds(
-                                    sounds, album_name, start, end, headers, int(choice), number, path))
+                                loop.run_until_complete(ximalaya.get_selected_sounds(sounds, album_name, start, end, headers, int(choice), number, path))
                                 break
                             else:
                                 print("输入有误，请重新输入！")

diff --git a/main.py b/main.py
@@ -69,9 +69,9 @@ def analyze_sound(self, sound_id, headers):
         for encrypted_url in encrypted_url_list:
             if encrypted_url["type"] == "M4A_128":
                 sound_info[2] = self.decrypt_url(encrypted_url["url"])
-            elif encrypted_url["type"] == "M4A_64":
+            elif encrypted_url["type"] == "MP3_64":
                 sound_info[1] = self.decrypt_url(encrypted_url["url"])
-            elif encrypted_url["type"] == "M4A_24":
+            elif encrypted_url["type"] == "MP3_32":
                 sound_info[0] = self.decrypt_url(encrypted_url["url"])
         logger.debug(f'ID为{sound_id}的声音解析成功！')
         return sound_info
@@ -136,9 +136,9 @@ async def async_analyze_sound(self, sound_id, session, headers):
         for encrypted_url in encrypted_url_list:
             if encrypted_url["type"] == "M4A_128":
                 sound_info[2] = self.decrypt_url(encrypted_url["url"])
-            elif encrypted_url["type"] == "M4A_64":
+            elif encrypted_url["type"] == "MP3_64":
                 sound_info[1] = self.decrypt_url(encrypted_url["url"])
-            elif encrypted_url["type"] == "M4A_24":
+            elif encrypted_url["type"] == "MP3_32":
                 sound_info[0] = self.decrypt_url(encrypted_url["url"])
         logger.debug(f'ID为{sound_id}的声音解析成功！')
         return sound_info
@@ -155,7 +155,11 @@ def replace_invalid_chars(self, name):
     def get_sound(self, sound_name, sound_url, path):
         retries = 3
         sound_name = self.replace_invalid_chars(sound_name)
-        if os.path.exists(f"{path}/{sound_name}.m4a"):
+        if '?' in sound_url:
+            type = sound_url.split('?')[0][-3:]
+        else:
+            type = sound_url[-3:]
+        if os.path.exists(f"{path}/{sound_name}.{type}"):
             print(f'{sound_name}已存在！')
             return
         while retries > 0:
@@ -174,7 +178,7 @@ def get_sound(self, sound_name, sound_url, path):
         sound_file = response.content
         if not os.path.exists(path):
             os.makedirs(path)
-        with open(f"{path}/{sound_name}.m4a", mode="wb") as f:
+        with open(f"{path}/{sound_name}.{type}", mode="wb") as f:
             f.write(sound_file)
         print(f'{sound_name}下载完成！')
         logger.debug(f'{sound_name}下载完成！')
@@ -188,15 +192,19 @@ async def async_get_sound(self, sound_name, sound_url, album_name, session, path
         else:
             sound_name = f"{num}-{sound_name}"
             sound_name = self.replace_invalid_chars(sound_name)
+        if '?' in sound_url:
+            type = sound_url.split('?')[0][-3:]
+        else:
+            type = sound_url[-3:]
         album_name = self.replace_invalid_chars(album_name)
         if not os.path.exists(f"{path}/{album_name}"):
             os.makedirs(f"{path}/{album_name}")
-        if os.path.exists(f"{path}/{album_name}/{sound_name}.m4a"):
+        if os.path.exists(f"{path}/{album_name}/{sound_name}.{type}"):
             print(f'{sound_name}已存在！')
         while retries > 0:
             try:
                 async with session.get(sound_url, headers=self.default_headers, timeout=120) as response:
-                    async with aiofiles.open(f"{path}/{album_name}/{sound_name}.m4a", mode="wb") as f:
+                    async with aiofiles.open(f"{path}/{album_name}/{sound_name}.{type}", mode="wb") as f:
                         await f.write(await response.content.read())
                 print(f'{sound_name}下载完成！')
                 logger.debug(f'{sound_name}下载完成！')