diff --git a/py/tests/test_乌云影视.py b/py/tests/test_乌云影视.py index 86a5a68..95c3046 100644 --- a/py/tests/test_乌云影视.py +++ b/py/tests/test_乌云影视.py @@ -1,6 +1,7 @@ import unittest from importlib.machinery import SourceFileLoader from pathlib import Path +from unittest.mock import patch ROOT = Path(__file__).resolve().parents[1] @@ -108,6 +109,90 @@ class TestWooyunSpider(unittest.TestCase): self.assertEqual(body["sortCode"], "hot") self.assertEqual(body["topCode"], "movie") + 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") + if __name__ == "__main__": unittest.main() diff --git a/py/乌云影视.py b/py/乌云影视.py index 5d38b76..084f8e2 100644 --- a/py/乌云影视.py +++ b/py/乌云影视.py @@ -1,5 +1,6 @@ # coding=utf-8 import json +import math import sys from base.spider import Spider as BaseSpider @@ -184,7 +185,90 @@ class Spider(BaseSpider): "topCode": self._map_top_code(category_id), } + 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): + if isinstance(home_blocks, dict): + blocks = self._ensure_list(home_blocks.get("records")) + else: + blocks = self._ensure_list(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): + media_type = item.get("mediaType") or {} + 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(media_type.get("code")), + "type_name": self._stringify(media_type.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 homeContent(self, filter): menu = self._request_json("/movie/category/menu") classes = self._build_classes(menu) return {"class": classes, "filters": self._build_filters(menu, classes)} + + def homeVideoContent(self): + data = self._request_json(f"/movie/media/home/custom/classify/1/3?limit={self.home_limit}") + items = self._extract_home_list(data)[: self.home_limit] + return {"list": [self._map_vod(item) for item in items]} + + 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}