From 5cab55eba6df31e62caaccfe8b7dc9da1b7176f7 Mon Sep 17 00:00:00 2001 From: Harold <8866033@gmail.com> Date: Mon, 20 Apr 2026 11:52:53 +0800 Subject: [PATCH] feat: add heimao player parsing --- py/tests/test_黑猫APP.py | 60 ++++++++++++++++++++++++++++++++++++++++ py/黑猫APP.py | 43 ++++++++++++++++++++++++++++ 2 files changed, 103 insertions(+) diff --git a/py/tests/test_黑猫APP.py b/py/tests/test_黑猫APP.py index c95a626..1a520c8 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 @@ -190,6 +191,65 @@ class TestHeiMaoAppSpider(unittest.TestCase): "$$$蓝光$高清线路12@@direct@@https://parser-c/?url=,https%3A%2F%2Fplay-c,token+tk-c,2,2", ) + def test_player_content_returns_direct_url_for_parse_type_zero(self): + result = self.spider.playerContent( + "🐈‍⬛高清线路", + "高清线路1@@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( + "🐈‍⬛高清线路", + "高清线路1@@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(status_code=200, text='{"url":"https://video.example/direct.m3u8"}') + result = self.spider.playerContent( + "🐈‍⬛高清线路", + "高清线路1@@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( + "🐈‍⬛高清线路", + "高清线路1@@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") + self.assertEqual(mock_api_post.call_args.args[1]["parse_api"], "https://parser/?url=") + self.assertEqual(mock_api_post.call_args.args[1]["token"], "abc") + if __name__ == "__main__": unittest.main() diff --git a/py/黑猫APP.py b/py/黑猫APP.py index 9b4d1b1..70bca09 100644 --- a/py/黑猫APP.py +++ b/py/黑猫APP.py @@ -3,6 +3,7 @@ import base64 import json import sys from datetime import datetime +from urllib.parse import unquote from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes from cryptography.hazmat.primitives.padding import PKCS7 @@ -261,6 +262,9 @@ class Spider(BaseSpider): 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: @@ -319,3 +323,42 @@ class Spider(BaseSpider): } ) 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 == "auto": + 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) + data = json.loads(response.text or "{}") + if data.get("url"): + return {"parse": 0, "jx": 0, "url": data.get("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 {} + try: + inner = json.loads(result.get("json", "{}")) + except Exception: + inner = {} + return {"parse": 0, "jx": 0, "url": inner.get("url", ""), "header": {}}