|
|
@@ -6,6 +6,7 @@ import shutil
|
|
|
import threading
|
|
|
import uuid
|
|
|
import base64
|
|
|
+import re
|
|
|
import urllib.error
|
|
|
import urllib.parse
|
|
|
import urllib.request
|
|
|
@@ -24,6 +25,18 @@ LOCAL_MUSIC_DIR = BASE_DIR / "mp3file"
|
|
|
CLOUD_MUSIC_DIR = Path("/mnt/baiducloud/百度网盘/mp3file")
|
|
|
CLOUD_WEBDAV_BASE = "http://110.42.102.94:5244/dav"
|
|
|
CLOUD_ALIST_BASE = "http://110.42.102.94:5244"
|
|
|
+def _nat_sort_key(name: str) -> tuple:
|
|
|
+ """Sort key that treats numeric substrings as integers for natural ordering."""
|
|
|
+ parts = re.split(r'(\d+)', name.lower())
|
|
|
+ result = []
|
|
|
+ for part in parts:
|
|
|
+ if part.isdigit():
|
|
|
+ result.append((0, int(part), part))
|
|
|
+ else:
|
|
|
+ result.append((1, 0, part))
|
|
|
+ return result
|
|
|
+
|
|
|
+
|
|
|
CLOUD_WEBDAV_USER = "sequoia00"
|
|
|
CLOUD_WEBDAV_PASSWORD = "792199bb"
|
|
|
MUSIC_ROOTS: list[tuple[str, Path, str]] = [
|
|
|
@@ -56,6 +69,7 @@ app.mount("/static", StaticFiles(directory=BASE_DIR / "static"), name="static")
|
|
|
templates = Jinja2Templates(directory=str(BASE_DIR / "templates"))
|
|
|
CLOUD_REFRESH_STATE = {"running": False}
|
|
|
CLOUD_REFRESH_LOCK = threading.Lock()
|
|
|
+_ALIST_TOKEN_CACHE: dict[str, tuple[str, float]] = {}
|
|
|
|
|
|
|
|
|
class FolderCreateRequest(BaseModel):
|
|
|
@@ -166,6 +180,66 @@ def cloud_alist_api_path(relative_path: str) -> str:
|
|
|
return f"/百度网盘/mp3file/{inner_path}" if inner_path else "/百度网盘/mp3file"
|
|
|
|
|
|
|
|
|
+def _alist_list_dir(alist_path: str, token: str) -> list[dict[str, Any]]:
|
|
|
+ """List files in an alist directory, handling pagination."""
|
|
|
+ items: list[dict[str, Any]] = []
|
|
|
+ page = 1
|
|
|
+ while True:
|
|
|
+ payload = json.dumps({"path": alist_path, "password": "", "page": page, "per_page": 200}).encode("utf-8")
|
|
|
+ request = urllib.request.Request(
|
|
|
+ f"{CLOUD_ALIST_BASE.rstrip('/')}/api/fs/list",
|
|
|
+ data=payload,
|
|
|
+ headers={
|
|
|
+ "Content-Type": "application/json",
|
|
|
+ "Authorization": token,
|
|
|
+ "User-Agent": "MusicWebPlayer/1.0",
|
|
|
+ },
|
|
|
+ )
|
|
|
+ try:
|
|
|
+ with urllib.request.urlopen(request, timeout=20) as response:
|
|
|
+ result = json.loads(response.read().decode("utf-8"))
|
|
|
+ except urllib.error.HTTPError:
|
|
|
+ break
|
|
|
+ except urllib.error.URLError:
|
|
|
+ break
|
|
|
+ data = result.get("data")
|
|
|
+ if not data or not isinstance(data, dict):
|
|
|
+ break
|
|
|
+ content_list = data.get("content", [])
|
|
|
+ if not content_list:
|
|
|
+ break
|
|
|
+ items.extend(content_list)
|
|
|
+ if not data.get("has_more"):
|
|
|
+ break
|
|
|
+ page += 1
|
|
|
+ return items
|
|
|
+
|
|
|
+
|
|
|
+def _get_alist_token() -> str | None:
|
|
|
+ """Get alist login token, cached briefly to avoid repeated logins."""
|
|
|
+ now = datetime.now(timezone.utc).timestamp()
|
|
|
+ cached = _ALIST_TOKEN_CACHE.get("token")
|
|
|
+ if cached and cached[1] > now:
|
|
|
+ return cached[0]
|
|
|
+ payload = json.dumps(
|
|
|
+ {"username": CLOUD_WEBDAV_USER, "password": CLOUD_WEBDAV_PASSWORD}
|
|
|
+ ).encode("utf-8")
|
|
|
+ request = urllib.request.Request(
|
|
|
+ f"{CLOUD_ALIST_BASE.rstrip('/')}/api/auth/login",
|
|
|
+ data=payload,
|
|
|
+ headers={"Content-Type": "application/json", "User-Agent": "MusicWebPlayer/1.0"},
|
|
|
+ )
|
|
|
+ try:
|
|
|
+ with urllib.request.urlopen(request, timeout=20) as response:
|
|
|
+ result = json.loads(response.read().decode("utf-8"))
|
|
|
+ except (urllib.error.HTTPError, urllib.error.URLError):
|
|
|
+ return None
|
|
|
+ token = result.get("data", {}).get("token")
|
|
|
+ if token:
|
|
|
+ _ALIST_TOKEN_CACHE["token"] = (token, now + 8000)
|
|
|
+ return token
|
|
|
+
|
|
|
+
|
|
|
def alist_login_token() -> str:
|
|
|
payload = json.dumps(
|
|
|
{
|
|
|
@@ -272,7 +346,7 @@ def build_tree(root: Path, prefix: str = "", label: str = "音乐库") -> dict[s
|
|
|
"folders": [],
|
|
|
"tracks": [],
|
|
|
}
|
|
|
- for child in sorted(root.iterdir(), key=lambda item: (item.is_file(), item.name.lower())):
|
|
|
+ for child in sorted(root.iterdir(), key=lambda item: (item.is_file(), _nat_sort_key(item.name))):
|
|
|
if child.is_dir():
|
|
|
child_rel = child.relative_to(root).as_posix()
|
|
|
child_prefix = f"{prefix}/{child_rel}" if prefix else child_rel
|
|
|
@@ -290,6 +364,36 @@ def collect_tracks(root: Path, prefix: str = "") -> list[dict[str, str]]:
|
|
|
return tracks
|
|
|
|
|
|
|
|
|
+def _build_cloud_tree_from_alist(alist_path: str, prefix: str, label: str, token: str) -> dict[str, Any]:
|
|
|
+ """Build a library tree node by querying the alist API recursively."""
|
|
|
+ node = {
|
|
|
+ "name": label,
|
|
|
+ "path": prefix,
|
|
|
+ "cover": None,
|
|
|
+ "folders": [],
|
|
|
+ "tracks": [],
|
|
|
+ }
|
|
|
+ try:
|
|
|
+ items = _alist_list_dir(alist_path, token)
|
|
|
+ except Exception:
|
|
|
+ return node
|
|
|
+ folders: list[dict[str, Any]] = []
|
|
|
+ tracks: list[dict[str, str]] = []
|
|
|
+ for item in sorted(items, key=lambda i: (i.get("is_dir", 0), _nat_sort_key(i.get("name", "")))):
|
|
|
+ name = item.get("name", "")
|
|
|
+ is_dir = bool(item.get("is_dir"))
|
|
|
+ if is_dir:
|
|
|
+ child_prefix = f"{prefix}/{name}" if prefix else name
|
|
|
+ child_path = f"{alist_path.rstrip('/')}/{name}" if alist_path else name
|
|
|
+ folders.append(_build_cloud_tree_from_alist(child_path, child_prefix, name, token))
|
|
|
+ else:
|
|
|
+ track_path = f"{prefix}/{name}" if prefix else name
|
|
|
+ tracks.append(build_track(track_path))
|
|
|
+ node["folders"] = folders
|
|
|
+ node["tracks"] = tracks
|
|
|
+ return node
|
|
|
+
|
|
|
+
|
|
|
def root_entry(source: str) -> tuple[str, Path, str]:
|
|
|
for prefix, root, label in MUSIC_ROOTS:
|
|
|
if prefix == source:
|
|
|
@@ -359,7 +463,36 @@ def write_cloud_library_cache(payload: dict[str, Any]) -> dict[str, Any]:
|
|
|
|
|
|
|
|
|
def build_cloud_library_payload() -> dict[str, Any]:
|
|
|
- return build_library_payload("cloud")
|
|
|
+ ensure_storage()
|
|
|
+ token = _get_alist_token()
|
|
|
+ if not token:
|
|
|
+ return {
|
|
|
+ "tree": {"name": "音乐库", "path": "", "cover": None, "folders": [], "tracks": []},
|
|
|
+ "all_tracks": [],
|
|
|
+ "playlists": load_playlists(),
|
|
|
+ }
|
|
|
+ try:
|
|
|
+ cloud_tree = _build_cloud_tree_from_alist("/百度网盘/mp3file", "cloud", "百度网盘", token)
|
|
|
+ except Exception:
|
|
|
+ cloud_tree = {"name": "百度网盘", "path": "cloud", "cover": None, "folders": [], "tracks": []}
|
|
|
+ all_tracks: list[dict[str, str]] = []
|
|
|
+ def _collect(node: dict[str, Any]) -> None:
|
|
|
+ all_tracks.extend(node.get("tracks", []))
|
|
|
+ for child in node.get("folders", []):
|
|
|
+ _collect(child)
|
|
|
+ _collect(cloud_tree)
|
|
|
+ sanitize_library_tree(cloud_tree)
|
|
|
+ return {
|
|
|
+ "tree": {
|
|
|
+ "name": "音乐库",
|
|
|
+ "path": "",
|
|
|
+ "cover": None,
|
|
|
+ "folders": [cloud_tree],
|
|
|
+ "tracks": [],
|
|
|
+ },
|
|
|
+ "all_tracks": all_tracks,
|
|
|
+ "playlists": load_playlists(),
|
|
|
+ }
|
|
|
|
|
|
|
|
|
def cloud_library_payload_from_cache() -> dict[str, Any] | None:
|
