nas/MoviePilot
1
0
Fork 0
mirror of https://github.com/jxxghp/MoviePilot.git synced 2026-10-31 04:14:38 +08:00
Code Issues Packages Projects Releases Wiki Activity

fix(tmdb): read list creation results from JSON dictionaries

Browse Source
This commit is contained in:
jxxghp
2026-10-16 23:48:29 +08:00
parent 907276ca2f
commit dafc0c1fb0
3 changed files with 89 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files

56
app/modules/themoviedb/tmdbv3api/objs/list.py
View File

@@ -1,7 +1,10 @@
from ..tmdb import TMDb
from ..exceptions import TMDbException
from ..tmdb import TMDb, TMDbJson
class List(TMDb):
"""提供同步与异步的 TMDb 片单查询和维护接口。"""
_urls = {
"details": "/list/%s",
"check_status": "/list/%s/item_status",
@@ -12,7 +15,7 @@ class List(TMDb):
"delete_list": "/list/%s",
}
def details(self, list_id):
def details(self, list_id: int | str) -> TMDbJson:
"""
Get list details by id.
:param list_id: int
@@ -20,23 +23,24 @@ class List(TMDb):
"""
return self._request_obj(self._urls["details"] % list_id, key="items")
def check_item_status(self, list_id, movie_id):
def check_item_status(self, list_id: int | str, movie_id: int | str) -> TMDbJson:
"""
You can use this method to check if a movie has already been added to the list.
:param list_id: int
:param movie_id: int
:return:
"""
return self._request_obj(self._urls["check_status"] % list_id, params="movie_id=%s" % movie_id)["item_present"]
result = self._request_obj(self._urls["check_status"] % list_id, params="movie_id=%s" % movie_id)
return self._required_field(result, "item_present")
def create_list(self, name, description):
def create_list(self, name: str, description: str) -> TMDbJson:
"""
You can use this method to check if a movie has already been added to the list.
Create a list and return its identifier.
:param name: str
:param description: str
:return:
"""
return self._request_obj(
result = self._request_obj(
self._urls["create"],
params="session_id=%s" % self.session_id,
method="POST",
@@ -45,9 +49,10 @@ class List(TMDb):
"description": description,
"language": self.language
}
).list_id
)
return self._required_field(result, "list_id")
def add_movie(self, list_id, movie_id):
def add_movie(self, list_id: int | str, movie_id: int | str) -> None:
"""
Add a movie to a list.
:param list_id: int
@@ -60,7 +65,7 @@ class List(TMDb):
json={"media_id": movie_id}
)
def remove_movie(self, list_id, movie_id):
def remove_movie(self, list_id: int | str, movie_id: int | str) -> None:
"""
Remove a movie from a list.
:param list_id: int
@@ -73,7 +78,7 @@ class List(TMDb):
json={"media_id": movie_id}
)
def clear_list(self, list_id):
def clear_list(self, list_id: int | str) -> None:
"""
Clear all of the items from a list.
:param list_id: int
@@ -84,7 +89,7 @@ class List(TMDb):
method="POST"
)
def delete_list(self, list_id):
def delete_list(self, list_id: int | str) -> None:
"""
Delete a list.
:param list_id: int
@@ -96,7 +101,7 @@ class List(TMDb):
)
# 异步版本方法
async def async_details(self, list_id):
async def async_details(self, list_id: int | str) -> TMDbJson:
"""
Get list details by id.（异步版本）
:param list_id: int
@@ -104,7 +109,7 @@ class List(TMDb):
"""
return await self._async_request_obj(self._urls["details"] % list_id, key="items")
async def async_check_item_status(self, list_id, movie_id):
async def async_check_item_status(self, list_id: int | str, movie_id: int | str) -> TMDbJson:
"""
You can use this method to check if a movie has already been added to the list.（异步版本）
:param list_id: int
@@ -112,11 +117,11 @@ class List(TMDb):
:return:
"""
result = await self._async_request_obj(self._urls["check_status"] % list_id, params="movie_id=%s" % movie_id)
return result["item_present"]
return self._required_field(result, "item_present")
async def async_create_list(self, name, description):
async def async_create_list(self, name: str, description: str) -> TMDbJson:
"""
You can use this method to check if a movie has already been added to the list.（异步版本）
Create a list and return its identifier.（异步版本）
:param name: str
:param description: str
:return:
@@ -131,9 +136,9 @@ class List(TMDb):
"language": self.language
}
)
return result.list_id
return self._required_field(result, "list_id")
async def async_add_movie(self, list_id, movie_id):
async def async_add_movie(self, list_id: int | str, movie_id: int | str) -> None:
"""
Add a movie to a list.（异步版本）
:param list_id: int
@@ -146,7 +151,7 @@ class List(TMDb):
json={"media_id": movie_id}
)
async def async_remove_movie(self, list_id, movie_id):
async def async_remove_movie(self, list_id: int | str, movie_id: int | str) -> None:
"""
Remove a movie from a list.（异步版本）
:param list_id: int
@@ -159,7 +164,7 @@ class List(TMDb):
json={"media_id": movie_id}
)
async def async_clear_list(self, list_id):
async def async_clear_list(self, list_id: int | str) -> None:
"""
Clear all of the items from a list.（异步版本）
:param list_id: int
@@ -170,7 +175,7 @@ class List(TMDb):
method="POST"
)
async def async_delete_list(self, list_id):
async def async_delete_list(self, list_id: int | str) -> None:
"""
Delete a list.（异步版本）
:param list_id: int
@@ -180,3 +185,10 @@ class List(TMDb):
params="session_id=%s" % self.session_id,
method="DELETE"
)
@staticmethod
def _required_field(response: TMDbJson, field: str) -> TMDbJson:
"""读取响应必需字段，保留字段缺失时的 KeyError。"""
if not isinstance(response, dict):
raise TMDbException("TheMovieDb 返回数据格式异常：期望JSON对象")
return response[field]

1
mypy.ini
View File

@@ -174,4 +174,5 @@ files =
app/api/dependencies/subscription.py,
app/modules/themoviedb/tmdbv3api/tmdb.py,
app/modules/themoviedb/tmdbv3api/objs/auth.py,
app/modules/themoviedb/tmdbv3api/objs/list.py,
scripts/architecture/async_blocking.py

54
tests/test_tmdb_list_response_contract.py Normal file
View File

@@ -0,0 +1,54 @@
"""片单接口必须从 JSON 字典读取创建结果，并保留双入口一致性。"""
import asyncio
from unittest.mock import AsyncMock, Mock, call
import pytest
from app.modules.themoviedb.tmdbv3api.exceptions import TMDbException
from app.modules.themoviedb.tmdbv3api.objs.list import List
@pytest.mark.parametrize("asynchronous", [False, True])
def test_list_creation_reads_id_and_preserves_false_item_status(monkeypatch, asynchronous):
"""创建片单读取 list_id；电影尚未加入时原样返回 False。"""
request = (AsyncMock if asynchronous else Mock)(side_effect=[
{"success": True, "list_id": 735}, {"item_present": False},
])
monkeypatch.setattr(List, "_async_request_obj" if asynchronous else "_request_obj", request)
client = List(language="zh-CN")
client.session_id = "session"
try:
if asynchronous:
created = asyncio.run(client.async_create_list("收藏", "说明"))
present = asyncio.run(client.async_check_item_status(735, 12))
else:
created = client.create_list("收藏", "说明")
present = client.check_item_status(735, 12)
assert created == 735
assert present is False
assert request.call_args_list == [
call("/list", params="session_id=session", method="POST", json={
"name": "收藏", "description": "说明", "language": "zh-CN",
}),
call("/list/735/item_status", params="movie_id=12"),
]
finally:
client.close()
@pytest.mark.parametrize("asynchronous", [False, True])
@pytest.mark.parametrize("payload,error", [({}, KeyError), ([], TMDbException), (None, TMDbException)])
def test_list_status_rejects_missing_required_data(monkeypatch, asynchronous, payload, error):
"""缺失状态字段仍抛 KeyError，非对象响应不能冒充有效片单状态。"""
request = (AsyncMock if asynchronous else Mock)(return_value=payload)
monkeypatch.setattr(List, "_async_request_obj" if asynchronous else "_request_obj", request)
client = List()
try:
with pytest.raises(error):
if asynchronous:
asyncio.run(client.async_check_item_status(735, 12))
else:
client.check_item_status(735, 12)
finally:
client.close()