diff --git a/py/docs/superpowers/specs/2026-04-20-mofang-app-spider-design.md b/py/docs/superpowers/specs/2026-04-20-mofang-app-spider-design.md new file mode 100644 index 0000000..d59ea3e --- /dev/null +++ b/py/docs/superpowers/specs/2026-04-20-mofang-app-spider-design.md @@ -0,0 +1,382 @@ +# 魔方 APP Python 爬虫设计 + +## 目标 + +在当前 Python 仓库中新增一个符合 `base.spider.Spider` 接口的魔方 APP 爬虫，覆盖以下能力： + +- 首页分类与筛选 +- 分类列表 +- 搜索 +- 详情解析 +- 播放解析 + +实现以用户提供的 Node/JS 版本为行为参考，但落地形式遵循当前仓库的单文件 Spider 约定。 + +## 范围 + +本次实现包含： + +- 新增独立脚本，文件名为 `魔方APP.py` +- 使用魔方 APP 的 AES-CBC 接口协议 +- 固定 host 为 `http://www.613mf4k12.top` +- 支持 `home/category/detail/search/player` 全链路 +- 支持分类排序、分类屏蔽和分类重命名 +- 支持地区融合和年份自动补充 +- 支持线路排序、线路重命名和线路屏蔽 +- 保留搜索验证码分支，并在需要时尝试 OCR +- 为新增行为补齐 `unittest` + +本次实现不包含： + +- 动态 `site` 下发 host +- Node/Fastify 路由层 +- 真实 OCR 服务联调 +- 修改 `base/` 公共层 +- 多解析器 `auto` 优先级调度 +- 本地缓存和重试机制 + +## 方案选择 + +采用仓库现有的“单站点单文件 + 单测”方案，并以 `黑猫APP.py` 的结构为底稿实现魔方版本。 + +选择该方案的原因： + +- `黑猫APP.py` 已具备 AES、筛选、详情、播放的完整链路骨架 +- 魔方的 JS 逻辑与黑猫同属 APP AES 接口族，迁移成本最低 +- 本次任务目标是新增一个稳定 Spider，不是抽象公共基类 + +本次不抽取共享基类的原因： + +- 会扩大改动面，增加回归范围 +- 现有多个同类 Spider 仍有站点级差异，抽象收益不足 +- 用户当前需求是尽快落地单蜘蛛实现 + +## 模块边界 + +新增脚本只在站点文件内部维护逻辑，不修改 `base/`。 + +脚本内部职责拆分如下： + +- `init` + - 初始化 host、API path、AES key/iv、UA、分类配置、地区融合配置、线路配置、搜索验证码状态 +- `homeContent` + - 拉取初始化接口，输出分类、筛选和首页推荐 +- `homeVideoContent` + - 返回空列表 +- `categoryContent` + - 请求分类筛选接口，必要时执行大陆地区聚合 +- `searchContent` + - 请求搜索接口，执行验证码处理与本地结果过滤 +- `detailContent` + - 请求详情接口并整理影片元数据与播放列表 +- `playerContent` + - 根据线路模式、解析类型和 `vodParse` 结果输出播放信息 +- 私有辅助函数 + - AES 加解密 + - API 请求 + - 年份补全 + - 地区融合 + - 线路整理 + - 验证码获取与 OCR 文本修正 + +## Host 与请求策略 + +固定主域： + +- `http://www.613mf4k12.top` + +接口路径： + +- `api=1`，因此实际请求路径为 `/api.php/getappapi.index/` + +初始化与搜索端点默认值： + +- `initV119` +- `searchList` + +请求策略： + +- 接口请求统一走 `POST` +- 请求头默认带 `User-Agent: okhttp/3.10.0` 和 `Accept-Encoding: gzip` +- 返回体中的 `data` 字段先做 AES-CBC 解密，再解析 JSON +- `version` 暂为空；只有后续站点明确要求时才注入 + +初始化策略： + +- `init` 只做一次 +- 通过初始化接口读取 `system_search_verify_status` +- 若该标记为真，则搜索前进入验证码分支 + +异常策略： + +- 单次请求失败时尽量返回空结果，不向上抛出未处理异常 +- 某个详情端点失败时允许切换备用端点继续尝试 +- 播放解析失败时返回空 URL 或空列表，而不是抛异常 + +## 配置设计 + +站点文件内部维护默认配置对象，至少包含： + +- `name` +- `url` +- `api` +- `dataKey` +- `dataIv` +- `init` +- `search` +- `version` +- `ua` +- `headers` +- `categories` +- `areaMerge` +- `ocr` + +分类管理配置： + +- `blockedNames = ["全部"]` +- `renameMap = {}` +- `forceOrder = ["电影", "连续剧", "综艺", "动漫", "短剧", "直播"]` + +地区融合配置： + +- `enabled = True` +- `displayName = "大陆"` +- `mergeList = ["中国大陆", "大陆", "内地"]` + +OCR 配置： + +- `enabled = False` +- `api = "http://154.222.22.188:9898/ocr/b64/text"` + +线路管理配置： + +- 以内置线路表匹配原始线路名 +- 支持显示名、排序权重、解析模式、是否启用 +- 直接使用用户提供的魔方线路表，包括禁用项如 `水印资源`、`YY` + +## 分类与筛选设计 + +`homeContent` 的输入来源是初始化接口返回的 `type_list`。 + +分类处理规则： + +- 屏蔽名称包含在 `blockedNames` 里的分类，例如 `全部` +- 对命中的分类名应用 `renameMap` +- 若配置了 `forceOrder`，则按配置顺序重排已保留分类 + +筛选转换规则： + +- 将接口中的 `class/area/lang/year/sort` 转成仓库常用的筛选结构 +- `sort` 对外统一映射为 `by` +- 显示名映射为 `类型/地区/语言/年份/排序` + +特殊筛选处理： + +- 地区筛选启用融合时，把 `中国大陆/大陆/内地` 合并显示为单个 `大陆` +- 年份筛选自动补入当前年份；若列表里无当前年份，则插入到 `全部` 后面 + +首页返回字段： + +- `class` +- `filters` +- `list` + +其中 `list` 直接复用初始化返回的分类推荐数据平铺结果。 + +## 列表与搜索设计 + +分类列表使用接口： + +- `typeFilterVodList` + +请求参数包括： + +- `type_id` +- `page` +- `area` +- `year` +- `sort` +- `lang` +- `class` + +对外筛选字段中的 `by` 在请求时还原成接口所需的 `sort`。 + +大陆地区融合开启且用户选择融合值时： + +- 不直接请求一次 `area=大陆` +- 改为依次请求 `中国大陆/大陆/内地` +- 按 `vod_id` 去重后合并结果 + +搜索使用接口： + +- `searchList`，也允许通过配置覆盖 + +搜索规则： + +- 若初始化要求验证码，则在搜索前获取验证码图片、调用 OCR 并修正常见误识别字符 +- 若 OCR 失败，则返回空列表，并附带错误消息 +- 本地过滤掉 `vod_class` 包含 `屏蔽预留` 的结果 +- 如果有搜索词，则只保留标题、备注或分类文本中包含关键词的结果 +- 搜索结果映射为 `vod_id/vod_name/vod_pic/vod_remarks` +- `vod_remarks` 使用 `vod_year + vod_class` 拼接 + +分页返回字段遵循仓库当前约定： + +- `page` +- `limit` +- `total` +- `list` + +不返回 `pagecount`。 + +## 验证码设计 + +验证码仅在初始化标记要求时启用。 + +处理流程： + +- 请求 `/verify/create?key=` 获取图片 +- 将图片转为 base64 文本 +- 发送到 OCR 接口 +- 对 OCR 返回文本执行字符替换修正 +- 仅当结果是 4 位数字时才作为有效验证码提交 + +字符修正规则沿用参考 JS 中的映射，例如： + +- `y -> 9` +- `口 -> 0` +- `q/u/o/d/D -> 0` +- `b -> 8` +- `已 -> 2` +- `五 -> 5` + +实现要求： + +- 优先使用标准库方式生成 `uuid` +- 测试中只验证本地逻辑与请求参数，不访问真实 OCR 服务 + +## 详情设计 + +详情优先尝试两个端点： + +- `vodDetail` +- `vodDetail2` + +详情字段映射： + +- `vod_name` +- `vod_pic` +- `vod_remarks` +- `vod_content` +- `vod_actor` +- `vod_director` +- `vod_year` +- `vod_area` +- `vod_play_from` +- `vod_play_url` + +数据整理规则： + +- `vod_actor` 去掉前缀 `演员` +- `vod_director` 去掉前缀 `导演` +- `vod_year` 有值时补后缀 `年` + +线路处理规则： + +- 如果原始线路名含 `防走丢/群/防失群/官网`，则替换为递增兜底名 `1线/2线/...` +- 若线路名重复，则对重复项追加序号，避免冲突 +- 若线路被配置为禁用，则直接跳过 +- 每条线路按配置映射展示名与排序权重 + +播放串编码格式： + +- `集名$线路名@@模式@@parse_api,url,token+,player_parse_type,parse_type` + +输出时： + +- `vod_play_from` 使用排序后的展示线路名，以 `$$$` 拼接 +- `vod_play_url` 使用排序后的集列表，以 `$$$` 拼接 + +## 播放解析设计 + +`playerContent` 输入是详情阶段编码好的播放串。 + +处理顺序： + +- 先解析出线路名、模式和载荷 +- 若线路已禁用，直接返回空地址 +- 当前只支持 `direct` 模式；`auto` 不实现动态 parser 调度 + +解析分支： + +- `parse_type == "0"` + - 直接返回解码后的原始播放地址 + - `parse = 0`，`jx = 0` +- `parse_type == "2"` + - 返回 `parse_api + 播放地址` + - `parse = 1`，`jx = 1` +- `player_parse_type == "2"` + - 先请求 `parse_api + 播放地址` + - 若响应 JSON 里存在 `url`，则直接返回该直链 +- 其他情况 + - 对播放地址执行 AES 加密 + - 请求 `vodParse` + - 从返回的 `json` 字段中提取最终地址 + +播放器返回结构沿用仓库现有 APP Spider 约定： + +- `parse` +- `jx` +- `url` +- `header` + +直连与需二次解析分支均带 Android/Dalvik UA 头；其余分支默认返回空 header。 + +## 测试设计 + +新增测试文件： + +- `tests/test_魔方APP.py` + +测试覆盖目标： + +- AES 加解密可逆 +- `_api_post` 能正确解密响应 +- `homeContent` 会过滤分类、处理地区融合并补当前年份 +- `categoryContent` 能正确映射 `by -> sort` +- `categoryContent` 在选择 `大陆` 时会多次请求并按 `vod_id` 去重 +- `searchContent` 会过滤 `屏蔽预留` 且只保留关键词命中的结果 +- 搜索验证码开启但 OCR 失败时返回空列表 +- `detailContent` 会回退第二详情端点，并按线路规则构造 `vod_play_from` 与 `vod_play_url` +- `playerContent` 覆盖 `parse_type == 0` +- `playerContent` 覆盖 `parse_type == 2` +- `playerContent` 覆盖 `player_parse_type == 2` +- `playerContent` 覆盖 `vodParse` 回退分支 + +测试策略： + +- 使用 `unittest` 和 `unittest.mock` +- 只 mock `fetch/post/_api_post` 等边界调用 +- 不访问真实站点和 OCR 服务 +- 先跑单模块测试，再视需要扩展到更大范围 + +## 验证计划 + +实现完成后的验证顺序： + +1. `python -m unittest tests.test_魔方APP -v` +2. 如果实现过程中复用了或改动了相邻逻辑，再补跑相关模块测试 + +## 风险与取舍 + +主要风险： + +- 魔方站点实际返回字段可能与参考 JS 记录存在细节差异 +- 验证码图片格式或 OCR 服务返回格式可能不稳定 +- 某些线路名可能与当前内置线路表存在别名差异 + +本次取舍： + +- 优先保证与参考 JS 的主要行为一致 +- 不为未使用的 `auto` 解析模式增加复杂度 +- 不提前抽公共基类，避免影响其他 Spider diff --git a/py/tests/test_鲸鱼APP.py b/py/tests/test_鲸鱼APP.py deleted file mode 100644 index 857d03e..0000000 --- a/py/tests/test_鲸鱼APP.py +++ /dev/null @@ -1,255 +0,0 @@ -import json -import unittest -from importlib.machinery import SourceFileLoader -from pathlib import Path - -ROOT = Path(__file__).resolve().parents[1] -MODULE = SourceFileLoader("jingyu_spider", str(ROOT / "鲸鱼APP.py")).load_module() -Spider = MODULE.Spider - - -class TestJingyuSpider(unittest.TestCase): - def setUp(self): - Spider._instance = None - self.spider = Spider() - - def test_aes_encrypt_decrypt_roundtrip(self): - plaintext = '{"key":"value"}' - encrypted = self.spider._aes_encrypt(plaintext) - decrypted = self.spider._aes_decrypt(encrypted) - self.assertEqual(decrypted, plaintext) - - def test_aes_encrypt_produces_base64(self): - import re - encrypted = self.spider._aes_encrypt("hello") - self.assertTrue(re.match(r'^[A-Za-z0-9+/=]+$', encrypted)) - - def test_api_post_decrypts_response(self): - payload = {"result": "ok"} - encrypted = self.spider._aes_encrypt(json.dumps(payload)) - - class FakeResponse: - status_code = 200 - encoding = "utf-8" - def json(self): - return {"data": encrypted} - - self.spider.post = lambda url, **kwargs: FakeResponse() - self.spider.host = "http://test.com" - result = self.spider._api_post("someEndpoint") - self.assertEqual(result, payload) - - def test_init_fetches_host_from_site_url(self): - init_encrypted = self.spider._aes_encrypt(json.dumps({"type_list": []})) - - class FakeInitResponse: - status_code = 200 - encoding = "utf-8" - text = "http://example.com" - def json(self): - return {"data": init_encrypted} - - self.spider.fetch = lambda url, **kwargs: FakeInitResponse() - self.spider.post = lambda url, **kwargs: FakeInitResponse() - self.spider.init() - self.assertEqual(self.spider.host, "http://example.com") - - def test_process_classes_blocks_and_sorts(self): - type_list = [ - {"type_id": "0", "type_name": "全部"}, - {"type_id": "1", "type_name": "电影"}, - {"type_id": "3", "type_name": "综艺"}, - {"type_id": "2", "type_name": "电视剧"}, - {"type_id": "4", "type_name": "动漫"}, - ] - classes = self.spider._process_classes(type_list) - names = [c["type_name"] for c in classes] - self.assertNotIn("全部", names) - self.assertEqual(names, ["电影", "电视剧", "综艺", "动漫"]) - - def test_process_area_filter_merges_mainland_areas(self): - areas = ["全部", "中国大陆", "大陆", "内地", "美国", "日本"] - result = self.spider._process_area_filter(areas) - self.assertIn("大陆", result) - self.assertNotIn("中国大陆", result) - self.assertNotIn("内地", result) - self.assertIn("美国", result) - - def test_convert_filters_adds_current_year(self): - import datetime - type_list = [{ - "type_id": "1", - "filter_type_list": [ - {"name": "year", "list": ["全部", "2024"]}, - {"name": "area", "list": ["全部", "中国大陆"]}, - ] - }] - current = str(datetime.datetime.now().year) - filters = self.spider._convert_filters(type_list) - year_values = [v["v"] for v in filters["1"][0]["value"]] - self.assertIn(current, year_values) - area_values = [v["v"] for v in filters["1"][1]["value"]] - self.assertIn("大陆", area_values) - self.assertNotIn("中国大陆", area_values) - - def test_category_content_posts_filter_payload(self): - encrypted_empty = self.spider._aes_encrypt(json.dumps({"recommend_list": []})) - - class FakeRsp: - status_code = 200 - encoding = "utf-8" - def json(self): - return {"data": encrypted_empty} - - self.spider.post = lambda url, **kwargs: FakeRsp() - self.spider.host = "http://test.com" - self.spider.init_data = {} - result = self.spider.categoryContent("1", "2", False, {"area": "美国", "year": "2025"}) - self.assertEqual(result["page"], 2) - self.assertNotIn("pagecount", result) - self.assertEqual(result["list"], []) - - def test_category_content_merges_area_when_mainland_selected(self): - items_a = [{"vod_id": "1", "vod_name": "A", "vod_pic": "", "vod_remarks": ""}] - items_b = [{"vod_id": "2", "vod_name": "B", "vod_pic": "", "vod_remarks": ""}, - {"vod_id": "1", "vod_name": "A", "vod_pic": "", "vod_remarks": ""}] - - call_idx = {"n": 0} - - def fake_api_post(endpoint, payload=None): - call_idx["n"] += 1 - if call_idx["n"] == 1: - return {"recommend_list": items_a} - return {"recommend_list": items_b} - - self.spider._api_post = fake_api_post - self.spider.host = "http://test.com" - self.spider.init_data = {} - result = self.spider.categoryContent("1", "1", False, {"area": "大陆"}) - ids = [v["vod_id"] for v in result["list"]] - self.assertEqual(len(ids), 2) - self.assertIn("1", ids) - self.assertIn("2", ids) - - def test_detail_content_builds_play_lines(self): - detail_data = { - "vod": { - "vod_name": "测试影片", - "vod_pic": "http://pic.jpg", - "vod_remarks": "HD", - "vod_content": "简介", - "vod_actor": "演员张三", - "vod_director": "导演李四", - "vod_year": "2025", - "vod_area": "大陆", - }, - "vod_play_list": [ - { - "player_info": { - "show": "线路一", - "parse": "http://parse.com/", - "player_parse_type": "1", - "parse_type": "1", - }, - "urls": [ - {"name": "第1集", "url": "http://play/1", "token": "tok1"}, - {"name": "第2集", "url": "http://play/2", "token": "tok2"}, - ], - }, - ], - } - - self.spider._api_post = lambda ep, payload=None: detail_data if "vodDetail" in ep else None - self.spider.host = "http://test.com" - self.spider.init_data = {} - result = self.spider.detailContent(["123"]) - vod = result["list"][0] - self.assertEqual(vod["vod_name"], "测试影片") - self.assertEqual(vod["vod_year"], "2025年") - self.assertIn("线路一", vod["vod_play_from"]) - self.assertIn("第1集$", vod["vod_play_url"]) - self.assertIn("第2集$", vod["vod_play_url"]) - - def test_detail_content_filters_junk_lines(self): - detail_data = { - "vod": {"vod_name": "X", "vod_pic": "", "vod_remarks": "", "vod_content": "", - "vod_actor": "", "vod_director": "", "vod_year": "", "vod_area": ""}, - "vod_play_list": [ - { - "player_info": {"show": "防走丢群", "parse": "", "player_parse_type": "0", "parse_type": "0"}, - "urls": [{"name": "链接", "url": "http://x", "token": ""}], - }, - { - "player_info": {"show": "正式线路", "parse": "", "player_parse_type": "0", "parse_type": "0"}, - "urls": [{"name": "第1集", "url": "http://y", "token": ""}], - }, - ], - } - - self.spider._api_post = lambda ep, payload=None: detail_data - self.spider.host = "http://test.com" - self.spider.init_data = {} - result = self.spider.detailContent(["1"]) - vod = result["list"][0] - self.assertIn("1线", vod["vod_play_from"]) - self.assertIn("正式线路", vod["vod_play_from"]) - - def test_search_content_filters_and_maps_results(self): - search_data = { - "search_list": [ - {"vod_id": "1", "vod_name": "繁花", "vod_pic": "http://p1.jpg", - "vod_remarks": "更新中", "vod_class": "都市", "vod_year": "2024"}, - {"vod_id": "2", "vod_name": "花繁", "vod_pic": "http://p2.jpg", - "vod_remarks": "", "vod_class": "屏蔽预留", "vod_year": ""}, - ] - } - - self.spider._api_post = lambda ep, payload=None: search_data - self.spider.host = "http://test.com" - self.spider.init_data = {} - result = self.spider.searchContent("繁花", False, "1") - self.assertEqual(len(result["list"]), 1) - self.assertEqual(result["list"][0]["vod_name"], "繁花") - self.assertNotIn("pagecount", result) - - def test_player_direct_parse_type_0(self): - play_id = "线路一@@direct@@http://parse.com,http://video.m3u8,token+t1,1,0" - result = self.spider.playerContent("线路一", play_id, {}) - self.assertEqual(result["parse"], 0) - self.assertEqual(result["url"], "http://video.m3u8") - - def test_player_prefix_parse_type_2(self): - play_id = "线路二@@direct@@http://parse.com,http://video.m3u8,token+t1,1,2" - result = self.spider.playerContent("线路二", play_id, {}) - self.assertEqual(result["parse"], 1) - self.assertEqual(result["url"], "http://parse.comhttp://video.m3u8") - - def test_player_vod_parse_default(self): - play_id = "线路三@@direct@@http://parse.com,http://video.m3u8,token+t1,1,1" - inner = json.dumps({"url": "http://decrypted.m3u8"}) - encrypted_inner = self.spider._aes_encrypt(json.dumps({"json": inner})) - - class FakeRsp: - status_code = 200 - encoding = "utf-8" - def json(self): - return {"data": encrypted_inner} - - self.spider.post = lambda url, **kwargs: FakeRsp() - self.spider.host = "http://test.com" - result = self.spider.playerContent("线路三", play_id, {}) - self.assertEqual(result["parse"], 0) - self.assertEqual(result["url"], "http://decrypted.m3u8") - - def test_player_fetch_parse_player_parse_type_2(self): - play_id = "线路四@@direct@@,http://video.m3u8,token+t1,2,1" - - class FakeFetchRsp: - def json(self): - return {"url": "http://fetched.m3u8"} - - self.spider.fetch = lambda url, **kwargs: FakeFetchRsp() - self.spider.host = "http://test.com" - result = self.spider.playerContent("线路四", play_id, {}) - self.assertEqual(result["parse"], 0) - self.assertEqual(result["url"], "http://fetched.m3u8") diff --git a/py/鲸鱼APP.py b/py/鲸鱼APP.py deleted file mode 100644 index 5ed92e4..0000000 --- a/py/鲸鱼APP.py +++ /dev/null @@ -1,409 +0,0 @@ -# coding=utf-8 -import base64 -import json -import re -import sys -import time - -from Crypto.Cipher import AES -from Crypto.Util.Padding import pad, unpad -from base.spider import Spider as BaseSpider - -sys.path.append("..") - -AES_KEY = "AAdgrdghjfgsABC1" -AES_IV = "AAdgrdghjfgsABC1" - - -class Spider(BaseSpider): - def __init__(self): - self.name = "鲸鱼APP" - self.host = "" - self.ua = "okhttp/3.10.0" - self.api_path = "/api.php/qijiappapi.index" - self.init_endpoint = "initV122" - self.search_endpoint = "searchList" - self.search_verify = False - self.init_data = None - - def getName(self): - return self.name - - SITE_URL = "https://jingyu4k-1312635929.cos.ap-nanjing.myqcloud.com/juyu3.json" - - def init(self, extend=""): - if self.host: - return - headers = {"User-Agent": self.ua} - try: - rsp = self.fetch(self.SITE_URL, headers=headers, timeout=10, verify=False) - host = rsp.text.strip().rstrip("/") - if not host.startswith("http"): - host = "http://" + host - self.host = host - except Exception as e: - self.log(f"获取host失败: {e}") - raise - try: - data = self._api_post(self.init_endpoint) - if data and data.get("config", {}).get("system_search_verify_status"): - self.search_verify = True - self.init_data = data - except Exception as e: - self.log(f"初始化数据失败: {e}") - - def _aes_encrypt(self, plaintext): - key = AES_KEY.encode("utf-8") - iv = AES_IV.encode("utf-8") - cipher = AES.new(key, AES.MODE_CBC, iv) - padded = pad(plaintext.encode("utf-8"), AES.block_size) - encrypted = cipher.encrypt(padded) - return base64.b64encode(encrypted).decode("utf-8") - - def _aes_decrypt(self, ciphertext): - key = AES_KEY.encode("utf-8") - iv = AES_IV.encode("utf-8") - raw = base64.b64decode(ciphertext) - cipher = AES.new(key, AES.MODE_CBC, iv) - decrypted = unpad(cipher.decrypt(raw), AES.block_size) - return decrypted.decode("utf-8") - - def _api_post(self, endpoint, payload=None): - if payload is None: - payload = {} - ep = f"/{endpoint}" if not endpoint.startswith("/") else endpoint - url = f"{self.host}{self.api_path}{ep}" - headers = { - "User-Agent": self.ua, - "Accept-Encoding": "gzip", - } - rsp = self.post(url, json=payload, headers=headers, timeout=15, verify=False) - data = rsp.json().get("data") - if not data: - return None - try: - return json.loads(self._aes_decrypt(data)) - except Exception as e: - self.log(f"JSON解析失败: {e}") - return None - - CATEGORY_BLOCKED = ["全部"] - CATEGORY_FORCE_ORDER = ["电影", "电视剧", "综艺", "动漫", "短剧"] - AREA_MERGE_DISPLAY = "大陆" - AREA_MERGE_LIST = ["中国大陆", "大陆", "内地"] - - def _process_classes(self, type_list): - order_map = {n: i for i, n in enumerate(self.CATEGORY_FORCE_ORDER)} - classes = [ - {"type_id": t["type_id"], "type_name": t["type_name"]} - for t in type_list - if t["type_name"] not in self.CATEGORY_BLOCKED - ] - classes.sort(key=lambda c: order_map.get(c["type_name"], 999)) - return classes - - def _process_area_filter(self, area_list): - if not area_list: - return area_list - merge_set = set(self.AREA_MERGE_LIST) - filtered = [a for a in area_list if a not in merge_set] - has_merge = any(a in merge_set for a in area_list) - if has_merge: - try: - idx = filtered.index("全部") - filtered.insert(idx + 1, self.AREA_MERGE_DISPLAY) - except ValueError: - filtered.insert(0, self.AREA_MERGE_DISPLAY) - return filtered - - def _convert_filters(self, type_list): - name_map = {"class": "类型", "area": "地区", "lang": "语言", "year": "年份", "sort": "排序"} - current_year = str(time.localtime().tm_year) - filters = {} - for t in type_list: - arr = [] - for f in t.get("filter_type_list", []): - key = "by" if f["name"] == "sort" else f["name"] - values = list(f.get("list", [])) - if f["name"] == "area": - values = self._process_area_filter(values) - if f["name"] == "year" and current_year not in values: - try: - idx = values.index("全部") - values.insert(idx + 1, current_year) - except ValueError: - values.insert(0, current_year) - arr.append({ - "key": key, - "name": name_map.get(f["name"], f["name"]), - "value": [{"n": v, "v": v} for v in values], - }) - filters[t["type_id"]] = arr - return filters - - def homeContent(self, filter): - self.init() - data = self.init_data - if not data: - return {"class": [], "filters": {}} - classes = self._process_classes(data.get("type_list", [])) - filters = self._convert_filters(data.get("type_list", [])) - return {"class": classes, "filters": filters} - - def homeVideoContent(self): - self.init() - if not self.init_data: - return {"list": []} - videos = [] - for t in self.init_data.get("type_list", []): - for item in t.get("recommend_list", []): - videos.append({ - "vod_id": str(item.get("vod_id", "")), - "vod_name": item.get("vod_name", ""), - "vod_pic": item.get("vod_pic", ""), - "vod_remarks": item.get("vod_remarks", ""), - }) - return {"list": videos} - - def _merge_area_search(self, type_id, page, base_filters): - all_results = [] - seen = set() - for area_val in self.AREA_MERGE_LIST: - try: - payload = {"type_id": type_id, "page": page, "area": area_val} - payload.update(base_filters) - res = self._api_post("typeFilterVodList", payload) - for item in (res or {}).get("recommend_list", []): - vid = str(item.get("vod_id", "")) - if vid not in seen: - seen.add(vid) - all_results.append({ - "vod_id": vid, - "vod_name": item.get("vod_name", ""), - "vod_pic": item.get("vod_pic", ""), - "vod_remarks": item.get("vod_remarks", ""), - }) - except Exception as e: - self.log(f"聚合搜索地区[{area_val}]失败: {e}") - return all_results - - def categoryContent(self, tid, pg, filter, extend): - self.init() - page = int(pg) - ext = extend or {} - - if ext.get("area") == self.AREA_MERGE_DISPLAY: - base = {k: v for k, v in ext.items() if k != "area"} - base.setdefault("year", "全部") - base.setdefault("sort", "最新") - base.setdefault("lang", "全部") - base.setdefault("class", "全部") - merged = self._merge_area_search(tid, page, base) - return {"list": merged, "page": page, "limit": 90, "total": 999999} - - payload = { - "type_id": tid, - "page": page, - "area": ext.get("area", "全部"), - "year": ext.get("year", "全部"), - "sort": ext.get("by", ext.get("sort", "最新")), - "lang": ext.get("lang", "全部"), - "class": ext.get("class", "全部"), - } - res = self._api_post("typeFilterVodList", payload) - items = [] - for item in (res or {}).get("recommend_list", []): - items.append({ - "vod_id": str(item.get("vod_id", "")), - "vod_name": item.get("vod_name", ""), - "vod_pic": item.get("vod_pic", ""), - "vod_remarks": item.get("vod_remarks", ""), - }) - return {"list": items, "page": page, "limit": 90, "total": 999999} - - JUNK_KEYWORDS = ["防走丢", "群", "防失群", "官网"] - - def detailContent(self, ids): - self.init() - vod_id = ids[0] - data = None - for endpoint in ["vodDetail", "vodDetail2"]: - try: - data = self._api_post(endpoint, {"vod_id": vod_id}) - if data: - break - except Exception: - continue - if not data: - return {"list": []} - - vod = data.get("vod", {}) - lines = [] - name_count = {} - line_id = 1 - - for line in data.get("vod_play_list", []): - info = line.get("player_info", {}) - name = info.get("show", "") - - if any(kw in name for kw in self.JUNK_KEYWORDS): - name = f"{line_id}线" - info["show"] = name - - count = name_count.get(name, 0) + 1 - name_count[name] = count - if count > 1: - name = f"{name}{count}" - info["show"] = name - - urls = line.get("urls", []) - if not urls: - line_id += 1 - continue - - play_items = [] - for ep in urls: - payload = ",".join([ - info.get("parse", ""), - ep.get("url", ""), - "token+" + ep.get("token", ""), - str(info.get("player_parse_type", "")), - str(info.get("parse_type", "")), - ]) - play_items.append(f"{ep.get('name', '')}${name}@@direct@@{payload}") - - if play_items: - lines.append({ - "display": name, - "urls": "#".join(play_items), - }) - line_id += 1 - - return { - "list": [{ - "vod_id": vod_id, - "vod_name": vod.get("vod_name", ""), - "vod_pic": vod.get("vod_pic", ""), - "vod_remarks": vod.get("vod_remarks", ""), - "vod_content": vod.get("vod_content", ""), - "vod_actor": (vod.get("vod_actor") or "").replace("演员", ""), - "vod_director": (vod.get("vod_director") or "").replace("导演", ""), - "vod_year": (vod.get("vod_year") or "") + "年" if vod.get("vod_year") else "", - "vod_area": vod.get("vod_area", ""), - "vod_play_from": "$$$".join(l["display"] for l in lines), - "vod_play_url": "$$$".join(l["urls"] for l in lines), - }] - } - - def _get_verify_code(self): - import uuid - try: - uid = str(uuid.uuid4()) - verify_url = f"{self.host}{self.api_path}/verify/create?key={uid}" - rsp = self.fetch(verify_url, headers={"User-Agent": self.ua}, - timeout=10, verify=False) - b64_img = base64.b64encode(rsp.content).decode("utf-8") - ocr_url = "http://154.222.22.188:9898/ocr/b64/text" - ocr_rsp = self.post(ocr_url, data=b64_img, - headers={"User-Agent": self.ua, "Content-Type": "text/plain"}, - timeout=10, verify=False) - code = (ocr_rsp.text or "").strip() - if not code: - return None - replacements = { - "y": "9", "口": "0", "q": "0", "u": "0", "o": "0", - ">": "1", "d": "0", "b": "8", "已": "2", "D": "0", "五": "5", - } - code = "".join(replacements.get(c, c) for c in code) - if not re.match(r"^\d{4}$", code): - return None - return {"uuid": uid, "code": code} - except Exception as e: - self.log(f"验证码获取失败: {e}") - return None - - def searchContent(self, key, quick, pg="1"): - self.init() - payload = {"keywords": key, "type_id": "0", "page": str(pg)} - if self.search_verify: - verify = self._get_verify_code() - if verify: - payload["code"] = verify["code"] - payload["key"] = verify["uuid"] - res = self._api_post(self.search_endpoint, payload) - if not res: - return {"list": [], "page": int(pg), "limit": 90, "total": 999999} - raw = res.get("search_list", []) - filtered = [i for i in raw if "屏蔽预留" not in (i.get("vod_class") or "").lower()] - kw = (key or "").strip().lower() - if kw: - filtered = [ - i for i in filtered - if kw in " ".join([ - i.get("vod_name", ""), - i.get("vod_remarks", ""), - i.get("vod_class", ""), - ]).lower() - ] - items = [{ - "vod_id": str(i.get("vod_id", "")), - "vod_name": i.get("vod_name", ""), - "vod_pic": i.get("vod_pic", ""), - "vod_remarks": f"{i.get('vod_year', '')} {i.get('vod_class', '')}".strip(), - } for i in filtered] - return {"list": items, "page": int(pg), "limit": 90, "total": 999999} - - def playerContent(self, flag, id, vipFlags): - try: - parts = id.split("@@") - if len(parts) < 3: - return {"parse": 0, "jx": 0, "url": ""} - payload = parts[2] - - arr = payload.split(",") - if len(arr) < 5: - return {"parse": 0, "jx": 0, "url": ""} - - parse_api = arr[0] - kurl = arr[1] - token = arr[2].replace("token+", "") if arr[2].startswith("token+") else arr[2] - player_parse_type = arr[3] - parse_type = arr[4] - - try: - from urllib.parse import unquote - kurl = unquote(kurl) - except Exception: - pass - - header = {"User-Agent": "Dalvik/2.1.0 (Linux; Android 14)"} - - if parse_type == "0": - return {"parse": 0, "jx": 0, "url": kurl, "header": header} - - if parse_type == "2": - return {"parse": 1, "jx": 1, "url": parse_api + kurl, "header": header} - - if player_parse_type == "2": - try: - rsp = self.fetch(parse_api + kurl, headers={"User-Agent": self.ua}, timeout=10, verify=False) - fetched = rsp.json() - if fetched.get("url"): - return {"parse": 0, "jx": 0, "url": fetched["url"]} - except Exception: - pass - - encrypted_url = self._aes_encrypt(kurl) - res = self._api_post("vodParse", { - "parse_api": parse_api, - "url": encrypted_url, - "player_parse_type": player_parse_type, - "token": token, - }) - if not res or not res.get("json"): - return {"parse": 0, "jx": 0, "url": ""} - inner = json.loads(res["json"]) - return {"parse": 0, "jx": 0, "url": inner.get("url", "")} - except Exception as e: - self.log(f"播放解析失败: {e}") - return {"parse": 0, "jx": 0, "url": ""}