fix: remove tmdb manual gzip fallback

app/modules/themoviedb/tmdbv3api/tmdb.py

@@ -1,8 +1,6 @@
 # -*- coding: utf-8 -*-
 
 import asyncio
-import gzip
-import json as jsonlib
 import logging
 import time
 from copy import deepcopy
@@ -21,7 +19,6 @@ logger = logging.getLogger(__name__)
 
 class TMDb(object):
     _RESPONSE_SNAPSHOT_MARKER = "__mp_tmdb_response_snapshot__"
-    _JSON_DECODE_FAILED = object()
 
     def __init__(self, session=None, language=None):
         self._api_key = settings.TMDB_API_KEY
@@ -166,42 +163,8 @@ class TMDb(object):
         try:
             return response.json()
         except (ValueError, UnicodeDecodeError) as err:
-            # httpx.Response.json() 在响应体是压缩字节或错误编码时会直接抛 UnicodeDecodeError，
-            # 先尝试兼容未被客户端解压的 gzip JSON，仍失败时再收敛成 TMDbException。
-            json_data = cls._decode_compressed_response_json(response)
-            if json_data is not cls._JSON_DECODE_FAILED:
-                return json_data
             raise TMDbException(cls._build_invalid_json_message(response, err)) from err
 
-    @classmethod
-    def _decode_compressed_response_json(cls, response):
-        """
-        尝试解析未被HTTP客户端自动解压的压缩JSON响应。
-        """
-        response_content = getattr(response, "content", b"") or b""
-        if isinstance(response_content, str):
-            response_content = response_content.encode("utf-8")
-        if not isinstance(response_content, (bytes, bytearray)):
-            return cls._JSON_DECODE_FAILED
-
-        content_bytes = bytes(response_content)
-        content_encoding = cls._get_header_value(
-            getattr(response, "headers", {}) or {},
-            "Content-Encoding",
-        ) or ""
-        encodings = {
-            encoding.strip().lower()
-            for encoding in str(content_encoding).split(",")
-            if encoding.strip()
-        }
-        if "gzip" not in encodings and not content_bytes.startswith(b"\x1f\x8b"):
-            return cls._JSON_DECODE_FAILED
-
-        try:
-            return jsonlib.loads(gzip.decompress(content_bytes))
-        except (OSError, EOFError, ValueError, UnicodeDecodeError):
-            return cls._JSON_DECODE_FAILED
-
     @staticmethod
     def _get_header_value(headers, name):
         """
@@ -255,7 +218,7 @@ class TMDb(object):
         if content_encoding:
             message_parts.append(f"Content-Encoding：{content_encoding}")
         if is_encoding_error:
-            message_parts.append("响应内容因编码错误已省略")
+            message_parts.append("响应内容编码异常，已省略原始内容")
         elif response_text:
             message_parts.append(f"响应内容：{response_text!r}")
         else:

tests/test_tmdb_response_cache.py

@@ -1,7 +1,5 @@
 import asyncio
-import gzip
 import importlib.util
-import json
 import pickle
 import sys
 from contextlib import asynccontextmanager, contextmanager
@@ -162,18 +160,6 @@ class _UnicodeDecodeErrorResponse:
         raise UnicodeDecodeError("utf-8", b"\x8b", 1, 2, "invalid start byte")
 
 
-class _GzipJsonResponse(_UnicodeDecodeErrorResponse):
-    """
-    模拟响应对象带着尚未解压的 gzip JSON 字节。
-    """
-
-    def __init__(self, payload):
-        """
-        将JSON载荷压缩成 gzip 字节，复现代理返回原始压缩体的情况。
-        """
-        super().__init__(gzip.compress(json.dumps(payload).encode("utf-8")))
-
-
 class TmdbResponseCacheTest(TestCase):
     def test_request_returns_pickleable_snapshot(self):
         tmdb = TMDb()
@@ -233,25 +219,11 @@ class TmdbResponseCacheTest(TestCase):
 
         with self.assertRaisesRegex(
                 TMDbException,
-                "不是有效JSON.*Content-Encoding：gzip.*响应内容因编码错误已省略",
+                "不是有效JSON.*Content-Encoding：gzip.*响应内容编码异常，已省略原始内容",
         ) as cm:
             TMDb.request.__wrapped__(tmdb, "GET", "https://example.com", None, None)
         self.assertNotIn("乱码内容", str(cm.exception))
 
-    def test_request_decodes_raw_gzip_json_response(self):
-        """
-        客户端未自动解压 gzip JSON 时，应手动解压后正常进入响应快照。
-        """
-        tmdb = TMDb()
-        tmdb._req.get_res = lambda *args, **kwargs: _GzipJsonResponse(
-            {"page": 1, "results": [{"id": 100}]}
-        )
-
-        result = TMDb.request.__wrapped__(tmdb, "GET", "https://example.com", None, None)
-
-        self.assertTrue(result[TMDb._RESPONSE_SNAPSHOT_MARKER])
-        self.assertEqual(result["json"]["results"], [{"id": 100}])
-
     def test_get_response_json_rejects_invalid_live_response(self):
         """
         未缓存的实时响应解析失败时也应输出统一诊断信息。