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": {}}