fix: handle tmdb unicode decode errors

This commit is contained in:
jxxghp
2026-05-20 09:21:36 +08:00
parent 07f51c5d94
commit 6edb627145
2 changed files with 37 additions and 2 deletions

View File

@@ -158,11 +158,13 @@ class TMDb(object):
@classmethod
def _decode_response_json(cls, response):
"""
解析TMDB响应JSON并把空响应代理错误页统一转换为TMDB异常。
解析TMDB响应JSON并把空响应代理错误页或错误编码的响应统一转换为TMDB异常。
"""
try:
return response.json()
except ValueError as err:
except (ValueError, UnicodeDecodeError) as err:
# httpx.Response.json() 在响应体是压缩字节或错误编码时会直接抛 UnicodeDecodeError
# 这里统一收敛成 TMDbException避免上层把脏响应当作未捕获异常。
raise TMDbException(cls._build_invalid_json_message(response)) from err
@staticmethod
@@ -192,6 +194,9 @@ class TMDb(object):
message_parts.append(f"HTTP状态码{status_code}")
if content_type:
message_parts.append(f"Content-Type{content_type}")
content_encoding = headers.get("content-encoding") or headers.get("Content-Encoding")
if content_encoding:
message_parts.append(f"Content-Encoding{content_encoding}")
if response_text:
message_parts.append(f"响应内容:{response_text!r}")
else:

View File

@@ -139,6 +139,26 @@ class _FakeResponse:
return self._payload
class _UnicodeDecodeErrorResponse:
"""
模拟 httpx.Response.json() 直接抛 UnicodeDecodeError 的异常响应。
"""
def __init__(self):
"""
初始化一个带有压缩响应特征的伪响应对象。
"""
self.headers = {"Content-Type": "application/json", "Content-Encoding": "gzip"}
self.status_code = 200
self.text = ""
def json(self):
"""
模拟 httpx.Response.json() 在遇到错误编码响应时直接抛出 UnicodeDecodeError。
"""
raise UnicodeDecodeError("utf-8", b"\x8b", 1, 2, "invalid start byte")
class TmdbResponseCacheTest(TestCase):
def test_request_returns_pickleable_snapshot(self):
tmdb = TMDb()
@@ -187,6 +207,16 @@ class TmdbResponseCacheTest(TestCase):
with self.assertRaisesRegex(TMDbException, "不是有效JSON.*HTTP状态码502.*bad gateway"):
TMDb.request.__wrapped__(tmdb, "GET", "https://example.com", None, None)
def test_request_rejects_unicode_decode_error_response(self):
"""
错误编码的响应体也应转换为TMDbException避免UnicodeDecodeError直接冒泡。
"""
tmdb = TMDb()
tmdb._req.get_res = lambda *args, **kwargs: _UnicodeDecodeErrorResponse()
with self.assertRaisesRegex(TMDbException, "不是有效JSON.*Content-Encodinggzip"):
TMDb.request.__wrapped__(tmdb, "GET", "https://example.com", None, None)
def test_get_response_json_rejects_invalid_live_response(self):
"""
未缓存的实时响应解析失败时也应输出统一诊断信息。