diff --git a/py/tests/test_黑猫APP.py b/py/tests/test_黑猫APP.py index cd81e00..18edcec 100644 --- a/py/tests/test_黑猫APP.py +++ b/py/tests/test_黑猫APP.py @@ -1,6 +1,8 @@ import unittest +from datetime import datetime from importlib.machinery import SourceFileLoader from pathlib import Path +from unittest.mock import patch ROOT = Path(__file__).resolve().parents[1] @@ -31,6 +33,104 @@ class TestHeiMaoAppSpider(unittest.TestCase): def test_replace_code_normalizes_common_ocr_misreads(self): self.assertEqual(self.spider._replace_code("5y6口"), "5960") + @patch.object(Spider, "_api_post") + def test_home_content_filters_categories_merges_area_and_inserts_current_year(self, mock_api_post): + mock_api_post.return_value = { + "type_list": [ + { + "type_id": "6", + "type_name": "伦理", + "recommend_list": [], + "filter_type_list": [], + }, + { + "type_id": "3", + "type_name": "综艺", + "recommend_list": [{"vod_id": "z1"}], + "filter_type_list": [ + {"name": "area", "list": ["全部", "中国大陆", "香港"]}, + {"name": "year", "list": ["全部", "2025", "2024"]}, + {"name": "sort", "list": ["最新", "最热"]}, + ], + }, + { + "type_id": "1", + "type_name": "电影", + "recommend_list": [{"vod_id": "m1"}], + "filter_type_list": [], + }, + ], + "config": {}, + } + content = self.spider.homeContent(False) + self.assertEqual([item["type_name"] for item in content["class"]], ["电影", "综艺"]) + self.assertEqual(content["list"], [{"vod_id": "z1"}, {"vod_id": "m1"}]) + area_values = content["filters"]["3"][0]["value"] + year_values = content["filters"]["3"][1]["value"] + self.assertEqual(area_values[1], {"n": "大陆", "v": "大陆"}) + self.assertIn({"n": str(datetime.now().year), "v": str(datetime.now().year)}, year_values) + self.assertEqual(content["filters"]["3"][2]["key"], "by") + + @patch.object(Spider, "_api_post") + def test_category_content_maps_by_to_sort_and_omits_pagecount(self, mock_api_post): + mock_api_post.return_value = { + "recommend_list": [{"vod_id": "movie-1", "vod_name": "分类片"}] + } + result = self.spider.categoryContent( + "1", + "2", + False, + {"area": "香港", "year": "2025", "by": "最热", "lang": "粤语", "class": "动作"}, + ) + self.assertEqual( + mock_api_post.call_args.args, + ( + "typeFilterVodList", + { + "type_id": "1", + "page": "2", + "area": "香港", + "year": "2025", + "sort": "最热", + "lang": "粤语", + "class": "动作", + }, + ), + ) + self.assertEqual(result["page"], 2) + self.assertEqual(result["list"][0]["vod_id"], "movie-1") + self.assertNotIn("pagecount", result) + + @patch.object(Spider, "_api_post") + def test_category_content_merges_mainland_results_and_deduplicates(self, mock_api_post): + mock_api_post.side_effect = [ + {"recommend_list": [{"vod_id": "a"}, {"vod_id": "b"}]}, + {"recommend_list": [{"vod_id": "b"}, {"vod_id": "c"}]}, + {"recommend_list": [{"vod_id": "c"}, {"vod_id": "d"}]}, + ] + result = self.spider.categoryContent("1", "1", False, {"area": "大陆"}) + self.assertEqual([item["vod_id"] for item in result["list"]], ["a", "b", "c", "d"]) + self.assertEqual( + [call.args[1]["area"] for call in mock_api_post.call_args_list], + ["中国大陆", "大陆", "内地"], + ) + + @patch.object(Spider, "_api_post") + def test_search_content_filters_ethics_and_non_matching_items(self, mock_api_post): + mock_api_post.return_value = { + "search_list": [ + {"vod_id": "1", "vod_name": "繁花", "vod_pic": "p1", "vod_year": "2024", "vod_class": "剧情"}, + {"vod_id": "2", "vod_name": "别的片", "vod_pic": "p2", "vod_year": "2024", "vod_class": "伦理"}, + {"vod_id": "3", "vod_name": "完全无关", "vod_pic": "p3", "vod_year": "2023", "vod_class": "动作"}, + ] + } + result = self.spider.searchContent("繁花", False, "1") + self.assertEqual( + result["list"], + [{"vod_id": "1", "vod_name": "繁花", "vod_pic": "p1", "vod_remarks": "2024 剧情"}], + ) + self.assertNotIn("pagecount", result) + if __name__ == "__main__": unittest.main() diff --git a/py/黑猫APP.py b/py/黑猫APP.py index 11a71b5..50d3e33 100644 --- a/py/黑猫APP.py +++ b/py/黑猫APP.py @@ -1,6 +1,8 @@ # coding=utf-8 import base64 +import json import sys +from datetime import datetime from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes from cryptography.hazmat.primitives.padding import PKCS7 @@ -18,6 +20,16 @@ class Spider(BaseSpider): self.user_agent = "okhttp/3.10.0" self.aes_key = "VwsHxkCViDXEExWa" self.aes_iv = "VwsHxkCViDXEExWa" + self.category_config = { + "blockedNames": ["伦理"], + "renameMap": {}, + "forceOrder": ["电影", "连续剧", "综艺", "动漫", "短剧", "直播"], + } + self.area_merge_config = { + "enabled": True, + "displayName": "大陆", + "mergeList": ["中国大陆", "大陆", "内地"], + } def init(self, extend=""): self.search_api = "searchList" @@ -28,6 +40,12 @@ class Spider(BaseSpider): def getName(self): return self.name + def _headers(self): + return { + "User-Agent": self.user_agent, + "Accept-Encoding": "gzip", + } + def homeVideoContent(self): return {"list": []} @@ -35,6 +53,20 @@ class Spider(BaseSpider): path = str(endpoint or "").lstrip("/") return f"{self.host}{self.api_path}.index/{path}" + def _api_post(self, endpoint, payload=None): + data = {} if payload is None else dict(payload) + response = self.post( + self._build_api_url(endpoint), + data=data, + headers=self._headers(), + timeout=10, + verify=False, + ) + body = json.loads(response.text or "{}") + if not body.get("data"): + return None + return json.loads(self._aes_decrypt(body["data"])) + def _aes_encrypt(self, text): padder = PKCS7(128).padder() data = padder.update(str(text).encode("utf-8")) + padder.finalize() @@ -65,3 +97,142 @@ class Spider(BaseSpider): "五": "5", } return "".join(replacements.get(char, char) for char in str(text or "")) + + def _process_classes(self, type_list): + blocked = set(self.category_config.get("blockedNames", [])) + rename_map = self.category_config.get("renameMap", {}) + order_map = {name: index for index, name in enumerate(self.category_config.get("forceOrder", []))} + classes = [ + { + "type_id": item.get("type_id", ""), + "type_name": rename_map.get(item.get("type_name", ""), item.get("type_name", "")), + } + for item in type_list + if item.get("type_name") not in blocked + ] + classes.sort(key=lambda item: order_map.get(item["type_name"], 999)) + return classes + + def _merge_area_filter(self, values): + merge_values = set(self.area_merge_config.get("mergeList", [])) + raw_values = list(values) + if not self.area_merge_config.get("enabled") or not any(value in merge_values for value in raw_values): + return raw_values + result = [value for value in raw_values if value not in merge_values] + index = result.index("全部") + 1 if "全部" in result else 0 + result.insert(index, self.area_merge_config.get("displayName", "大陆")) + return result + + def _convert_filters(self, type_list): + name_map = {"class": "类型", "area": "地区", "lang": "语言", "year": "年份", "sort": "排序"} + current_year = str(datetime.now().year) + filters = {} + for item in type_list: + entries = [] + for config in item.get("filter_type_list", []): + raw_name = config.get("name", "") + values = list(config.get("list", [])) + if raw_name == "area": + values = self._merge_area_filter(values) + if raw_name == "year" and current_year not in values: + index = values.index("全部") + 1 if "全部" in values else 0 + values.insert(index, current_year) + entries.append( + { + "key": "by" if raw_name == "sort" else raw_name, + "name": name_map.get(raw_name, raw_name), + "value": [{"n": value, "v": value} for value in values], + } + ) + filters[item.get("type_id", "")] = entries + return filters + + def homeContent(self, filter): + init_data = self._api_post(self.init_api) or {"type_list": [], "config": {}} + self.search_verify = bool(init_data.get("config", {}).get("system_search_verify_status")) + return { + "class": self._process_classes(init_data.get("type_list", [])), + "filters": self._convert_filters(init_data.get("type_list", [])), + "list": [vod for item in init_data.get("type_list", []) for vod in item.get("recommend_list", [])], + } + + def _page_result(self, items, pg, limit=90, total=None): + page = int(pg) + return { + "page": page, + "limit": limit, + "total": len(items) if total is None else total, + "list": items, + } + + def _merge_area_search(self, tid, pg, extend): + seen = set() + merged = [] + for area in self.area_merge_config.get("mergeList", []): + payload = { + "type_id": str(tid), + "page": str(pg), + "area": area, + "year": extend.get("year", "全部"), + "sort": extend.get("by", "最新"), + "lang": extend.get("lang", "全部"), + "class": extend.get("class", "全部"), + } + result = self._api_post("typeFilterVodList", payload) or {} + for item in result.get("recommend_list", []): + vod_id = item.get("vod_id") + if vod_id in seen: + continue + seen.add(vod_id) + merged.append(item) + return merged + + def categoryContent(self, tid, pg, filter, extend): + options = dict(extend or {}) + if options.get("area") == self.area_merge_config.get("displayName"): + return self._page_result(self._merge_area_search(tid, pg, options), pg) + payload = { + "type_id": str(tid), + "page": str(pg), + "area": options.get("area", "全部"), + "year": options.get("year", "全部"), + "sort": options.get("by", "最新"), + "lang": options.get("lang", "全部"), + "class": options.get("class", "全部"), + } + result = self._api_post("typeFilterVodList", payload) or {} + return self._page_result(result.get("recommend_list", []), pg) + + def searchContent(self, key, quick, pg="1"): + payload = { + "keywords": str(key or ""), + "type_id": "0", + "page": str(pg), + } + result = self._api_post(self.search_api, payload) or {} + keyword = str(key or "").strip().lower() + items = [] + for item in result.get("search_list", []): + vod_class = str(item.get("vod_class", "")) + text = " ".join( + [ + str(item.get("vod_name", "")), + str(item.get("vod_remarks", "")), + vod_class, + ] + ).lower() + if "伦理" in vod_class: + continue + if keyword and keyword not in text: + continue + items.append( + { + "vod_id": item.get("vod_id", ""), + "vod_name": item.get("vod_name", ""), + "vod_pic": item.get("vod_pic", ""), + "vod_remarks": " ".join( + value for value in [str(item.get("vod_year", "")), vod_class] if value + ).strip(), + } + ) + return self._page_result(items, pg)