diff --git a/py/tests/test_四万影视.py b/py/tests/test_四万影视.py new file mode 100644 index 0000000..5fd903c --- /dev/null +++ b/py/tests/test_四万影视.py @@ -0,0 +1,205 @@ +import json +import unittest +from importlib.machinery import SourceFileLoader +from pathlib import Path +from types import SimpleNamespace +from unittest.mock import patch + + +ROOT = Path(__file__).resolve().parents[1] +MODULE = SourceFileLoader("siwan_spider", str(ROOT / "四万影视.py")).load_module() +Spider = MODULE.Spider + + +class TestSiWanSpider(unittest.TestCase): + def setUp(self): + Spider._instance = None + self.spider = Spider() + self.spider.init() + + def _response(self, status_code=200, payload=None, text=None): + body = text if text is not None else json.dumps(payload or {}, ensure_ascii=False) + return SimpleNamespace(status_code=status_code, text=body) + + def test_home_content_exposes_expected_classes_and_filters(self): + content = self.spider.homeContent(False) + self.assertEqual( + [item["type_id"] for item in content["class"]], + ["20", "30", "39", "45", "32"], + ) + self.assertEqual( + [item["key"] for item in content["filters"]["20"]], + ["subType", "year", "sort"], + ) + self.assertEqual(content["filters"]["20"][0]["init"], "20") + self.assertEqual(content["filters"]["20"][1]["value"][0], {"n": "全部", "v": ""}) + self.assertEqual(content["filters"]["20"][2]["value"][0], {"n": "时间", "v": "time"}) + + def test_home_video_content_returns_empty_list(self): + self.assertEqual(self.spider.homeVideoContent(), {"list": []}) + + @patch.object(Spider, "fetch") + def test_api_get_requests_maccms_endpoint_with_browser_headers(self, mock_fetch): + mock_fetch.return_value = self._response(payload={"list": []}) + data = self.spider._api_get({"ac": "detail", "pg": 1}) + self.assertEqual(data, {"list": []}) + self.assertEqual(mock_fetch.call_args.args[0], "https://40000.me/api/maccms") + self.assertEqual(mock_fetch.call_args.kwargs["params"], {"ac": "detail", "pg": 1}) + self.assertEqual(mock_fetch.call_args.kwargs["headers"]["Referer"], "https://40000.me/") + + @patch.object(Spider, "fetch") + def test_api_get_rejects_non_200_and_non_object_body(self, mock_fetch): + mock_fetch.return_value = self._response(status_code=500, payload={"msg": "bad"}) + with self.assertRaises(ValueError): + self.spider._api_get({"ac": "detail"}) + + mock_fetch.return_value = self._response(text="[]") + with self.assertRaises(ValueError): + self.spider._api_get({"ac": "detail"}) + + def test_normalize_vod_applies_type_name_actor_cleanup_and_fallback_cover(self): + self.assertEqual( + self.spider._normalize_vod( + { + "vod_id": 7, + "vod_name": "示例影片", + "vod_pic": "", + "type_id": "20", + "vod_remarks": "HD", + "vod_year": "2025", + "vod_actor": "O&#039;Brien , 张三", + } + ), + { + "vod_id": "7", + "vod_name": "示例影片", + "vod_pic": "https://40000.me/public/favicon.png", + "vod_remarks": "HD", + "vod_year": "2025", + "type_name": "电影", + "vod_actor": "O'Brien, 张三", + }, + ) + + @patch.object(Spider, "_api_get") + def test_category_content_maps_subtype_year_sort_and_page(self, mock_api_get): + mock_api_get.return_value = { + "page": 2, + "total": 55, + "list": [ + {"vod_id": 100, "vod_name": "分类影片", "vod_pic": "/p.jpg", "type_id": "21", "vod_remarks": "更新中"} + ], + } + result = self.spider.categoryContent("20", "2", False, {"subType": "21", "year": "2025", "sort": "hits"}) + self.assertEqual( + mock_api_get.call_args.args[0], + {"ac": "detail", "t": "21", "pg": 2, "h": "2025", "by": "hits"}, + ) + self.assertEqual(result["page"], 2) + self.assertEqual(result["total"], 55) + self.assertEqual(result["limit"], 1) + self.assertEqual(result["list"][0]["vod_id"], "100") + self.assertNotIn("pagecount", result) + + @patch.object(Spider, "_api_get") + def test_search_content_maps_keyword_and_page(self, mock_api_get): + mock_api_get.return_value = { + "page": 3, + "total": 10, + "list": [ + {"vod_id": 200, "vod_name": "搜索结果", "vod_pic": "", "type_id": "30", "vod_remarks": "完结"} + ], + } + result = self.spider.searchContent("繁花", False, "3") + self.assertEqual(mock_api_get.call_args.args[0], {"ac": "detail", "wd": "繁花", "pg": 3}) + self.assertEqual(result["page"], 3) + self.assertEqual(result["list"][0]["type_name"], "电视剧") + + def test_search_content_returns_empty_list_for_blank_keyword(self): + self.assertEqual(self.spider.searchContent("", False, "1"), {"page": 1, "limit": 0, "total": 0, "list": []}) + + def test_parse_play_groups_builds_repo_style_play_fields(self): + self.assertEqual( + self.spider._parse_play_groups( + { + "vod_play_from": "量子$$$", + "vod_play_url": "第1集$https://cdn.example/1.m3u8#第2集$abc$$$solo", + } + ), + { + "vod_play_from": "量子$$$线路2", + "vod_play_url": "第1集$https://cdn.example/1.m3u8#第2集$abc$$$第1集$solo", + }, + ) + + @patch.object(Spider, "_api_get") + def test_detail_content_builds_detail_and_play_fields(self, mock_api_get): + mock_api_get.return_value = { + "list": [ + { + "vod_id": 300, + "vod_name": "详情影片", + "vod_pic": "/poster.jpg", + "type_id": "30", + "type_name": "", + "vod_year": "2024", + "vod_area": "大陆", + "vod_director": "导演甲", + "vod_actor": "A&#039;B , 演员乙", + "vod_blurb": "一段简介", + "vod_remarks": "更新至2集", + "vod_play_from": "量子$$$非凡", + "vod_play_url": "第1集$https://cdn.example/1.m3u8#第2集$play-2$$$正片$play-3", + } + ] + } + result = self.spider.detailContent(["300"]) + vod = result["list"][0] + self.assertEqual(vod["vod_id"], "300") + self.assertEqual(vod["type_name"], "电视剧") + self.assertEqual(vod["vod_actor"], "A'B, 演员乙") + self.assertEqual(vod["vod_play_from"], "量子$$$非凡") + self.assertEqual( + vod["vod_play_url"], + "第1集$https://cdn.example/1.m3u8#第2集$play-2$$$正片$play-3", + ) + + @patch.object(Spider, "_api_get") + def test_category_content_returns_empty_payload_on_api_error(self, mock_api_get): + mock_api_get.side_effect = ValueError("boom") + self.assertEqual( + self.spider.categoryContent("20", "1", False, {}), + {"page": 1, "limit": 0, "total": 0, "list": []}, + ) + + @patch.object(Spider, "_api_get") + def test_detail_content_returns_empty_list_on_api_error(self, mock_api_get): + mock_api_get.side_effect = ValueError("boom") + self.assertEqual(self.spider.detailContent(["9"]), {"list": []}) + + def test_player_content_returns_direct_url_for_http_play_id(self): + result = self.spider.playerContent("量子", "https://cdn.example/test.m3u8", {}) + self.assertEqual(result["parse"], 0) + self.assertEqual(result["jx"], 0) + self.assertEqual(result["url"], "https://cdn.example/test.m3u8") + self.assertEqual(result["header"]["Referer"], "https://40000.me/") + + def test_player_content_marks_non_http_play_id_for_parse(self): + result = self.spider.playerContent("量子", "play-2", {}) + self.assertEqual(result["parse"], 1) + self.assertEqual(result["jx"], 1) + self.assertEqual(result["url"], "play-2") + + def test_player_content_returns_empty_parse_payload_for_blank_id(self): + self.assertEqual( + self.spider.playerContent("量子", "", {}), + { + "parse": 1, + "jx": 1, + "url": "", + "header": { + "User-Agent": self.spider.headers["User-Agent"], + "Referer": "https://40000.me/", + }, + }, + ) diff --git a/py/四万影视.py b/py/四万影视.py new file mode 100644 index 0000000..6baadb4 --- /dev/null +++ b/py/四万影视.py @@ -0,0 +1,295 @@ +# 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://40000.me" + self.api = self.host + "/api/maccms" + self.fallback_pic = self.host + "/public/favicon.png" + self.headers = { + "User-Agent": ( + "Mozilla/5.0 (Windows NT 10.0; Win64; x64) " + "AppleWebKit/537.36 (KHTML, like Gecko) " + "Chrome/146.0.0.0 Safari/537.36 Edg/146.0.0.0" + ), + "Referer": self.host + "/", + "Accept": "*/*", + } + self.classes = [ + {"type_id": "20", "type_name": "电影"}, + {"type_id": "30", "type_name": "电视剧"}, + {"type_id": "39", "type_name": "动漫"}, + {"type_id": "45", "type_name": "综艺"}, + {"type_id": "32", "type_name": "欧美"}, + ] + self.filter_type_options = { + "20": [ + {"n": "全部", "v": "20"}, + {"n": "动作片", "v": "21"}, + {"n": "喜剧片", "v": "22"}, + {"n": "恐怖片", "v": "23"}, + {"n": "科幻片", "v": "24"}, + {"n": "爱情片", "v": "25"}, + {"n": "剧情片", "v": "26"}, + {"n": "战争片", "v": "27"}, + {"n": "纪录片", "v": "28"}, + {"n": "理论片", "v": "29"}, + {"n": "预告片", "v": "52"}, + {"n": "电影解说", "v": "51"}, + ], + "30": [ + {"n": "全部", "v": "30"}, + {"n": "国产剧", "v": "31"}, + {"n": "欧美剧", "v": "32"}, + {"n": "香港剧", "v": "33"}, + {"n": "韩国剧", "v": "34"}, + {"n": "台湾剧", "v": "35"}, + {"n": "日本剧", "v": "36"}, + {"n": "海外剧", "v": "37"}, + {"n": "泰国剧", "v": "38"}, + {"n": "短剧大全", "v": "58"}, + ], + "39": [ + {"n": "全部", "v": "39"}, + {"n": "国产动漫", "v": "40"}, + {"n": "日韩动漫", "v": "41"}, + {"n": "欧美动漫", "v": "42"}, + {"n": "港台动漫", "v": "43"}, + {"n": "海外动漫", "v": "44"}, + {"n": "动画片", "v": "50"}, + ], + "45": [ + {"n": "全部", "v": "45"}, + {"n": "大陆综艺", "v": "46"}, + {"n": "港台综艺", "v": "47"}, + {"n": "日韩综艺", "v": "48"}, + {"n": "欧美综艺", "v": "49"}, + ], + "32": [ + {"n": "全部", "v": "32"}, + {"n": "欧美剧", "v": "32"}, + {"n": "欧美动漫", "v": "42"}, + {"n": "欧美综艺", "v": "49"}, + {"n": "海外剧", "v": "37"}, + ], + } + self.type_name_map = {item["type_id"]: item["type_name"] for item in self.classes} + self.subtype_parent_map = { + "20": "20", + "21": "20", + "22": "20", + "23": "20", + "24": "20", + "25": "20", + "26": "20", + "27": "20", + "28": "20", + "29": "20", + "51": "20", + "52": "20", + "30": "30", + "31": "30", + "32": "32", + "33": "30", + "34": "30", + "35": "30", + "36": "30", + "37": "32", + "38": "30", + "58": "30", + "39": "39", + "40": "39", + "41": "39", + "42": "32", + "43": "39", + "44": "39", + "50": "39", + "45": "45", + "46": "45", + "47": "45", + "48": "45", + "49": "32", + } + self.filters = self._build_filters() + + def init(self, extend=""): + return None + + def getName(self): + return self.name + + def _build_filters(self): + years = [{"n": "全部", "v": ""}] + for year in range(2026, 1999, -1): + years.append({"n": str(year), "v": str(year)}) + sort_values = [ + {"n": "时间", "v": "time"}, + {"n": "人气", "v": "hits"}, + {"n": "评分", "v": "score"}, + {"n": "点赞", "v": "up"}, + ] + filters = {} + for item in self.classes: + type_id = item["type_id"] + filters[type_id] = [ + {"key": "subType", "name": "分类", "init": type_id, "value": self.filter_type_options[type_id]}, + {"key": "year", "name": "年代", "init": "", "value": years}, + {"key": "sort", "name": "排序", "init": "time", "value": list(sort_values)}, + ] + return filters + + def homeContent(self, filter): + return {"class": self.classes, "filters": self.filters} + + def homeVideoContent(self): + return {"list": []} + + def _type_name_by_id(self, type_id): + raw = str(type_id or "") + parent = self.subtype_parent_map.get(raw, raw) + return self.type_name_map.get(parent, "") + + def _normalize_actor(self, value): + return str(value or "").replace("&#039;", "'").replace(" , ", ", ") + + def _normalize_vod(self, item): + return { + "vod_id": str(item.get("vod_id", "")), + "vod_name": str(item.get("vod_name", "")), + "vod_pic": str(item.get("vod_pic") or self.fallback_pic), + "vod_remarks": str(item.get("vod_remarks", "")), + "vod_year": str(item.get("vod_year", "")), + "type_name": str(item.get("type_name") or self._type_name_by_id(item.get("type_id"))), + "vod_actor": self._normalize_actor(item.get("vod_actor", "")), + } + + def _page_result(self, items, page, total=0): + return { + "page": int(page), + "limit": len(items), + "total": int(total) if total else len(items), + "list": items, + } + + def _parse_play_groups(self, item): + from_list = str(item.get("vod_play_from", "")).split("$$$") + url_list = str(item.get("vod_play_url", "")).split("$$$") + final_from = [] + final_url = [] + for index, raw_urls in enumerate(url_list): + raw_urls = str(raw_urls or "").strip() + if not raw_urls: + continue + line_name = str(from_list[index] if index < len(from_list) else "").strip() or f"线路{index + 1}" + episodes = [] + for episode_index, part in enumerate(raw_urls.split("#")): + chunks = str(part or "").split("$", 1) + if len(chunks) > 1: + title = chunks[0].strip() + play_id = chunks[1].strip() + else: + title = "" + play_id = chunks[0].strip() if chunks else "" + title = title or f"第{episode_index + 1}集" + if play_id: + episodes.append(f"{title}${play_id}") + if episodes: + final_from.append(line_name) + final_url.append("#".join(episodes)) + return {"vod_play_from": "$$$".join(final_from), "vod_play_url": "$$$".join(final_url)} + + def _api_get(self, params): + response = self.fetch( + self.api, + params=params, + headers=dict(self.headers), + timeout=10, + ) + if response.status_code != 200: + raise ValueError("api request failed") + data = json.loads(response.text or "{}") + if not isinstance(data, dict): + raise ValueError("api response is not object") + return data + + def categoryContent(self, tid, pg, filter, extend): + page = max(int(pg), 1) + values = dict(extend or {}) + params = { + "ac": "detail", + "t": str(values.get("subType") or tid), + "pg": page, + "by": str(values.get("sort") or "time"), + } + year = str(values.get("year") or "").strip() + if year: + params["h"] = year + try: + data = self._api_get(params) + except ValueError: + return {"page": page, "limit": 0, "total": 0, "list": []} + items = [self._normalize_vod(item) for item in data.get("list", [])] + return self._page_result(items, data.get("page", page), data.get("total", 0)) + + def searchContent(self, key, quick, pg="1"): + page = max(int(pg), 1) + keyword = str(key or "").strip() + if not keyword: + return {"page": page, "limit": 0, "total": 0, "list": []} + try: + data = self._api_get({"ac": "detail", "wd": keyword, "pg": page}) + except ValueError: + return {"page": page, "limit": 0, "total": 0, "list": []} + items = [self._normalize_vod(item) for item in data.get("list", [])] + return self._page_result(items, data.get("page", page), data.get("total", 0)) + + def detailContent(self, ids): + vod_id = str(ids[0] if isinstance(ids, list) and ids else ids or "").strip() + if not vod_id: + return {"list": []} + try: + data = self._api_get({"ac": "detail", "ids": vod_id}) + except ValueError: + return {"list": []} + item = (data.get("list") or [None])[0] + if not item: + return {"list": []} + normalized = self._normalize_vod(item) + play_data = self._parse_play_groups(item) + return { + "list": [ + { + "vod_id": normalized["vod_id"], + "vod_name": normalized["vod_name"], + "vod_pic": normalized["vod_pic"], + "type_name": normalized["type_name"], + "vod_year": str(item.get("vod_year", "")), + "vod_area": str(item.get("vod_area", "")), + "vod_director": str(item.get("vod_director", "")), + "vod_actor": normalized["vod_actor"], + "vod_content": str(item.get("vod_blurb") or item.get("vod_content") or ""), + "vod_remarks": str(item.get("vod_remarks", "")), + "vod_play_from": play_data["vod_play_from"], + "vod_play_url": play_data["vod_play_url"], + } + ] + } + + def playerContent(self, flag, id, vipFlags): + play_id = str(id or "").strip() + header = { + "User-Agent": self.headers["User-Agent"], + "Referer": self.host + "/", + } + if not play_id: + return {"parse": 1, "jx": 1, "url": "", "header": header} + if play_id.startswith("http://") or play_id.startswith("https://"): + return {"parse": 0, "jx": 0, "url": play_id, "header": header} + return {"parse": 1, "jx": 1, "url": play_id, "header": header}