fix: 修复音乐路径设置后找不到音乐的问题

xiaomusic/httpserver.py

@@ -1,5 +1,6 @@
 import asyncio
 import json
+from pathlib import Path
 import os
 import secrets
 from contextlib import asynccontextmanager
@@ -73,7 +74,6 @@ def reset_http_server():
         app.dependency_overrides[verification] = no_verification
     else:
         app.dependency_overrides = {}
-    app.mount("/music", StaticFiles(directory=config.music_path), name="music")
 
 
 def HttpInit(_xiaomusic):
@@ -86,6 +86,24 @@ def HttpInit(_xiaomusic):
     reset_http_server()
 
 
+@app.get("/music/{file_path:path}")
+async def read_music_file(file_path: str):
+    base_dir = Path(config.music_path).resolve()
+    real_path = os.path.join(base_dir, file_path)
+    file_location = Path(real_path).resolve()
+    log.info(f"read_music_file. file_path:{file_path} real_path:{real_path}")
+    if not file_location.exists() or not file_location.is_file():
+        raise HTTPException(status_code=404, detail="File not found")
+
+    # 确保请求的文件在我们的基础目录下
+    if base_dir not in file_location.parents:
+        raise HTTPException(
+            status_code=403, detail="Access to this file is not permitted"
+        )
+
+    return FileResponse(file_location)
+
+
 @app.get("/")
 async def read_index():
     return FileResponse("xiaomusic/static/index.html")

xiaomusic/xiaomusic.py

@@ -145,9 +145,9 @@ def setup_logger(self):
     async def poll_latest_ask(self):
         async with ClientSession() as session:
             while True:
-                #self.log.debug(
+                # self.log.debug(
                 #    f"Listening new message, timestamp: {self.last_timestamp}"
-                #)
+                # )
                 session._cookie_jar = self.cookie_jar
 
                 # 拉取所有音箱的对话记录
@@ -158,11 +158,11 @@ async def poll_latest_ask(self):
                 await asyncio.gather(*tasks)
 
                 start = time.perf_counter()
-                #self.log.debug(f"Polling_event, timestamp: {self.last_timestamp}")
+                # self.log.debug(f"Polling_event, timestamp: {self.last_timestamp}")
                 await self.polling_event.wait()
                 if (d := time.perf_counter() - start) < 1:
                     # sleep to avoid too many request
-                    #self.log.debug(f"Sleep {d}, timestamp: {self.last_timestamp}")
+                    # self.log.debug(f"Sleep {d}, timestamp: {self.last_timestamp}")
                     await asyncio.sleep(1 - d)
 
     async def init_all_data(self, session):
@@ -269,7 +269,7 @@ async def get_latest_ask_from_xiaoai(self, session, device_id):
                     hardware=hardware,
                     timestamp=str(int(time.time() * 1000)),
                 )
-                #self.log.debug(f"url:{url} device_id:{device_id} hardware:{hardware}")
+                # self.log.debug(f"url:{url} device_id:{device_id} hardware:{hardware}")
                 r = await session.get(url, timeout=timeout, cookies=cookies)
             except Exception as e:
                 self.log.exception(f"Execption {e}")