From 2d26930b0b66c54ea82fcb77b4a265bb9dd23a53 Mon Sep 17 00:00:00 2001 From: Harold <8866033@gmail.com> Date: Sun, 19 Apr 2026 13:37:00 +0800 Subject: [PATCH] feat: add wooyun player resolution --- py/tests/test_乌云影视.py | 60 +++++++++++++++++++++++++++++++++++++++ py/乌云影视.py | 55 ++++++++++++++++++++++++++++++++++- 2 files changed, 114 insertions(+), 1 deletion(-) diff --git a/py/tests/test_乌云影视.py b/py/tests/test_乌云影视.py index 77ac048..da41ebc 100644 --- a/py/tests/test_乌云影视.py +++ b/py/tests/test_乌云影视.py @@ -272,6 +272,66 @@ class TestWooyunSpider(unittest.TestCase): self.assertEqual(vod["vod_play_from"], "乌云影视") self.assertIn("第2集 超前$", vod["vod_play_url"]) + def test_find_play_url_matches_video_id_then_ep_no(self): + seasons = [ + { + "seasonNo": 1, + "videoList": [ + {"id": 10, "epNo": 1, "playUrl": "/ep1.m3u8"}, + {"id": 11, "epNo": 2, "playUrl": "/ep2.m3u8"}, + ], + } + ] + self.assertEqual(self.spider._find_play_url(seasons, {"seasonNo": 1, "videoId": 11}), "/ep2.m3u8") + self.assertEqual(self.spider._find_play_url(seasons, {"seasonNo": 1, "epNo": 1}), "/ep1.m3u8") + + def test_player_content_returns_direct_url_when_payload_has_play_url(self): + play_id = self.spider._encode_play_payload( + { + "mediaId": "100", + "seasonNo": 1, + "epNo": 1, + "videoId": 10, + "playUrl": "/direct.m3u8", + "name": "第1集", + } + ) + result = self.spider.playerContent("乌云影视", play_id, {}) + self.assertEqual(result["parse"], 0) + self.assertEqual(result["jx"], 0) + self.assertEqual(result["url"], "https://wooyun.tv/direct.m3u8") + self.assertEqual(result["header"]["Referer"], "https://wooyun.tv/") + + @patch.object(Spider, "_request_json") + def test_player_content_refetches_video_list_when_payload_missing_play_url(self, mock_request_json): + mock_request_json.return_value = [ + { + "seasonNo": 1, + "videoList": [ + {"id": 77, "epNo": 3, "playUrl": "/refetched.m3u8"}, + ], + } + ] + play_id = self.spider._encode_play_payload( + { + "mediaId": "100", + "seasonNo": 1, + "epNo": 3, + "videoId": 77, + "playUrl": "", + "name": "第3集", + } + ) + result = self.spider.playerContent("乌云影视", play_id, {}) + self.assertEqual(result["parse"], 0) + self.assertEqual(result["url"], "https://wooyun.tv/refetched.m3u8") + + def test_player_content_falls_back_to_external_parse_when_no_url_found(self): + result = self.spider.playerContent("乌云影视", "raw-play-id", {}) + self.assertEqual(result["parse"], 1) + self.assertEqual(result["jx"], 1) + self.assertEqual(result["url"], "raw-play-id") + if __name__ == "__main__": unittest.main() diff --git a/py/乌云影视.py b/py/乌云影视.py index 4dfe22a..89fe25c 100644 --- a/py/乌云影视.py +++ b/py/乌云影视.py @@ -54,6 +54,18 @@ class Spider(BaseSpider): except Exception: return {} + def _full_url(self, path): + raw = self._stringify(path).strip() + if not raw: + return "" + if raw.startswith(("http://", "https://")): + return raw + if raw.startswith("//"): + return "https:" + raw + if raw.startswith("/"): + return self.host + raw + return self.host + "/" + raw + def _request_json(self, path, method="GET", data=None, headers=None): url = path if str(path).startswith("http") else self.host + path merged_headers = dict(self.headers) @@ -284,7 +296,7 @@ class Spider(BaseSpider): try: return json.loads(base64.urlsafe_b64decode((raw + padding).encode("utf-8")).decode("utf-8")) except Exception: - return {"mediaId": raw, "playUrl": raw} + return {"mediaId": "", "playUrl": ""} def _build_play_sources(self, seasons, media_id): season_list = self._ensure_list(seasons) @@ -352,3 +364,44 @@ class Spider(BaseSpider): } ) return result + + def _find_play_url(self, seasons, target): + for season in self._ensure_list(seasons): + if target.get("seasonNo") is not None and int(season.get("seasonNo") or 0) != int(target.get("seasonNo")): + continue + for video in self._ensure_list(season.get("videoList")): + if target.get("videoId") is not None and int(video.get("id") or 0) == int(target.get("videoId")): + return self._stringify(video.get("playUrl")) + if target.get("epNo") is not None and int(video.get("epNo") or 0) == int(target.get("epNo")): + return self._stringify(video.get("playUrl")) + + first_season = self._ensure_list(seasons)[0] if self._ensure_list(seasons) else {} + first_video = ( + self._ensure_list(first_season.get("videoList"))[0] + if self._ensure_list(first_season.get("videoList")) + else {} + ) + return self._stringify(first_video.get("playUrl")) + + def playerContent(self, flag, id, vipFlags): + payload = self._decode_play_id(self._stringify(id).strip()) + final_url = self._stringify(payload.get("playUrl")) + if not final_url and payload.get("mediaId"): + seasons = self._request_json( + f"/movie/media/video/list?mediaId={payload['mediaId']}&lineName=&resolutionCode=" + ) or [] + final_url = self._find_play_url(seasons, payload) + + if not final_url: + return {"parse": 1, "jx": 1, "url": self._stringify(id), "header": dict(self.headers)} + + return { + "parse": 0, + "jx": 0, + "url": self._full_url(final_url), + "header": { + "Referer": self.host + "/", + "Origin": self.host, + "User-Agent": self.headers["User-Agent"], + }, + }