From a1d170adb5cba95af30685c6151474aef5e461f4 Mon Sep 17 00:00:00 2001 From: Harold <8866033@gmail.com> Date: Thu, 23 Apr 2026 19:09:42 +0800 Subject: [PATCH] feat: add jikanyingshi home flows --- py/tests/test_即看影视.py | 64 ++++++++++++++++++++++ py/即看影视.py | 108 ++++++++++++++++++++++++++++++++++++++ 2 files changed, 172 insertions(+) diff --git a/py/tests/test_即看影视.py b/py/tests/test_即看影视.py index 0bfef33..d344e3c 100644 --- a/py/tests/test_即看影视.py +++ b/py/tests/test_即看影视.py @@ -47,6 +47,70 @@ class TestJiKanYingShiSpider(unittest.TestCase): decoded = base64.b64decode(result.encode("utf-8")).decode("utf-8") self.assertTrue(decoded) + @patch.object(Spider, "fetch") + @patch.object(Spider, "post") + def test_ensure_ready_fetches_host_token_and_config_once(self, mock_post, mock_fetch): + mock_fetch.side_effect = [ + SimpleNamespace(text="https://appsky2025999.ideasz.net\n"), + SimpleNamespace(text=build_sk_payload('{"direct_link":[]}', self.spider.aes_key, self.spider.aes_iv)), + ] + mock_post.return_value = SimpleNamespace( + text=build_sk_payload("token-123", self.spider.aes_key, self.spider.aes_iv) + ) + + self.spider._ensure_ready() + self.spider._ensure_ready() + + self.assertEqual(self.spider.api_host, "https://appsky2025999.ideasz.net") + self.assertEqual(self.spider.auth_token, "token-123") + self.assertEqual(self.spider.config_data, {"direct_link": []}) + self.assertEqual(mock_post.call_count, 1) + + @patch.object(Spider, "_fetch_filters") + @patch.object(Spider, "_fetch_api") + def test_home_content_returns_classes_filters_and_home_list(self, mock_fetch_api, mock_fetch_filters): + mock_fetch_api.side_effect = [ + {"data": [{"type_id": "1", "type_name": "电影"}, {"type_id": "2", "type_name": "剧集"}]}, + {"data": [{"vod_id": "v1", "vod_name": "首页推荐"}]}, + ] + mock_fetch_filters.return_value = { + "1": [{"key": "year", "name": "年份", "value": [{"n": "全部", "v": ""}]}] + } + + content = self.spider.homeContent(False) + + self.assertEqual( + content["class"], + [{"type_id": "1", "type_name": "电影"}, {"type_id": "2", "type_name": "剧集"}], + ) + self.assertEqual(content["filters"]["1"][0]["key"], "year") + self.assertEqual(content["list"], [{"vod_id": "v1", "vod_name": "首页推荐"}]) + + @patch.object(Spider, "_fetch_api") + def test_home_video_content_requests_randomlikeindex(self, mock_fetch_api): + mock_fetch_api.return_value = {"data": [{"vod_id": "x1", "vod_name": "推荐片"}]} + + result = self.spider.homeVideoContent() + + self.assertEqual( + mock_fetch_api.call_args.args, + ("/sk-api/vod/list", {"page": 1, "limit": 12, "type": "randomlikeindex"}), + ) + self.assertEqual(result["list"][0]["vod_id"], "x1") + + @patch.object(Spider, "_fetch_api") + def test_fetch_filters_maps_extend_fields_and_adds_sort(self, mock_fetch_api): + mock_fetch_api.return_value = { + "code": 200, + "data": {"extendtype": "动作,喜剧", "area": "大陆,香港", "lang": "国语", "year": "2025,2024"}, + } + + filters = self.spider._fetch_filters([{"type_id": "1", "type_name": "电影"}]) + + self.assertEqual(filters["1"][0]["key"], "class") + self.assertEqual(filters["1"][0]["value"][1], {"n": "动作", "v": "动作"}) + self.assertEqual(filters["1"][-1]["value"][-1], {"n": "评分", "v": "score"}) + if __name__ == "__main__": unittest.main() diff --git a/py/即看影视.py b/py/即看影视.py index d1b52cf..b15d32c 100644 --- a/py/即看影视.py +++ b/py/即看影视.py @@ -1,7 +1,10 @@ # coding=utf-8 import base64 import hashlib +import json import sys +import time +from urllib.parse import quote_plus try: from Crypto.Cipher import AES @@ -61,3 +64,108 @@ class Spider(BaseSpider): cipher = AES.new(self.ck_key.encode("utf-8"), AES.MODE_CBC, self.ck_iv.encode("utf-8")) encrypted = cipher.encrypt(pad(second, AES.block_size)).hex() return base64.b64encode(encrypted.encode("utf-8")).decode("utf-8") + + def _safe_json(self, text, default=None): + try: + return json.loads(text) + except (TypeError, json.JSONDecodeError): + return {} if default is None else default + + def _fetch_auth_token(self): + sign = self._md5(self.aes_key + self.aes_iv) + ck = self._ck_encrypt(f"{self.api_host}##5483##{int(time.time() * 1000)}##ckzmbc") + response = self.post( + f"{self.api_host}/get_config", + json={"sign": sign, "ck": ck}, + headers=dict(self.headers, **{"Content-Type": "application/json"}), + timeout=15, + verify=False, + ) + self.auth_token = self._sk_decrypt(getattr(response, "text", "")) + + def _fetch_app_config(self): + response = self.fetch( + f"{self.api_host}/app/config", + headers=dict(self.headers, **{"Authorization": f"Bearer {self.auth_token}"}), + timeout=15, + verify=False, + ) + self.config_data = self._safe_json(self._sk_decrypt(getattr(response, "text", "")), default={}) + + def _ensure_ready(self): + if self.is_ready: + return + self.api_host = self._resolve_api_host() + if len(self.aes_key) != 16 or len(self.aes_iv) != 16: + raise ValueError("invalid aes config") + self._fetch_auth_token() + self._fetch_app_config() + self.is_ready = True + + def _fetch_api(self, path, params=None): + self._ensure_ready() + query = [] + for key, value in (params or {}).items(): + if value in (None, ""): + continue + query.append(f"{quote_plus(str(key))}={quote_plus(str(value))}") + url = f"{self.api_host}{path}" + if query: + url += "?" + "&".join(query) + response = self.fetch( + url, + headers=dict(self.headers, **{"Authorization": f"Bearer {self.auth_token}"}), + timeout=15, + verify=False, + ) + return self._safe_json(self._sk_decrypt(getattr(response, "text", "")), default={}) + + def _fetch_filters(self, classes): + result = {} + mappings = { + "extendtype": ("class", "类型"), + "area": ("area", "地区"), + "lang": ("lang", "语言"), + "year": ("year", "年份"), + } + for item in classes or []: + payload = self._fetch_api("/sk-api/type/alltypeextend", {"typeId": item.get("type_id")}) + data = payload.get("data") or {} + entries = [] + for source_key, (target_key, label) in mappings.items(): + raw = str(data.get(source_key) or "").strip() + if not raw: + continue + values = [{"n": "全部", "v": ""}] + values.extend({"n": value.strip(), "v": value.strip()} for value in raw.split(",") if value.strip()) + entries.append({"key": target_key, "name": label, "init": "", "value": values}) + entries.append( + { + "key": "sort", + "name": "排序", + "init": "updateTime", + "value": [ + {"n": "最新", "v": "updateTime"}, + {"n": "人气", "v": "hot"}, + {"n": "评分", "v": "score"}, + ], + } + ) + result[str(item.get("type_id"))] = entries + return result + + def homeVideoContent(self): + data = self._fetch_api("/sk-api/vod/list", {"page": 1, "limit": 12, "type": "randomlikeindex"}) + return {"list": data.get("data") or []} + + def homeContent(self, filter): + home = self._fetch_api("/sk-api/type/list", {}) + classes = [] + for item in home.get("data") or []: + if isinstance(item, dict) and item.get("type_id"): + classes.append({"type_id": item.get("type_id"), "type_name": item.get("type_name", "")}) + return { + "class": classes, + "filters": self._fetch_filters(classes), + "list": self.homeVideoContent()["list"], + }