fmt

xiaomusic/api/dependencies.py

@@ -217,7 +217,9 @@ class AuthStaticFiles(StaticFiles):
     async def __call__(self, scope, receive, send) -> None:
         request = Request(scope, receive)
         # 系统提示音，不走任何校验，直接允许访问(修复启用安全验证后，无法播放系统提示音的问题)
-        if request.url.path.endswith(("/xiaomusic_ok.mp3", "/xiaomusic_error.mp3", "/silence.mp3", "/search.mp3")):
+        if request.url.path.endswith(
+            ("/xiaomusic_ok.mp3", "/xiaomusic_error.mp3", "/silence.mp3", "/search.mp3")
+        ):
             await super().__call__(scope, receive, send)
             return
         if not config.disable_httpauth:

xiaomusic/config.py

@@ -150,7 +150,9 @@ class Config:
     )
     keywords_play: str = os.getenv("XIAOMUSIC_KEYWORDS_PLAY", "播放歌曲,放歌曲")
     keywords_online_play: str = os.getenv("XIAOMUSIC_KEYWORDS_ONLINE_PLAY", "在线播放")
-    keywords_online_playlist_play: str = os.getenv("XIAOMUSIC_KEYWORDS_ONLINE_PLAYLIST", "在线歌单,搜索歌单")
+    keywords_online_playlist_play: str = os.getenv(
+        "XIAOMUSIC_KEYWORDS_ONLINE_PLAYLIST", "在线歌单,搜索歌单"
+    )
     keywords_singer_play: str = os.getenv("XIAOMUSIC_KEYWORDS_SINGER_PLAY", "播放歌手")
     keywords_stop: str = os.getenv("XIAOMUSIC_KEYWORDS_STOP", "关机,暂停,停止,停止播放")
     keywords_playlist: str = os.getenv(

xiaomusic/device_player.py

@@ -367,9 +367,13 @@ class XiaoMusicDevice:
                         # 200是正常，206是部分内容(也是正常)
                         if resp.status in (200, 206):
                             is_url_ok = True
-                            self.log.info(f"探路成功！代理服务器存活，状态码: {resp.status}")
+                            self.log.info(
+                                f"探路成功！代理服务器存活，状态码: {resp.status}"
+                            )
                         else:
-                            self.log.warning(f"探路发现死链！代理服务器报错，状态码: {resp.status}")
+                            self.log.warning(
+                                f"探路发现死链！代理服务器报错，状态码: {resp.status}"
+                            )
             except Exception as e:
                 self.log.warning(f"探路超时或网络异常(插件解析失败): {e}")
 
@@ -382,7 +386,9 @@ class XiaoMusicDevice:
                     self.log.error("连续 5 次获取歌曲死链，触发第一层终极熔断！")
                     self._url_failed_cnt = 0
                     # 调用大管家的万能报错机制！
-                    await self.xiaomusic.handle_fatal_error(self.did, "连续多次获取歌曲失败，已为您停止")
+                    await self.xiaomusic.handle_fatal_error(
+                        self.did, "连续多次获取歌曲失败，已为您停止"
+                    )
                     return
 
                 # 没到 5 次，静默切下一首
@@ -400,9 +406,9 @@ class XiaoMusicDevice:
             self.log.info(f"播放 {name} 失败. 失败次数: {self._play_failed_cnt}")
             await asyncio.sleep(1)
             if (
-                    self.is_playing
-                    and self._last_cmd != "stop"
-                    and self._play_failed_cnt < 10
+                self.is_playing
+                and self._last_cmd != "stop"
+                and self._play_failed_cnt < 10
             ):
                 self._play_failed_cnt = self._play_failed_cnt + 1
                 await self._play_next()

xiaomusic/js_plugin_manager.py

@@ -345,7 +345,7 @@ class JSPluginManager:
                 if voice_playlist_strategy is not None:
                     config_data["voice_playlist_strategy"] = {
                         "desc": "语音搜单策略: default(首条), max_songs(歌曲最多), max_plays(播放最多), random(随机)",
-                        "value": voice_playlist_strategy
+                        "value": voice_playlist_strategy,
                     }
 
                 with open(self.plugins_config_path, "w", encoding="utf-8") as f:
@@ -1109,7 +1109,14 @@ class JSPluginManager:
             self.log.error(f"Failed to read enabled plugins from config: {e}")
             return False
 
-    def search(self, plugin_name: str, keyword: str, page: int = 1, limit: int = 20, type_: str = "music"):
+    def search(
+        self,
+        plugin_name: str,
+        keyword: str,
+        page: int = 1,
+        limit: int = 20,
+        type_: str = "music",
+    ):
         """搜索音乐/歌单"""
         if plugin_name not in self.plugins:
             raise ValueError(f"Plugin {plugin_name} not found or not loaded")
@@ -1122,7 +1129,12 @@ class JSPluginManager:
                 "action": "search",
                 "pluginName": plugin_name,
                 # 把 type_ 参数传给底层 JS
-                "params": {"keywords": keyword, "page": page, "limit": limit, "type": type_},
+                "params": {
+                    "keywords": keyword,
+                    "page": page,
+                    "limit": limit,
+                    "type": type_,
+                },
             }
         )
 
@@ -2618,11 +2630,11 @@ class JSPluginManager:
             self.log.error(f"Failed to update plugin config: {e}")
 
     async def plugin_playlist_search(
-            self,
-            plugin_name: str,
-            keyword: str,
-            page: int = 1,
-            limit: int = 20,
+        self,
+        plugin_name: str,
+        keyword: str,
+        page: int = 1,
+        limit: int = 20,
     ):
         """调用 MusicFree 插件进行歌单搜索，并对齐 LX 格式
 
@@ -2636,7 +2648,7 @@ class JSPluginManager:
         """
         try:
             # 显式传入 type_='sheet' 调用底层插件搜索
-            result = self.search(plugin_name, keyword, page, limit, type_='sheet')
+            result = self.search(plugin_name, keyword, page, limit, type_="sheet")
             data_list = result.get("data", [])
 
             # 补齐平台字段
@@ -2655,16 +2667,16 @@ class JSPluginManager:
                     "total": result.get("total", len(data_list)),
                     "page": page,
                     "limit": limit,
-                }
+                },
             }
         except Exception as e:
             self.log.error(f"插件 {plugin_name} 歌单搜索执行失败: {e}")
             raise
 
     async def plugin_playlist_detail(
-            self,
-            plugin_name: str,
-            b64_id: str,
+        self,
+        plugin_name: str,
+        b64_id: str,
     ):
         """解析 Base64 歌单对象并调用插件获取详情（全量歌曲）
 
@@ -2676,10 +2688,10 @@ class JSPluginManager:
         """
         try:
             # 1. 容错处理 Base64 损坏
-            b64_id = b64_id.replace(' ', '+')
+            b64_id = b64_id.replace(" ", "+")
             missing_padding = len(b64_id) % 4
             if missing_padding:
-                b64_id += '=' * (4 - missing_padding)
+                b64_id += "=" * (4 - missing_padding)
 
             # 2. 还原完整的 JSON 歌单对象
             json_str = base64.b64decode(b64_id).decode("utf-8")
@@ -2691,9 +2703,13 @@ class JSPluginManager:
 
             # 3. 循环翻页，直到拿完所有歌曲
             while current_page <= max_pages:
-                self.log.info(f"正在获取 {plugin_name} 歌单详情，第 {current_page} 页...")
+                self.log.info(
+                    f"正在获取 {plugin_name} 歌单详情，第 {current_page} 页..."
+                )
 
-                result = self.get_music_sheet_info(plugin_name, playlist_info, page=current_page)
+                result = self.get_music_sheet_info(
+                    plugin_name, playlist_info, page=current_page
+                )
 
                 if not result or not result.get("musicList"):
                     break
@@ -2712,7 +2728,11 @@ class JSPluginManager:
                 current_page += 1
 
             if not data_list:
-                return {"success": False, "error": "歌单解析为空或格式不支持", "data": []}
+                return {
+                    "success": False,
+                    "error": "歌单解析为空或格式不支持",
+                    "data": [],
+                }
 
             # 4. 补齐平台字段，确保后端播歌逻辑能识别来源
             for item in data_list:
@@ -2721,11 +2741,7 @@ class JSPluginManager:
 
             self.log.info(f"歌单全量拉取完成，共 {len(data_list)} 首歌。")
 
-            return {
-                "success": True,
-                "data": data_list,
-                "total": len(data_list)
-            }
+            return {"success": True, "data": data_list, "total": len(data_list)}
         except Exception as e:
             self.log.error(f"插件 {plugin_name} 歌单详情解析执行失败: {e}")
             raise
@@ -2750,6 +2766,7 @@ class JSPluginManager:
             return playlists[0]
         if strategy == "random":
             return random.choice(playlists)
+
         # 定义提取权重的辅助闭包
         def get_count(item, keys):
             for k in keys:
@@ -2760,11 +2777,18 @@ class JSPluginManager:
                     except:
                         continue
             return 0
+
         if strategy == "max_songs":
-            return max(playlists,
-                       key=lambda x: get_count(x, ["worksNum", "worksNums", "songCount", "song_count", "total"]))
+            return max(
+                playlists,
+                key=lambda x: get_count(
+                    x, ["worksNum", "worksNums", "songCount", "song_count", "total"]
+                ),
+            )
         if strategy == "max_plays":
-            return max(playlists, key=lambda x: get_count(x, ["playCount", "play_count"]))
+            return max(
+                playlists, key=lambda x: get_count(x, ["playCount", "play_count"])
+            )
         return playlists[0]
 
     def reset_restart_limit(self):

xiaomusic/music_library.py

@@ -1109,14 +1109,14 @@ class MusicLibrary:
 
         # 是否需要代理
         # 对 bilibili 页面 URL，优先解析为真实音频/CDN URL；成功后再走 proxy，避免 /proxy 只拿到 HTML 页面。
-        is_bilibili_page = (
-            isinstance(url, str)
-            and ("bilibili.com/video/" in url or "b23.tv/" in url)
+        is_bilibili_page = isinstance(url, str) and (
+            "bilibili.com/video/" in url or "b23.tv/" in url
         )
         resolved_origin_url = url
         if is_bilibili_page:
             try:
                 import asyncio
+
                 cmd = [
                     "yt-dlp",
                     "-f",
@@ -1135,7 +1135,11 @@ class MusicLibrary:
                     stderr=asyncio.subprocess.PIPE,
                 )
                 stdout, stderr = await proc.communicate()
-                lines = [line.strip() for line in stdout.decode(errors="replace").splitlines() if line.strip()]
+                lines = [
+                    line.strip()
+                    for line in stdout.decode(errors="replace").splitlines()
+                    if line.strip()
+                ]
                 resolved = lines[0] if lines else ""
                 if proc.returncode == 0 and resolved:
                     self.log.info(

xiaomusic/online_music.py

@@ -348,16 +348,16 @@ class OnlineMusicService:
     async def _execute_plugin_playlist_search(self, plugin, keyword, page, limit):
         """执行 MusicFree 插件歌单搜索具体业务"""
         if plugin == "all":
-            return {"success": False, "error": "请选择具体的插件平台(如 qq)进行搜单，暂不支持聚合搜单"}
+            return {
+                "success": False,
+                "error": "请选择具体的插件平台(如 qq)进行搜单，暂不支持聚合搜单",
+            }
 
         if self.js_plugin_manager:
             try:
                 # 同 LX 一样调用执行层的适配方法
                 return await self.js_plugin_manager.plugin_playlist_search(
-                    plugin_name=plugin,
-                    keyword=keyword,
-                    page=page,
-                    limit=limit
+                    plugin_name=plugin, keyword=keyword, page=page, limit=limit
                 )
             except Exception as e:
                 return {"success": False, "error": str(e)}
@@ -370,8 +370,7 @@ class OnlineMusicService:
             try:
                 # 同 LX 一样调用执行层的适配方法
                 return await self.js_plugin_manager.plugin_playlist_detail(
-                    plugin_name=plugin,
-                    b64_id=b64_id
+                    plugin_name=plugin, b64_id=b64_id
                 )
             except Exception as e:
                 return {"success": False, "error": f"解析失败: {str(e)}"}
@@ -613,7 +612,9 @@ class OnlineMusicService:
             name = search_key
         # 没有歌名
         if not name:
-            return await self.xiaomusic.handle_fatal_error(did, "小Music没有听到搜索歌名哦，请重试。")
+            return await self.xiaomusic.handle_fatal_error(
+                did, "小Music没有听到搜索歌名哦，请重试。"
+            )
         self.reset_singer_add_page()
         self.log.info(f"搜索关键字{search_key},提取的歌名{name}")
 
@@ -621,19 +622,25 @@ class OnlineMusicService:
             res = await self.search_top_one_play(did, search_key, name)
             # 接口明确返回了失败或者没找到歌曲
             if res and isinstance(res, dict) and not res.get("success"):
-                return await self.xiaomusic.handle_fatal_error(did, f"小Music没找到关于{name}的歌曲")
+                return await self.xiaomusic.handle_fatal_error(
+                    did, f"小Music没找到关于{name}的歌曲"
+                )
             return res
 
         except Exception:
             # 异常兜底，代码崩溃
-            return await self.xiaomusic.handle_fatal_error(did, "小Music搜歌过程中出了点小故障")
+            return await self.xiaomusic.handle_fatal_error(
+                did, "小Music搜歌过程中出了点小故障"
+            )
 
     async def online_playlist_play(self, did="", arg1="", **kwargs):
         """执行语音搜歌单并播放"""
         await self._before_play()
         search_key = str(arg1).strip() if arg1 else ""
         if not search_key:
-            return await self.xiaomusic.handle_fatal_error(did, "小Music没有听到搜索歌单内容，请重试。")
+            return await self.xiaomusic.handle_fatal_error(
+                did, "小Music没有听到搜索歌单内容，请重试。"
+            )
         self.log.info(f"语音搜索歌单: {search_key}")
 
         try:
@@ -648,7 +655,9 @@ class OnlineMusicService:
                     enabled = self.js_plugin_manager.get_enabled_plugins()
                     pref = enabled[0] if enabled else "qq"
             # 发起搜索 (拿第 1 页)
-            search_res = await self.get_playlist_online(plugin=pref, keyword=search_key, page=1)
+            search_res = await self.get_playlist_online(
+                plugin=pref, keyword=search_key, page=1
+            )
 
             # 极致柔性脱壳，对齐前端的 "dataObj = resJson.data || resJson" 兜底逻辑
             data_obj = search_res.get("data") or search_res
@@ -661,29 +670,43 @@ class OnlineMusicService:
                 playlists = []
 
             if not playlists:
-                return await self.xiaomusic.handle_fatal_error(did, f"小Music没找到关于{search_key}的歌单")
+                return await self.xiaomusic.handle_fatal_error(
+                    did, f"小Music没找到关于{search_key}的歌单"
+                )
             # 选出最优歌单
             best_playlist = self.js_plugin_manager.pick_best_playlist(playlists)
-            pl_name = best_playlist.get("title") or best_playlist.get("name") or "未知歌单"
+            pl_name = (
+                best_playlist.get("title") or best_playlist.get("name") or "未知歌单"
+            )
             pl_id = best_playlist.get("id")
             # 准备详情请求参数 (MF 歌单需要特殊处理)
             api_type = 2 if self.js_plugin_manager.is_lx_server() else 1
             target_id = pl_id
             if api_type == 1:
                 # MF 详情需要 Base64 包装对象
-                target_id = base64.b64encode(json.dumps(best_playlist).encode("utf-8")).decode("utf-8")
+                target_id = base64.b64encode(
+                    json.dumps(best_playlist).encode("utf-8")
+                ).decode("utf-8")
 
             # 获取歌单内部全量歌曲
-            detail_res = await self.get_playlist_detail_online(id=target_id, plugin=pref, api_type=api_type)
+            detail_res = await self.get_playlist_detail_online(
+                id=target_id, plugin=pref, api_type=api_type
+            )
             songs = detail_res.get("data", [])
             if not songs:
-                return await self.xiaomusic.handle_fatal_error(did, f"小Music获取歌单{pl_name}，里面一首歌都没有")
+                return await self.xiaomusic.handle_fatal_error(
+                    did, f"小Music获取歌单{pl_name}，里面一首歌都没有"
+                )
             # 推送到音箱 (归并到 _online_iwebplayer_search)
             self.log.info(f"成功选中歌单【{pl_name}】，共 {len(songs)} 首歌")
-            return await self.push_music_list_play(did, songs, "_online_iwebplayer_search")
+            return await self.push_music_list_play(
+                did, songs, "_online_iwebplayer_search"
+            )
         except Exception as e:
             self.log.error(f"语音搜歌单失败: {e}")
-            return await self.xiaomusic.handle_fatal_error(did, "小Music搜索歌单过程中出了点小故障")
+            return await self.xiaomusic.handle_fatal_error(
+                did, "小Music搜索歌单过程中出了点小故障"
+            )
 
     async def singer_play(self, did="", arg1="", **kwargs):
         await self._before_play()
@@ -693,20 +716,26 @@ class OnlineMusicService:
         if not name:
             name = search_key
         if not name:
-            return await self.xiaomusic.handle_fatal_error(did, "小Music没有听到歌手名哦，请重试。")
+            return await self.xiaomusic.handle_fatal_error(
+                did, "小Music没有听到歌手名哦，请重试。"
+            )
         self.reset_singer_add_page()
         self.log.info(f"搜索关键字{search_key},搜索歌手名{name}")
         try:
             res = await self.search_singer_play(did, search_key, name)
             # 没搜到这位歌手的歌
             if res and isinstance(res, dict) and not res.get("success"):
-                return await self.xiaomusic.handle_fatal_error(did, f"小Music没找到歌手{name}的歌曲")
+                return await self.xiaomusic.handle_fatal_error(
+                    did, f"小Music没找到歌手{name}的歌曲"
+                )
             return res
 
         except Exception as e:
             self.log.error(f"语音搜歌手失败: {e}")
             # 异常兜底：代码崩溃
-            return await self.xiaomusic.handle_fatal_error(did, "小Music搜歌手时出了点小故障")
+            return await self.xiaomusic.handle_fatal_error(
+                did, "小Music搜歌手时出了点小故障"
+            )
 
     # 处理推送的歌单并播放
     async def push_music_list_play(

xiaomusic/utils/music_utils.py

@@ -589,9 +589,7 @@ def extract_audio_metadata(file_path: str, save_root: str) -> dict:
                 tags["metadata_block_picture"][0]
             )
             if picture_data:
-                metadata.picture = _save_picture(
-                    picture_data, save_root, file_path
-                )
+                metadata.picture = _save_picture(picture_data, save_root, file_path)
 
     elif isinstance(audio, ASF):
         metadata.title = _get_tag_value(tags, "Title")

xiaomusic/xiaomusic.py

@@ -382,7 +382,7 @@ class XiaoMusic:
 
     # 在线获取歌单列表
     async def get_playlist_online(
-            self, plugin="all", keyword="", page=1, limit=20, **kwargs
+        self, plugin="all", keyword="", page=1, limit=20, **kwargs
     ):
         """委托给 online_music_service"""
         return await self.online_music_service.get_playlist_online(
@@ -716,6 +716,7 @@ class XiaoMusic:
         await self.play_url(did, error_url)
         # 在第 3 秒时提前下发 stop，把硬件断流的咔声藏在静音里
         import asyncio
+
         await asyncio.sleep(3)
         # 停止
         await self.stop(did, "notts")