From d991d99695d37672af371994d9f0fa24770566d2 Mon Sep 17 00:00:00 2001 From: Harold <8866033@gmail.com> Date: Mon, 20 Apr 2026 13:25:20 +0800 Subject: [PATCH 1/3] feat: scaffold mofang app spider --- py/tests/test_魔方APP.py | 32 ++++++++++++++++++++ py/魔方APP.py | 64 ++++++++++++++++++++++++++++++++++++++++ 2 files changed, 96 insertions(+) create mode 100644 py/tests/test_魔方APP.py create mode 100644 py/魔方APP.py diff --git a/py/tests/test_魔方APP.py b/py/tests/test_魔方APP.py new file mode 100644 index 0000000..e0b1909 --- /dev/null +++ b/py/tests/test_魔方APP.py @@ -0,0 +1,32 @@ +import unittest +from importlib.machinery import SourceFileLoader +from pathlib import Path + + +ROOT = Path(__file__).resolve().parents[1] +MODULE = SourceFileLoader("mofang_spider", str(ROOT / "魔方APP.py")).load_module() +Spider = MODULE.Spider + + +class TestMoFangAppSpider(unittest.TestCase): + def setUp(self): + Spider._instance = None + self.spider = Spider() + self.spider.init() + + def test_home_video_content_returns_empty_list(self): + self.assertEqual(self.spider.homeVideoContent(), {"list": []}) + + def test_build_api_url_uses_default_getappapi_path(self): + self.assertEqual( + self.spider._build_api_url("initV119"), + "http://www.613mf4k12.top/api.php/getappapi.index/initV119", + ) + + def test_aes_encrypt_and_decrypt_round_trip(self): + encrypted = self.spider._aes_encrypt('{"msg":"ok"}') + self.assertNotEqual(encrypted, '{"msg":"ok"}') + self.assertEqual(self.spider._aes_decrypt(encrypted), '{"msg":"ok"}') + + def test_replace_code_normalizes_common_ocr_misreads(self): + self.assertEqual(self.spider._replace_code("5y6口"), "5960") diff --git a/py/魔方APP.py b/py/魔方APP.py new file mode 100644 index 0000000..cb92f4e --- /dev/null +++ b/py/魔方APP.py @@ -0,0 +1,64 @@ +# coding=utf-8 +import base64 +import sys + +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 = "魔方APP" + self.host = "http://www.613mf4k12.top" + self.api_path = "/api.php/getappapi" + self.user_agent = "okhttp/3.10.0" + self.aes_key = "1234567887654321" + self.aes_iv = "1234567887654321" + + def init(self, extend=""): + self.init_api = "initV119" + self.search_api = "searchList" + self.search_verify = False + return None + + def getName(self): + return self.name + + def homeVideoContent(self): + return {"list": []} + + def _build_api_url(self, endpoint): + path = str(endpoint or "").lstrip("/") + return f"{self.host}{self.api_path}.index/{path}" + + def _aes_encrypt(self, text): + cipher = AES.new(self.aes_key.encode("utf-8"), AES.MODE_CBC, self.aes_iv.encode("utf-8")) + return base64.b64encode(cipher.encrypt(pad(str(text).encode("utf-8"), AES.block_size))).decode("utf-8") + + def _aes_decrypt(self, text): + cipher = AES.new(self.aes_key.encode("utf-8"), AES.MODE_CBC, self.aes_iv.encode("utf-8")) + return unpad(cipher.decrypt(base64.b64decode(str(text or ""))), AES.block_size).decode("utf-8") + + def _replace_code(self, text): + replacements = { + "y": "9", + "口": "0", + "q": "0", + "u": "0", + "o": "0", + ">": "1", + "d": "0", + "b": "8", + "已": "2", + "D": "0", + "五": "5", + } + return "".join(replacements.get(char, char) for char in str(text or "")) From 70b1994104789e15e60debf55d12431433be43ee Mon Sep 17 00:00:00 2001 From: Harold <8866033@gmail.com> Date: Mon, 20 Apr 2026 13:27:07 +0800 Subject: [PATCH 2/3] feat: add mofang app home category and search support --- py/tests/test_魔方APP.py | 108 +++++++++++++++++++++++ py/魔方APP.py | 179 +++++++++++++++++++++++++++++++++++++++ 2 files changed, 287 insertions(+) diff --git a/py/tests/test_魔方APP.py b/py/tests/test_魔方APP.py index e0b1909..d0803b8 100644 --- a/py/tests/test_魔方APP.py +++ b/py/tests/test_魔方APP.py @@ -1,6 +1,8 @@ import unittest +from datetime import datetime from importlib.machinery import SourceFileLoader from pathlib import Path +from unittest.mock import patch ROOT = Path(__file__).resolve().parents[1] @@ -30,3 +32,109 @@ class TestMoFangAppSpider(unittest.TestCase): def test_replace_code_normalizes_common_ocr_misreads(self): self.assertEqual(self.spider._replace_code("5y6口"), "5960") + + @patch.object(Spider, "_api_post") + def test_home_content_filters_categories_merges_area_and_inserts_current_year(self, mock_api_post): + mock_api_post.return_value = { + "type_list": [ + { + "type_id": "0", + "type_name": "全部", + "recommend_list": [], + "filter_type_list": [], + }, + { + "type_id": "3", + "type_name": "综艺", + "recommend_list": [{"vod_id": "z1"}], + "filter_type_list": [ + {"name": "area", "list": ["全部", "中国大陆", "香港"]}, + {"name": "year", "list": ["全部", "2025", "2024"]}, + {"name": "sort", "list": ["最新", "最热"]}, + ], + }, + { + "type_id": "1", + "type_name": "电影", + "recommend_list": [{"vod_id": "m1"}], + "filter_type_list": [], + }, + ], + "config": {}, + } + content = self.spider.homeContent(False) + self.assertEqual([item["type_name"] for item in content["class"]], ["电影", "综艺"]) + self.assertEqual(content["list"], [{"vod_id": "z1"}, {"vod_id": "m1"}]) + area_values = content["filters"]["3"][0]["value"] + year_values = content["filters"]["3"][1]["value"] + self.assertEqual(area_values[1], {"n": "大陆", "v": "大陆"}) + self.assertIn({"n": str(datetime.now().year), "v": str(datetime.now().year)}, year_values) + self.assertEqual(content["filters"]["3"][2]["key"], "by") + + @patch.object(Spider, "_api_post") + def test_category_content_maps_by_to_sort_and_omits_pagecount(self, mock_api_post): + mock_api_post.return_value = {"recommend_list": [{"vod_id": "movie-1", "vod_name": "分类片"}]} + result = self.spider.categoryContent( + "1", + "2", + False, + {"area": "香港", "year": "2025", "by": "最热", "lang": "粤语", "class": "动作"}, + ) + self.assertEqual( + mock_api_post.call_args.args, + ( + "typeFilterVodList", + { + "type_id": "1", + "page": "2", + "area": "香港", + "year": "2025", + "sort": "最热", + "lang": "粤语", + "class": "动作", + }, + ), + ) + self.assertEqual(result["page"], 2) + self.assertEqual(result["list"][0]["vod_id"], "movie-1") + self.assertNotIn("pagecount", result) + + @patch.object(Spider, "_api_post") + def test_category_content_merges_mainland_results_and_deduplicates(self, mock_api_post): + mock_api_post.side_effect = [ + {"recommend_list": [{"vod_id": "a"}, {"vod_id": "b"}]}, + {"recommend_list": [{"vod_id": "b"}, {"vod_id": "c"}]}, + {"recommend_list": [{"vod_id": "c"}, {"vod_id": "d"}]}, + ] + result = self.spider.categoryContent("1", "1", False, {"area": "大陆"}) + self.assertEqual([item["vod_id"] for item in result["list"]], ["a", "b", "c", "d"]) + self.assertEqual( + [call.args[1]["area"] for call in mock_api_post.call_args_list], + ["中国大陆", "大陆", "内地"], + ) + + @patch.object(Spider, "_api_post") + def test_search_content_filters_reserved_category_and_non_matching_items(self, mock_api_post): + mock_api_post.return_value = { + "search_list": [ + {"vod_id": "1", "vod_name": "繁花", "vod_pic": "p1", "vod_year": "2024", "vod_class": "剧情"}, + {"vod_id": "2", "vod_name": "别的片", "vod_pic": "p2", "vod_year": "2024", "vod_class": "屏蔽预留"}, + {"vod_id": "3", "vod_name": "完全无关", "vod_pic": "p3", "vod_year": "2023", "vod_class": "动作"}, + ] + } + result = self.spider.searchContent("繁花", False, "1") + self.assertEqual( + result["list"], + [{"vod_id": "1", "vod_name": "繁花", "vod_pic": "p1", "vod_remarks": "2024 剧情"}], + ) + self.assertNotIn("pagecount", result) + + @patch.object(Spider, "_get_verification_code") + @patch.object(Spider, "_api_post") + def test_search_content_returns_empty_list_when_verify_required_but_ocr_fails(self, mock_api_post, mock_verify): + mock_api_post.return_value = {"config": {"system_search_verify_status": 1}, "type_list": []} + mock_verify.return_value = None + self.spider.search_verify = True + result = self.spider.searchContent("繁花", False, "1") + self.assertEqual(result["list"], []) + self.assertEqual(result.get("msg"), "验证码获取失败") diff --git a/py/魔方APP.py b/py/魔方APP.py index cb92f4e..8ffacfd 100644 --- a/py/魔方APP.py +++ b/py/魔方APP.py @@ -1,6 +1,8 @@ # coding=utf-8 import base64 +import json import sys +from datetime import datetime try: from Crypto.Cipher import AES @@ -22,6 +24,17 @@ class Spider(BaseSpider): self.user_agent = "okhttp/3.10.0" self.aes_key = "1234567887654321" self.aes_iv = "1234567887654321" + self.category_config = { + "blockedNames": ["全部"], + "renameMap": {}, + "forceOrder": ["电影", "连续剧", "综艺", "动漫", "短剧", "直播"], + } + self.area_merge_config = { + "enabled": True, + "displayName": "大陆", + "mergeList": ["中国大陆", "大陆", "内地"], + } + self.ocr_api = "http://154.222.22.188:9898/ocr/b64/text" def init(self, extend=""): self.init_api = "initV119" @@ -39,6 +52,12 @@ class Spider(BaseSpider): path = str(endpoint or "").lstrip("/") return f"{self.host}{self.api_path}.index/{path}" + def _headers(self): + return { + "User-Agent": self.user_agent, + "Accept-Encoding": "gzip", + } + def _aes_encrypt(self, text): cipher = AES.new(self.aes_key.encode("utf-8"), AES.MODE_CBC, self.aes_iv.encode("utf-8")) return base64.b64encode(cipher.encrypt(pad(str(text).encode("utf-8"), AES.block_size))).decode("utf-8") @@ -62,3 +81,163 @@ class Spider(BaseSpider): "五": "5", } return "".join(replacements.get(char, char) for char in str(text or "")) + + def _api_post(self, endpoint, payload=None): + data = {} if payload is None else dict(payload) + response = self.post( + self._build_api_url(endpoint), + data=data, + headers=self._headers(), + timeout=10, + verify=False, + ) + body = json.loads(response.text or "{}") + if not body.get("data"): + return None + return json.loads(self._aes_decrypt(body["data"])) + + def _process_classes(self, type_list): + blocked = set(self.category_config.get("blockedNames", [])) + rename_map = self.category_config.get("renameMap", {}) + order_map = {name: index for index, name in enumerate(self.category_config.get("forceOrder", []))} + classes = [ + { + "type_id": item.get("type_id", ""), + "type_name": rename_map.get(item.get("type_name", ""), item.get("type_name", "")), + } + for item in type_list + if item.get("type_name") not in blocked + ] + classes.sort(key=lambda item: order_map.get(item["type_name"], 999)) + return classes + + def _merge_area_filter(self, values): + merge_values = set(self.area_merge_config.get("mergeList", [])) + raw_values = list(values) + if not self.area_merge_config.get("enabled") or not any(value in merge_values for value in raw_values): + return raw_values + result = [value for value in raw_values if value not in merge_values] + index = result.index("全部") + 1 if "全部" in result else 0 + result.insert(index, self.area_merge_config.get("displayName", "大陆")) + return result + + def _convert_filters(self, type_list): + name_map = {"class": "类型", "area": "地区", "lang": "语言", "year": "年份", "sort": "排序"} + current_year = str(datetime.now().year) + filters = {} + for item in type_list: + entries = [] + for config in item.get("filter_type_list", []): + raw_name = config.get("name", "") + values = list(config.get("list", [])) + if raw_name == "area": + values = self._merge_area_filter(values) + if raw_name == "year" and current_year not in values: + index = values.index("全部") + 1 if "全部" in values else 0 + values.insert(index, current_year) + entries.append( + { + "key": "by" if raw_name == "sort" else raw_name, + "name": name_map.get(raw_name, raw_name), + "value": [{"n": value, "v": value} for value in values], + } + ) + filters[item.get("type_id", "")] = entries + return filters + + def homeContent(self, filter): + init_data = self._api_post(self.init_api) or {"type_list": [], "config": {}} + self.search_verify = bool(init_data.get("config", {}).get("system_search_verify_status")) + return { + "class": self._process_classes(init_data.get("type_list", [])), + "filters": self._convert_filters(init_data.get("type_list", [])), + "list": [vod for item in init_data.get("type_list", []) for vod in item.get("recommend_list", [])], + } + + def _page_result(self, items, pg, limit=90, total=None): + page = int(pg) + return { + "page": page, + "limit": limit, + "total": len(items) if total is None else total, + "list": items, + } + + def _merge_area_search(self, tid, pg, extend): + seen = set() + merged = [] + for area in self.area_merge_config.get("mergeList", []): + payload = { + "type_id": str(tid), + "page": str(pg), + "area": area, + "year": extend.get("year", "全部"), + "sort": extend.get("by", "最新"), + "lang": extend.get("lang", "全部"), + "class": extend.get("class", "全部"), + } + result = self._api_post("typeFilterVodList", payload) or {} + for item in result.get("recommend_list", []): + vod_id = item.get("vod_id") + if vod_id in seen: + continue + seen.add(vod_id) + merged.append(item) + return merged + + def categoryContent(self, tid, pg, filter, extend): + options = dict(extend or {}) + if options.get("area") == self.area_merge_config.get("displayName"): + return self._page_result(self._merge_area_search(tid, pg, options), pg) + payload = { + "type_id": str(tid), + "page": str(pg), + "area": options.get("area", "全部"), + "year": options.get("year", "全部"), + "sort": options.get("by", "最新"), + "lang": options.get("lang", "全部"), + "class": options.get("class", "全部"), + } + result = self._api_post("typeFilterVodList", payload) or {} + return self._page_result(result.get("recommend_list", []), pg) + + def _get_verification_code(self): + return None + + def searchContent(self, key, quick, pg="1"): + if self.search_verify: + verify = self._get_verification_code() + if not verify: + return { + "page": int(pg), + "limit": 90, + "total": 0, + "list": [], + "msg": "验证码获取失败", + } + result = self._api_post( + self.search_api, + {"keywords": str(key or ""), "type_id": "0", "page": str(pg)}, + ) or {} + keyword = str(key or "").strip().lower() + items = [] + for item in result.get("search_list", []): + vod_class = str(item.get("vod_class", "")) + text = " ".join( + [str(item.get("vod_name", "")), str(item.get("vod_remarks", "")), vod_class] + ).lower() + if "屏蔽预留" in vod_class: + continue + if keyword and keyword not in text: + continue + items.append( + { + "vod_id": item.get("vod_id", ""), + "vod_name": item.get("vod_name", ""), + "vod_pic": item.get("vod_pic", ""), + "vod_remarks": " ".join( + value for value in [str(item.get("vod_year", "")), vod_class] if value + ).strip(), + } + ) + return self._page_result(items, pg) From 3f1b781846012703b74fd5923d64cc7672770d91 Mon Sep 17 00:00:00 2001 From: Harold <8866033@gmail.com> Date: Mon, 20 Apr 2026 13:28:55 +0800 Subject: [PATCH 3/3] feat: add mofang app detail and play parsing --- py/tests/test_魔方APP.py | 125 +++++++++++++++++++++++++++++ py/魔方APP.py | 166 ++++++++++++++++++++++++++++++++++++++- 2 files changed, 290 insertions(+), 1 deletion(-) diff --git a/py/tests/test_魔方APP.py b/py/tests/test_魔方APP.py index d0803b8..5dcd073 100644 --- a/py/tests/test_魔方APP.py +++ b/py/tests/test_魔方APP.py @@ -2,6 +2,7 @@ import unittest from datetime import datetime from importlib.machinery import SourceFileLoader from pathlib import Path +from types import SimpleNamespace from unittest.mock import patch @@ -138,3 +139,127 @@ class TestMoFangAppSpider(unittest.TestCase): result = self.spider.searchContent("繁花", False, "1") self.assertEqual(result["list"], []) self.assertEqual(result.get("msg"), "验证码获取失败") + + @patch.object(Spider, "_api_post") + def test_detail_content_falls_back_to_second_endpoint_sorts_lines_and_blocks_disabled_lines(self, mock_api_post): + mock_api_post.side_effect = [ + None, + { + "vod": { + "vod_name": "示例影片", + "vod_pic": "poster.jpg", + "vod_remarks": "更新至10集", + "vod_content": "一段剧情", + "vod_actor": "演员甲/演员乙", + "vod_director": "导演甲", + "vod_year": "2025", + "vod_area": "中国大陆", + }, + "vod_play_list": [ + { + "player_info": { + "show": "腾讯视频", + "parse": "https://parser-a/?url=", + "player_parse_type": "1", + "parse_type": "1", + }, + "urls": [{"name": "正片", "url": "https%3A%2F%2Fplay-a", "token": "tk-a"}], + }, + { + "player_info": { + "show": "防走丢加群", + "parse": "https://parser-b/?url=", + "player_parse_type": "1", + "parse_type": "0", + }, + "urls": [{"name": "备用", "url": "https%3A%2F%2Fplay-b", "token": "tk-b"}], + }, + { + "player_info": { + "show": "YY", + "parse": "https://parser-c/?url=", + "player_parse_type": "2", + "parse_type": "2", + }, + "urls": [{"name": "蓝光", "url": "https%3A%2F%2Fplay-c", "token": "tk-c"}], + }, + ], + }, + ] + result = self.spider.detailContent(["123"]) + vod = result["list"][0] + self.assertEqual(vod["vod_id"], "123") + self.assertEqual(vod["vod_name"], "示例影片") + self.assertEqual(vod["vod_year"], "2025年") + self.assertEqual(vod["vod_play_from"], "⭐️腾讯视频$$$1线") + self.assertEqual( + vod["vod_play_url"], + "正片$腾讯视频@@direct@@https://parser-a/?url=,https%3A%2F%2Fplay-a,token+tk-a,1,1" + "$$$备用$1线@@direct@@https://parser-b/?url=,https%3A%2F%2Fplay-b,token+tk-b,1,0", + ) + + @patch.object(Spider, "fetch") + def test_get_verification_code_fetches_image_calls_ocr_and_returns_digits(self, mock_fetch): + image_response = SimpleNamespace(content=b"fake-image") + ocr_response = SimpleNamespace(text="5y6口") + mock_fetch.side_effect = [image_response, ocr_response] + result = self.spider._get_verification_code() + self.assertEqual(result["code"], "5960") + self.assertIn("uuid", result) + + def test_player_content_returns_direct_url_for_parse_type_zero(self): + result = self.spider.playerContent( + "⭐️腾讯视频", + "腾讯视频@@direct@@https://parser/?url=,https%3A%2F%2Fvideo.example%2Fraw.m3u8,token+abc,1,0", + "", + ) + self.assertEqual( + result, + { + "parse": 0, + "jx": 0, + "url": "https://video.example/raw.m3u8", + "header": {"User-Agent": "Dalvik/2.1.0 (Linux; Android 14)"}, + }, + ) + + def test_player_content_returns_parser_url_for_parse_type_two(self): + result = self.spider.playerContent( + "⭐️腾讯视频", + "腾讯视频@@direct@@https://parser/?url=,https%3A%2F%2Fvideo.example%2Fneed-jx.m3u8,token+abc,1,2", + "", + ) + self.assertEqual( + result, + { + "parse": 1, + "jx": 1, + "url": "https://parser/?url=https://video.example/need-jx.m3u8", + "header": {"User-Agent": "Dalvik/2.1.0 (Linux; Android 14)"}, + }, + ) + + @patch.object(Spider, "fetch") + def test_player_content_uses_player_parse_type_two_direct_json(self, mock_fetch): + mock_fetch.return_value = SimpleNamespace(text='{"url":"https://video.example/direct.m3u8"}') + result = self.spider.playerContent( + "⭐️腾讯视频", + "腾讯视频@@direct@@https://parser/?url=,https%3A%2F%2Fvideo.example%2Fwrapped,token+abc,2,1", + "", + ) + self.assertEqual(result["parse"], 0) + self.assertEqual(result["jx"], 0) + self.assertEqual(result["url"], "https://video.example/direct.m3u8") + + @patch.object(Spider, "_api_post") + def test_player_content_falls_back_to_vod_parse(self, mock_api_post): + mock_api_post.return_value = {"json": '{"url":"https://video.example/final.m3u8"}'} + result = self.spider.playerContent( + "⭐️腾讯视频", + "腾讯视频@@direct@@https://parser/?url=,https%3A%2F%2Fvideo.example%2Fencrypted,token+abc,1,1", + "", + ) + self.assertEqual(result["parse"], 0) + self.assertEqual(result["jx"], 0) + self.assertEqual(result["url"], "https://video.example/final.m3u8") + self.assertEqual(mock_api_post.call_args.args[0], "vodParse") diff --git a/py/魔方APP.py b/py/魔方APP.py index 8ffacfd..ae76836 100644 --- a/py/魔方APP.py +++ b/py/魔方APP.py @@ -2,7 +2,9 @@ import base64 import json import sys +import uuid from datetime import datetime +from urllib.parse import unquote try: from Crypto.Cipher import AES @@ -35,6 +37,26 @@ class Spider(BaseSpider): "mergeList": ["中国大陆", "大陆", "内地"], } self.ocr_api = "http://154.222.22.188:9898/ocr/b64/text" + self.line_config = { + "4K线路db4": {"displayName": "🌷4K高清1", "order": 101, "mode": "direct", "enabled": True}, + "4K线路3U": {"displayName": "🌷4K高清2", "order": 102, "mode": "direct", "enabled": True}, + "4K极清": {"displayName": "🌷4K高清3", "order": 103, "mode": "direct", "enabled": True}, + "4K线路": {"displayName": "🌷4K高清4", "order": 104, "mode": "direct", "enabled": True}, + "腾讯视频": {"displayName": "⭐️腾讯视频", "order": 201, "mode": "direct", "enabled": True}, + "芒果视频": {"displayName": "⭐️芒果视频", "order": 202, "mode": "direct", "enabled": True}, + "奇艺视频": {"displayName": "⭐️奇艺视频", "order": 203, "mode": "direct", "enabled": True}, + "优酷视频": {"displayName": "⭐️优酷视频", "order": 204, "mode": "direct", "enabled": True}, + "电影天堂": {"displayName": "❤️电影天堂", "order": 301, "mode": "direct", "enabled": True}, + "如意资源": {"displayName": "❤️如意资源", "order": 306, "mode": "direct", "enabled": True}, + "红牛在线": {"displayName": "❤️红牛在线", "order": 303, "mode": "direct", "enabled": True}, + "豆瓣在线": {"displayName": "❤️豆瓣在线", "order": 304, "mode": "direct", "enabled": True}, + "暴风资源": {"displayName": "❤️暴风资源", "order": 305, "mode": "direct", "enabled": True}, + "极速资源": {"displayName": "❤️极速资源", "order": 307, "mode": "direct", "enabled": True}, + "茅台资源": {"displayName": "❤️茅台资源", "order": 308, "mode": "direct", "enabled": True}, + "水印资源": {"displayName": "❤️水印资源", "order": 309, "mode": "direct", "enabled": False}, + "无尽": {"displayName": "❤️无尽资源", "order": 310, "mode": "direct", "enabled": True}, + "YY": {"displayName": "❤️YY资源", "order": 311, "mode": "direct", "enabled": False}, + } def init(self, extend=""): self.init_api = "initV119" @@ -202,7 +224,27 @@ class Spider(BaseSpider): return self._page_result(result.get("recommend_list", []), pg) def _get_verification_code(self): - return None + uuid_value = str(uuid.uuid4()) + image_response = self.fetch( + f"{self.host}{self.api_path}.index/verify/create?key={uuid_value}", + headers={"User-Agent": self.user_agent}, + timeout=10, + verify=False, + ) + image_b64 = base64.b64encode(image_response.content).decode("utf-8") + ocr_response = self.fetch( + self.ocr_api, + data=image_b64, + headers={"User-Agent": self.user_agent, "Content-Type": "text/plain"}, + timeout=10, + verify=False, + ) + code = self._replace_code(getattr(ocr_response, "text", "")) + if len(code) == 3: + code = code.replace("566", "5066").replace("066", "1666") + if not code.isdigit() or len(code) != 4: + return None + return {"uuid": uuid_value, "code": code} def searchContent(self, key, quick, pg="1"): if self.search_verify: @@ -241,3 +283,125 @@ class Spider(BaseSpider): } ) return self._page_result(items, pg) + + def _get_line_config(self, name): + for key, value in self.line_config.items(): + if key in str(name or ""): + return value + return None + + def _line_display(self, name): + config = self._get_line_config(name) + return config.get("displayName", name) if config else name + + def _line_order(self, name): + config = self._get_line_config(name) + return config.get("order", 999) if config else 999 + + def _line_mode(self, name): + config = self._get_line_config(name) + return config.get("mode", "direct") if config else "direct" + + def _line_blocked(self, name): + config = self._get_line_config(name) + return bool(config) and config.get("enabled") is False + + def _player_headers(self): + return {"User-Agent": "Dalvik/2.1.0 (Linux; Android 14)"} + + def detailContent(self, ids): + items = [] + for vod_id in ids: + payload = {"vod_id": str(vod_id)} + data = self._api_post("vodDetail", payload) or self._api_post("vodDetail2", payload) + if not data: + continue + lines = [] + name_count = {} + fallback_index = 1 + for line in data.get("vod_play_list", []): + player_info = line.get("player_info", {}) + raw_name = str(player_info.get("show", "")) + if any(keyword in raw_name for keyword in ["防走丢", "群", "防失群", "官网"]): + raw_name = f"{fallback_index}线" + fallback_index += 1 + name_count[raw_name] = name_count.get(raw_name, 0) + 1 + line_name = raw_name if name_count[raw_name] == 1 else f"{raw_name}{name_count[raw_name]}" + if self._line_blocked(line_name): + continue + play_items = [] + for episode in line.get("urls", []): + payload_text = ",".join( + [ + str(player_info.get("parse", "")), + str(episode.get("url", "")), + "token+" + str(episode.get("token", "")), + str(player_info.get("player_parse_type", "")), + str(player_info.get("parse_type", "")), + ] + ) + play_items.append( + f"{episode.get('name', '')}${line_name}@@{self._line_mode(line_name)}@@{payload_text}" + ) + if play_items: + lines.append( + { + "display": self._line_display(line_name), + "order": self._line_order(line_name), + "urls": "#".join(play_items), + } + ) + lines.sort(key=lambda item: item["order"]) + vod = data.get("vod", {}) + items.append( + { + "vod_id": str(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": str(vod.get("vod_actor", "")).replace("演员", ""), + "vod_director": str(vod.get("vod_director", "")).replace("导演", ""), + "vod_year": f"{vod.get('vod_year', '')}年" if vod.get("vod_year") else "", + "vod_area": vod.get("vod_area", ""), + "vod_play_from": "$$$".join(line["display"] for line in lines), + "vod_play_url": "$$$".join(line["urls"] for line in lines), + } + ) + return {"list": items} + + def playerContent(self, flag, id, vipFlags): + parts = str(id or "").split("@@", 2) + line_name = parts[0] if len(parts) > 0 else "" + mode = parts[1] if len(parts) > 1 else "direct" + payload = parts[2] if len(parts) > 2 else "" + if self._line_blocked(line_name): + return {"parse": 0, "jx": 0, "url": "", "header": {}} + if mode != "direct": + return {"parse": 0, "jx": 0, "url": "", "header": {}} + fields = payload.split(",", 4) + while len(fields) < 5: + fields.append("") + parse_api, play_url, token_with_prefix, player_parse_type, parse_type = fields + real_url = unquote(play_url) + token = token_with_prefix.replace("token+", "", 1) + if parse_type == "0": + return {"parse": 0, "jx": 0, "url": real_url, "header": self._player_headers()} + if parse_type == "2": + return {"parse": 1, "jx": 1, "url": parse_api + real_url, "header": self._player_headers()} + if player_parse_type == "2": + response = self.fetch(parse_api + real_url, headers=self._headers(), timeout=10, verify=False) + direct = json.loads(response.text or "{}") + if direct.get("url"): + return {"parse": 0, "jx": 0, "url": direct["url"], "header": {}} + result = self._api_post( + "vodParse", + { + "parse_api": parse_api, + "url": self._aes_encrypt(real_url), + "player_parse_type": player_parse_type, + "token": token, + }, + ) or {} + inner = json.loads(result.get("json", "{}")) + return {"parse": 0, "jx": 0, "url": inner.get("url", ""), "header": {}}