feat(media): support extensible media sources

app/api/endpoints/music.py

@@ -42,11 +42,6 @@ MusicAlbumTypeParam = Annotated[
     Optional[str],
     Query(pattern="^(album|single|ep|broadcast|other|compilation|soundtrack|live|remix)$"),
 ]
-_MUSIC_DETAIL_SOURCES = frozenset({
-    MediaSource.MusicBrainz,
-    MediaSource.TheAudioDB,
-    MediaSource.DoubanMusic,
-})
 _MUSIC_EXPLORE_SOURCES = frozenset({
     MediaSource.MusicBrainz,
     MediaSource.DoubanMusic,
@@ -55,14 +50,14 @@ _MUSIC_EXPLORE_SOURCES = frozenset({
 
 def _validate_music_source(
         media_source: MediaSource,
-        allowed_sources: frozenset[MediaSource],
+        allowed_sources: Optional[frozenset[MediaSource]] = None,
 ) -> MediaSource:
-    """将 HTTP 或直接调用参数规范为音乐来源枚举，并拒绝不支持的来源。"""
+    """规范音乐来源；仅来源专属端点额外限制内置来源集合。"""
     try:
         normalized_source = MediaSource(media_source)
     except (TypeError, ValueError) as err:
         raise HTTPException(status_code=422, detail="无效的媒体来源") from err
-    if normalized_source not in allowed_sources:
+    if allowed_sources is not None and normalized_source not in allowed_sources:
         raise HTTPException(status_code=422, detail="该媒体来源不支持此音乐接口")
     return normalized_source
 
@@ -228,7 +223,7 @@ async def music_album(
         _: schemas.TokenPayload = Depends(verify_token),
 ) -> schemas.MusicAlbumInfo:
     """按专辑标准 ID 返回专辑详情、曲目列表和发行版本。"""
-    media_source = _validate_music_source(media_source, _MUSIC_DETAIL_SOURCES)
+    media_source = _validate_music_source(media_source)
     info = await MediaChain().async_get_music_album(
         media_source=media_source, media_id=album_id
     )
@@ -249,7 +244,7 @@ async def music_album_related(
         _: schemas.TokenPayload = Depends(verify_token),
 ) -> list[schemas.MusicInfo]:
     """按来源和专辑 ID 返回可继续浏览的关联专辑。"""
-    media_source = _validate_music_source(media_source, _MUSIC_DETAIL_SOURCES)
+    media_source = _validate_music_source(media_source)
     results = await MediaChain().async_get_music_album_related(
         media_source=media_source,
         media_id=album_id,
@@ -272,7 +267,7 @@ async def music_artist_albums(
         _: schemas.TokenPayload = Depends(verify_token),
 ) -> list[schemas.MusicInfo]:
     """按艺术家标准 ID 分页返回其专辑、EP 和单曲。"""
-    media_source = _validate_music_source(media_source, _MUSIC_DETAIL_SOURCES)
+    media_source = _validate_music_source(media_source)
     results = await MediaChain().async_get_music_artist_albums(
         media_source=media_source,
         media_id=artist_id,
@@ -295,7 +290,7 @@ async def music_artist_related(
         _: schemas.TokenPayload = Depends(verify_token),
 ) -> list[schemas.MusicArtistInfo]:
     """按艺术家关系返回可继续浏览的关联艺术家。"""
-    media_source = _validate_music_source(media_source, _MUSIC_DETAIL_SOURCES)
+    media_source = _validate_music_source(media_source)
     results = await MediaChain().async_get_music_artist_related(
         media_source=media_source,
         media_id=artist_id,
@@ -315,7 +310,7 @@ async def music_artist(
         _: schemas.TokenPayload = Depends(verify_token),
 ) -> schemas.MusicArtistInfo:
     """按艺术家标准 ID 返回艺术家详情。"""
-    media_source = _validate_music_source(media_source, _MUSIC_DETAIL_SOURCES)
+    media_source = _validate_music_source(media_source)
     info = await MediaChain().async_get_music_artist(
         media_source=media_source, media_id=artist_id
     )

app/chain/media.py

@@ -10,7 +10,7 @@ from app import schemas
 from app.chain import ChainBase
 from app.chain.acoustid import AcoustIdChain
 from app.chain.douban import DoubanChain
-from app.chain.musicbrainz import MusicBrainzChain
+from app.chain.musicbrainz import MusicBrainzChain, _MusicMetadataSourceChain
 from app.chain.theaudiodb import TheAudioDbChain
 from app.core.cache import async_fresh, fresh
 from app.core.config import settings
@@ -61,23 +61,29 @@ class MediaChain(ChainBase, metaclass=Singleton):
     @staticmethod
     def _music_source_chain(
             media_source: MediaSource,
-    ) -> Optional[MusicBrainzChain | TheAudioDbChain | DoubanChain]:
-        """按固定音乐来源返回对应来源链。"""
+    ) -> Optional[_MusicMetadataSourceChain | DoubanChain]:
+        """返回内置来源专用链，或绑定插件扩展来源的通用音乐端口。"""
         source = normalize_media_source(media_source)
+        if not source:
+            return None
         chains = {
             MediaSource.MusicBrainz: MusicBrainzChain,
             MediaSource.TheAudioDB: TheAudioDbChain,
             MediaSource.DoubanMusic: DoubanChain,
         }
         chain_type = chains.get(source)
-        return chain_type() if chain_type else None
+        if chain_type:
+            return chain_type()
+        plugin_chain = _MusicMetadataSourceChain()
+        plugin_chain.source = source
+        return plugin_chain
 
     @classmethod
     def _music_search_sources(
             cls,
             media_source: Optional[MediaSourceSelection],
     ) -> list[MediaSource]:
-        """解析有序音乐搜索来源集合，忽略未知来源和重复项。"""
+        """解析有序音乐搜索来源集合，保留合法插件扩展来源并去重。"""
         if not media_source:
             return [cls._music_primary_source]
         raw_sources = (
@@ -87,13 +93,14 @@ class MediaChain(ChainBase, metaclass=Singleton):
         )
         sources: list[MediaSource] = []
         for raw_source in raw_sources:
-            if is_music_media_source(raw_source) and raw_source not in sources:
-                sources.append(raw_source)
+            source = normalize_media_source(raw_source)
+            if source and source not in sources:
+                sources.append(source)
         return sources
 
     @staticmethod
     async def _async_search_music_source(
-            chain: MusicBrainzChain | TheAudioDbChain | DoubanChain,
+            chain: _MusicMetadataSourceChain | DoubanChain,
             source: MediaSource,
             meta: MetaMusic,
             limit: int,
@@ -1480,7 +1487,15 @@ class MediaChain(ChainBase, metaclass=Singleton):
                 mtype=mtype or MediaInfo.get_bangumi_media_type(source_info),
                 season=season if season is not None else meta.begin_season,
             )
-        return None
+        event_data = schemas.MediaRecognizeConvertEventData(
+            media_source=media_source,
+            media_id=media_id,
+            target_media_source=target_source,
+        )
+        event = eventmanager.send_event(
+            ChainEventType.MediaRecognizeConvert, event_data,
+        )
+        return event_data.media_dict if event and event_data.media_dict else None
 
 
     @staticmethod
@@ -2018,4 +2033,12 @@ class MediaChain(ChainBase, metaclass=Singleton):
                 mtype=mtype or MediaInfo.get_bangumi_media_type(source_info),
                 season=season if season is not None else meta.begin_season,
             )
-        return None
+        event_data = schemas.MediaRecognizeConvertEventData(
+            media_source=media_source,
+            media_id=media_id,
+            target_media_source=target_source,
+        )
+        event = await eventmanager.async_send_event(
+            ChainEventType.MediaRecognizeConvert, event_data,
+        )
+        return event_data.media_dict if event and event_data.media_dict else None

app/chain/musicbrainz.py

@@ -168,21 +168,23 @@ class _MusicMetadataSourceChain(ChainBase):
         normalized = str(media_id).strip() if media_id is not None else ""
         return normalized if normalized and normalized != "0" else None
 
-    @classmethod
-    def _music_infos(cls, result: Any, limit: Optional[int] = None) -> list[MusicInfo]:
-        """将模块或插件结果统一转换为音乐候选列表。"""
+    def _music_infos(
+            self,
+            result: Any,
+            limit: Optional[int] = None,
+    ) -> list[MusicInfo]:
+        """将模块或插件结果统一转换为当前来源的音乐候选列表。"""
         candidates = result if isinstance(result, list) else []
         infos = [
             item if isinstance(item, MusicInfo) else MusicInfo.from_dict(item)
             for item in candidates
             if isinstance(item, (MusicInfo, dict))
         ]
-        infos = [info for info in infos if info.media_source == cls.source]
+        infos = [info for info in infos if info.media_source == self.source]
         return infos[:limit] if limit else infos
 
-    @classmethod
     def _music_info(
-            cls,
+            self,
             result: Any,
             media_id: Optional[str] = None,
     ) -> Optional[MusicInfo]:
@@ -193,15 +195,14 @@ class _MusicMetadataSourceChain(ChainBase):
             info = MusicInfo.from_dict(result)
         else:
             return None
-        if info.media_source and info.media_source != cls.source:
+        if info.media_source and info.media_source != self.source:
             return None
-        if media_id and (info.media_source != cls.source or info.media_id != media_id):
+        if media_id and (info.media_source != self.source or info.media_id != media_id):
             return None
         return info
 
-    @classmethod
     def _music_album(
-            cls,
+            self,
             result: Any,
             media_id: Optional[str] = None,
     ) -> Optional[MusicAlbumInfo]:
@@ -212,15 +213,14 @@ class _MusicMetadataSourceChain(ChainBase):
             album = MusicAlbumInfo.from_dict(result)
         else:
             return None
-        if album.media_source != cls.source:
+        if album.media_source != self.source:
             return None
         if media_id and album.media_id != media_id:
             return None
         return album
 
-    @classmethod
     def _music_artist(
-            cls,
+            self,
             result: Any,
             media_id: Optional[str] = None,
     ) -> Optional[MusicArtistInfo]:
@@ -231,15 +231,14 @@ class _MusicMetadataSourceChain(ChainBase):
             artist = MusicArtistInfo.from_dict(result)
         else:
             return None
-        if artist.media_source != cls.source:
+        if artist.media_source != self.source:
             return None
         if media_id and artist.media_id != media_id:
             return None
         return artist
 
-    @classmethod
     def _music_artists(
-            cls,
+            self,
             result: Any,
             limit: Optional[int] = None,
     ) -> list[MusicArtistInfo]:
@@ -250,7 +249,7 @@ class _MusicMetadataSourceChain(ChainBase):
             for item in candidates
             if isinstance(item, (MusicArtistInfo, dict))
         ]
-        artists = [artist for artist in artists if artist.media_source == cls.source]
+        artists = [artist for artist in artists if artist.media_source == self.source]
         return artists[:limit] if limit else artists
 
 

app/db/models/media_identity.py

@@ -1,21 +1,17 @@
 from sqlalchemy import CheckConstraint
 
-from app.schemas.types import MediaSource
-
-
-MEDIA_SOURCE_SQL_VALUES = ", ".join(
-    f"'{media_source.value}'" for media_source in MediaSource
-)
 MEDIA_IDENTITY_CHECK_SQL = (
     "(media_source IS NULL AND media_id IS NULL) OR "
     "(media_source IS NOT NULL AND "
-    f"media_source IN ({MEDIA_SOURCE_SQL_VALUES}) AND "
+    "trim(media_source) <> '' AND media_source = lower(trim(media_source)) AND "
+    "length(media_source) <= 64 AND media_source NOT LIKE '%:%' AND "
+    "media_source NOT LIKE '% %' AND "
     "media_id IS NOT NULL AND trim(media_id) <> '' AND trim(media_id) <> '0')"
 )
 
 
 def media_identity_constraint(table_name: str) -> CheckConstraint:
-    """构造通用媒体表使用的来源枚举与身份成对数据库约束。"""
+    """构造允许插件扩展来源且保证身份成对的数据库约束。"""
     return CheckConstraint(
         MEDIA_IDENTITY_CHECK_SQL,
         name=f"ck_{table_name}_media_identity",

app/schemas/event.py

@@ -482,12 +482,12 @@ class DiscoverMediaSource(BaseModel):
     探索媒体数据源的基类。
 
     ``mediaid_prefix`` 是既有插件与前端标签使用的稳定标识；
-    ``media_source`` 是新的规范媒体来源。模型同时输出两者，并在输入时互相补齐，
-    以兼容尚未升级的已安装插件。
+    ``media_source`` 是新的规范媒体来源，可以是内置常量或插件扩展成员。模型同时
+    输出两者，并在输入时互相补齐，以兼容尚未升级的已安装插件。
     """
 
     name: str = Field(..., description="数据源名称")
-    media_source: MediaSource = Field(..., description="媒体来源枚举")
+    media_source: MediaSource = Field(..., description="内置或插件扩展媒体来源")
     mediaid_prefix: str = Field(..., description="兼容插件使用的媒体ID前缀")
     api_path: str = Field(..., description="媒体数据源API地址")
     filter_params: Optional[Dict[str, JsonData]] = Field(
@@ -517,7 +517,7 @@ class DiscoverMediaSource(BaseModel):
 
     @staticmethod
     def _media_source_from_prefix(mediaid_prefix: str) -> MediaSource:
-        """将旧插件使用的历史前缀映射为规范媒体来源枚举。"""
+        """将旧插件前缀映射为内置或插件扩展媒体来源。"""
         aliases = {
             "mangguo": MediaSource.MangoTV,
             "tencentvideo": MediaSource.TencentVideo,
@@ -573,9 +573,9 @@ class MediaRecognizeConvertEventData(RequiredMediaIdentityMixin, ChainEventData)
 
     Attributes:
         # 输入参数
-        media_source (MediaSource): 输入媒体来源
+        media_source (MediaSource): 输入内置或插件扩展媒体来源
         media_id (str): 数据源原生 ID
-        target_media_source (MediaSource): 需要转换到的目标媒体来源
+        target_media_source (MediaSource): 需要转换到的内置或插件扩展媒体来源
 
         # 输出参数
         media_dict (dict): TheMovieDb/豆瓣的媒体数据

app/schemas/media.py

@@ -4,7 +4,7 @@ from app.schemas.types import MediaSource
 
 
 class OptionalMediaIdentityMixin:
-    """为可选媒体身份模型统一校验来源枚举与原生 ID 的成对约束。"""
+    """为可选媒体身份模型统一校验内置或插件来源与原生 ID 的成对约束。"""
 
     @model_validator(mode="before")
     @classmethod
@@ -52,7 +52,7 @@ class OptionalMediaIdentityMixin:
 
 
 class RequiredMediaIdentityMixin:
-    """为必填媒体身份模型统一校验来源枚举与原生 ID。"""
+    """为必填媒体身份模型统一校验内置或插件来源与原生 ID。"""
 
     @model_validator(mode="after")
     def _validate_required_media_identity(self):

app/schemas/transfer.py

@@ -224,7 +224,7 @@ class EpisodeFormatRecommendItem(BaseModel):
 
 
 class ManualTransferItem(OptionalMediaIdentityMixin, BaseModel):
-    """手动整理请求，媒体身份只接受来源枚举与原生 ID。"""
+    """手动整理请求，媒体身份接受内置或插件来源与原生 ID。"""
 
     # 文件项
     fileitem: FileItem = None

app/schemas/types.py

@@ -1,6 +1,10 @@
+import re
 from enum import Enum
 from typing import Literal, Optional, Tuple, Union
 
+from pydantic import GetJsonSchemaHandler
+from pydantic_core import CoreSchema
+
 
 # 音乐实体命名空间由公共类型模块统一持有，避免模型、接口和工具层重复定义。
 MUSIC_ENTITY_RECORDING = "recording"
@@ -46,8 +50,20 @@ class MediaType(Enum):
         }.get(self, self.value)
 
 
+MEDIA_SOURCE_IDENTIFIER_PATTERN = r"^[a-z][a-z0-9._-]{0,63}$"
+_MEDIA_SOURCE_IDENTIFIER_RE = re.compile(MEDIA_SOURCE_IDENTIFIER_PATTERN)
+_MEDIA_SOURCE_VALUE_ALIASES = {
+    "tmdb": "themoviedb",
+    "audio_db": "theaudiodb",
+    "douban_music": "doubanmusic",
+    "mango_tv": "mangguodiscover",
+    "migu_video": "migu",
+    "tencent_video": "tencentvideodiscover",
+}
+
+
 class MediaSource(str, Enum):
-    """媒体主身份的数据来源。"""
+    """媒体主身份的数据来源，内置来源为常量，插件来源为动态扩展成员。"""
 
     TMDB = "themoviedb"
     Douban = "douban"
@@ -67,8 +83,37 @@ class MediaSource(str, Enum):
         """返回可直接用于 API 和数据库的规范值。"""
         return self.value
 
+    @classmethod
+    def _missing_(cls, value: object) -> Optional["MediaSource"]:
+        """将合法插件来源标识解析为动态枚举成员，并规范化内置别名。"""
+        if not isinstance(value, str):
+            return None
+        normalized = value.strip().casefold()
+        normalized = _MEDIA_SOURCE_VALUE_ALIASES.get(normalized, normalized)
+        known_member = cls._value2member_map_.get(normalized)
+        if known_member:
+            return known_member
+        if not _MEDIA_SOURCE_IDENTIFIER_RE.fullmatch(normalized):
+            return None
+        member = str.__new__(cls, normalized)
+        member._name_ = normalized
+        member._value_ = normalized
+        cls._value2member_map_.setdefault(normalized, member)
+        return cls._value2member_map_[normalized]
 
-# 搜索可以选择一个或多个来源，但集合中的每一项都必须是固定枚举。
+    @classmethod
+    def __get_pydantic_json_schema__(
+            cls, core_schema: CoreSchema, handler: GetJsonSchemaHandler,
+    ) -> dict:
+        """在 OpenAPI 中声明可扩展标识格式，避免把内置成员误写成完整白名单。"""
+        schema = handler(core_schema)
+        schema.pop("enum", None)
+        schema["pattern"] = MEDIA_SOURCE_IDENTIFIER_PATTERN
+        schema["examples"] = [source.value for source in cls]
+        return schema
+
+
+# 搜索可以选择一个或多个内置或插件扩展来源。
 MediaSourceSelection = Union[MediaSource, Tuple[MediaSource, ...]]
 
 

app/utils/media.py

@@ -75,13 +75,19 @@ def is_music_media_source(
 def normalize_media_source(
         source: Optional[Union[MediaSource, str]],
 ) -> Optional[MediaSource]:
-    """将来源别名规范化为固定枚举，未知来源返回 None。"""
+    """将内置别名或插件扩展标识规范化为 MediaSource。"""
     if not source:
         return None
     if isinstance(source, MediaSource):
         return source
     normalized = str(source).strip().casefold()
-    return MEDIA_SOURCE_ALIASES.get(normalized)
+    builtin_source = MEDIA_SOURCE_ALIASES.get(normalized)
+    if builtin_source:
+        return builtin_source
+    try:
+        return MediaSource(normalized)
+    except ValueError:
+        return None
 
 
 def parse_media_source_selection(value: Optional[str]) -> Tuple[MediaSource, ...]:
@@ -90,7 +96,7 @@ def parse_media_source_selection(value: Optional[str]) -> Tuple[MediaSource, ...
 
     :param value: 逗号分隔的来源值；空值表示未显式选择来源
     :return: 去重后的媒体来源枚举元组
-    :raises ValueError: 包含固定枚举之外的来源
+    :raises ValueError: 包含格式非法的来源标识
     """
     if not value:
         return ()
@@ -241,5 +247,5 @@ def build_media_key(
     normalized_id = str(media_id).strip() if media_id is not None else ""
     if not normalized_source or not normalized_id or normalized_id == "0":
         return ""
-    prefix = MEDIA_SOURCE_PREFIXES[normalized_source]
+    prefix = MEDIA_SOURCE_PREFIXES.get(normalized_source, normalized_source.value)
     return f"{prefix}:{normalized_id}"

database/versions/8a4c7e1d2f90_3_0_0.py

@@ -7,6 +7,7 @@ Create Date: 2026-08-12
 """
 
 from collections.abc import Iterable
+import re
 
 from alembic import op
 import sqlalchemy as sa
@@ -69,16 +70,15 @@ SOURCE_ALIASES = {
     "tencentvideodiscover": "tencentvideodiscover",
     "tencent_video": "tencentvideodiscover",
 }
-MEDIA_SOURCE_VALUES = frozenset(SOURCE_ALIASES.values())
-MEDIA_SOURCE_SQL_VALUES = ", ".join(
-    f"'{source}'" for source in sorted(MEDIA_SOURCE_VALUES)
-)
 MEDIA_IDENTITY_CHECK_SQL = (
     "(media_source IS NULL AND media_id IS NULL) OR "
     "(media_source IS NOT NULL AND "
-    f"media_source IN ({MEDIA_SOURCE_SQL_VALUES}) AND "
+    "trim(media_source) <> '' AND media_source = lower(trim(media_source)) AND "
+    "length(media_source) <= 64 AND media_source NOT LIKE '%:%' AND "
+    "media_source NOT LIKE '% %' AND "
     "media_id IS NOT NULL AND trim(media_id) <> '' AND trim(media_id) <> '0')"
 )
+MEDIA_SOURCE_PATTERN = re.compile(r"^[a-z][a-z0-9._-]{0,63}$")
 
 
 def _inspector() -> sa.Inspector:
@@ -118,7 +118,7 @@ def _identity_missing(table: sa.TableClause):
 
 
 def _normalize_existing_sources(table_name: str) -> None:
-    """把旧版本允许的来源别名规范化为当前枚举值。"""
+    """规范内置来源别名，并保留插件注册的扩展来源标识。"""
     table = sa.table(
         table_name,
         sa.column("media_source", sa.String()),
@@ -133,7 +133,7 @@ def _normalize_existing_sources(table_name: str) -> None:
     connection.execute(
         table.update()
         .where(table.c.media_source.is_not(None))
-        .values(media_source=sa.func.trim(table.c.media_source))
+        .values(media_source=sa.func.lower(sa.func.trim(table.c.media_source)))
     )
 
 
@@ -147,9 +147,9 @@ def _clear_invalid_or_partial_identity(table_name: str) -> None:
     invalid_identity = sa.or_(
         table.c.media_source.is_(None),
         sa.func.trim(table.c.media_source) == "",
-        sa.func.lower(sa.func.trim(table.c.media_source)).not_in(
-            MEDIA_SOURCE_VALUES
-        ),
+        sa.func.length(sa.func.trim(table.c.media_source)) > 64,
+        sa.func.trim(table.c.media_source).contains(":"),
+        sa.func.trim(table.c.media_source).contains(" "),
         table.c.media_id.is_(None),
         sa.func.trim(table.c.media_id) == "",
         sa.func.trim(table.c.media_id) == "0",
@@ -167,11 +167,12 @@ def _clear_invalid_or_partial_identity(table_name: str) -> None:
 
 
 def _backfill_prefixed_media_id(table_name: str, columns: set[str]) -> None:
-    """从旧的 ``prefix:id`` 组合字段回填规范身份。"""
+    """从旧的 ``prefix:id`` 组合字段回填内置或插件扩展身份。"""
     if "mediaid" not in columns:
         return
     table = sa.table(
         table_name,
+        sa.column("id", sa.Integer()),
         sa.column("mediaid", sa.String()),
         sa.column("media_source", sa.String()),
         sa.column("media_id", sa.String()),
@@ -205,6 +206,32 @@ def _backfill_prefixed_media_id(table_name: str, columns: set[str]) -> None:
             )
         )
 
+    # 插件来源无法预先枚举，已知别名批量回填后再解析剩余合法前缀。
+    connection = op.get_bind()
+    rows = connection.execute(
+        sa.select(table.c.id, table.c.mediaid)
+        .where(_identity_missing(table))
+        .where(table.c.mediaid.is_not(None))
+    ).mappings().all()
+    for row in rows:
+        raw_media_id = str(row["mediaid"]).strip()
+        raw_source, separator, raw_native_id = raw_media_id.partition(":")
+        media_source = raw_source.strip().casefold()
+        media_id = raw_native_id.strip()
+        if (
+                not separator
+                or not MEDIA_SOURCE_PATTERN.fullmatch(media_source)
+                or not media_id
+                or media_id == "0"
+        ):
+            continue
+        connection.execute(
+            table.update()
+            .where(table.c.id == row["id"])
+            .where(_identity_missing(table))
+            .values(media_source=media_source, media_id=media_id)
+        )
+
 
 def _backfill_source_columns(table_name: str, columns: set[str]) -> None:
     """按确定优先级从旧的来源专用字段回填规范身份。"""
@@ -302,7 +329,7 @@ def _ensure_identity_indexes() -> None:
 
 
 def _ensure_identity_constraints() -> None:
-    """为六张通用媒体表建立来源枚举与身份成对数据库约束。"""
+    """为六张通用媒体表建立可扩展来源与身份成对数据库约束。"""
     for table_name in LEGACY_COLUMNS:
         if not _has_table(table_name):
             continue

database/versions/b3d7e9f1a2c4_3_0_0.py

@@ -0,0 +1,74 @@
+"""3.0.0
+允许插件扩展媒体来源
+
+Revision ID: b3d7e9f1a2c4
+Revises: e3d9f4b7c806
+Create Date: 2026-08-13
+"""
+
+from alembic import op
+import sqlalchemy as sa
+
+
+revision = "b3d7e9f1a2c4"
+down_revision = "e3d9f4b7c806"
+branch_labels = None
+depends_on = None
+
+
+MEDIA_TABLES = (
+    "subscribe",
+    "subscribehistory",
+    "downloadhistory",
+    "transferhistory",
+    "downloadfailure",
+    "mediaserveritem",
+)
+EXTENSIBLE_IDENTITY_CHECK_SQL = (
+    "(media_source IS NULL AND media_id IS NULL) OR "
+    "(media_source IS NOT NULL AND "
+    "trim(media_source) <> '' AND media_source = lower(trim(media_source)) AND "
+    "length(media_source) <= 64 AND media_source NOT LIKE '%:%' AND "
+    "media_source NOT LIKE '% %' AND "
+    "media_id IS NOT NULL AND trim(media_id) <> '' AND trim(media_id) <> '0')"
+)
+BUILTIN_IDENTITY_CHECK_SQL = (
+    "(media_source IS NULL AND media_id IS NULL) OR "
+    "(media_source IS NOT NULL AND media_source IN ("
+    "'anilist', 'bangumi', 'bilibili', 'douban', 'doubanmusic', 'imdb', "
+    "'mangguodiscover', 'migu', 'musicbrainz', 'tencentvideodiscover', "
+    "'theaudiodb', 'themoviedb', 'tvdb') AND "
+    "media_id IS NOT NULL AND trim(media_id) <> '' AND trim(media_id) <> '0')"
+)
+
+
+def _inspector() -> sa.Inspector:
+    """返回使用当前迁移连接的数据库检查器。"""
+    return sa.inspect(op.get_bind())
+
+
+def _replace_constraints(check_sql: str) -> None:
+    """在现有媒体表上以批处理方式替换统一身份约束。"""
+    table_names = set(_inspector().get_table_names())
+    for table_name in MEDIA_TABLES:
+        if table_name not in table_names:
+            continue
+        constraint_name = f"ck_{table_name}_media_identity"
+        existing = {
+            constraint.get("name")
+            for constraint in _inspector().get_check_constraints(table_name)
+        }
+        with op.batch_alter_table(table_name) as batch_op:
+            if constraint_name in existing:
+                batch_op.drop_constraint(constraint_name, type_="check")
+            batch_op.create_check_constraint(constraint_name, check_sql)
+
+
+def upgrade() -> None:
+    """把固定内置来源白名单替换为允许插件来源的格式约束。"""
+    _replace_constraints(EXTENSIBLE_IDENTITY_CHECK_SQL)
+
+
+def downgrade() -> None:
+    """恢复只允许当前内置来源的旧约束。"""
+    _replace_constraints(BUILTIN_IDENTITY_CHECK_SQL)

docs/cli.md

@@ -489,7 +489,7 @@ moviepilot tool run search_torrents media_type=movie media_source=themoviedb med
 - `tool list` 用于动态发现当前服务可调用的工具
 - `tool show` 会输出参数名、类型和描述
 - `tool run` 参数格式固定为 `key=value`
-- 涉及精确媒体身份的通用工具统一使用 `media_source` + `media_id`；`media_source` 必须是工具 Schema 列出的 `MediaSource` 枚举值，两个字段必须成对传递并复用搜索结果。TMDB 等单数据源专属工具按各自 Schema 保留原生 ID 参数
+- 涉及精确媒体身份的通用工具统一使用 `media_source` + `media_id`；内置来源使用工具 Schema 中的 `MediaSource` 常量，插件可以注册符合 Schema 格式的扩展来源，两个字段必须成对传递并复用搜索结果。TMDB 等单数据源专属工具按各自 Schema 保留原生 ID 参数
 - `read_file`、`write_file`、`edit_file` 和 `execute_command`
   属于内置 Agent 的本地敏感能力，不通过 MCP/`moviepilot tool` 暴露；插件开发时
   由 Agent 按当前用户权限直接调用这些工具。

docs/mcp-api.md

@@ -131,13 +131,13 @@ FastAPI 的 HTTP 异常和参数校验异常统一使用 `message`，不再返
 
 #### 媒体识别 / 整理
 
-媒体识别、搜索和手动整理统一使用 `media_source` + `media_id` 表示媒体主身份。`media_source` 必须是 `MediaSource` 枚举值：`themoviedb`、`douban`、`bangumi`、`anilist`、`imdb`、`tvdb`、`musicbrainz`、`theaudiodb`、`doubanmusic`、`bilibili`、`mangguodiscover`、`migu` 或 `tencentvideodiscover`；`media_id` 是该来源的原生 ID，不添加 `tmdb:` 等前缀。需要精确身份时两个字段必须同时提供，不能只传其中一个。
+媒体识别、搜索和手动整理统一使用 `media_source` + `media_id` 表示媒体主身份。内置来源通过 `MediaSource` 提供 `themoviedb`、`douban`、`bangumi`、`anilist`、`imdb`、`tvdb`、`musicbrainz`、`theaudiodb`、`doubanmusic`、`bilibili`、`mangguodiscover`、`migu` 和 `tencentvideodiscover` 等常量；该列表不是插件来源白名单，插件可以注册符合 OpenAPI 格式约束的稳定扩展标识。`media_id` 是该来源的原生 ID，不添加 `tmdb:` 等前缀。需要精确身份时两个字段必须同时提供，不能只传其中一个。
 
 影视自动识别在未指定来源时只使用 TMDB，未命中时不会继续查询其它影视源。音乐路径识别严格按 AcoustID 音频指纹、文件标签、文件名三级依次执行；指纹或标签直接提供 MusicBrainz Recording ID 时，会直接查询 MusicBrainz 详情，标签和文件名标题识别也只使用 MusicBrainz。其它元数据源仅在手动操作通过请求级 `media_source`，或通过完整的 `media_source` + `media_id` 精确指定时使用，不修改系统默认值，也不会跨来源兜底。`MediaInfo` 响应仍可能包含 `tmdb_id`、`douban_id`、`bangumi_id`、`anilist_id` 等跨源映射辅助字段，但这些字段不是通用请求入口。明确归属 `/tmdb`、`/douban`、`/bangumi`、`/anilist` 的接口，以及固定使用 TMDB 的剧集组和排期接口，仍可按其单数据源契约接收原生 ID。
 
 | 方法 | 路径 | 说明 |
 | :--- | :--- | :--- |
-| GET | `/api/v1/media/search` | 按标题搜索媒体、合集、人物或音乐，参数：`title`、`type`、`page`、`count`，可重复传入可选 `media_source`；不同搜索类型仅接受其支持的 `MediaSource` 枚举值，旧客户端的逗号格式仅在输入边界兼容 |
+| GET | `/api/v1/media/search` | 按标题搜索媒体、合集、人物或音乐，参数：`title`、`type`、`page`、`count`，可重复传入可选 `media_source`；内置模块只处理自身支持的来源，插件模块可以处理其注册的扩展来源，旧客户端的逗号格式仅在输入边界兼容 |
 | GET | `/api/v1/media/recognize` | 识别标题，参数：`title`、`subtitle`、`custom_words`，可选 `media_source`；当 `title` 为含目录的媒体文件路径时，会合并父目录中的名称、年份等信息 |
 | GET | `/api/v1/media/recognize_file` | 识别文件路径，参数：`path`，可选 `media_source` |
 | GET | `/api/v1/media/{media_id}` | 按原生 ID 查询媒体详情；必填参数：`media_source`、`type_name`，其中 `media_source` 与路径中的 `media_id` 组成统一媒体身份 |
@@ -296,7 +296,7 @@ TMDB 缓存查询响应的 `data` 包含 `count`、`recognized`、`unrecognized`
 其中 `read_file` 单次最多返回 50KB 文件内容；超出时会截断并提示 Agent 使用
 `start_line`、`end_line` 指定更小的行号范围继续读取。
 
-媒体相关 MCP 工具以 `MediaSource` 枚举 `media_source` + 来源原生 `media_id` 传递精确身份。`query_media_detail`、`search_torrents`、`query_library_exists` 必须提供完整字段对；`add_subscribe`、`transfer_file`、`scrape_metadata` 在显式指定身份时也必须成对提供。`search_media` 和 `recognize_media` 是按标题或路径发现身份的入口，其结果中的字段对可直接用于后续工具。音乐调用还使用 `media_type=music` 与 `music_type=recording|album|artist`；其中艺术家只允许搜索和详情浏览。工具响应中的专用 ID 仅是跨源映射辅助输出，不应再作为上述通用工具的输入。TMDB 专用的 `query_episode_schedule` 仍使用 `tmdb_id`，因为它直接调用单一 TMDB 剧集接口。
+媒体相关 MCP 工具以 `media_source` + 来源原生 `media_id` 传递精确身份；内置来源使用 `MediaSource` 常量，插件来源使用注册的稳定扩展标识。`query_media_detail`、`search_torrents`、`query_library_exists` 必须提供完整字段对；`add_subscribe`、`transfer_file`、`scrape_metadata` 在显式指定身份时也必须成对提供。`search_media` 和 `recognize_media` 是按标题或路径发现身份的入口，其结果中的字段对可直接用于后续工具。音乐调用还使用 `media_type=music` 与 `music_type=recording|album|artist`；其中艺术家只允许搜索和详情浏览。工具响应中的专用 ID 仅是跨源映射辅助输出，不应再作为上述通用工具的输入。TMDB 专用的 `query_episode_schedule` 仍使用 `tmdb_id`，因为它直接调用单一 TMDB 剧集接口。
 
 Agent 音乐流程与影视共用同一采集管线，但实体边界不同：单曲通过 `music_type=recording` 按一个文件处理；专辑通过 `music_type=album` 类似电视剧整季包，按一个目录/资源处理并校验总曲目数；艺术家不是采集目标。`add_subscribe` / `update_subscribe` 支持音乐音质筛选字段和 `best_version` 音质洗版；`query_subscribes` 会返回筛选条件及当前音质快照。`scrape_metadata(media_type="music")` 会按策略写音频标签、封面和歌词，并返回歌词新增、已存在、未匹配和失败数量。
 

docs/rules/03-commands.md

@@ -229,9 +229,10 @@ moviepilot scheduler run subscribe_refresh
 ```
 
 **Media identity rule:** Generic media tools use the complete `media_source` +
-`media_id` pair returned by media search. `media_source` must be a `MediaSource`
-enum value. A source-owned tool such as `query_episode_schedule` may retain its
-native ID parameter because its schema and implementation are single-source.
+`media_id` pair returned by media search. Built-in sources use `MediaSource`
+constants; plugins may register a schema-valid extension identifier. A
+source-owned tool such as `query_episode_schedule` may retain its native ID
+parameter because its schema and implementation are single-source.
 
 ---
 

tests/test_agent_music_tools.py

@@ -291,8 +291,8 @@ def test_scrape_album_uses_unified_entity_recognition(tmp_path):
     assert async_recognize.await_args.kwargs["music_type"] == "album"
 
 
-def test_scrape_metadata_rejects_unknown_media_source_before_file_access(tmp_path):
-    """Agent 直接调用工具时也必须拒绝固定枚举之外的媒体来源。"""
+def test_scrape_metadata_rejects_invalid_media_source_before_file_access(tmp_path):
+    """Agent 直接调用工具时也必须拒绝格式非法的媒体来源。"""
     audio_file = tmp_path / "unknown-source.flac"
     audio_file.write_bytes(b"audio")
     tool = ScrapeMetadataTool(session_id="session-1", user_id="10001")
@@ -300,7 +300,7 @@ def test_scrape_metadata_rejects_unknown_media_source_before_file_access(tmp_pat
     result = asyncio.run(tool.run(
         path=str(audio_file),
         media_type="music",
-        media_source="plugin-source",
+        media_source="plugin source:invalid",
         media_id="recording-1",
     ))
 

tests/test_bangumi_media_type.py

@@ -1,5 +1,6 @@
 import asyncio
 from typing import Optional
+from unittest.mock import AsyncMock, Mock, patch
 
 import pytest
 
@@ -97,8 +98,8 @@ def test_bangumi_movie_conversion_uses_movie_type() -> None:
     assert chain.douban_mtype == MediaType.MOVIE
 
 
-def test_media_identity_conversion_rejects_invalid_or_unsupported_pairs() -> None:
-    """跨源转换只接受完整非零 pair 和受支持的来源组合。"""
+def test_media_identity_conversion_rejects_invalid_pair_without_plugin_handler() -> None:
+    """跨源转换拒绝无效 pair，且没有插件处理器时返回空结果。"""
     chain = _SyncBangumiMediaChain()
 
     assert MediaChain.convert_media_identity(
@@ -107,16 +108,38 @@ def test_media_identity_conversion_rejects_invalid_or_unsupported_pairs() -> Non
         media_source=MediaSource.Bangumi,
         media_id="0",
     ) is None
-    assert MediaChain.convert_media_identity(
-        chain,
-        target_source=MediaSource.TheAudioDB,
-        media_source=MediaSource.Bangumi,
-        media_id="1",
-    ) is None
+    with patch("app.chain.media.eventmanager.send_event", return_value=None):
+        assert MediaChain.convert_media_identity(
+            chain,
+            target_source=MediaSource.TheAudioDB,
+            media_source=MediaSource.Bangumi,
+            media_id="1",
+        ) is None
     assert chain.tmdb_mtype is None
     assert chain.douban_mtype is None
 
 
+def test_media_identity_conversion_dispatches_plugin_source() -> None:
+    """内置转换无匹配时应把动态来源交给插件转换事件。"""
+    chain = _SyncBangumiMediaChain()
+    result = {"media_source": MediaSource.TMDB, "media_id": "550"}
+
+    def handle_event(_event_type, event_data):
+        """模拟插件在链式事件中写入转换结果。"""
+        event_data.media_dict.update(result)
+        return Mock(event_data=event_data)
+
+    with patch("app.chain.media.eventmanager.send_event", side_effect=handle_event):
+        converted = MediaChain.convert_media_identity(
+            chain,
+            target_source=MediaSource.TMDB,
+            media_source=MediaSource("acme.video"),
+            media_id="custom-1",
+        )
+
+    assert converted == result
+
+
 class _AsyncBangumiMediaChain:
     """异步Bangumi跨数据源转换测试桩。"""
 
@@ -176,3 +199,27 @@ def test_async_bangumi_movie_conversion_uses_movie_type() -> None:
     assert douban_info == {"id": "200"}
     assert chain.tmdb_mtype == MediaType.MOVIE
     assert chain.douban_mtype == MediaType.MOVIE
+
+
+def test_async_media_identity_conversion_dispatches_plugin_source() -> None:
+    """异步内置转换无匹配时也应分派插件转换事件。"""
+    chain = _AsyncBangumiMediaChain()
+    result = {"media_source": MediaSource.Douban, "media_id": "1295644"}
+
+    async def handle_event(_event_type, event_data):
+        """模拟异步插件在链式事件中写入转换结果。"""
+        event_data.media_dict.update(result)
+        return Mock(event_data=event_data)
+
+    with patch(
+        "app.chain.media.eventmanager.async_send_event",
+        new=AsyncMock(side_effect=handle_event),
+    ):
+        converted = asyncio.run(MediaChain.async_convert_media_identity(
+            chain,
+            target_source=MediaSource.Douban,
+            media_source=MediaSource("acme.video"),
+            media_id="custom-1",
+        ))
+
+    assert converted == result

tests/test_database_index_migration.py

@@ -143,8 +143,8 @@ with Engine.connect() as connection:
             "media_idisnull",
             "media_sourceisnotnull",
             "media_idisnotnull",
-            "'themoviedb'",
-            "'anilist'",
+            "length(media_source)",
+            "media_sourcenotlike'%:%'",
         ):
             assert fragment in normalized_sql, (
                 table_name,
@@ -174,12 +174,12 @@ with Engine.connect() as connection:
                     "INSERT INTO mediaserveritem (media_source, media_id) "
                     "VALUES (:media_source, :media_id)"
                 ),
-                {{"media_source": "invalid_source", "media_id": "1"}},
+                {{"media_source": "invalid:source", "media_id": "1"}},
             )
     except IntegrityError as error:
         assert constraint_name in str(error.orig), str(error.orig)
     else:
-        raise AssertionError("非法媒体身份未被具名检查约束拒绝")
+        raise AssertionError("格式非法的媒体身份未被具名检查约束拒绝")
 """.format(
     media_tables=MEDIA_TABLES,
     legacy_identity_columns=LEGACY_IDENTITY_COLUMNS,

tests/test_explicit_response_models.py

@@ -153,8 +153,15 @@ def test_discover_media_source_keeps_legacy_prefix_compatible():
         mediaid_prefix="mangguo",
         api_path="plugin/MangoTVDiscover/discover",
     )
+    plugin_source = DiscoverMediaSource(
+        name="Acme Video",
+        media_source=MediaSource("acme.video"),
+        api_path="plugin/AcmeVideo/discover",
+    )
 
     assert legacy.media_source is MediaSource.Bilibili
     assert legacy.model_dump(mode="json")["mediaid_prefix"] == "bilibili"
     assert current.mediaid_prefix == MediaSource.TencentVideo.value
     assert historical_alias.media_source is MediaSource.MangoTV
+    assert plugin_source.media_source == MediaSource("acme.video")
+    assert plugin_source.mediaid_prefix == "acme.video"

tests/test_extensible_media_source_migration.py

@@ -0,0 +1,66 @@
+"""插件扩展媒体来源数据库迁移测试。"""
+
+import importlib
+
+from alembic.migration import MigrationContext
+from alembic.operations import Operations
+import pytest
+import sqlalchemy as sa
+
+
+def _operations(connection: sa.Connection) -> Operations:
+    """为内存 SQLite 连接构造 Alembic 操作对象。"""
+    return Operations(MigrationContext.configure(connection))
+
+
+def _create_fixed_constraint_table(connection: sa.Connection) -> None:
+    """创建模拟已执行旧固定白名单 revision 的订阅表。"""
+    metadata = sa.MetaData()
+    sa.Table(
+        "subscribe",
+        metadata,
+        sa.Column("id", sa.Integer, primary_key=True),
+        sa.Column("media_source", sa.String),
+        sa.Column("media_id", sa.String),
+        sa.CheckConstraint(
+            "(media_source IS NULL AND media_id IS NULL) OR "
+            "(media_source IN ('themoviedb', 'douban') AND media_id IS NOT NULL "
+            "AND trim(media_id) <> '' AND trim(media_id) <> '0')",
+            name="ck_subscribe_media_identity",
+        ),
+    )
+    metadata.create_all(connection)
+
+
+def test_upgrade_replaces_fixed_source_whitelist(monkeypatch) -> None:
+    """升级后应保留原数据、允许插件来源并继续拒绝非法身份。"""
+    migration = importlib.import_module(
+        "database.versions.b3d7e9f1a2c4_3_0_0"
+    )
+    engine = sa.create_engine("sqlite://")
+
+    with engine.begin() as connection:
+        _create_fixed_constraint_table(connection)
+        connection.execute(sa.text(
+            "INSERT INTO subscribe (id, media_source, media_id) "
+            "VALUES (1, 'themoviedb', '550')"
+        ))
+        monkeypatch.setattr(migration, "op", _operations(connection))
+
+        migration.upgrade()
+
+        connection.execute(sa.text(
+            "INSERT INTO subscribe (id, media_source, media_id) "
+            "VALUES (2, 'acme.video', 'custom-1')"
+        ))
+        with pytest.raises(sa.exc.IntegrityError):
+            with connection.begin_nested():
+                connection.execute(sa.text(
+                    "INSERT INTO subscribe (id, media_source, media_id) "
+                    "VALUES (3, 'invalid:source', 'custom-2')"
+                ))
+        rows = connection.execute(sa.text(
+            "SELECT media_source, media_id FROM subscribe ORDER BY id"
+        )).all()
+
+    assert rows == [("themoviedb", "550"), ("acme.video", "custom-1")]

tests/test_media_identity_cleanup_migration.py

@@ -171,6 +171,14 @@ def test_cleanup_migration_keeps_one_complete_identity_and_drops_legacy_columns(
                 "doubanid": None,
                 "mediaid": None,
             },
+            {
+                "id": 7,
+                "media_source": None,
+                "media_id": None,
+                "tmdbid": None,
+                "doubanid": None,
+                "mediaid": "acme.video:custom-7",
+            },
         ])
         connection.execute(tables["subscribehistory"].insert(), {
             "id": 1, "mediaid": "bangumi:400602",
@@ -225,7 +233,7 @@ def test_cleanup_migration_keeps_one_complete_identity_and_drops_legacy_columns(
             "themoviedb", "1396",
         )
         assert (subscribe_rows[3]["media_source"], subscribe_rows[3]["media_id"]) == (
-            "douban", "35209731",
+            "plugin_source", "custom-1",
         )
         assert (subscribe_rows[4]["media_source"], subscribe_rows[4]["media_id"]) == (
             "douban", "1295644",
@@ -233,6 +241,9 @@ def test_cleanup_migration_keeps_one_complete_identity_and_drops_legacy_columns(
         assert (subscribe_rows[5]["media_source"], subscribe_rows[5]["media_id"]) == (
             "musicbrainz", "release-group-1",
         )
+        assert (subscribe_rows[6]["media_source"], subscribe_rows[6]["media_id"]) == (
+            "acme.video", "custom-7",
+        )
         assert identities["subscribehistory"]["media_source"] == "bangumi"
         assert identities["downloadhistory"]["media_source"] == "douban"
         assert identities["transferhistory"]["media_source"] == "anilist"
@@ -257,9 +268,9 @@ def test_cleanup_migration_backfills_every_supported_mediaid_prefix(monkeypatch)
                     "mediaid": f"{prefix}:native-{index}",
                 }
                 for index, (prefix, _) in enumerate(PREFIXED_IDENTITIES, start=1)
-            ] + [{
-                "id": len(PREFIXED_IDENTITIES) + 1,
-                "mediaid": "audioXdb:must-not-match-alias",
+        ] + [{
+            "id": len(PREFIXED_IDENTITIES) + 1,
+            "mediaid": "audioXdb:must-not-match-alias",
             }],
         )
 
@@ -283,11 +294,13 @@ def test_cleanup_migration_backfills_every_supported_mediaid_prefix(monkeypatch)
         (source, f"native-{index}")
         for index, (_, source) in enumerate(PREFIXED_IDENTITIES, start=1)
     ]
-    assert (rows[-1]["media_source"], rows[-1]["media_id"]) == (None, None)
+    assert (rows[-1]["media_source"], rows[-1]["media_id"]) == (
+        "audioxdb", "must-not-match-alias",
+    )
 
 
 def test_cleanup_migration_rejects_invalid_database_identity_pairs(monkeypatch) -> None:
-    """升级后的数据库应拒绝半对、未知来源和零值身份。"""
+    """升级后的数据库应允许插件来源，并拒绝半对、非法来源和零值身份。"""
     migration = importlib.import_module(
         "database.versions.8a4c7e1d2f90_3_0_0"
     )
@@ -303,7 +316,7 @@ def test_cleanup_migration_rejects_invalid_database_identity_pairs(monkeypatch)
         for identity in (
             {"media_source": "themoviedb", "media_id": None},
             {"media_source": None, "media_id": "550"},
-            {"media_source": "plugin_source", "media_id": "550"},
+            {"media_source": "invalid:source", "media_id": "550"},
             {"media_source": "themoviedb", "media_id": "0"},
         ):
             with pytest.raises(sa.exc.IntegrityError):
@@ -313,6 +326,12 @@ def test_cleanup_migration_rejects_invalid_database_identity_pairs(monkeypatch)
                         **identity,
                     })
 
+        connection.execute(subscribe.insert(), {
+            "name": "plugin",
+            "media_source": "plugin_source",
+            "media_id": "custom-1",
+        })
+
         connection.execute(subscribe.insert(), [
             {"name": "empty", "media_source": None, "media_id": None},
             {

tests/test_media_search_source_selection.py

@@ -46,7 +46,7 @@ def test_media_search_endpoint_forwards_source(
 
 
 def test_media_search_endpoint_forwards_multi_source() -> None:
-    """媒体搜索接口应将逗号分隔来源解析为固定枚举元组。"""
+    """媒体搜索接口应将逗号分隔来源解析为规范来源元组。"""
     chain = Mock()
     chain.async_search = AsyncMock(return_value=(Mock(), []))
 
@@ -69,7 +69,7 @@ def test_media_search_endpoint_forwards_multi_source() -> None:
 
 @pytest.mark.anyio
 async def test_media_search_route_accepts_comma_separated_music_sources() -> None:
-    """真实路由在旧逗号格式兼容边界后应把每项转换为固定枚举。"""
+    """真实路由在旧逗号格式兼容边界后应把每项转换为 MediaSource。"""
     chain = Mock()
     chain.async_search_music = AsyncMock(return_value=[])
     app = FastAPI()
@@ -166,8 +166,8 @@ async def test_media_search_route_deduplicates_repeated_sources() -> None:
 
 
 @pytest.mark.anyio
-async def test_media_search_route_rejects_unknown_source() -> None:
-    """真实搜索路由应在进入处理链前拒绝固定枚举之外的数据源。"""
+async def test_media_search_route_forwards_plugin_source() -> None:
+    """真实搜索路由应把插件扩展来源传入完整模块调度。"""
     chain = Mock()
     chain.async_search = AsyncMock(return_value=(Mock(), []))
     app = FastAPI()
@@ -184,11 +184,11 @@ async def test_media_search_route_rejects_unknown_source() -> None:
                 params={"title": "测试", "media_source": "plugin-source"},
             )
 
-    assert response.status_code == 422
-    detail = response.json()["detail"][0]
-    assert detail["type"] == "enum"
-    assert detail["input"] == "plugin-source"
-    chain.async_search.assert_not_awaited()
+    assert response.status_code == 200
+    chain.async_search.assert_awaited_once_with(
+        title="测试",
+        media_source=(MediaSource("plugin-source"),),
+    )
 
 
 @pytest.mark.parametrize(

tests/test_media_source_routing.py

@@ -22,11 +22,19 @@ def test_generic_source_id_resolves_as_fixed_enum() -> None:
     ) == (MediaSource.AniList, "154587")
 
 
-def test_unknown_plugin_source_is_rejected() -> None:
-    """固定枚举以外的来源不能进入通用识别链。"""
+def test_plugin_source_is_preserved_as_dynamic_enum() -> None:
+    """格式合法的插件来源应作为动态枚举成员进入通用识别链。"""
     assert resolve_media_identity(
         media_source="plugin_source",
         media_id="custom-1",
+    ) == (MediaSource("plugin_source"), "custom-1")
+
+
+def test_invalid_source_identifier_is_rejected() -> None:
+    """包含空格或分隔符的来源标识不得进入通用识别链。"""
+    assert resolve_media_identity(
+        media_source="Plugin Source:Invalid",
+        media_id="custom-1",
     ) == (None, None)
 
 

tests/test_message_media_identity.py

@@ -29,8 +29,8 @@ def test_message_context_contains_primary_and_auxiliary_media_fields() -> None:
     assert media.to_dict()["media_id"] == "170942"
 
 
-def test_media_info_rejects_unknown_plugin_source_identity() -> None:
-    """核心媒体对象应丢弃枚举以外的插件自定义来源。"""
+def test_media_info_preserves_plugin_source_identity() -> None:
+    """核心媒体对象应保留格式合法的插件自定义来源。"""
     media = MediaInfo(
         media_source="plugin_source",
         media_id="custom-100",
@@ -38,5 +38,6 @@ def test_media_info_rejects_unknown_plugin_source_identity() -> None:
         title="插件电影",
     )
 
-    assert media.media_source is None
-    assert media.media_id is None
+    assert media.media_source == MediaSource("plugin_source")
+    assert media.media_id == "custom-100"
+    assert media.to_dict()["media_source"] == "plugin_source"

tests/test_music_metadata_sources.py

@@ -509,6 +509,15 @@ async def test_media_chain_aggregates_music_sources_and_isolates_source_failure(
     ])
     source_chains = {
         "musicbrainz": musicbrainz,
+        "acme.music": Mock(
+            async_search_music=AsyncMock(return_value=[
+                MusicInfo(
+                    media_source="acme.music",
+                    media_id="plugin-1",
+                    title="Yellow",
+                )
+            ])
+        ),
         "theaudiodb": theaudiodb,
         "doubanmusic": douban,
     }
@@ -520,6 +529,7 @@ async def test_media_chain_aggregates_music_sources_and_isolates_source_failure(
         limit=30,
         media_source=(
             MediaSource.MusicBrainz,
+            MediaSource("acme.music"),
             MediaSource.TheAudioDB,
             MediaSource.DoubanMusic,
             MediaSource.MusicBrainz,
@@ -528,10 +538,12 @@ async def test_media_chain_aggregates_music_sources_and_isolates_source_failure(
 
     assert [(str(item.media_source), item.media_id) for item in results] == [
         ("musicbrainz", "recording-1"),
+        ("acme.music", "plugin-1"),
         ("doubanmusic", "album-1"),
     ]
     assert [str(item.args[0]) for item in select_chain.call_args_list] == [
         "musicbrainz",
+        "acme.music",
         "theaudiodb",
         "doubanmusic",
     ]

tests/test_music_plugin_recognize.py

@@ -405,8 +405,8 @@ def test_chain_async_supplement_media_recognize():
     assert result.media_id == "song-1"
 
 
-def test_chain_supplement_rejects_unknown_media_source():
-    """插件返回非固定枚举来源时不得进入统一识别链。"""
+def test_chain_supplement_accepts_plugin_media_source():
+    """插件返回的规范扩展来源应进入统一识别链。"""
     chain = ChainBase()
     event = Event(
         ChainEventType.MusicMediaRecognize,
@@ -429,4 +429,5 @@ def test_chain_supplement_rejects_unknown_media_source():
             mediainfo=None,
         )
 
-    assert result is None
+    assert result.media_source == MediaSource("qqmusic")
+    assert result.media_id == "song-1"

tests/test_search_media_sources.py

@@ -13,10 +13,11 @@ from app.schemas.types import MediaSource, MediaType
 from app.utils.media import normalize_media_source
 
 
-def test_media_source_normalization_rejects_unknown_source() -> None:
-    """固定枚举之外的来源不能进入统一身份链路。"""
-    assert normalize_media_source("plugin_source") is None
+def test_media_source_normalization_accepts_plugin_source() -> None:
+    """来源规范化应同时支持内置别名和插件扩展标识。"""
+    assert normalize_media_source(" Plugin_Source ") == MediaSource("plugin_source")
     assert normalize_media_source("tmdb") == MediaSource.TMDB
+    assert normalize_media_source("plugin source:invalid") is None
 
 
 def test_resolve_anilist_search_params_preserves_identity() -> None:

tests/test_subscribe_endpoint.py

@@ -1358,7 +1358,7 @@ def test_subscribe_preserves_explicit_zero_and_numeric_string_values():
         {"media_id": "123"},
         {"media_source": ""},
         {"media_id": ""},
-        {"media_source": "unknown", "media_id": "123"},
+        {"media_source": "invalid source:", "media_id": "123"},
         {"media_source": MediaSource.TMDB, "media_id": "0"},
         {"media_source": MediaSource.TMDB, "media_id": "   "},
     ],

tests/test_transfer_tmdb_category.py

@@ -7,8 +7,8 @@ from app.schemas import FileItem, TransferDirectoryConf, TransferTask
 from app.schemas.types import MediaSource, MediaType
 
 
-def test_transfer_rejects_partial_or_unknown_explicit_identity() -> None:
-    """整理公共入口不得把半套身份或未知来源传入后台任务。"""
+def test_transfer_rejects_partial_or_invalid_explicit_identity() -> None:
+    """整理公共入口不得把半套身份或格式非法来源传入后台任务。"""
     chain = object.__new__(TransferChain)
     fileitem = FileItem(
         storage="local",
@@ -20,16 +20,16 @@ def test_transfer_rejects_partial_or_unknown_explicit_identity() -> None:
         fileitem=fileitem,
         media_source=MediaSource.TMDB,
     )
-    unknown_state, unknown_message = chain.do_transfer(
+    invalid_state, invalid_message = chain.do_transfer(
         fileitem=fileitem,
-        media_source="plugin-source",
+        media_source="plugin source:invalid",
         media_id="1234",
     )
 
     assert not partial_state
     assert "media_source" in partial_message
-    assert not unknown_state
-    assert "media_source" in unknown_message
+    assert not invalid_state
+    assert "media_source" in invalid_message
 
 
 def test_transfer_resolves_complete_identity_before_building_tasks(monkeypatch) -> None: