feat: import Netease playlists directly by explicit ID

README.en.md

@@ -135,6 +135,14 @@ Music source dependency notes:
 
 ## Netease Support (Optional, via `NeteaseCloudMusicApi`)
 
+### TeamSpeak playlist commands
+
+- `playlist <keywords>` searches Netease playlists and shows up to five results.
+- `select <number>` adds a result from your latest search; results are scoped to your TeamSpeak user and expire after five minutes.
+- `playlist id:<ID>` adds a Netease playlist directly, without searching first; for example, `playlist id:123456789`. The ID must be a positive integer. The Chinese alias is `歌单 id:<ID>`.
+- A bare numeric argument, such as `playlist 123456789`, remains a keyword search. Direct ID import preserves your previous search results for subsequent `select` commands.
+- Both import paths use the same queue behavior: available tracks are added, and playback starts if the player is idle. Playlist visibility and playable tracks depend on the music account and upstream permissions.
+
 This project does **not** directly call Netease official APIs. Instead, it forwards through your own `NeteaseCloudMusicApi` deployment.
 
 - **NPM**: https://www.npmjs.com/package/NeteaseCloudMusicApi

README.md

@@ -144,6 +144,7 @@ TSBot 的目标是把边界重新划清：
 | 中文指令 | 英文指令 | 行为 |
 | --- | --- | --- |
 | `歌单 <关键词>` | `playlist <keywords>` | 仅检索网易云歌单，显示前 5 个结果；结果按当前 TeamSpeak 用户隔离并保留 5 分钟。 |
+| `歌单 id:<ID>` | `playlist id:<ID>` | 直接按网易云歌单 ID 加入队列，无需先搜索；例如 `歌单 id:123456789`。ID 必须为正整数。 |
 | `选择 <编号>` | `select <number>` | 从当前用户最近一次歌单搜索结果中选择歌单，将可用曲目加入播放队列；空闲时自动开始播放。 |
 | `清空` | `clear` | 清空播放队列、停止当前播放、取消正在加载的曲目并重置随机队列。 |
 | `随机` / `随机播放` | `random` / `shuffle` | 切换为随机播放；空闲时立即从当前队列随机选择曲目。 |
@@ -153,6 +154,8 @@ TSBot 的目标是把边界重新划清：
 
 可发送 `帮助`、`菜单`、`指令`、`help` 或 `?` 查看机器人返回的完整指令帮助。
 
+纯数字参数（如 `歌单 123456789`）仍作为关键词搜索；直接指定歌单时请使用 `id:` 前缀。直接导入不会覆盖上一次搜索结果，之后仍可用 `选择 <编号>`。
+
 ## 网易云音乐支持（可选，依赖 `NeteaseCloudMusicApi`）
 
 本项目 **不直接** 调用网易云官方接口；而是通过你自行部署的 `NeteaseCloudMusicApi` 服务转发/封装。

backend/main.py

@@ -3133,7 +3133,7 @@ def _format_help() -> str:
         "增加|add <song_id|keywords> - add to queue\n"
         "播放|play [song_id|keywords] - play now; no argument plays the first queue item\n"
         "队列|queue - show queue\n"
-        "歌单|playlist <keywords> - search Netease playlists\n"
+        "歌单|playlist <keywords> / id:<ID> - search Netease playlists or add directly by ID\n"
         "选择|select <number> - add a playlist from the last playlist search\n"
         "清空|clear - clear the current queue\n"
         "顺序播放|order / 随机播放|random - switch queue playback mode\n"
@@ -3694,9 +3694,16 @@ async def reply(text: str) -> None:
             await reply(f"已切换为{mode}")
             return
 
-        if cmd == "playlist":
+        selected_playlist: dict[str, str] | None = None
+        if cmd == "playlist" and arg[:3].lower() == "id:":
+            playlist_id = arg[3:].strip()
+            if not re.fullmatch(r"[0-9]+", playlist_id) or not playlist_id.lstrip("0"):
+                await reply("歌单 ID 必须为正整数，用法: playlist id:<ID>")
+                return
+            selected_playlist = {"id": playlist_id.lstrip("0"), "name": ""}
+        elif cmd == "playlist":
             if not arg:
-                await reply("用法: playlist <歌单关键词>")
+                await reply("用法: playlist <歌单关键词> 或 playlist id:<ID>")
                 return
             raw = await netease.search(keywords=arg, limit=5, type_=1000)
             playlists = _extract_playlist_search_items(raw)
@@ -3724,7 +3731,6 @@ async def reply(text: str) -> None:
                 return
             playlists = _get_ts_playlist_results(invoker_key)
             selection_token = arg.split()[0].strip()
-            selected_playlist: dict[str, str] | None = None
             try:
                 selection = int(selection_token)
             except ValueError:
@@ -3739,6 +3745,8 @@ async def reply(text: str) -> None:
             if selected_playlist is None:
                 await reply("歌单编号无效，请先使用 playlist 搜索")
                 return
+
+        if selected_playlist is not None:
             playlist_id = selected_playlist["id"]
 
             cookie = _get_admin_cookie_or_none()

tests/test_playlist_id.py

@@ -0,0 +1,78 @@
+from contextlib import ExitStack
+import unittest
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, patch
+
+from sqlalchemy import create_engine, select
+from sqlalchemy.orm import sessionmaker
+
+from backend import main
+from backend.db import Base
+from backend.models import QueueItem
+
+
+class PlaylistIdTests(unittest.IsolatedAsyncioTestCase):
+    def setUp(self):
+        engine = create_engine("sqlite:///:memory:")
+        self.addCleanup(engine.dispose)
+        Base.metadata.create_all(engine)
+        self.sessions = sessionmaker(bind=engine)
+        patches = ExitStack()
+        self.addCleanup(patches.close)
+        patches.enter_context(patch.object(main, "new_session", side_effect=self.sessions))
+        patches.enter_context(patch.object(main, "_schedule_ts_description_update"))
+        patches.enter_context(patch.object(main, "_ts_playlist_results", {}))
+        patches.enter_context(patch.object(main, "_get_admin_cookie_or_none", return_value="test-cookie"))
+        self.search = patches.enter_context(patch.object(main.netease, "search", AsyncMock(return_value={"result": {"playlists": []}})))
+        self.detail = patches.enter_context(patch.object(main.netease, "playlist_detail", AsyncMock(return_value={
+            "playlist": {"name": "By ID", "tracks": [
+                {"id": 11, "name": "First", "ar": [{"name": "Artist"}], "al": {"name": "Album"}},
+                {"id": 22, "name": "Second", "ar": [{"name": "Artist"}], "al": {"name": "Album"}},
+            ]}
+        })))
+        patches.enter_context(patch.object(main.voice, "get_status", AsyncMock(return_value=SimpleNamespace(state="STATE_PLAYING"))))
+        self.notice = patches.enter_context(patch.object(main.voice, "send_notice", AsyncMock()))
+
+    def queued(self):
+        with self.sessions() as session:
+            return [(row.track_id, row.title) for row in session.execute(select(QueueItem).order_by(QueueItem.id)).scalars()]
+
+    async def test_direct_id_enqueues_without_prior_search(self):
+        await main._handle_chat_command("Alice", "playlist id:123456789", invoker_unique_id="alice")
+        self.assertEqual([("netease:11", "First"), ("netease:22", "Second")], self.queued())
+        self.detail.assert_awaited_once_with(playlist_id="123456789", cookie="test-cookie")
+        self.search.assert_not_awaited()
+        self.assertIn("加入 2 首歌曲", self.notice.await_args.args[0])
+
+    async def test_chinese_alias_accepts_case_insensitive_id_prefix(self):
+        await main._handle_chat_command("Alice", "歌单 ID: 00123", invoker_unique_id="alice")
+        self.assertEqual(2, len(self.queued()))
+        self.detail.assert_awaited_once_with(playlist_id="123", cookie="test-cookie")
+
+    async def test_invalid_explicit_ids_do_not_make_music_api_requests(self):
+        for value in ["", "0", "000", "-1", "+123", "12x", "123 456", "１２３"]:
+            with self.subTest(value=value):
+                await main._handle_chat_command("Alice", "playlist id:" + value, invoker_unique_id="alice")
+                self.assertIn("ID", self.notice.await_args.args[0])
+        self.assertEqual([], self.queued())
+        self.detail.assert_not_awaited()
+        self.search.assert_not_awaited()
+
+    async def test_numeric_keyword_still_searches(self):
+        await main._handle_chat_command("Alice", "playlist 123456789", invoker_unique_id="alice")
+        self.search.assert_awaited_once_with(keywords="123456789", limit=5, type_=1000)
+        self.detail.assert_not_awaited()
+        self.assertEqual([], self.queued())
+
+    async def test_direct_id_preserves_previous_search_selection(self):
+        main._remember_ts_playlist_results("uid:alice", [{"id": "987", "name": "Previous"}])
+        await main._handle_chat_command("Alice", "playlist id:123", invoker_unique_id="alice")
+        await main._handle_chat_command("Alice", "select 1", invoker_unique_id="alice")
+        self.assertEqual(4, len(self.queued()))
+        self.assertEqual("987", self.detail.await_args.kwargs["playlist_id"])
+
+    async def test_empty_playlist_reports_no_tracks_without_changing_queue(self):
+        self.detail.return_value = {"playlist": {"name": "Empty", "tracks": []}}
+        await main._handle_chat_command("Alice", "playlist id:123", invoker_unique_id="alice")
+        self.assertEqual([], self.queued())
+        self.assertIn("歌单为空", self.notice.await_args.args[0])