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

docs: add jikanyingshi spider design

Browse Source
This commit is contained in:
Harold
2026-04-23 18:55:13 +08:00
parent 3332a470ba
commit 86b6e3a0db
3 changed files with 263 additions and 493 deletions
Download Patch File Download Diff File Expand all files Collapse all files
py/docs/superpowers/specs/2026-04-23-jikanyingshi-spider-design.md
+263
View File
@@ -0,0 +1,263 @@
# 即看影视 Python 蜘蛛设计
## 目标
在当前 Python 仓库中新增一个符合 `base.spider.Spider` 接口的即看影视蜘蛛，覆盖以下能力：
- 首页分类与筛选
- 首页推荐列表
- 分类列表
- 搜索
- 详情解析
- 播放解析
实现以用户提供的 Node/JS 版本为行为参考，但落地形式遵循当前仓库的单文件 Spider 约定。
## 范围
本次实现包含：
- 新增独立脚本，文件名为 `即看影视.py`
- 使用即看影视的 `SKApp` 加密协议
- 启动时通过 `https://skyappdata-1321528676.cos.accelerate.myqcloud.com/4kapp/appipr.txt` 获取真实 `API_HOST`
- 通过 `/get_config` 获取授权 token
- 通过 `/app/config` 获取应用配置
- 支持 `home/homeVideo/category/search/detail/player` 全链路
- 为新增行为补齐 `unittest`
本次实现不包含：
- 参考代码中的 HTTP 路由导出层
- 公共 `SKApp` helper 抽象
- 动态播放规则配置
- 重试、缓存持久化和多域名探活
## 方案选择
候选方案有三种：
1. 单文件简化版
2. 单文件加初始化缓存
3. 抽公共 `SKApp` helper
选用方案 2。
原因：
- 保持与仓库现有蜘蛛的一致结构
- 避免在一次 Spider 生命周期内重复拉取 `appipr.txt`、`/get_config` 和 `/app/config`
- 不扩大本次任务范围到公共层改造
## 模块边界
新增脚本只在站点文件内部维护逻辑，不修改 `base/`。
文件内部按职责拆分为以下私有 helper：
- `_ensure_ready`
- 负责初始化 `API_HOST`、加密参数、token 和应用配置
- `_resolve_api_host`
- 请求 `appipr.txt` 并提取真实 host
- `_fetch_auth_token`
- 调用 `/get_config` 获取 Bearer token
- `_fetch_app_config`
- 调用 `/app/config` 获取播放相关配置
- `_sk_decrypt`
- 处理 `FROMSKZZJM` 前缀响应
- `_ck_encrypt`
- 生成 `/get_config` 所需的 `ck`
- `_fetch_api`
- 统一拼接查询参数、附加认证头、发起请求和解密 JSON
- `_fetch_filters`
- 逐分类请求 `/sk-api/type/alltypeextend` 并转换筛选结构
Spider 对外方法：
- `init`
- `homeContent`
- `homeVideoContent`
- `categoryContent`
- `searchContent`
- `detailContent`
- `playerContent`
## 初始化与鉴权设计
固定配置保留在站点文件内部：
- `host_config_url`
- `https://skyappdata-1321528676.cos.accelerate.myqcloud.com/4kapp/appipr.txt`
- `aes_key`
- `ygcnbckhcuvygdyb`
- `aes_iv`
- `4023892775143708`
- `ck_key`
- `ygcnbcrvaervztmw`
- `ck_iv`
- `1212164105143708`
- `user_agent`
- `Dart/2.10 (dart:io)`
当前 `appipr.txt` 解析出的真实地址为：
- `https://appsky2025999.ideasz.net`
初始化流程：
1. 读取 `appipr.txt` 得到 `API_HOST`
2. 校验 AES key/iv 长度为 16
3. 计算 `sign=md5(aes_key + aes_iv)`
4. 生成 `ck=<API_HOST>##5483##<timestamp>##ckzmbc` 后做 `ckEncrypt`
5. POST `/get_config` 获取授权 token
6. GET `/app/config` 获取应用配置
7. 将 `API_HOST`、`AUTH_TOKEN`、`CONFIG_DATA` 缓存在 Spider 实例上
`_ensure_ready` 在首次请求时执行，后续方法复用缓存。
## 数据映射设计
### 首页
`homeContent` 请求 `/sk-api/type/list`。
分类映射：
- 输出 `type_id`
- 输出 `type_name`
筛选映射：
- 逐分类请求 `/sk-api/type/alltypeextend`
- 将 `extendtype/area/lang/year` 分别映射成 `类型/地区/语言/年份`
- 增加固定排序项：
- `最新 -> updateTime`
- `人气 -> hot`
- `评分 -> score`
首页最终返回：
- `class`
- `filters`
- `list`
其中 `list` 由 `homeVideoContent` 提供。
### 首页推荐
`homeVideoContent` 请求：
- `/sk-api/vod/list?page=1&limit=12&type=randomlikeindex`
返回：
- `{"list": data.data or []}`
### 分类
`categoryContent` 请求 `/sk-api/vod/list`，参数为：
- `typeId=tid`
- `page=pg`
- `limit=18`
- `type=extend.sort or updateTime`
- `area=extend.area or ""`
- `lang=extend.lang or ""`
- `year=extend.year or ""`
- `mtype=""`
- `extendtype=extend.class or ""`
返回结构遵循仓库约定：
- `page`
- `limit`
- `total`
- `list`
不返回 `pagecount`。
### 搜索
`searchContent` 请求 `/sk-api/search/pages`，参数为：
- `keyword`
- `page`
- `limit=10`
- `typeId=-1`
返回结构同分类列表，不返回 `pagecount`。
### 详情
`detailContent` 请求 `/sk-api/vod/one?vodId=<id>`。
优先策略：
- 如果上游 `data.data` 已经是仓库可直接消费的标准 `vod` 结构，则直接返回
- 如果上游字段不完整，再做最小必要字段补齐
详情阶段不引入额外二次端点回退，保持与参考实现一致。
## 播放设计
用户确认的固定规则是：
- 总是先请求 `/sk-api/vod/skjson`
- 失败后回退原始 `id`
`playerContent` 逻辑：
1. 调用 `/sk-api/vod/skjson?url=<id>&skjsonindex=0`
2. 如果返回 `data.url` 且为 `http` 链接，则返回：
- `parse=0`
- `jx=0`
- `url=<playUrl>`
3. 如果解析失败，则回退原始 `id`
4. 如果原始 `id` 命中 `iqiyi/qq/youku/mgtv/bilibili` 域名，则返回：
- `parse=0`
- `jx=1`
- `url=<id>`
5. 其他情况返回：
- `parse=0`
- `jx=0`
- `url=<id>`
播放器请求头固定为移动 Safari UA，以保持与参考实现一致。
本次不实现基于 `/app/config` 中 `direct_json_link` 或 `direct_link` 的动态分支，`CONFIG_DATA` 仅作为初始化链路保留。
## 容错策略
- `_sk_decrypt` 只在响应带 `FROMSKZZJM` 前缀时尝试解密，否则按明文返回
- `appipr.txt` 为空、解密失败、JSON 解析失败时，抛出明确异常给站点内部处理
- `homeContent` 获取筛选时，单个分类失败仅跳过该分类筛选
- 列表和搜索接口异常时返回空结果结构
- 详情接口异常时返回空 `list`
- 播放解析异常时回退原始 `id`
## 测试设计
新增 `tests/test_即看影视.py`，使用 `unittest` 和 `unittest.mock`，不访问真实网络。
首轮测试覆盖：
1. 读取 `appipr.txt` 并解析真实 `API_HOST`
2. `_sk_decrypt` 对明文和加密前缀响应的处理
3. `_ck_encrypt` 返回非空密文
4. `homeContent` 正确输出分类和筛选
5. `homeVideoContent` 返回推荐列表
6. `categoryContent` 参数映射正确且不返回 `pagecount`
7. `searchContent` 返回标准分页结构
8. `detailContent` 返回 `{"list": [vod]}` 或空列表
9. `playerContent` 优先使用 `skjson` 结果
10. `playerContent` 在 `skjson` 失败时回退原始链接，并对站外视频域名设置 `jx=1`
测试顺序：
- 先跑单模块测试
- 再确认该模块无回归
## 风险与边界
- `appipr.txt` 返回的真实域名可能变化，测试中必须 mock，不依赖当前线上值
- `/app/config` 当前在简化版里不驱动播放分支，后续如果站点改成强依赖配置，播放器逻辑需要补齐
- 若上游详情字段不是标准 `vod` 结构，可能需要在实现阶段补一个轻量映射层，但范围仅限当前站点文件
py/tests/test_不太灵.py
-185
View File
@@ -1,185 +0,0 @@
import unittest
from importlib.machinery import SourceFileLoader
from pathlib import Path
from unittest.mock import patch
ROOT = Path(__file__).resolve().parents[1]
MODULE = SourceFileLoader("butailing_spider", str(ROOT / "不太灵.py")).load_module()
Spider = MODULE.Spider
class FakeResponse:
def __init__(self, text, status_code=200):
self.text = text
self.status_code = status_code
self.encoding = "utf-8"
class TestBuTaiLingSpider(unittest.TestCase):
def setUp(self):
Spider._instance = None
self.spider = Spider()
self.spider.init()
def test_home_content_returns_fixed_categories_and_filters(self):
with patch.object(
Spider,
"_request_api",
return_value={"t1": [{"title": "动作"}], "t2": [{"title": "中国"}], "t3": [], "t4": [], "t5": []},
):
content = self.spider.homeContent(False)
self.assertEqual(
[item["type_id"] for item in content["class"]],
["1", "2", "3", "4", "5"],
)
self.assertIn("1", content["filters"])
self.assertIn("2", content["filters"])
self.assertEqual(content["filters"]["1"][0]["key"], "sc")
@patch.object(Spider, "_request_api")
def test_home_video_content_requests_recent_hot_list(self, mock_request_api):
mock_request_api.return_value = [
{"doub_id": 11, "title": "示例电影", "image": "https://img.test/poster.jpg", "ejs": "HD"}
]
content = self.spider.homeVideoContent()
self.assertEqual(mock_request_api.call_args.args[0], "getVideoList")
self.assertEqual(mock_request_api.call_args.args[1]["sc"], "3")
self.assertEqual(content["list"][0]["vod_id"], "11")
self.assertEqual(content["list"][0]["vod_remarks"], "HD")
def test_build_api_url_appends_credentials(self):
url = self.spider._build_api_url("getVideoList", {"page": 2, "limit": 24})
self.assertIn("getVideoList", url)
self.assertIn("app_id=83768d9ad4", url)
self.assertIn("identity=23734adac0301bccdcb107c4aa21f96c", url)
self.assertIn("page=2", url)
@patch("requests.get")
def test_request_api_parses_json_and_callback_payload(self, mock_get):
mock_get.return_value = FakeResponse('callback({"success":true,"code":200,"data":{"data":[{"doub_id": 1}]}})')
result = self.spider._request_api("getVideoList", {"page": 1})
self.assertEqual(result, [{"doub_id": 1}])
def test_parse_ext_object_supports_plain_urlencoded_and_base64_json(self):
plain = self.spider._parse_ext_object('{"sc":"动作"}')
encoded = self.spider._parse_ext_object("%7B%22sd%22%3A%22中国%22%7D")
wrapped = self.spider._parse_ext_object("eyJzZSI6IjIwMjYifQ==")
self.assertEqual(plain["sc"], "动作")
self.assertEqual(encoded["sd"], "中国")
self.assertEqual(wrapped["se"], "2026")
@patch.object(Spider, "_request_api")
def test_category_content_movie_uses_filter_params(self, mock_request_api):
mock_request_api.return_value = [
{"doub_id": 21, "title": "电影A", "image": "a.jpg", "ejs": "4K"},
{"doub_id": 21, "title": "电影A", "image": "a.jpg", "ejs": "4K"},
]
result = self.spider.categoryContent("1", "2", False, '{"sc":"动作","sd":"中国","iswp":"1","status":"更新中"}')
self.assertEqual(mock_request_api.call_args.args[0], "getVideoMovieList")
self.assertEqual(
mock_request_api.call_args.args[1],
{"sa": 1, "page": 2, "sc": "动作", "sd": "中国", "se": "", "sf": "", "sh": "", "sg": "1", "iswp": 1},
)
self.assertEqual(result["page"], 2)
self.assertEqual(len(result["list"]), 1)
self.assertNotIn("pagecount", result)
@patch.object(Spider, "_request_api")
def test_category_content_hot_uses_local_dedupe_and_pagination(self, mock_request_api):
mock_request_api.return_value = [
{"doub_id": 1, "title": "A"},
{"doub_id": 1, "title": "A"},
{"doub_id": 2, "title": "B"},
]
result = self.spider.categoryContent("3", "1", False, {})
self.assertEqual(mock_request_api.call_args.args[0], "getVideoList")
self.assertEqual(result["total"], 2)
self.assertEqual(len(result["list"]), 2)
@patch.object(Spider, "_request_api")
def test_search_content_filters_by_name_and_dedupes(self, mock_request_api):
mock_request_api.return_value = [
{"doub_id": 1, "title": "繁花", "image": "1.jpg"},
{"doub_id": 2, "title": "繁花幕后", "image": "2.jpg"},
{"doub_id": 2, "title": "繁花幕后", "image": "2.jpg"},
{"doub_id": 3, "title": "别的内容", "image": "3.jpg"},
]
result = self.spider.searchContent("繁花", False, "1")
self.assertEqual(mock_request_api.call_args.args[0], "getVideoList")
self.assertEqual(result["total"], 2)
self.assertEqual([item["vod_id"] for item in result["list"]], ["1", "2"])
def test_detect_drive_type_by_url(self):
self.assertEqual(self.spider._detect_drive_type_by_url("https://pan.baidu.com/s/abc"), "baidu")
self.assertEqual(self.spider._detect_drive_type_by_url("https://pan.quark.cn/s/abc"), "quark")
self.assertEqual(self.spider._detect_drive_type_by_url("https://pan.xunlei.com/s/abc"), "xunlei")
self.assertEqual(self.spider._detect_drive_type_by_url("https://www.alipan.com/s/abc"), "aliyun")
self.assertEqual(self.spider._detect_drive_type_by_url("https://www.123pan.com/s/abc"), "a123")
self.assertEqual(self.spider._detect_drive_type_by_url("https://example.com/file"), "other")
def test_extract_pan_sources_creates_independent_lines_and_ignores_magnet_groups(self):
detail = {
"title": "示例详情",
"movies_online_seed": {
"夸克": [
{"seed_name": "夸克资源A", "link": "https://pan.quark.cn/s/q1"},
{"seed_name": "夸克资源A", "link": "https://pan.quark.cn/s/q1"},
],
"百度网盘": [{"seed_name": "百度资源", "link": "https://pan.baidu.com/s/b1"}],
"未知": [{"seed_name": "未知资源", "link": "https://example.com/file"}],
},
"ecca": {"WEB-4K": [{"zlink": "magnet:?xt=1"}]},
}
play_from, play_url = self.spider._extract_pan_sources(detail)
self.assertEqual(play_from, "baidu#1$$$quark#1$$$other#1")
self.assertEqual(
play_url,
"baidu#1$https://pan.baidu.com/s/b1$$$quark#1$https://pan.quark.cn/s/q1$$$other#1$https://example.com/file",
)
@patch.object(Spider, "_request_api")
def test_detail_content_maps_metadata_and_pan_lines(self, mock_request_api):
mock_request_api.return_value = {
"doub_id": 88,
"title": "详情标题",
"image": "https://img.test/detail.jpg",
"ejs": "全 12 集",
"years": "2026",
"abstract": "这是一段剧情简介",
"performer": "演员甲/演员乙",
"director": "导演甲",
"production_area": "中国",
"movies_online_seed": {
"阿里": [{"seed_name": "阿里资源", "link": "https://www.alipan.com/s/ali1"}]
},
}
result = self.spider.detailContent(["88"])
vod = result["list"][0]
self.assertEqual(mock_request_api.call_args.args[0], "getVideoDetail")
self.assertEqual(mock_request_api.call_args.args[1], {"id": "88"})
self.assertEqual(vod["vod_id"], "88")
self.assertEqual(vod["vod_name"], "详情标题")
self.assertEqual(vod["vod_play_from"], "aliyun#1")
self.assertEqual(vod["vod_play_url"], "aliyun#1$https://www.alipan.com/s/ali1")
def test_player_content_passthroughs_share_links_with_tvbox_shape(self):
self.assertEqual(
self.spider.playerContent("quark#1", "https://pan.quark.cn/s/q1", {}),
{"parse": 0, "playUrl": "", "url": "https://pan.quark.cn/s/q1"},
)
self.assertEqual(
self.spider.playerContent("quark#1", "push://https://pan.quark.cn/s/q1", {}),
{"parse": 0, "playUrl": "", "url": "https://pan.quark.cn/s/q1"},
)
def test_player_content_returns_empty_url_for_non_pan_links(self):
self.assertEqual(
self.spider.playerContent("other#1", "https://example.com/file", {}),
{"parse": 0, "playUrl": "", "url": ""},
)
if __name__ == "__main__":
unittest.main()
py/不太灵.py
-308
View File
@@ -1,308 +0,0 @@
# coding=utf-8
import base64
import json
import sys
from urllib.parse import unquote, urlencode, urljoin
import requests
from base.spider import Spider as BaseSpider
sys.path.append("..")
class Spider(BaseSpider):
DRIVE_ORDER = {"baidu": 1, "quark": 2, "xunlei": 3, "aliyun": 4, "a123": 5, "other": 99}
def __init__(self):
self.name = "不太灵"
self.api_base = "https://web5.mukaku.com/prod/api/v1/"
self.app_id = "83768d9ad4"
self.identity = "23734adac0301bccdcb107c4aa21f96c"
self.timeout = 10
self.headers = {
"User-Agent": (
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
"AppleWebKit/537.36 (KHTML, like Gecko) "
"Chrome/124.0.0.0 Safari/537.36"
),
"Referer": self.api_base,
}
self.classes = [
{"type_id": "1", "type_name": "电影"},
{"type_id": "2", "type_name": "电视剧"},
{"type_id": "3", "type_name": "近日热门"},
{"type_id": "4", "type_name": "本周热门"},
{"type_id": "5", "type_name": "本月热门"},
]
def init(self, extend=""):
return None
def getName(self):
return self.name
def _build_api_url(self, endpoint, params=None):
query = {"app_id": self.app_id, "identity": self.identity}
query.update(params or {})
return urljoin(self.api_base, endpoint) + "?" + urlencode(query)
def _parse_payload(self, text):
raw = str(text or "").strip()
if raw.startswith("callback(") and raw.endswith(")"):
raw = raw[9:-1]
return json.loads(raw) if raw else {}
def _request_api(self, endpoint, params=None):
response = requests.get(
self._build_api_url(endpoint, params),
headers=dict(self.headers),
timeout=self.timeout,
verify=False,
)
if response.status_code != 200:
return [] if endpoint in {"getVideoList", "getVideoMovieList"} else None
payload = self._parse_payload(response.text)
if not payload or payload.get("success") is not True or payload.get("code") != 200:
return [] if endpoint in {"getVideoList", "getVideoMovieList"} else None
data = payload.get("data")
if endpoint in {"getVideoList", "getVideoMovieList"}:
if isinstance(data, dict) and isinstance(data.get("data"), list):
return data.get("data")
if isinstance(data, dict) and isinstance(data.get("list"), list):
return data.get("list")
return []
return data
def _map_filter_values(self, items):
values = [{"n": "不限", "v": "0"}]
for item in items or []:
title = str((item or {}).get("title") or "").strip()
if title:
values.append({"n": title, "v": title})
return values
def _build_filters(self, type_data):
if not isinstance(type_data, dict):
return {}
common = [
{"key": "sc", "name": "影视类型", "value": self._map_filter_values(type_data.get("t1"))},
{"key": "sd", "name": "制片地区", "value": self._map_filter_values(type_data.get("t2"))},
{"key": "se", "name": "上映年份", "value": self._map_filter_values(type_data.get("t3"))},
{"key": "sf", "name": "资源画质", "value": self._map_filter_values(type_data.get("t4"))},
{"key": "sh", "name": "影视标签", "value": self._map_filter_values(type_data.get("t5"))},
]
return {
"1": common,
"2": common + [{"key": "status", "name": "剧集状态", "value": [{"n": "不限", "v": "0"}]}],
}
def _normalize_video(self, item):
data = item or {}
return {
"vod_id": str(data.get("doub_id") or data.get("id") or ""),
"vod_name": data.get("title") or "未知标题",
"vod_pic": data.get("image") or data.get("epic") or "",
"vod_remarks": data.get("ejs") or data.get("zqxd") or "",
"vod_year": data.get("years") or "",
"vod_content": data.get("abstract") or "",
"vod_actor": data.get("performer") or "",
"vod_director": data.get("director") or "",
"vod_area": data.get("production_area") or "",
}
def _parse_ext_object(self, ext):
candidates = [ext]
try:
decoded = base64.b64decode(str(ext or "")).decode("utf-8")
if decoded and decoded != ext:
candidates.append(decoded)
except Exception:
pass
for raw in list(candidates):
try:
decoded = unquote(str(raw))
if decoded != raw:
candidates.append(decoded)
except Exception:
pass
for raw in candidates:
try:
value = json.loads(raw)
if isinstance(value, dict):
return value
except Exception:
continue
return {}
def _normalize_filter_value(self, value):
text = str(value or "").strip()
return "" if text in {"", "0", "不限", "all"} else text
def _to01(self, value, fallback=0):
text = str(value if value is not None else fallback).strip().lower()
if text in {"1", "true", "yes", "on"}:
return 1
if text in {"0", "false", "no", "off", ""}:
return 0
return fallback
def _dedupe_by_vod_id(self, items):
result = []
seen = set()
for item in items or []:
vod_id = str((item or {}).get("doub_id") or (item or {}).get("id") or "")
if not vod_id or vod_id in seen:
continue
seen.add(vod_id)
result.append(item)
return result
def _paginate_items(self, items, page, limit):
page_num = max(1, int(page))
size = max(1, int(limit))
total = len(items)
start = (page_num - 1) * size
return {"page": page_num, "limit": size, "total": total, "list": items[start:start + size]}
def _build_movie_params(self, tid, pg, ext):
ext_obj = self._parse_ext_object(ext)
return {
"sa": int(tid),
"page": int(pg),
"sc": self._normalize_filter_value(ext_obj.get("sc")),
"sd": self._normalize_filter_value(ext_obj.get("sd")),
"se": self._normalize_filter_value(ext_obj.get("se")),
"sf": self._normalize_filter_value(ext_obj.get("sf")),
"sh": self._normalize_filter_value(ext_obj.get("sh")),
"sg": self._normalize_filter_value(ext_obj.get("sg")) or "1",
"iswp": self._to01(ext_obj.get("iswp"), 0),
}
def _normalize_drive_key(self, value):
text = str(value or "").strip().lower()
if text in {"alipan", "aliyun", "aliyundrive", "ali", "阿里"}:
return "aliyun"
if "quark" in text or "夸克" in text:
return "quark"
if "baidu" in text or "百度" in text:
return "baidu"
if "xunlei" in text or "迅雷" in text:
return "xunlei"
if "123" in text:
return "a123"
return "other"
def _detect_drive_type_by_url(self, url):
text = str(url or "").lower()
if "pan.quark.cn" in text:
return "quark"
if "pan.baidu.com" in text:
return "baidu"
if "pan.xunlei.com" in text:
return "xunlei"
if "alipan.com" in text or "aliyundrive.com" in text:
return "aliyun"
if "123865.com" in text or "123684.com" in text or "123pan.com" in text:
return "a123"
return "other"
def _is_supported_pan_url(self, url):
text = str(url or "").strip().lower()
return any(
token in text
for token in [
"pan.quark.cn",
"pan.baidu.com",
"pan.xunlei.com",
"alipan.com",
"aliyundrive.com",
"123865.com",
"123684.com",
"123pan.com",
"drive.uc.cn",
]
)
def _extract_pan_sources(self, detail):
groups = (detail or {}).get("movies_online_seed")
if not isinstance(groups, dict):
return "", ""
seen = set()
counters = {}
entries = []
for group_name, items in groups.items():
for item in items or []:
link = str((item or {}).get("link") or "").strip()
if not link or link in seen:
continue
seen.add(link)
kind = self._detect_drive_type_by_url(link)
if kind == "other":
kind = self._normalize_drive_key(group_name)
counters[kind] = counters.get(kind, 0) + 1
line_name = f"{kind}@{counters[kind]}"
entries.append((line_name, link))
entries.sort(
key=lambda pair: (
self.DRIVE_ORDER.get(pair[0].split("@")[0], 999),
int(pair[0].split("@")[1]),
pair[0],
)
)
return "$$$".join(name for name, _ in entries), "$$$".join(f"{name}${url}" for name, url in entries)
def homeContent(self, filter):
return {"class": self.classes, "filters": self._build_filters(self._request_api("getVideoTypeList", {}))}
def homeVideoContent(self):
items = self._request_api("getVideoList", {"sc": "3", "limit": 20}) or []
return {"list": [self._normalize_video(item) for item in items]}
def categoryContent(self, tid, pg, filter, extend):
page = int(pg)
if str(tid) in {"1", "2"}:
items = self._dedupe_by_vod_id(
self._request_api("getVideoMovieList", self._build_movie_params(tid, page, extend)) or []
)
return {
"page": page,
"limit": len(items) or 20,
"total": page * 20 + len(items),
"list": [self._normalize_video(item) for item in items],
}
items = self._dedupe_by_vod_id(self._request_api("getVideoList", {"sc": str(tid), "page": 1, "limit": 300}) or [])
paged = self._paginate_items(items, page, 20)
paged["list"] = [self._normalize_video(item) for item in paged["list"]]
return paged
def searchContent(self, key, quick, pg="1"):
keyword = str(key or "").strip()
page = int(pg)
if not keyword:
return {"page": page, "total": 0, "list": []}
items = self._request_api("getVideoList", {"sb": keyword, "page": 1, "limit": 300}) or []
matched = [item for item in items if keyword.lower() in str((item or {}).get("title") or "").lower()]
paged = self._paginate_items(self._dedupe_by_vod_id(matched), page, 20)
paged["list"] = [self._normalize_video(item) for item in paged["list"]]
return paged
def detailContent(self, ids):
vod_id = str(ids[0] if isinstance(ids, list) and ids else ids or "").strip()
if not vod_id:
return {"list": []}
data = self._request_api("getVideoDetail", {"id": vod_id})
if not isinstance(data, dict):
return {"list": []}
vod = self._normalize_video(data)
play_from, play_url = self._extract_pan_sources(data)
vod["vod_play_from"] = play_from
vod["vod_play_url"] = play_url
return {"list": [vod]}
def playerContent(self, flag, id, vipFlags):
raw = str(id or "").strip()
url = raw[7:] if raw.startswith("push://") else raw
if self._is_supported_pan_url(url):
return {"parse": 0, "playUrl": "", "url": url}
return {"parse": 0, "playUrl": "", "url": ""}