From 4e1d725f08242f20cb5f86a7e57d4dd1b8e2009c Mon Sep 17 00:00:00 2001 From: Harold <8866033@gmail.com> Date: Thu, 23 Apr 2026 19:23:20 +0800 Subject: [PATCH] =?UTF-8?q?=E9=85=B7=E6=88=91=E5=90=AC=E4=B9=A6.py?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- py/tests/test_即看影视.py | 208 ------------------------------- py/即看影视.py | 255 -------------------------------------- 2 files changed, 463 deletions(-) delete mode 100644 py/tests/test_即看影视.py delete mode 100644 py/即看影视.py diff --git a/py/tests/test_即看影视.py b/py/tests/test_即看影视.py deleted file mode 100644 index 43005bb..0000000 --- a/py/tests/test_即看影视.py +++ /dev/null @@ -1,208 +0,0 @@ -import base64 -import unittest -from importlib.machinery import SourceFileLoader -from pathlib import Path -from types import SimpleNamespace -from unittest.mock import patch - -try: - from Crypto.Cipher import AES - from Crypto.Util.Padding import pad -except ModuleNotFoundError: - from Cryptodome.Cipher import AES - from Cryptodome.Util.Padding import pad - - -ROOT = Path(__file__).resolve().parents[1] -MODULE = SourceFileLoader("jikanyingshi_spider", str(ROOT / "即看影视.py")).load_module() -Spider = MODULE.Spider - - -def build_sk_payload(value, key, iv): - cipher = AES.new(key.encode("utf-8"), AES.MODE_CBC, iv.encode("utf-8")) - encrypted = cipher.encrypt(pad(value.encode("utf-8"), AES.block_size)).hex() - return "FROMSKZZJM" + encrypted - - -class TestJiKanYingShiSpider(unittest.TestCase): - def setUp(self): - Spider._instance = None - self.spider = Spider() - self.spider.init() - - @patch.object(Spider, "fetch") - def test_resolve_api_host_reads_appipr_txt(self, mock_fetch): - mock_fetch.return_value = SimpleNamespace(text="https://appsky2025999.ideasz.net\n") - self.assertEqual(self.spider._resolve_api_host(), "https://appsky2025999.ideasz.net") - - def test_sk_decrypt_returns_plain_text_without_prefix(self): - self.assertEqual(self.spider._sk_decrypt("plain-text"), "plain-text") - - def test_sk_decrypt_decodes_prefixed_ciphertext(self): - encrypted = build_sk_payload('{"msg":"ok"}', self.spider.aes_key, self.spider.aes_iv) - self.assertEqual(self.spider._sk_decrypt(encrypted), '{"msg":"ok"}') - - def test_ck_encrypt_returns_base64_ascii(self): - result = self.spider._ck_encrypt("https://host##5483##1700000000000##ckzmbc") - 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"}) - - @patch.object(Spider, "_fetch_api") - def test_category_content_maps_extend_to_sk_api_params(self, mock_fetch_api): - mock_fetch_api.return_value = {"data": [{"vod_id": "c1", "vod_name": "分类片"}]} - - result = self.spider.categoryContent( - "2", - "3", - False, - {"sort": "hot", "area": "香港", "lang": "粤语", "year": "2025", "class": "动作"}, - ) - - self.assertEqual( - mock_fetch_api.call_args.args, - ( - "/sk-api/vod/list", - { - "typeId": "2", - "page": 3, - "limit": 18, - "type": "hot", - "area": "香港", - "lang": "粤语", - "year": "2025", - "mtype": "", - "extendtype": "动作", - }, - ), - ) - self.assertEqual(result["page"], 3) - self.assertNotIn("pagecount", result) - - @patch.object(Spider, "_fetch_api") - def test_search_content_returns_standard_page_payload(self, mock_fetch_api): - mock_fetch_api.return_value = {"data": [{"vod_id": "s1", "vod_name": "繁花"}]} - - result = self.spider.searchContent("繁花", False, "4") - - self.assertEqual( - mock_fetch_api.call_args.args, - ("/sk-api/search/pages", {"keyword": "繁花", "page": 4, "limit": 10, "typeId": -1}), - ) - self.assertEqual(result["page"], 4) - self.assertEqual(result["list"][0]["vod_id"], "s1") - self.assertNotIn("pagecount", result) - - @patch.object(Spider, "_fetch_api") - def test_category_content_returns_empty_list_on_api_error(self, mock_fetch_api): - mock_fetch_api.side_effect = RuntimeError("boom") - - result = self.spider.categoryContent("1", "1", False, {}) - - self.assertEqual(result["list"], []) - self.assertEqual(result["total"], 0) - - @patch.object(Spider, "_fetch_api") - def test_detail_content_wraps_single_vod_in_list(self, mock_fetch_api): - mock_fetch_api.return_value = {"data": {"vod_id": "d1", "vod_name": "详情片"}} - - result = self.spider.detailContent(["d1"]) - - self.assertEqual(mock_fetch_api.call_args.args, ("/sk-api/vod/one", {"vodId": "d1"})) - self.assertEqual(result, {"list": [{"vod_id": "d1", "vod_name": "详情片"}]}) - - @patch.object(Spider, "_fetch_api") - def test_player_content_prefers_skjson_url(self, mock_fetch_api): - mock_fetch_api.return_value = {"data": {"url": "https://media.example/final.m3u8"}} - - result = self.spider.playerContent("", "https://wrapped.example/play", "") - - self.assertEqual( - mock_fetch_api.call_args.args, - ("/sk-api/vod/skjson", {"url": "https://wrapped.example/play", "skjsonindex": 0}), - ) - self.assertEqual(result["url"], "https://media.example/final.m3u8") - self.assertEqual(result["jx"], 0) - - @patch.object(Spider, "_fetch_api") - def test_player_content_falls_back_to_original_id_and_sets_jx_for_video_sites(self, mock_fetch_api): - mock_fetch_api.side_effect = RuntimeError("parse failed") - - result = self.spider.playerContent("", "https://v.qq.com/x/cover/abcd", "") - - self.assertEqual(result["url"], "https://v.qq.com/x/cover/abcd") - self.assertEqual(result["jx"], 1) - self.assertEqual(result["parse"], 0) - - @patch.object(Spider, "_fetch_api") - def test_detail_content_returns_empty_list_when_api_has_no_data(self, mock_fetch_api): - mock_fetch_api.return_value = {"data": None} - - self.assertEqual(self.spider.detailContent(["404"]), {"list": []}) - - -if __name__ == "__main__": - unittest.main() diff --git a/py/即看影视.py b/py/即看影视.py deleted file mode 100644 index fd598ca..0000000 --- a/py/即看影视.py +++ /dev/null @@ -1,255 +0,0 @@ -# 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 - from Crypto.Util.Padding import pad, unpad -except ModuleNotFoundError: - from Cryptodome.Cipher import AES - from Cryptodome.Util.Padding import pad, unpad - -from base.spider import Spider as BaseSpider - -sys.path.append("..") - - -class Spider(BaseSpider): - def __init__(self): - self.name = "即看影视" - self.host_config_url = "https://skyappdata-1321528676.cos.accelerate.myqcloud.com/4kapp/appipr.txt" - self.aes_key = "ygcnbckhcuvygdyb" - self.aes_iv = "4023892775143708" - self.ck_key = "ygcnbcrvaervztmw" - self.ck_iv = "1212164105143708" - self.user_agent = "Dart/2.10 (dart:io)" - self.headers = { - "User-Agent": self.user_agent, - "Accept-Encoding": "gzip", - } - self.api_host = "" - self.auth_token = "" - self.config_data = {} - self.is_ready = False - - def init(self, extend=""): - return None - - def getName(self): - return self.name - - def _resolve_api_host(self): - response = self.fetch(self.host_config_url, headers=self.headers, timeout=10, verify=False) - return str(getattr(response, "text", "")).strip().rstrip("/") - - def _md5(self, text): - return hashlib.md5(str(text).encode("utf-8")).hexdigest() - - def _sk_decrypt(self, text): - value = str(text or "") - prefix = "FROMSKZZJM" - if not value.startswith(prefix): - return value - encrypted_hex = value[len(prefix):] - cipher = AES.new(self.aes_key.encode("utf-8"), AES.MODE_CBC, self.aes_iv.encode("utf-8")) - return unpad(cipher.decrypt(bytes.fromhex(encrypted_hex)), AES.block_size).decode("utf-8") - - def _ck_encrypt(self, text): - first = base64.b64encode(str(text).encode("utf-8")) - second = base64.b64encode(first) - 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"], - } - - def _page_result(self, items, pg, limit): - values = list(items or []) - return { - "page": int(pg), - "limit": limit, - "total": len(values), - "list": values, - } - - def categoryContent(self, tid, pg, filter, extend): - options = dict(extend or {}) - try: - data = self._fetch_api( - "/sk-api/vod/list", - { - "typeId": str(tid), - "page": int(pg), - "limit": 18, - "type": options.get("sort", "updateTime"), - "area": options.get("area", ""), - "lang": options.get("lang", ""), - "year": options.get("year", ""), - "mtype": "", - "extendtype": options.get("class", ""), - }, - ) - return self._page_result(data.get("data") or [], pg, 18) - except Exception: - return self._page_result([], pg, 18) - - def searchContent(self, key, quick, pg="1"): - try: - data = self._fetch_api( - "/sk-api/search/pages", - {"keyword": str(key or ""), "page": int(pg), "limit": 10, "typeId": -1}, - ) - return self._page_result(data.get("data") or [], pg, 10) - except Exception: - return self._page_result([], pg, 10) - - def detailContent(self, ids): - vod_id = ids[0] if isinstance(ids, (list, tuple)) else ids - try: - data = self._fetch_api("/sk-api/vod/one", {"vodId": vod_id}) - except Exception: - return {"list": []} - vod = data.get("data") - return {"list": [vod]} if isinstance(vod, dict) else {"list": []} - - def _player_headers(self): - return { - "User-Agent": ( - "Mozilla/5.0 (iPhone; CPU iPhone OS 13_2_3 like Mac OS X) " - "AppleWebKit/605.1.15 (KHTML, like Gecko) Version/13.0.3 " - "Mobile/15E148 Safari/604.1" - ) - } - - def playerContent(self, flag, id, vipFlags): - try: - data = self._fetch_api("/sk-api/vod/skjson", {"url": id, "skjsonindex": 0}) - play_url = str((data.get("data") or {}).get("url") or "") - if play_url.startswith("http"): - return { - "parse": 0, - "jx": 0, - "url": play_url, - "header": self._player_headers(), - } - except Exception: - pass - - target = str(id or "") - jx = 1 if any( - domain in target.lower() - for domain in ["iqiyi.com", "qq.com", "youku.com", "mgtv.com", "bilibili.com"] - ) else 0 - return { - "parse": 0, - "jx": jx, - "url": target, - "header": self._player_headers(), - }