feat: add mofang app detail and play parsing

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")

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