From fac29bb064273fb40e7bdb45a181538e94d0acbb Mon Sep 17 00:00:00 2001 From: Harold <8866033@gmail.com> Date: Sun, 19 Apr 2026 13:32:13 +0800 Subject: [PATCH] feat: scaffold wooyun spider menu parsing --- py/tests/test_乌云影视.py | 113 +++++++++++++++++++++++ py/乌云影视.py | 190 ++++++++++++++++++++++++++++++++++++++ 2 files changed, 303 insertions(+) create mode 100644 py/tests/test_乌云影视.py create mode 100644 py/乌云影视.py diff --git a/py/tests/test_乌云影视.py b/py/tests/test_乌云影视.py new file mode 100644 index 0000000..86a5a68 --- /dev/null +++ b/py/tests/test_乌云影视.py @@ -0,0 +1,113 @@ +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") + + +if __name__ == "__main__": + unittest.main() diff --git a/py/乌云影视.py b/py/乌云影视.py new file mode 100644 index 0000000..5d38b76 --- /dev/null +++ b/py/乌云影视.py @@ -0,0 +1,190 @@ +# coding=utf-8 +import json +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 _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)}