feat: add jikanyingshi home flows

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()

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"],
+        }