diff --git a/app/api/endpoints/media.py b/app/api/endpoints/media.py index c816cb25..08d56b8e 100644 --- a/app/api/endpoints/media.py +++ b/app/api/endpoints/media.py @@ -9,7 +9,7 @@ from app.chain.tmdb import TmdbChain from app.core.config import settings from app.core.context import Context from app.core.event import eventmanager -from app.core.metainfo import MetaInfo +from app.core.metainfo import MetaInfo, MetaInfoPath from app.core.security import verify_token, verify_apitoken from app.db.models import User from app.db.user_oper import get_current_active_user, get_current_active_superuser @@ -154,26 +154,64 @@ async def search( def scrape( fileitem: schemas.FileItem, storage: Optional[str] = "local", + media_source: Optional[MediaSource] = None, + media_id: Optional[str] = None, + type_name: Optional[MediaType] = None, _: schemas.TokenPayload = Depends(verify_token), ) -> Any: """ - 刮削媒体信息 + 刮削媒体信息，可按请求指定媒体数据源及其原生ID + + :param fileitem: 待刮削文件项 + :param storage: 文件所在存储 + :param media_source: 请求级媒体数据源 + :param media_id: 数据源原生ID + :param type_name: 媒体类型 + :param _: Token校验 """ if not fileitem or not fileitem.path: return schemas.Response(success=False, message="刮削路径无效") + normalized_media_id = media_id.strip() if media_id else None + if normalized_media_id and not media_source: + return schemas.Response( + success=False, message="指定媒体ID时必须同时指定媒体数据源" + ) + if normalized_media_id and not normalized_media_id.isdigit(): + return schemas.Response(success=False, message="媒体ID格式无效") + chain = MediaChain() - # 识别媒体信息 - context = chain.recognize_by_path(fileitem.path, obtain_images=True) - if not context or not context.media_info: + if normalized_media_id: + meta_info = MetaInfoPath(Path(fileitem.path)) + media_info = chain.recognize_media( + meta=meta_info, + mtype=type_name, + source=media_source, + mediaid=normalized_media_id, + ) + if media_info: + media_info.scrape_source = media_source + chain.obtain_images(mediainfo=media_info) + else: + context = chain.recognize_by_path( + fileitem.path, + source=media_source, + obtain_images=True, + ) + meta_info = context.meta_info if context else None + media_info = context.media_info if context else None + + if not media_info: return schemas.Response(success=False, message="刮削失败，无法识别媒体信息") + if media_source: + media_info.scrape_source = media_source if storage == "local": if not Path(fileitem.path).exists(): return schemas.Response(success=False, message="刮削路径不存在") # 手动刮削 (暂时使用同步版本，可以后续优化为异步) chain.scrape_metadata( fileitem=fileitem, - meta=context.meta_info, - mediainfo=context.media_info, + meta=meta_info, + mediainfo=media_info, overwrite=True, ) return schemas.Response(success=True, message=f"{fileitem.path} 刮削完成") diff --git a/docs/mcp-api.md b/docs/mcp-api.md index e3384a9f..fda38f01 100644 --- a/docs/mcp-api.md +++ b/docs/mcp-api.md @@ -120,6 +120,7 @@ FastAPI 异常响应保留 `detail` 字段，并在错误详情为文本时返 | GET | `/api/v1/media/recognize` | 识别标题，参数：`title`、`subtitle`、`custom_words`，可选 `source` | | GET | `/api/v1/media/recognize_file` | 识别文件路径，参数：`path`，可选 `source` | | GET | `/api/v1/media/{mediaid}` | 查询媒体详情，`mediaid` 支持 `tmdb:`、`douban:`、`bangumi:`、`anilist:` 前缀 | +| POST | `/api/v1/media/scrape/{storage}` | 刮削媒体元数据；请求体为 `FileItem`，可选查询参数 `media_source`、`media_id`、`type_name`（电影/电视剧）可指定本次刮削媒体 | | POST | `/api/v1/transfer/manual/target-path` | 匹配手动整理目标路径；请求体可用 `media_source` + `media_id` 指定数据源原生ID | | POST | `/api/v1/transfer/manual` | 手动整理；请求体可用 `media_source` + `media_id` 指定本次识别与刮削数据源，同时兼容 `tmdbid`、`doubanid` | diff --git a/skills/moviepilot-api/SKILL.md b/skills/moviepilot-api/SKILL.md index 3ac4ea8c..ae3034a7 100644 --- a/skills/moviepilot-api/SKILL.md +++ b/skills/moviepilot-api/SKILL.md @@ -96,7 +96,7 @@ All endpoints are under the base URL `{MP_HOST}`. Path parameters are shown as ` | GET | `/api/v1/media/recognize2` | Recognize media (API_TOKEN auth, use `--token-param`). Params: `title`, `subtitle`, `custom_words`, optional `source` | | GET | `/api/v1/media/recognize_file` | Recognize media from file path. Params: `path` (required), optional `source` | | GET | `/api/v1/media/recognize_file2` | Recognize file (API_TOKEN auth). Params: `path`, optional `source` | -| POST | `/api/v1/media/scrape/{storage}` | Scrape media metadata. Body: FileItem JSON | +| POST | `/api/v1/media/scrape/{storage}` | Scrape media metadata. Body: FileItem JSON. Optional params: `media_source`, `media_id`, `type_name` (`电影`/`电视剧`) | | GET | `/api/v1/media/category/config` | Get category strategy config | | POST | `/api/v1/media/category/config` | Save category strategy config. Body: CategoryConfig | | GET | `/api/v1/media/category` | Get auto-categorization config | diff --git a/tests/test_media_scrape_endpoint.py b/tests/test_media_scrape_endpoint.py new file mode 100644 index 00000000..9c8fa1e5 --- /dev/null +++ b/tests/test_media_scrape_endpoint.py @@ -0,0 +1,82 @@ +from unittest.mock import Mock, patch + +from app.api.endpoints.media import scrape +from app.core.context import Context, MediaInfo +from app.core.meta import MetaBase +from app.schemas import FileItem, MediaType + + +def test_scrape_uses_explicit_media_source_and_id() -> None: + """手动刮削应使用请求指定的数据源原生ID，并传给后续刮削流程。""" + fileitem = FileItem(storage="alist", path="/movies/Test Movie (2026).mkv", type="file") + media_info = MediaInfo(title="测试电影", type=MediaType.MOVIE) + chain = Mock() + chain.recognize_media.return_value = media_info + + with patch("app.api.endpoints.media.MediaChain", return_value=chain): + result = scrape( + fileitem=fileitem, + storage="alist", + media_source="douban", + media_id="123456", + type_name=MediaType.MOVIE, + _=Mock(), + ) + + assert result.success is True + chain.recognize_by_path.assert_not_called() + recognize_kwargs = chain.recognize_media.call_args.kwargs + assert recognize_kwargs["source"] == "douban" + assert recognize_kwargs["mediaid"] == "123456" + assert recognize_kwargs["mtype"] == MediaType.MOVIE + chain.obtain_images.assert_called_once_with(mediainfo=media_info) + assert media_info.scrape_source == "douban" + scrape_kwargs = chain.scrape_metadata.call_args.kwargs + assert scrape_kwargs["fileitem"] is fileitem + assert scrape_kwargs["mediainfo"] is media_info + assert scrape_kwargs["overwrite"] is True + + +def test_scrape_keeps_automatic_recognition_compatible() -> None: + """未指定媒体ID时应继续按路径识别，并允许仅限定请求级数据源。""" + fileitem = FileItem(storage="alist", path="/tv/Test Show S01E01.mkv", type="file") + meta_info = MetaBase("Test Show S01E01") + media_info = MediaInfo(title="测试剧集", type=MediaType.TV) + chain = Mock() + chain.recognize_by_path.return_value = Context(meta_info=meta_info, media_info=media_info) + + with patch("app.api.endpoints.media.MediaChain", return_value=chain): + result = scrape( + fileitem=fileitem, + storage="alist", + media_source="bangumi", + _=Mock(), + ) + + assert result.success is True + chain.recognize_by_path.assert_called_once_with( + fileitem.path, + source="bangumi", + obtain_images=True, + ) + chain.recognize_media.assert_not_called() + assert media_info.scrape_source == "bangumi" + chain.scrape_metadata.assert_called_once_with( + fileitem=fileitem, + meta=meta_info, + mediainfo=media_info, + overwrite=True, + ) + + +def test_scrape_rejects_media_id_without_source() -> None: + """原生媒体ID缺少所属数据源时应直接返回明确错误。""" + result = scrape( + fileitem=FileItem(storage="alist", path="/movies/Test.mkv", type="file"), + storage="alist", + media_id="123456", + _=Mock(), + ) + + assert result.success is False + assert result.message == "指定媒体ID时必须同时指定媒体数据源"