yeyinly/tvboxzt
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

酷我听书.py

Browse Source
This commit is contained in:
Harold
2026-04-23 19:23:20 +08:00
parent bc7471737a
commit 4e1d725f08
2 changed files with 0 additions and 463 deletions
Download Patch File Download Diff File Expand all files Collapse all files
py/tests/test_即看影视.py
-208
View File
@@ -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()
py/即看影视.py
-255
View File
@@ -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(),
}