diff --git a/py/docs/superpowers/specs/2026-04-23-jikanyingshi-spider-design.md b/py/docs/superpowers/specs/2026-04-23-jikanyingshi-spider-design.md new file mode 100644 index 0000000..ea220d6 --- /dev/null +++ b/py/docs/superpowers/specs/2026-04-23-jikanyingshi-spider-design.md @@ -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=##5483####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=`。 + +优先策略: + +- 如果上游 `data.data` 已经是仓库可直接消费的标准 `vod` 结构,则直接返回 +- 如果上游字段不完整,再做最小必要字段补齐 + +详情阶段不引入额外二次端点回退,保持与参考实现一致。 + +## 播放设计 + +用户确认的固定规则是: + +- 总是先请求 `/sk-api/vod/skjson` +- 失败后回退原始 `id` + +`playerContent` 逻辑: + +1. 调用 `/sk-api/vod/skjson?url=&skjsonindex=0` +2. 如果返回 `data.url` 且为 `http` 链接,则返回: + - `parse=0` + - `jx=0` + - `url=` +3. 如果解析失败,则回退原始 `id` +4. 如果原始 `id` 命中 `iqiyi/qq/youku/mgtv/bilibili` 域名,则返回: + - `parse=0` + - `jx=1` + - `url=` +5. 其他情况返回: + - `parse=0` + - `jx=0` + - `url=` + +播放器请求头固定为移动 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` 结构,可能需要在实现阶段补一个轻量映射层,但范围仅限当前站点文件 diff --git a/py/tests/test_不太灵.py b/py/tests/test_不太灵.py deleted file mode 100644 index 78b2a07..0000000 --- a/py/tests/test_不太灵.py +++ /dev/null @@ -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() diff --git a/py/不太灵.py b/py/不太灵.py deleted file mode 100644 index 6fa2d2e..0000000 --- a/py/不太灵.py +++ /dev/null @@ -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": ""}