diff --git a/py/tests/test_四万影视.py b/py/tests/test_四万影视.py index b551bf8..5f6f775 100644 --- a/py/tests/test_四万影视.py +++ b/py/tests/test_四万影视.py @@ -56,3 +56,110 @@ class TestSiWanSpider(unittest.TestCase): 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", + ) diff --git a/py/四万影视.py b/py/四万影视.py index 217f9cc..2c615c4 100644 --- a/py/四万影视.py +++ b/py/四万影视.py @@ -151,6 +151,60 @@ class Spider(BaseSpider): 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, @@ -164,3 +218,66 @@ class Spider(BaseSpider): 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"], + } + ] + }