diff --git a/py/tests/test_dbku.py b/py/tests/test_dbku.py index 472692a..0e01dfc 100644 --- a/py/tests/test_dbku.py +++ b/py/tests/test_dbku.py @@ -1,4 +1,5 @@ import unittest +import base64 from importlib.machinery import SourceFileLoader from pathlib import Path from unittest.mock import patch @@ -128,6 +129,32 @@ class TestDBKUSpider(unittest.TestCase): self.assertEqual(result["list"][0]["vod_id"], "https://www.dbku.tv/voddetail/200.html") self.assertEqual(result["list"][0]["vod_name"], "详情影片") + def test_parse_player_data_reads_json_block(self): + html = '' + data = self.spider._parse_player_data(html) + self.assertEqual(data["url"], "https://video.example/a.m3u8") + + def test_decode_play_url_by_encrypt_supports_modes_0_1_2_3(self): + self.assertEqual( + self.spider._decode_play_url_by_encrypt("https://video.example/raw.m3u8", 0), + "https://video.example/raw.m3u8", + ) + self.assertEqual( + self.spider._decode_play_url_by_encrypt("https%3A//video.example/escape.m3u8", 1), + "https://video.example/escape.m3u8", + ) + mode2 = base64.b64encode("https://video.example/base64.m3u8".encode("utf-8")).decode("utf-8") + self.assertEqual( + self.spider._decode_play_url_by_encrypt(mode2, 2), + "https://video.example/base64.m3u8", + ) + mode3_raw = "ABCDEFGHhttps://video.example/trimmed.m3u8HGFEDCBA" + mode3 = base64.b64encode(mode3_raw.encode("utf-8")).decode("utf-8") + self.assertEqual( + self.spider._decode_play_url_by_encrypt("12345678" + mode3, 3), + "https://video.example/trimmed.m3u8", + ) + if __name__ == "__main__": unittest.main() diff --git a/py/独播库.py b/py/独播库.py index 448ca0e..6e74d97 100644 --- a/py/独播库.py +++ b/py/独播库.py @@ -1,7 +1,9 @@ # coding=utf-8 +import base64 +import json import re import sys -from urllib.parse import quote +from urllib.parse import quote, unquote from lxml import etree @@ -212,3 +214,35 @@ class Spider(BaseSpider): vod_id = ids[0] html = self._request_html(vod_id, expect_xpath="//*[contains(@class,'title')]|//a[contains(@href,'/vodplay/')]") return self._parse_detail_page(html, vod_id) + + def _parse_player_data(self, html): + matched = re.search(r"var\s+player_data\s*=\s*(\{[\s\S]*?\})\s*;?\s*", html, re.I) + if not matched: + matched = re.search(r"var\s+player_[^=]*\s*=\s*(\{[\s\S]*?\})\s*;?\s*", html, re.I) + if not matched: + return None + try: + return json.loads(matched.group(1)) + except Exception: + return None + + def _decode_play_url_by_encrypt(self, value, encrypt): + raw = str(value or "") + mode = int(encrypt or 0) + if not raw: + return "" + try: + if mode == 1: + return unquote(raw) + if mode == 2: + decoded = base64.b64decode(raw).decode("utf-8") + return unquote(decoded) + if mode == 3: + text = raw[8:] if len(raw) > 16 else raw + text = base64.b64decode(text).decode("utf-8") + if len(text) > 16: + text = text[8:-8] + return text + return raw + except Exception: + return raw