From aaf2ae30df75b4215fd1de5c5ddb688d3a7364d9 Mon Sep 17 00:00:00 2001 From: jxxghp Date: Sat, 17 Oct 2026 00:11:53 +0800 Subject: [PATCH] fix(tmdb): accept numeric TVDB identities without coercing invalid values --- app/modules/themoviedb/tmdbv3api/objs/find.py | 52 +++++++++++-------- mypy.ini | 1 + tests/test_tmdb_find_identity.py | 42 +++++++++++++++ 3 files changed, 72 insertions(+), 23 deletions(-) create mode 100644 tests/test_tmdb_find_identity.py diff --git a/app/modules/themoviedb/tmdbv3api/objs/find.py b/app/modules/themoviedb/tmdbv3api/objs/find.py index 1df4dea51..4ef404c46 100644 --- a/app/modules/themoviedb/tmdbv3api/objs/find.py +++ b/app/modules/themoviedb/tmdbv3api/objs/find.py @@ -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 diff --git a/mypy.ini b/mypy.ini index b17a9d394..7f233ed0c 100644 --- a/mypy.ini +++ b/mypy.ini @@ -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 diff --git a/tests/test_tmdb_find_identity.py b/tests/test_tmdb_find_identity.py new file mode 100644 index 000000000..88688aa4d --- /dev/null +++ b/tests/test_tmdb_find_identity.py @@ -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()