nas/MoviePilot
1
0
Fork 0
mirror of https://github.com/jxxghp/MoviePilot.git synced 2026-11-03 22:04:57 +08:00
Code Issues Packages Projects Releases Wiki Activity

fix(tmdb): accept numeric TVDB identities without coercing invalid values

Browse Source
This commit is contained in:
jxxghp
2026-10-17 00:11:53 +08:00
parent 8f74a59fc2
commit aaf2ae30df
3 changed files with 72 additions and 23 deletions
Download Patch File Download Diff File Expand all files Collapse all files

52
app/modules/themoviedb/tmdbv3api/objs/find.py
View File

@@ -1,24 +1,28 @@
from ..tmdb import TMDb
from ..tmdb import TMDb, TMDbJson
class Find(TMDb):
"""通过外部来源身份查询关联的 TMDb 条目。"""
_urls = {
"find": "/find/%s"
}
def find(self, external_id, external_source):
def find(self, external_id: int | str, external_source: str) -> TMDbJson:
"""
The find method makes it easy to search for objects in our database by an external id. For example, an IMDB ID.
:param external_id: str
:param external_id: int | str
:param external_source str
:return:
"""
if isinstance(external_id, int):
external_id = str(external_id)
return self._request_obj(
self._urls["find"] % external_id.replace("/", "%2F"),
params="external_source=" + external_source
)
def find_by_imdb_id(self, imdb_id):
def find_by_imdb_id(self, imdb_id: str) -> TMDbJson:
"""
The find method makes it easy to search for objects in our database by an IMDB ID.
:param imdb_id: str
@@ -26,15 +30,15 @@ class Find(TMDb):
"""
return self.find(imdb_id, "imdb_id")
def find_by_tvdb_id(self, tvdb_id):
def find_by_tvdb_id(self, tvdb_id: int | str) -> TMDbJson:
"""
The find method makes it easy to search for objects in our database by a TVDB ID.
:param tvdb_id: int
:param tvdb_id: int | str
:return:
"""
return self.find(tvdb_id, "tvdb_id")
def find_by_freebase_mid(self, freebase_mid):
def find_by_freebase_mid(self, freebase_mid: str) -> TMDbJson:
"""
The find method makes it easy to search for objects in our database by a Freebase MID.
:param freebase_mid: str
@@ -42,7 +46,7 @@ class Find(TMDb):
"""
return self.find(freebase_mid, "freebase_mid")
def find_by_freebase_id(self, freebase_id):
def find_by_freebase_id(self, freebase_id: str) -> TMDbJson:
"""
The find method makes it easy to search for objects in our database by a Freebase ID.
:param freebase_id: str
@@ -50,7 +54,7 @@ class Find(TMDb):
"""
return self.find(freebase_id, "freebase_id")
def find_by_tvrage_id(self, tvrage_id):
def find_by_tvrage_id(self, tvrage_id: str) -> TMDbJson:
"""
The find method makes it easy to search for objects in our database by a TVRage ID.
:param tvrage_id: str
@@ -58,7 +62,7 @@ class Find(TMDb):
"""
return self.find(tvrage_id, "tvrage_id")
def find_by_facebook_id(self, facebook_id):
def find_by_facebook_id(self, facebook_id: str) -> TMDbJson:
"""
The find method makes it easy to search for objects in our database by a Facebook ID.
:param facebook_id: str
@@ -66,7 +70,7 @@ class Find(TMDb):
"""
return self.find(facebook_id, "facebook_id")
def find_by_instagram_id(self, instagram_id):
def find_by_instagram_id(self, instagram_id: str) -> TMDbJson:
"""
The find method makes it easy to search for objects in our database by a Instagram ID.
:param instagram_id: str
@@ -74,7 +78,7 @@ class Find(TMDb):
"""
return self.find(instagram_id, "instagram_id")
def find_by_twitter_id(self, twitter_id):
def find_by_twitter_id(self, twitter_id: str) -> TMDbJson:
"""
The find method makes it easy to search for objects in our database by a Twitter ID.
:param twitter_id: str
@@ -83,19 +87,21 @@ class Find(TMDb):
return self.find(twitter_id, "twitter_id")
# 异步版本方法
async def async_find(self, external_id, external_source):
async def async_find(self, external_id: int | str, external_source: str) -> TMDbJson:
"""
The find method makes it easy to search for objects in our database by an external id. For example, an IMDB ID.（异步版本）
:param external_id: str
:param external_id: int | str
:param external_source str
:return:
"""
if isinstance(external_id, int):
external_id = str(external_id)
return await self._async_request_obj(
self._urls["find"] % external_id.replace("/", "%2F"),
params="external_source=" + external_source
)
async def async_find_by_imdb_id(self, imdb_id):
async def async_find_by_imdb_id(self, imdb_id: str) -> TMDbJson:
"""
The find method makes it easy to search for objects in our database by an IMDB ID.（异步版本）
:param imdb_id: str
@@ -103,15 +109,15 @@ class Find(TMDb):
"""
return await self.async_find(imdb_id, "imdb_id")
async def async_find_by_tvdb_id(self, tvdb_id):
async def async_find_by_tvdb_id(self, tvdb_id: int | str) -> TMDbJson:
"""
The find method makes it easy to search for objects in our database by a TVDB ID.（异步版本）
:param tvdb_id: int
:param tvdb_id: int | str
:return:
"""
return await self.async_find(tvdb_id, "tvdb_id")
async def async_find_by_freebase_mid(self, freebase_mid):
async def async_find_by_freebase_mid(self, freebase_mid: str) -> TMDbJson:
"""
The find method makes it easy to search for objects in our database by a Freebase MID.（异步版本）
:param freebase_mid: str
@@ -119,7 +125,7 @@ class Find(TMDb):
"""
return await self.async_find(freebase_mid, "freebase_mid")
async def async_find_by_freebase_id(self, freebase_id):
async def async_find_by_freebase_id(self, freebase_id: str) -> TMDbJson:
"""
The find method makes it easy to search for objects in our database by a Freebase ID.（异步版本）
:param freebase_id: str
@@ -127,7 +133,7 @@ class Find(TMDb):
"""
return await self.async_find(freebase_id, "freebase_id")
async def async_find_by_tvrage_id(self, tvrage_id):
async def async_find_by_tvrage_id(self, tvrage_id: str) -> TMDbJson:
"""
The find method makes it easy to search for objects in our database by a TVRage ID.（异步版本）
:param tvrage_id: str
@@ -135,7 +141,7 @@ class Find(TMDb):
"""
return await self.async_find(tvrage_id, "tvrage_id")
async def async_find_by_facebook_id(self, facebook_id):
async def async_find_by_facebook_id(self, facebook_id: str) -> TMDbJson:
"""
The find method makes it easy to search for objects in our database by a Facebook ID.（异步版本）
:param facebook_id: str
@@ -143,7 +149,7 @@ class Find(TMDb):
"""
return await self.async_find(facebook_id, "facebook_id")
async def async_find_by_instagram_id(self, instagram_id):
async def async_find_by_instagram_id(self, instagram_id: str) -> TMDbJson:
"""
The find method makes it easy to search for objects in our database by a Instagram ID.（异步版本）
:param instagram_id: str
@@ -151,7 +157,7 @@ class Find(TMDb):
"""
return await self.async_find(instagram_id, "instagram_id")
async def async_find_by_twitter_id(self, twitter_id):
async def async_find_by_twitter_id(self, twitter_id: str) -> TMDbJson:
"""
The find method makes it easy to search for objects in our database by a Twitter ID.（异步版本）
:param twitter_id: str

1
mypy.ini
View File

@@ -186,4 +186,5 @@ files =
app/modules/themoviedb/tmdbv3api/objs/keyword.py,
app/modules/themoviedb/tmdbv3api/objs/review.py,
app/modules/themoviedb/tmdbv3api/objs/change.py,
app/modules/themoviedb/tmdbv3api/objs/find.py,
scripts/architecture/async_blocking.py

42
tests/test_tmdb_find_identity.py Normal file
View File

@@ -0,0 +1,42 @@
"""外部身份查询接受整数 TVDB ID，并保持字符串身份的路径编码。"""
import asyncio
from unittest.mock import AsyncMock, Mock
import pytest
from app.modules.themoviedb.tmdbv3api.objs.find import Find
@pytest.mark.parametrize("asynchronous", [False, True])
@pytest.mark.parametrize("method,identity,source,path", [
("find_by_tvdb_id", 353712, "tvdb_id", "353712"),
("find_by_tvdb_id", "353712", "tvdb_id", "353712"),
("find_by_imdb_id", "tt1234567", "imdb_id", "tt1234567"),
("find_by_freebase_mid", "/m/abc", "freebase_mid", "%2Fm%2Fabc"),
])
def test_find_preserves_external_identity_encoding(monkeypatch, asynchronous, method, identity, source, path):
"""整数查询不再调用 int.replace，字符串和斜杠身份仍发到同一个 TMDb 端点。"""
payload = {"movie_results": [], "tv_results": [{"id": 987}]}
request = (AsyncMock if asynchronous else Mock)(return_value=payload)
monkeypatch.setattr(Find, "_async_request_obj" if asynchronous else "_request_obj", request)
client = object.__new__(Find)
operation = getattr(client, f"async_{method}" if asynchronous else method)
result = asyncio.run(operation(identity)) if asynchronous else operation(identity)
assert result is payload
request.assert_called_once_with(f"/find/{path}", params=f"external_source={source}")
@pytest.mark.parametrize("asynchronous", [False, True])
@pytest.mark.parametrize("identity", [None, 1.5, object()])
def test_find_does_not_coerce_unsupported_identity(monkeypatch, asynchronous, identity):
"""只有整数 ID 新增支持；无效对象仍在请求前失败，不能被转换为路径文本。"""
request = AsyncMock() if asynchronous else Mock()
monkeypatch.setattr(Find, "_async_request_obj" if asynchronous else "_request_obj", request)
client = object.__new__(Find)
with pytest.raises(AttributeError):
if asynchronous:
asyncio.run(client.async_find_by_tvdb_id(identity))
else:
client.find_by_tvdb_id(identity)
request.assert_not_called()