From 95ac8957a14b99b4855ace334566f02e940e0343 Mon Sep 17 00:00:00 2001 From: Harold <8866033@gmail.com> Date: Sun, 19 Apr 2026 13:35:15 +0800 Subject: [PATCH] feat: add wooyun detail aggregation --- py/tests/test_乌云影视.py | 79 ++++++++++++++++++++++++++++++++++++++ py/乌云影视.py | 80 +++++++++++++++++++++++++++++++++++++++ 2 files changed, 159 insertions(+) diff --git a/py/tests/test_乌云影视.py b/py/tests/test_乌云影视.py index 95c3046..77ac048 100644 --- a/py/tests/test_乌云影视.py +++ b/py/tests/test_乌云影视.py @@ -193,6 +193,85 @@ class TestWooyunSpider(unittest.TestCase): self.assertEqual(body["topCode"], "") self.assertEqual(result["list"][0]["vod_id"], "15") + def test_build_play_sources_encodes_multiple_seasons(self): + seasons = [ + { + "seasonNo": 1, + "videoList": [ + {"id": 101, "epNo": 1, "remark": "", "playUrl": "/play-1.m3u8"}, + {"id": 102, "epNo": 2, "remark": "加更", "playUrl": "/play-2.m3u8"}, + ], + }, + { + "seasonNo": 2, + "videoList": [ + {"id": 201, "epNo": 1, "remark": "", "playUrl": "/play-3.m3u8"}, + ], + }, + ] + payload = self.spider._build_play_sources(seasons, "99") + self.assertEqual(payload["vod_play_from"], "第1季$$$第2季") + self.assertIn("第1集$", payload["vod_play_url"]) + self.assertIn("第2集 加更$", payload["vod_play_url"]) + + def test_decode_play_id_round_trips_base64url_payload(self): + encoded = self.spider._encode_play_payload( + { + "mediaId": "88", + "seasonNo": 1, + "epNo": 3, + "videoId": 12, + "playUrl": "/video.m3u8", + "name": "第3集", + } + ) + decoded = self.spider._decode_play_id(encoded) + self.assertEqual(decoded["mediaId"], "88") + self.assertEqual(decoded["seasonNo"], 1) + self.assertEqual(decoded["epNo"], 3) + self.assertEqual(decoded["playUrl"], "/video.m3u8") + + @patch.object(Spider, "_request_json") + def test_detail_content_merges_detail_apis_and_videos(self, mock_request_json): + mock_request_json.side_effect = [ + {"id": 300, "title": "基础标题", "posterUrl": "/base.jpg"}, + { + "id": 300, + "title": "完整标题", + "posterUrlS3": "https://img.example/detail.jpg", + "mediaType": {"code": "tv_series", "name": "电视剧"}, + "episodeStatus": "更新至2集", + "releaseYear": "2026", + "region": "大陆", + "actors": ["张三", "李四"], + "directors": ["王五"], + "overview": "一段简介", + "rating": "9.1", + }, + [ + { + "seasonNo": 1, + "videoList": [ + {"id": 901, "epNo": 1, "remark": "", "playUrl": "/ep1.m3u8"}, + {"id": 902, "epNo": 2, "remark": "超前", "playUrl": "/ep2.m3u8"}, + ], + } + ], + ] + result = self.spider.detailContent(["300"]) + vod = result["list"][0] + self.assertEqual(vod["vod_id"], "300") + self.assertEqual(vod["vod_name"], "完整标题") + self.assertEqual(vod["vod_pic"], "https://img.example/detail.jpg") + self.assertEqual(vod["type_name"], "电视剧") + self.assertEqual(vod["vod_area"], "大陆") + self.assertEqual(vod["vod_actor"], "张三/李四") + self.assertEqual(vod["vod_director"], "王五") + self.assertEqual(vod["vod_content"], "一段简介") + self.assertEqual(vod["vod_douban_score"], "9.1") + self.assertEqual(vod["vod_play_from"], "乌云影视") + self.assertIn("第2集 超前$", vod["vod_play_url"]) + if __name__ == "__main__": unittest.main() diff --git a/py/乌云影视.py b/py/乌云影视.py index 084f8e2..4dfe22a 100644 --- a/py/乌云影视.py +++ b/py/乌云影视.py @@ -1,4 +1,5 @@ # coding=utf-8 +import base64 import json import math import sys @@ -272,3 +273,82 @@ class Spider(BaseSpider): total = int(data.get("total") or 0) pagecount = int(data.get("pages") or (math.ceil(total / self.page_size) if total else (page if records else 0))) return {"page": page, "pagecount": pagecount, "total": total, "list": records} + + def _encode_play_payload(self, payload): + raw = json.dumps(payload, ensure_ascii=False).encode("utf-8") + return base64.urlsafe_b64encode(raw).decode("utf-8").rstrip("=") + + def _decode_play_id(self, play_id): + raw = self._stringify(play_id).strip() + padding = "=" * (-len(raw) % 4) + try: + return json.loads(base64.urlsafe_b64decode((raw + padding).encode("utf-8")).decode("utf-8")) + except Exception: + return {"mediaId": raw, "playUrl": raw} + + def _build_play_sources(self, seasons, media_id): + season_list = self._ensure_list(seasons) + multiple = len(season_list) > 1 + from_list = [] + url_list = [] + for index, season in enumerate(season_list, start=1): + season_no = season.get("seasonNo") or index + episode_entries = [] + for video_index, video in enumerate(self._ensure_list(season.get("videoList")), start=1): + ep_no = int(video.get("epNo") or video_index) + remark = self._stringify(video.get("remark")).strip() + if ep_no > 0: + name = f"第{ep_no}集" if not remark else f"第{ep_no}集 {remark}" + else: + name = remark or "正片" + payload = self._encode_play_payload( + { + "mediaId": self._stringify(media_id), + "seasonNo": season_no, + "epNo": ep_no, + "videoId": video.get("id"), + "playUrl": video.get("playUrl"), + "name": name, + } + ) + episode_entries.append(f"{name}${payload}") + if episode_entries: + from_list.append( + f"第{season_no}季" + if multiple + else self._stringify(season.get("lineName") or season.get("title") or self.name) + ) + url_list.append("#".join(episode_entries)) + return {"vod_play_from": "$$$".join(from_list), "vod_play_url": "$$$".join(url_list)} + + def detailContent(self, ids): + result = {"list": []} + for raw_id in ids: + media_id = self._stringify(raw_id).strip() + if not media_id: + continue + base_detail = self._request_json(f"/movie/media/base/detail?mediaId={media_id}") or {} + detail = self._request_json(f"/movie/media/detail?mediaId={media_id}") or {} + seasons = self._request_json( + f"/movie/media/video/list?mediaId={media_id}&lineName=&resolutionCode=" + ) or [] + merged = detail or base_detail + play_data = self._build_play_sources(seasons, media_id) + result["list"].append( + { + "vod_id": self._stringify(merged.get("id") or media_id), + "vod_name": self._stringify(merged.get("title") or base_detail.get("title")), + "vod_pic": self._pick_poster(merged or base_detail), + "type_id": self._stringify((merged.get("mediaType") or {}).get("code")), + "type_name": self._stringify((merged.get("mediaType") or {}).get("name")), + "vod_remarks": self._stringify(merged.get("episodeStatus")), + "vod_year": self._stringify(merged.get("releaseYear")), + "vod_area": self._stringify(merged.get("region")), + "vod_actor": self._join_text(merged.get("actors")), + "vod_director": self._join_text(merged.get("directors")), + "vod_content": self._stringify(merged.get("overview") or merged.get("description")), + "vod_douban_score": self._stringify(merged.get("rating")), + **play_data, + } + ) + return result