fix: parse video URLs by MP4 quality

Author: ink-developer

src/pymax/types/domain/attachments/video.py

@@ -62,18 +62,18 @@ class VideoRequest(CamelModel):
     :vartype external: str | bool | None
     :ivar cache: Использовать ли кеш.
     :vartype cache: bool
-    :ivar url: URL видео.
-    :vartype url: str
+    :ivar url: Прямой URL видео или ``None`` для внешнего видео.
+    :vartype url: str | None
     """
 
     external: str | bool | None = Field(default=None, alias="EXTERNAL")
     cache: bool
-    url: str
+    url: str | None = None
 
     @model_validator(mode="before")
     @classmethod
-    def unwrap_dynamic_url(cls, value: Any) -> Any:
-        """Нормализует динамический ключ URL в поле ``url``.
+    def select_video_url(cls, value: Any) -> Any:
+        """Выбирает прямой URL с максимальным доступным MP4-качеством.
 
         :param value: Значение, переданное в валидатор модели.
         :type value: Any
@@ -83,8 +83,29 @@ def unwrap_dynamic_url(cls, value: Any) -> Any:
         if not isinstance(value, dict) or "url" in value:
             return value
 
+        mp4_urls: list[tuple[int, str]] = []
         for key, url in value.items():
-            if key not in ("EXTERNAL", "cache"):
-                return {**value, "url": url}
+            if not isinstance(key, str) or not isinstance(url, str):
+                continue
+
+            normalized_key = key.upper()
+            if not normalized_key.startswith("MP4_"):
+                continue
+
+            try:
+                quality = int(normalized_key.removeprefix("MP4_"))
+            except ValueError:
+                continue
+
+            if quality > 0:
+                mp4_urls.append((quality, url))
+
+        if mp4_urls:
+            _, url = max(mp4_urls, key=lambda item: item[0])
+            return {**value, "url": url}
+
+        legacy_url = value.get("dynamicUrl", value.get("dynamic_url"))
+        if isinstance(legacy_url, str):
+            return {**value, "url": legacy_url}
 
         return value

tests/api/test_message_service.py

@@ -7,6 +7,7 @@
 from pymax.exceptions import UploadError
 from pymax.files import File, Photo, Video
 from pymax.protocol import Opcode
+from pymax.types.domain.attachments import VideoRequest
 from tests.conftest import FakeApp, frame, message_payload
 
 
@@ -372,7 +373,15 @@ async def test_reaction_methods_parse_optional_reaction_info() -> None:
 async def test_get_video_and_file_by_id_parse_request_models() -> None:
     app = FakeApp(
         [
-            frame({"cache": True, "dynamicUrl": "https://video.test"}),
+            frame(
+                {
+                    "cache": True,
+                    "FAILOVER_HOSTS": ["maxvd759.okcdn.ru"],
+                    "MP4_480": "https://video.test/480",
+                    "EXTERNAL": "https://m.ok.ru/video/1",
+                    "MP4_720": "https://video.test/720",
+                }
+            ),
             frame({"unsafe": False, "url": "https://file.test"}),
         ]
     )
@@ -381,7 +390,8 @@ async def test_get_video_and_file_by_id_parse_request_models() -> None:
     file = await app.api.messages.get_file_by_id(100, "10", 30)
 
     assert video is not None
-    assert video.url == "https://video.test"
+    assert video.url == "https://video.test/720"
+    assert video.external == "https://m.ok.ru/video/1"
     assert file is not None
     assert file.url == "https://file.test"
     assert [call.opcode for call in app.calls] == [
@@ -400,6 +410,19 @@ async def test_get_video_and_file_by_id_parse_request_models() -> None:
     }
 
 
+def test_video_request_supports_legacy_and_external_only_payloads() -> None:
+    legacy = VideoRequest.model_validate(
+        {"cache": True, "dynamicUrl": "https://video.test/legacy"}
+    )
+    external = VideoRequest.model_validate(
+        {"cache": True, "EXTERNAL": "https://m.ok.ru/video/1"}
+    )
+
+    assert legacy.url == "https://video.test/legacy"
+    assert external.url is None
+    assert external.external == "https://m.ok.ru/video/1"
+
+
 def test_next_cid_is_monotonic_when_clock_does_not_move(
     monkeypatch: pytest.MonkeyPatch,
 ) -> None: