From 38767d11eb7a6936d182b4cdccc7f9bfec1b464d Mon Sep 17 00:00:00 2001 From: Harold <8866033@gmail.com> Date: Sun, 19 Apr 2026 13:47:05 +0800 Subject: [PATCH] feat: add wooyun spider --- .../plans/2026-04-19-wooyun-spider.md | 842 ++++++++++++++++++ 1 file changed, 842 insertions(+) create mode 100644 py/docs/superpowers/plans/2026-04-19-wooyun-spider.md diff --git a/py/docs/superpowers/plans/2026-04-19-wooyun-spider.md b/py/docs/superpowers/plans/2026-04-19-wooyun-spider.md new file mode 100644 index 0000000..0ffdf03 --- /dev/null +++ b/py/docs/superpowers/plans/2026-04-19-wooyun-spider.md @@ -0,0 +1,842 @@ +# 乌云影视 Spider Implementation Plan + +> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. + +**Goal:** 在当前 Python 仓库中新增一个符合 `base.spider.Spider` 接口的乌云影视爬虫，支持 `home/homeVideo/category/detail/search/player` 全链路。 + +**Architecture:** 采用单文件站点脚本 `乌云影视.py` 承担乌云影视的 JSON API 适配逻辑，内部拆分为请求封装、分类与筛选构造、列表映射、详情聚合、播放载荷编解码与剧集回查几个 helper。测试沿用当前仓库 `unittest + SourceFileLoader + mock` 风格，优先覆盖纯函数与高层 mock 网络流程，不依赖真实站点网络。 + +**Tech Stack:** Python 3, `requests`, `unittest`, `unittest.mock`, `json`, `base64`, `math`, `urllib.parse`, `base.spider.Spider` + +--- + +## File Structure + +- Create: `乌云影视.py` + - 乌云影视站点实现，继承 `base.spider.Spider` + - 暴露 `init`、`homeContent`、`homeVideoContent`、`categoryContent`、`detailContent`、`searchContent`、`playerContent` + - 私有方法负责 JSON 请求、分类筛选构造、列表映射、详情聚合、播放载荷编解码、剧集直链回查 +- Create: `tests/test_乌云影视.py` + - 用 `SourceFileLoader` 加载 `乌云影视.py` + - 用 mock JSON 返回覆盖首页、分类、详情、搜索和播放解析 + +### Task 1: Scaffold Spider, Request Helpers, And Menu Parsing + +**Files:** +- Create: `tests/test_乌云影视.py` +- Create: `乌云影视.py` + +- [ ] **Step 1: Write the failing test** + +```python +import unittest +from importlib.machinery import SourceFileLoader +from pathlib import Path + + +ROOT = Path(__file__).resolve().parents[1] +MODULE = SourceFileLoader("wooyun_spider", str(ROOT / "乌云影视.py")).load_module() +Spider = MODULE.Spider + + +class TestWooyunSpider(unittest.TestCase): + def setUp(self): + Spider._instance = None + self.spider = Spider() + self.spider.init() + + def test_home_content_builds_classes_and_filters_from_menu(self): + menu = [ + {"code": "movie", "name": "电影"}, + {"code": "tv_series", "name": "电视剧"}, + {"code": "animation", "name": "动画"}, + {"code": "variety", "name": "综艺"}, + {"code": "short_drama", "name": "短剧"}, + { + "nameEn": "year", + "children": [ + {"code": "THIS_YEAR", "name": "今年"}, + {"code": "LAST_YEAR", "name": "去年"}, + {"code": "2024", "name": "2024"}, + ], + }, + { + "nameEn": "region", + "children": [{"code": "CN", "name": "大陆"}], + }, + { + "nameEn": "genre", + "children": [{"code": "COMEDY", "name": "喜剧"}], + }, + { + "nameEn": "language", + "children": [{"code": "ZH", "name": "国语"}], + }, + ] + + self.spider._request_json = lambda path, method="GET", data=None, headers=None: menu + content = self.spider.homeContent(False) + + self.assertEqual( + [item["type_id"] for item in content["class"]], + ["movie", "tv_series", "animation", "variety", "short_drama", "THIS_YEAR", "LAST_YEAR"], + ) + self.assertEqual(content["filters"]["movie"][0]["key"], "year") + self.assertEqual(content["filters"]["movie"][1]["key"], "region") + self.assertEqual(content["filters"]["movie"][2]["key"], "genre") + self.assertEqual(content["filters"]["movie"][3]["key"], "lang") + self.assertEqual(content["filters"]["movie"][4]["key"], "sort") + + def test_request_json_uses_post_for_json_payload(self): + calls = {} + + class FakeResponse: + def __init__(self): + self.status_code = 200 + self.text = "" + + def json(self): + return {"data": {"records": []}} + + def fake_post(url, json=None, headers=None, timeout=5, verify=True, stream=False, allow_redirects=True, params=None, data=None, cookies=None): + calls["url"] = url + calls["json"] = json + calls["headers"] = headers + return FakeResponse() + + self.spider.post = fake_post + result = self.spider._request_json( + "/movie/media/search", + method="POST", + data={"pageIndex": "1"}, + headers={"Content-Type": "application/json"}, + ) + + self.assertEqual(calls["url"], "https://wooyun.tv/movie/media/search") + self.assertEqual(calls["json"], {"pageIndex": "1"}) + self.assertEqual(calls["headers"]["Origin"], "https://wooyun.tv") + self.assertEqual(result, {"records": []}) + + def test_build_category_body_handles_year_shortcuts_and_filters(self): + body = self.spider._build_category_body( + "THIS_YEAR", + "3", + {"region": "CN", "year": "2025", "lang": "ZH", "sort": "hot"}, + ) + self.assertEqual(body["menuCodeList"], ["THIS_YEAR", "CN", "2025", "ZH"]) + self.assertEqual(body["pageIndex"], "3") + self.assertEqual(body["sortCode"], "hot") + self.assertEqual(body["topCode"], "movie") +``` + +- [ ] **Step 2: Run test to verify it fails** + +Run: `python -m unittest tests.test_乌云影视.TestWooyunSpider -v` +Expected: FAIL with `FileNotFoundError` for `乌云影视.py` or missing `Spider` attributes. + +- [ ] **Step 3: Write minimal implementation** + +```python +# coding=utf-8 +import base64 +import json +import math +import sys + +from base.spider import Spider as BaseSpider + +sys.path.append("..") + + +class Spider(BaseSpider): + def __init__(self): + self.name = "乌云影视" + self.host = "https://wooyun.tv" + self.page_size = 24 + self.home_limit = 12 + self.headers = { + "Accept": "application/json, text/plain, */*", + "Origin": self.host, + "Referer": self.host + "/", + "User-Agent": ( + "Mozilla/5.0 (Windows NT 10.0; Win64; x64) " + "AppleWebKit/537.36 (KHTML, like Gecko) " + "Chrome/124.0.0.0 Safari/537.36" + ), + } + self.sort_options = [ + {"n": "默认", "v": "default"}, + {"n": "最新", "v": "latest"}, + {"n": "最热", "v": "hot"}, + {"n": "评分", "v": "score"}, + ] + + def init(self, extend=""): + return None + + def getName(self): + return self.name + + def _stringify(self, value): + return "" if value is None else str(value) + + def _ensure_list(self, value): + return value if isinstance(value, list) else [] + + def _normalize_ext(self, extend): + if isinstance(extend, dict): + return extend + if not extend: + return {} + try: + return json.loads(str(extend)) + 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) + if headers: + merged_headers.update(headers) + + if method == "POST": + response = self.post(url, json=data, headers=merged_headers, timeout=15) + else: + response = self.fetch(url, headers=merged_headers, timeout=15) + + if response.status_code < 200 or response.status_code >= 300: + raise ValueError(f"{method} {path} => HTTP {response.status_code}") + + payload = response.json() + if payload.get("code") not in (None, 200) and payload.get("isSuccess") is False: + raise ValueError(payload.get("resultMsg") or str(payload.get("code"))) + return payload.get("data") + + def _build_classes(self, menu): + classes = [] + for code, fallback in [ + ("movie", "电影"), + ("tv_series", "电视剧"), + ("animation", "动画"), + ("variety", "综艺"), + ("short_drama", "短剧"), + ]: + found = next((item for item in self._ensure_list(menu) if item.get("code") == code), None) + classes.append({"type_id": code, "type_name": self._stringify((found or {}).get("name") or fallback)}) + + year_group = next((item for item in self._ensure_list(menu) if item.get("nameEn") == "year"), {}) + year_children = self._ensure_list(year_group.get("children")) + this_year = next((item for item in year_children if item.get("code") == "THIS_YEAR"), {}) + last_year = next((item for item in year_children if item.get("code") == "LAST_YEAR"), {}) + classes.append({"type_id": "THIS_YEAR", "type_name": self._stringify(this_year.get("name") or "今年")}) + classes.append({"type_id": "LAST_YEAR", "type_name": self._stringify(last_year.get("name") or "去年")}) + return classes + + def _build_filters(self, menu, classes): + option_map = {"genre": [], "region": [], "year": [], "language": []} + for group in self._ensure_list(menu): + values = [] + for item in self._ensure_list(group.get("children")): + code = self._stringify(item.get("code")) + name = self._stringify(item.get("name") or item.get("nameEn") or code) + if code and name and code not in ("THIS_YEAR", "LAST_YEAR"): + values.append({"n": name, "v": code}) + if group.get("nameEn") in option_map: + option_map[group.get("nameEn")] = values + + filters = {} + for cls in classes: + items = [] + if option_map["year"]: + items.append({"key": "year", "name": "年份", "init": "", "value": [{"n": "全部", "v": ""}] + option_map["year"]}) + if option_map["region"]: + items.append({"key": "region", "name": "地区", "init": "", "value": [{"n": "全部", "v": ""}] + option_map["region"]}) + if option_map["genre"] and cls["type_id"] in ("movie", "tv_series", "animation", "variety", "short_drama"): + items.append({"key": "genre", "name": "类型", "init": "", "value": [{"n": "全部", "v": ""}] + option_map["genre"]}) + if option_map["language"]: + items.append({"key": "lang", "name": "语言", "init": "", "value": [{"n": "全部", "v": ""}] + option_map["language"]}) + items.append({"key": "sort", "name": "排序", "init": "default", "value": self.sort_options}) + filters[cls["type_id"]] = items + return filters + + def _map_top_code(self, category_id): + mapping = { + "movie": "movie", + "tv_series": "tv_series", + "animation": "animation", + "variety": "variety", + "short_drama": "short_drama", + "THIS_YEAR": "movie", + "LAST_YEAR": "movie", + } + return mapping.get(category_id, "movie") + + def _build_category_body(self, category_id, page, extend): + filters = self._normalize_ext(extend) + menu_code_list = [] + if category_id not in ("movie", "tv_series", "animation", "variety", "short_drama"): + menu_code_list.append(category_id) + for key in ("genre", "region", "year", "lang", "other"): + value = self._stringify(filters.get(key)) + if value and value not in menu_code_list: + menu_code_list.append(value) + return { + "menuCodeList": menu_code_list, + "pageIndex": self._stringify(page or 1), + "pageSize": self.page_size, + "searchKey": "", + "sortCode": "" if filters.get("sort") in (None, "", "default") else self._stringify(filters.get("sort")), + "topCode": self._map_top_code(category_id), + } + + def homeContent(self, filter): + menu = self._request_json("/movie/category/menu") + classes = self._build_classes(menu) + return {"class": classes, "filters": self._build_filters(menu, classes)} +``` + +- [ ] **Step 4: Run test to verify it passes** + +Run: `python -m unittest tests.test_乌云影视.TestWooyunSpider -v` +Expected: PASS for the three new tests. + +- [ ] **Step 5: Commit** + +```bash +git add tests/test_乌云影视.py 乌云影视.py +git commit -m "feat: scaffold wooyun spider menu parsing" +``` + +### Task 2: Add Home Video, Category, And Search Flows + +**Files:** +- Modify: `tests/test_乌云影视.py` +- Modify: `乌云影视.py` + +- [ ] **Step 1: Write the failing test** + +```python +from unittest.mock import patch + + +class TestWooyunSpider(unittest.TestCase): + def test_extract_home_list_deduplicates_media_resources(self): + home_blocks = { + "records": [ + { + "mediaResources": [ + {"id": 1, "title": "影片A", "posterUrl": "/a.jpg"}, + {"id": 1, "title": "影片A", "posterUrl": "/a.jpg"}, + ] + }, + { + "mediaResources": [ + {"id": 2, "title": "影片B", "posterUrlS3": "https://img.example/b.jpg"}, + ] + }, + ] + } + items = self.spider._extract_home_list(home_blocks) + self.assertEqual([item["id"] for item in items], ["1", "2"]) + + def test_map_vod_picks_expected_fields(self): + item = { + "id": 7, + "title": "示例片", + "posterUrl": "/poster.jpg", + "mediaType": {"code": "movie", "name": "电影"}, + "episodeStatus": "更新至4集", + "releaseYear": "2026", + "rating": "8.8", + "actors": ["甲", "乙"], + "directors": ["丙"], + } + vod = self.spider._map_vod(item) + self.assertEqual(vod["vod_id"], "7") + self.assertEqual(vod["vod_pic"], "/poster.jpg") + self.assertEqual(vod["vod_actor"], "甲/乙") + self.assertEqual(vod["vod_director"], "丙") + + @patch.object(Spider, "_request_json") + def test_home_video_content_reads_home_blocks(self, mock_request_json): + mock_request_json.return_value = { + "records": [ + {"mediaResources": [{"id": 1, "title": "首页片", "posterUrl": "/a.jpg"}]} + ] + } + result = self.spider.homeVideoContent() + self.assertEqual(result["list"][0]["vod_id"], "1") + self.assertEqual(result["list"][0]["vod_name"], "首页片") + + @patch.object(Spider, "_request_json") + def test_category_content_posts_search_body_and_returns_page_data(self, mock_request_json): + mock_request_json.return_value = { + "records": [{"id": 9, "title": "分类片", "posterUrl": "/cate.jpg"}], + "total": 30, + "pages": 2, + } + result = self.spider.categoryContent("movie", "2", False, {"sort": "latest"}) + kwargs = mock_request_json.call_args.kwargs + self.assertEqual(kwargs["method"], "POST") + self.assertEqual(kwargs["data"]["pageIndex"], "2") + self.assertEqual(kwargs["data"]["sortCode"], "latest") + self.assertEqual(result["page"], 2) + self.assertEqual(result["pagecount"], 2) + self.assertEqual(result["limit"], 24) + self.assertEqual(result["total"], 30) + + @patch.object(Spider, "_request_json") + def test_search_content_returns_empty_result_for_blank_keyword(self, mock_request_json): + result = self.spider.searchContent("", False, "1") + self.assertEqual(result, {"page": 1, "pagecount": 0, "total": 0, "list": []}) + mock_request_json.assert_not_called() + + @patch.object(Spider, "_request_json") + def test_search_content_posts_keyword_body(self, mock_request_json): + mock_request_json.return_value = { + "records": [{"id": 15, "title": "搜索片", "posterUrl": "/search.jpg"}], + "total": 1, + "pages": 1, + } + result = self.spider.searchContent("繁花", False, "1") + body = mock_request_json.call_args.kwargs["data"] + self.assertEqual(body["searchKey"], "繁花") + self.assertEqual(body["topCode"], "") + self.assertEqual(result["list"][0]["vod_id"], "15") +``` + +- [ ] **Step 2: Run test to verify it fails** + +Run: `python -m unittest tests.test_乌云影视.TestWooyunSpider -v` +Expected: FAIL with missing `_extract_home_list` / `_map_vod` / `homeVideoContent` / `categoryContent` / `searchContent`. + +- [ ] **Step 3: Write minimal implementation** + +```python + def _join_text(self, value): + return "/".join([self._stringify(item) for item in self._ensure_list(value) if self._stringify(item)]) + + def _pick_poster(self, item): + return self._stringify( + item.get("posterUrlS3") or item.get("posterUrl") or item.get("thumbUrlS3") or item.get("thumbUrl") + ) + + def _extract_home_list(self, home_blocks): + blocks = self._ensure_list((home_blocks or {}).get("records") if isinstance(home_blocks, dict) else home_blocks) + seen = set() + items = [] + for block in blocks: + for item in self._ensure_list(block.get("mediaResources")): + vod_id = self._stringify(item.get("id")) + if not vod_id or vod_id in seen: + continue + seen.add(vod_id) + copied = dict(item) + copied["id"] = vod_id + items.append(copied) + return items + + def _map_vod(self, item): + return { + "vod_id": self._stringify(item.get("id")), + "vod_name": self._stringify(item.get("title")), + "vod_pic": self._pick_poster(item), + "type_id": self._stringify((item.get("mediaType") or {}).get("code")), + "type_name": self._stringify((item.get("mediaType") or {}).get("name")), + "vod_remarks": self._stringify(item.get("episodeStatus") or item.get("remark")), + "vod_year": self._stringify(item.get("releaseYear")), + "vod_douban_score": self._stringify(item.get("rating")), + "vod_actor": self._join_text(item.get("actors")), + "vod_director": self._join_text(item.get("directors")), + } + + def homeVideoContent(self): + data = self._request_json(f"/movie/media/home/custom/classify/1/3?limit={self.home_limit}") + return {"list": [self._map_vod(item) for item in self._extract_home_list(data)[: self.home_limit]]} + + def categoryContent(self, tid, pg, filter, extend): + page = int(pg) + data = self._request_json( + "/movie/media/search", + method="POST", + data=self._build_category_body(self._stringify(tid or "movie"), page, extend), + headers={"Content-Type": "application/json"}, + ) or {} + records = [self._map_vod(item) for item in self._ensure_list(data.get("records"))] + 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, "limit": self.page_size, "total": total, "list": records} + + def searchContent(self, key, quick, pg="1"): + page = int(pg) + keyword = self._stringify(key).strip() + if not keyword: + return {"page": page, "pagecount": 0, "total": 0, "list": []} + + data = self._request_json( + "/movie/media/search", + method="POST", + data={ + "menuCodeList": [], + "pageIndex": self._stringify(page), + "pageSize": self.page_size, + "searchKey": keyword, + "sortCode": "", + "topCode": "", + }, + headers={"Content-Type": "application/json"}, + ) or {} + records = [self._map_vod(item) for item in self._ensure_list(data.get("records"))] + 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} +``` + +- [ ] **Step 4: Run test to verify it passes** + +Run: `python -m unittest tests.test_乌云影视.TestWooyunSpider -v` +Expected: PASS for the new home/category/search tests. + +- [ ] **Step 5: Commit** + +```bash +git add tests/test_乌云影视.py 乌云影视.py +git commit -m "feat: add wooyun list and search flows" +``` + +### Task 3: Add Detail Aggregation And Multi-Season Playlists + +**Files:** +- Modify: `tests/test_乌云影视.py` +- Modify: `乌云影视.py` + +- [ ] **Step 1: Write the failing test** + +```python +class TestWooyunSpider(unittest.TestCase): + 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"]) +``` + +- [ ] **Step 2: Run test to verify it fails** + +Run: `python -m unittest tests.test_乌云影视.TestWooyunSpider -v` +Expected: FAIL with missing `_build_play_sources` / `_encode_play_payload` / `_decode_play_id` / `detailContent`. + +- [ ] **Step 3: Write minimal implementation** + +```python + 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 +``` + +- [ ] **Step 4: Run test to verify it passes** + +Run: `python -m unittest tests.test_乌云影视.TestWooyunSpider -v` +Expected: PASS for the new detail and playlist tests. + +- [ ] **Step 5: Commit** + +```bash +git add tests/test_乌云影视.py 乌云影视.py +git commit -m "feat: add wooyun detail aggregation" +``` + +### Task 4: Add Player Resolution And Fallback Logic + +**Files:** +- Modify: `tests/test_乌云影视.py` +- Modify: `乌云影视.py` + +- [ ] **Step 1: Write the failing test** + +```python +class TestWooyunSpider(unittest.TestCase): + 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") +``` + +- [ ] **Step 2: Run test to verify it fails** + +Run: `python -m unittest tests.test_乌云影视.TestWooyunSpider -v` +Expected: FAIL with missing `_find_play_url` or incorrect `playerContent` fallback behavior. + +- [ ] **Step 3: Write minimal implementation** + +```python + 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"], + }, + } +``` + +- [ ] **Step 4: Run test to verify it passes** + +Run: `python -m unittest tests.test_乌云影视.TestWooyunSpider -v` +Expected: PASS for the new player tests and all previous tests. + +- [ ] **Step 5: Commit** + +```bash +git add tests/test_乌云影视.py 乌云影视.py +git commit -m "feat: add wooyun player resolution" +``` + +### Task 5: Full Verification And Cleanup + +**Files:** +- Modify: `tests/test_乌云影视.py` +- Modify: `乌云影视.py` + +- [ ] **Step 1: Run the focused spider test suite** + +Run: `python -m unittest tests.test_乌云影视 -v` +Expected: All tests PASS with no import or syntax errors. + +- [ ] **Step 2: Run adjacent regression tests** + +Run: `python -m unittest tests.test_dbku tests.test_剧迷 tests.test_橘子TV tests.test_youknow tests.test_libvio -v` +Expected: Existing spider tests still PASS. + +- [ ] **Step 3: Run syntax verification** + +Run: `python -m py_compile 乌云影视.py tests/test_乌云影视.py` +Expected: Command exits successfully with no output. + +- [ ] **Step 4: Review git diff for scope** + +Run: `git diff -- 乌云影视.py tests/test_乌云影视.py docs/superpowers/plans/2026-04-19-wooyun-spider.md` +Expected: Diff only contains the new spider, its tests, and the implementation plan. + +- [ ] **Step 5: Commit** + +```bash +git add 乌云影视.py tests/test_乌云影视.py docs/superpowers/plans/2026-04-19-wooyun-spider.md +git commit -m "feat: add wooyun spider" +```