Merge pull request #6634 from dogodefi/codex/music-artist-collection-organize

app/api/endpoints/download.py

@@ -43,6 +43,8 @@ from app.schemas.token import TokenPayload as _SchemaTokenPayload
 from app.schemas.transfer import DownloaderTorrent as _SchemaDownloaderTorrent
 from app.schemas.transfer import MusicInfo as _SchemaMusicInfo
 from app.schemas.types import (
+    MUSIC_ARTIST_COLLECTION_CATEGORY,
+    MUSIC_ENTITY_ARTIST,
     MUSIC_ENTITY_RECORDING,
     MediaSource,
     MediaType,
@@ -253,6 +255,63 @@ def download(
     return _SchemaResponse(success=True, data={"download_id": did})
 
 
+@router.post(
+    "/artist-collection",
+    summary="添加艺术家合集下载",
+    response_model=_SchemaResponse[_SchemaDownloadAddedData],
+)
+def download_artist_collection(
+    artist_name: Annotated[str, Body(min_length=1)],
+    artist_id: Annotated[str, Body(min_length=1)],
+    media_source: Annotated[MediaSource, Body()],
+    torrent_in: _SchemaTorrentInfo,
+    downloader: Annotated[str | None, Body()] = None,
+    save_path: Annotated[str | None, Body()] = None,
+    current_user: ApiPrincipal = Depends(get_current_active_user),
+) -> Any:
+    """Add one artist-wide torrent without pretending that it is one album.
+
+    The artist identity remains available for history and subsequent collection
+    organization.  ``library_category`` is an intentional download-directory
+    classification snapshot: when resource category folders are enabled the
+    task starts under ``Artist Collection`` and therefore never needs an
+    out-of-band filesystem move that could break seeding.
+    """
+    name = artist_name.strip()
+    identity = artist_id.strip()
+    if not is_music_media_source(media_source):
+        return _SchemaResponse(success=False, message="艺术家合集只能使用音乐元数据源")
+
+    mediainfo = MusicInfo(
+        media_source=media_source,
+        media_id=identity,
+        music_type=MUSIC_ENTITY_ARTIST,
+        title=f"{name} 艺术家合集",
+        artists=[name],
+        album_artist=name,
+        album_type=MUSIC_ARTIST_COLLECTION_CATEGORY,
+        library_category=MUSIC_ARTIST_COLLECTION_CATEGORY,
+    )
+    metainfo = MetaInfo(
+        title=torrent_in.title,
+        subtitle=torrent_in.description,
+        mtype=MediaType.MUSIC,
+    )
+    torrentinfo = TorrentInfo()
+    torrentinfo.from_dict(torrent_in.model_dump())
+    torrentinfo.site_downloader = downloader
+    context = Context(meta_info=metainfo, media_info=mediainfo, torrent_info=torrentinfo)
+    did = DownloadChain().download_single(
+        context=context,
+        username=current_user.name,
+        save_path=save_path,
+        source="Manual",
+    )
+    if not did:
+        return _SchemaResponse(success=False, message="艺术家合集任务添加失败")
+    return _SchemaResponse(success=True, data={"download_id": did})
+
+
 @router.post(
     "/add",
     summary="添加下载（不含媒体信息）",

app/application/download/organization.py

@@ -10,7 +10,12 @@ from app.domain.classification.validation import validate_classification_categor
 from app.domain.meta.metabase import MetaBase
 from app.domain.meta.metamusic import MetaMusic
 from app.domain.metainfo import MetaInfo
-from app.schemas.types import MediaSource, MediaType, SystemConfigKey
+from app.schemas.types import (
+    MUSIC_ARTIST_COLLECTION_CATEGORY,
+    MediaSource,
+    MediaType,
+    SystemConfigKey,
+)
 
 _INVALID_NAME = re.compile(r'[<>:"/\\|?*\x00-\x1f]')
 _WINDOWS_DRIVE_PATH = re.compile(r"^[A-Za-z]:[\\/]")
@@ -57,6 +62,8 @@ def _local_path(value: Any, *, label: str, validate: bool = False) -> PurePath:
 
 def _normalize_music_category(media: Any) -> tuple[str, list[str]]:
     """优先使用已生效分类路径，缺失时兼容退回音乐主类型。"""
+    if str(getattr(media, "music_type", None) or "").casefold() == "artist":
+        return MUSIC_ARTIST_COLLECTION_CATEGORY, []
     classified_path = DirectoryHelper().resolve_media_category(media).path
     if classified_path:
         path = validate_classification_category_path(classified_path)
@@ -184,14 +191,20 @@ def _requested_category(value: Any, media_type: Any) -> str:
 
 def _root_name(media: Any) -> str:
     """生成跨电影、电视剧和音乐通用的规范任务根目录名。"""
+    is_artist_collection = False
     if getattr(media, "type", None) == MediaType.MUSIC:
-        title = getattr(media, "album", None) or getattr(media, "title", None)
-        artist = getattr(media, "album_artist", None) or getattr(media, "artist", None)
-        base = " - ".join(str(part).strip() for part in (artist, title) if str(part or "").strip())
+        if str(getattr(media, "music_type", None) or "").casefold() == "artist":
+            is_artist_collection = True
+            artist = getattr(media, "name", None) or getattr(media, "title", None)
+            base = f"{str(artist or '').strip()} - 艺术家合集"
+        else:
+            title = getattr(media, "album", None) or getattr(media, "title", None)
+            artist = getattr(media, "album_artist", None) or getattr(media, "artist", None)
+            base = " - ".join(str(part).strip() for part in (artist, title) if str(part or "").strip())
     else:
         base = str(getattr(media, "title", None) or "").strip()
     year = str(getattr(media, "year", None) or "").strip()
-    if year and f"({year})" not in base:
+    if year and not is_artist_collection and f"({year})" not in base:
         base = f"{base} ({year})"
     return _safe_relative_name(base, label="规范目录名")
 

app/schemas/download.py

@@ -4,7 +4,7 @@ from pydantic import BaseModel, Field
 from pydantic import model_validator as _model_validator
 
 from app.schemas.types import MediaSource as _MediaSource
-from app.schemas.types import MusicTargetEntityType as _MusicTargetEntityType
+from app.schemas.types import MusicEntityType as _MusicEntityType
 
 
 class DownloadTask(BaseModel):
@@ -90,7 +90,8 @@ class DownloadSourceClassificationRequest(BaseModel):  # type: ignore[misc]
     type_name: Optional[Literal["电影", "电视剧", "音乐"]] = None
     media_source: Optional[_MediaSource] = None
     media_id: Optional[str] = None
-    music_type: Optional[_MusicTargetEntityType] = None
+    # 已有任务资源归类额外支持 artist，用于单种子艺术家大合集。
+    music_type: Optional[_MusicEntityType] = None
     episode_group: Optional[str] = None
     media_category: Optional[str] = None
     smart_rename: bool = True

app/schemas/types.py

@@ -5,11 +5,12 @@ from typing import Literal, Optional, Tuple, Union
 from pydantic import GetJsonSchemaHandler
 from pydantic_core import CoreSchema
 
-
 # 音乐实体命名空间由公共类型模块统一持有，避免模型、接口和工具层重复定义。
 MUSIC_ENTITY_RECORDING = "recording"
 MUSIC_ENTITY_ALBUM = "album"
 MUSIC_ENTITY_ARTIST = "artist"
+# 艺术家大合集是下载资源包装分类，不是 MusicBrainz Release Group 类型。
+MUSIC_ARTIST_COLLECTION_CATEGORY = "Artist Collection"
 MusicEntityType = Literal["recording", "album", "artist"]
 MusicTargetEntityType = Literal["recording", "album"]
 MUSIC_ENTITY_TYPES = frozenset({

tests/test_download_source_organization.py

@@ -41,6 +41,7 @@ types_module = ModuleType("app.schemas.types")
 types_module.MediaSource = MediaSource
 types_module.MediaType = MediaType
 types_module.SystemConfigKey = SystemConfigKey
+types_module.MUSIC_ARTIST_COLLECTION_CATEGORY = "Artist Collection"
 
 with patch.dict(
     sys.modules,
@@ -171,6 +172,23 @@ class SourceOrganizationTests(unittest.TestCase):
         self.assertEqual(result["category"], "Album")
         self.assertEqual(result["target_save_path"], "/volume1/UT/Musics/Album")
 
+    def test_artist_collection_uses_one_dedicated_source_category(self):
+        self.request.music_type = "artist"
+        self.history.music_type = "artist"
+        self.media.music_type = "artist"
+        self.media.name = "许嵩"
+        self.media.classification_path = ("未分类",)
+        self.torrent.title = "许嵩[2006-2022]录音室专辑合集"
+
+        result = self.preview()
+
+        self.assertEqual(result["category"], "Artist Collection")
+        self.assertEqual(
+            result["target_save_path"],
+            "/volume1/UT/Musics/Artist Collection",
+        )
+        self.assertEqual(result["proposed_root_name"], "许嵩 - 艺术家合集")
+
     def test_preview_is_read_only_and_includes_qb_root_rename(self):
         result = self.preview()
         self.assertEqual(result["proposed_root_name"], "莫文蔚 - 含情脉脉 (2002)")

tests/test_media_classification_directory.py

@@ -408,6 +408,33 @@ def test_download_path_uses_safe_multilevel_automatic_category(
     assert target == Path("/downloads/电视剧/动漫/日番")
 
 
+def test_artist_collection_download_uses_dedicated_source_category(
+    directory_resolver: ClassificationCategoryResolver,
+) -> None:
+    """大合集作为一个种子只追加一层源目录分类，不伪装为单张 Album。"""
+    directory = TransferDirectoryConf(
+        download_path="/downloads/Musics",
+        media_type="音乐",
+        download_category_folder=True,
+    )
+    media = MusicInfo(
+        media_source="musicbrainz",
+        media_id="artist-1",
+        music_type="artist",
+        title="许嵩 艺术家合集",
+        artists=["许嵩"],
+        library_category="Artist Collection",
+    )
+
+    target = DownloadSubtitleOwner._append_download_classification(
+        root_path=Path("/downloads/Musics"),
+        dir_info=directory,
+        media_info=media,
+    )
+
+    assert target == Path("/downloads/Musics/Artist Collection")
+
+
 def test_fixed_download_category_does_not_append_duplicate_folder(
     directory_resolver: ClassificationCategoryResolver,
 ) -> None:

tests/test_music_download.py

@@ -7,7 +7,7 @@ import pytest
 from jinja2 import Template
 
 import app.chain.download.submission as download_submission
-from app.api.endpoints.download import add, download
+from app.api.endpoints.download import add, download, download_artist_collection
 from app.application.audio import AudioMetadataHelper
 from app.application.messaging.message import TemplateHelper
 from app.chain.download import DownloadChain
@@ -185,6 +185,36 @@ def test_download_endpoint_builds_music_context():
     assert context.meta_info.media_id is None
 
 
+def test_artist_collection_download_keeps_artist_identity_and_source_category():
+    """艺术家大合集不伪装成单张专辑，但为下载目录保存独立分类快照。"""
+    chain = Mock()
+    chain.download_single.return_value = "hash-collection"
+
+    with patch("app.api.endpoints.download.DownloadChain", return_value=chain):
+        response = download_artist_collection(
+            artist_name="许嵩",
+            artist_id="artist-1",
+            media_source="musicbrainz",
+            torrent_in=TorrentInfo(
+                title="许嵩[2006-2022]录音室专辑合集",
+                enclosure="https://example.com/collection.torrent",
+                category="音乐",
+            ),
+            downloader="qb",
+            save_path=None,
+            current_user=SimpleNamespace(name="admin"),
+        )
+
+    assert response.success is True
+    context = chain.download_single.call_args.kwargs["context"]
+    assert context.media_info.music_type == "artist"
+    assert context.media_info.media_id == "artist-1"
+    assert context.media_info.artists == ["许嵩"]
+    assert context.media_info.library_category == "Artist Collection"
+    assert context.media_info.category == "Artist Collection"
+    assert context.torrent_info.site_downloader == "qb"
+
+
 @pytest.mark.parametrize("music_type", ["recording", "album"])
 def test_download_history_restores_selected_music_album(music_type, monkeypatch):
     """手动下载保留独立种子证据，历史恢复仍能为无标签音频补齐已选专辑。"""