yeyinly/tvboxzt
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

feat: add letu player parsing

Browse Source
This commit is contained in:
Harold
2026-04-20 09:36:04 +08:00
parent 5aa57f5100
commit 15873e87dc
2 changed files with 96 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files
py/tests/test_乐兔.py
+42
View File
@@ -1,6 +1,8 @@
import unittest import unittest
import base64
from importlib.machinery import SourceFileLoader from importlib.machinery import SourceFileLoader
from pathlib import Path from pathlib import Path
from urllib.parse import quote
from unittest.mock import patch from unittest.mock import patch
@@ -135,6 +137,46 @@ class TestLeTuSpider(unittest.TestCase):
self.spider.detailContent(["detail/demo"]) self.spider.detailContent(["detail/demo"])
self.assertEqual(mock_request_html.call_args.args[0], "https://www.letu.me/detail/demo.html") self.assertEqual(mock_request_html.call_args.args[0], "https://www.letu.me/detail/demo.html")
@patch.object(Spider, "_request_html")
def test_player_content_returns_direct_json_url(self, mock_request_html):
mock_request_html.return_value = '{"code":200,"url":"https://video.example/direct.m3u8"}'
result = self.spider.playerContent("线路A", "play/123-1-1", {})
self.assertEqual(result["parse"], 0)
self.assertEqual(result["jx"], 0)
self.assertEqual(result["url"], "https://video.example/direct.m3u8")
@patch.object(Spider, "_request_html")
def test_player_content_decodes_rose_base64_url(self, mock_request_html):
encoded = quote(base64.b64encode(b"https://video.example/rose.m3u8").decode("utf-8"))
mock_request_html.return_value = '{"code":200,"url":"rose_' + encoded + '"}'
result = self.spider.playerContent("线路A", "play/123-1-1", {})
self.assertEqual(result["parse"], 0)
self.assertEqual(result["url"], "https://video.example/rose.m3u8")
def test_extract_player_data_reads_player_json(self):
html = '<script>var player_aaaa={"url":"https%3A%2F%2Fvideo.example%2Fenc1.m3u8","encrypt":"1"};</script>'
self.assertEqual(self.spider._extract_player_data(html)["encrypt"], "1")
@patch.object(Spider, "_request_html")
def test_player_content_supports_encrypt_1_and_encrypt_2(self, mock_request_html):
encoded = quote(base64.b64encode(b"https://video.example/enc2.m3u8").decode("utf-8"))
mock_request_html.side_effect = [
'<script>var player_aaaa={"url":"https%3A%2F%2Fvideo.example%2Fenc1.m3u8","encrypt":"1"};</script>',
'<script>var player_aaaa={"url":"' + encoded + '","encrypt":"2"};</script>',
]
encrypt_1 = self.spider.playerContent("线路A", "play/123-1-1", {})
encrypt_2 = self.spider.playerContent("线路A", "play/123-1-2", {})
self.assertEqual(encrypt_1["url"], "https://video.example/enc1.m3u8")
self.assertEqual(encrypt_2["url"], "https://video.example/enc2.m3u8")
@patch.object(Spider, "_request_html")
def test_player_content_falls_back_to_system_parse(self, mock_request_html):
mock_request_html.return_value = "<html></html>"
result = self.spider.playerContent("线路A", "play/123-1-1", {})
self.assertEqual(result["parse"], 1)
self.assertEqual(result["jx"], 1)
self.assertEqual(result["url"], "https://www.letu.me/play/123-1-1.html")
if __name__ == "__main__": if __name__ == "__main__":
unittest.main() unittest.main()
py/乐兔.py
+54 -1
View File
@@ -1,7 +1,9 @@
# coding=utf-8 # coding=utf-8
import base64
import json
import re import re
import sys import sys
from urllib.parse import quote, urljoin from urllib.parse import quote, unquote, urljoin
from base.spider import Spider as BaseSpider from base.spider import Spider as BaseSpider
@@ -203,3 +205,54 @@ class Spider(BaseSpider):
parsed = self._parse_detail_page(self._request_html(detail_url), vod_id) parsed = self._parse_detail_page(self._request_html(detail_url), vod_id)
result["list"].extend(parsed.get("list", [])) result["list"].extend(parsed.get("list", []))
return result return result
def _base64_decode(self, value):
text = re.sub(r"\s+", "", str(value or ""))
text += "=" * ((4 - len(text) % 4) % 4)
try:
return base64.b64decode(text).decode("utf-8")
except Exception:
return ""
def _decode_rose_url(self, value):
raw = str(value or "")[5:]
decoded = self._base64_decode(unquote(raw))
if decoded:
return decoded
return self._base64_decode(raw)
def _extract_player_data(self, html):
matched = re.search(r"player_[a-z0-9]+\s*=\s*(\{[\s\S]*?\})", str(html or ""), re.I)
if not matched:
return None
try:
return json.loads(matched.group(1))
except Exception:
return None
def playerContent(self, flag, id, vipFlags):
play_url = self._decode_play_id(id) or self._build_url(id)
body = self._request_html(play_url)
try:
payload = json.loads(body)
except Exception:
payload = None
if isinstance(payload, dict) and str(payload.get("code")) == "200" and payload.get("url"):
video_url = str(payload.get("url"))
if video_url.startswith("rose_"):
video_url = self._decode_rose_url(video_url)
elif video_url.startswith("/"):
video_url = self._build_url(video_url)
if video_url:
return {"parse": 0, "jx": 0, "url": video_url, "header": dict(self.headers)}
player = self._extract_player_data(body)
if player:
video_url = str(player.get("url") or "")
encrypt = str(player.get("encrypt") or "")
if encrypt == "1":
video_url = unquote(video_url)
elif encrypt == "2":
video_url = self._base64_decode(unquote(video_url))
if video_url:
return {"parse": 0, "jx": 0, "url": video_url, "header": dict(self.headers)}
return {"parse": 1, "jx": 1, "url": play_url, "header": dict(self.headers)}