feat: add wooyun detail aggregation

This commit is contained in:
Harold
2026-04-19 13:35:15 +08:00
parent 6fb92a7790
commit 95ac8957a1
2 changed files with 159 additions and 0 deletions
+79
View File
@@ -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()
+80
View File
@@ -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