fix(discover): 统一 V3 发现推荐分页

app/api/endpoints/douban.py

@@ -1,4 +1,4 @@
-from typing import Any, List, Optional
+from typing import Any, List, Optional, Sequence
 
 from fastapi import Depends
 
@@ -14,6 +14,14 @@ from app.schemas.workflow import MediaInfo as _SchemaMediaInfo
 router = ResponseAPIRouter()
 
 
+def _paginate_medias(medias: Sequence[Any], page: int, count: int) -> list[Any]:
+    """按 browse 端点统一的页码协议切片媒体结果。"""
+    normalized_page = max(page, 1)
+    normalized_count = max(count, 1)
+    start = (normalized_page - 1) * normalized_count
+    return list(medias[start : start + normalized_count])
+
+
 @router.get(
     "/person/{person_id}", summary="人物详情", response_model=_SchemaMediaPerson
 )
@@ -70,7 +78,11 @@ async def douban_credits(
     response_model=List[_SchemaMediaInfo],
 )
 async def douban_recommend(
-    doubanid: str, type_name: str, _: _SchemaTokenPayload = Depends(verify_token)
+    doubanid: str,
+    type_name: str,
+    page: int = 1,
+    count: int = 20,
+    _: _SchemaTokenPayload = Depends(verify_token),
 ) -> Any:
     """
     根据豆瓣ID查询推荐电影/电视剧，type_name: 电影/电视剧
@@ -83,7 +95,7 @@ async def douban_recommend(
     else:
         return []
     if medias:
-        return [media.to_dict() for media in medias]
+        return [media.to_dict() for media in _paginate_medias(medias, page, count)]
     return []
 
 

app/api/endpoints/tmdb.py

@@ -1,4 +1,4 @@
-from typing import Any, List, Optional
+from typing import Any, List, Optional, Sequence
 
 from fastapi import Depends
 
@@ -19,6 +19,14 @@ from app.schemas.workflow import MediaInfo as _SchemaMediaInfo
 router = ResponseAPIRouter()
 
 
+def _paginate_medias(medias: Sequence[Any], page: int, count: int) -> list[Any]:
+    """按 browse 端点统一的页码协议切片媒体结果。"""
+    normalized_page = max(page, 1)
+    normalized_count = max(count, 1)
+    start = (normalized_page - 1) * normalized_count
+    return list(medias[start : start + normalized_count])
+
+
 @router.get(
     "/cache",
     summary="查询 TheMovieDb 识别缓存",
@@ -93,7 +101,11 @@ async def tmdb_seasons(
     response_model=List[_SchemaMediaInfo],
 )
 async def tmdb_similar(
-    tmdbid: int, type_name: str, _: _SchemaTokenPayload = Depends(verify_token)
+    tmdbid: int,
+    type_name: str,
+    page: int = 1,
+    count: int = 20,
+    _: _SchemaTokenPayload = Depends(verify_token),
 ) -> Any:
     """
     根据TMDBID查询类似电影/电视剧，type_name: 电影/电视剧
@@ -106,7 +118,7 @@ async def tmdb_similar(
     else:
         return []
     if medias:
-        return [media.to_dict() for media in medias]
+        return [media.to_dict() for media in _paginate_medias(medias, page, count)]
     return []
 
 
@@ -116,7 +128,11 @@ async def tmdb_similar(
     response_model=List[_SchemaMediaInfo],
 )
 async def tmdb_recommend(
-    tmdbid: int, type_name: str, _: _SchemaTokenPayload = Depends(verify_token)
+    tmdbid: int,
+    type_name: str,
+    page: int = 1,
+    count: int = 20,
+    _: _SchemaTokenPayload = Depends(verify_token),
 ) -> Any:
     """
     根据TMDBID查询推荐电影/电视剧，type_name: 电影/电视剧
@@ -129,7 +145,7 @@ async def tmdb_recommend(
     else:
         return []
     if medias:
-        return [media.to_dict() for media in medias]
+        return [media.to_dict() for media in _paginate_medias(medias, page, count)]
     return []
 
 

tests/fixtures/architecture/mypy-baseline.json

@@ -827,7 +827,7 @@
   "app/api/endpoints/douban.py": {
     "attr-defined": 1,
     "misc": 5,
-    "no-untyped-call": 3
+    "no-untyped-call": 2
   },
   "app/api/endpoints/download.py": {
     "arg-type": 7,
@@ -969,7 +969,7 @@
   "app/api/endpoints/tmdb.py": {
     "attr-defined": 1,
     "misc": 11,
-    "no-untyped-call": 4,
+    "no-untyped-call": 2,
     "operator": 4,
     "type-arg": 3
   },

tests/test_discover_pagination.py

@@ -0,0 +1,61 @@
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, patch
+
+import pytest
+
+from app.api.endpoints.douban import douban_recommend
+from app.api.endpoints.tmdb import tmdb_recommend, tmdb_similar
+
+
+def _media(index: int) -> SimpleNamespace:
+    """构造只满足发现端点序列化边界的媒体替身。"""
+    return SimpleNamespace(to_dict=lambda: {"title": f"媒体 {index}"})
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    ("endpoint", "chain_name", "method_name", "identifier", "type_name"),
+    [
+        (tmdb_similar, "TmdbChain", "async_movie_similar", 100, "电影"),
+        (tmdb_recommend, "TmdbChain", "async_movie_recommend", 100, "电影"),
+    ],
+)
+async def test_tmdb_browse_recommendations_apply_page_and_count(
+    endpoint, chain_name, method_name, identifier, type_name
+):
+    """TMDB browse 推荐和类似列表必须返回请求页，不重复发送整组结果。"""
+    medias = [_media(index) for index in range(1, 6)]
+    with patch(f"app.api.endpoints.tmdb.{chain_name}") as chain_cls:
+        chain_method = AsyncMock(return_value=medias)
+        setattr(chain_cls.return_value, method_name, chain_method)
+
+        result = await endpoint(
+            tmdbid=identifier,
+            type_name=type_name,
+            page=2,
+            count=2,
+            _=None,
+        )
+
+    assert result == [{"title": "媒体 3"}, {"title": "媒体 4"}]
+    chain_method.assert_awaited_once_with(tmdbid=identifier)
+
+
+@pytest.mark.asyncio
+async def test_douban_browse_recommendations_return_empty_terminal_page():
+    """豆瓣推荐超出结果范围时返回空页，供前端无限列表结束加载。"""
+    medias = [_media(index) for index in range(1, 4)]
+    with patch("app.api.endpoints.douban.DoubanChain") as chain_cls:
+        chain_method = AsyncMock(return_value=medias)
+        chain_cls.return_value.async_movie_recommend = chain_method
+
+        result = await douban_recommend(
+            doubanid="db-1",
+            type_name="电影",
+            page=2,
+            count=3,
+            _=None,
+        )
+
+    assert result == []
+    chain_method.assert_awaited_once_with(doubanid="db-1")