diff --git a/py/tests/test_路漫漫.py b/py/tests/test_路漫漫.py
index e139697..48dbb57 100644
--- a/py/tests/test_路漫漫.py
+++ b/py/tests/test_路漫漫.py
@@ -1,8 +1,12 @@
+from base64 import b64encode
import unittest
from importlib.machinery import SourceFileLoader
from pathlib import Path
from unittest.mock import patch
+from Crypto.Cipher import AES
+from Crypto.Util.Padding import pad
+
ROOT = Path(__file__).resolve().parents[1]
MODULE = SourceFileLoader("lumman_spider", str(ROOT / "路漫漫.py")).load_module()
Spider = MODULE.Spider
@@ -119,6 +123,44 @@ class TestLuManManSpider(unittest.TestCase):
self.assertEqual(vod["vod_play_from"], "播放列表")
self.assertEqual(vod["vod_play_url"], "正片$vod/play/1-1-1.html")
+ def test_decrypt_token_returns_plaintext(self):
+ key = b"ejjooopppqqqrwww"
+ iv = b"1348987635684651"
+ cipher = AES.new(key, AES.MODE_CBC, iv)
+ token = b64encode(cipher.encrypt(pad(b"plain-token", AES.block_size))).decode("utf-8")
+ self.assertEqual(self.spider._decrypt_token(token), "plain-token")
+
+ def test_decrypt_token_handles_bad_ciphertext(self):
+ self.assertEqual(self.spider._decrypt_token("not-valid"), "")
+
+ @patch.object(Spider, "_resolve_player_url")
+ def test_player_content_returns_direct_media_without_parse(self, mock_resolve):
+ result = self.spider.playerContent("在线播放", "https://cdn.example.com/video.m3u8", {})
+ self.assertEqual(result["parse"], 0)
+ self.assertEqual(result["url"], "https://cdn.example.com/video.m3u8")
+ mock_resolve.assert_not_called()
+
+ @patch.object(Spider, "_resolve_player_url")
+ @patch.object(Spider, "_get_html")
+ def test_player_content_decodes_encrypt_1_and_encrypt_2(self, mock_html, mock_resolve):
+ encoded = b64encode("https://cdn.example.com/e2.m3u8".encode("utf-8")).decode("utf-8")
+ mock_html.side_effect = [
+ '',
+ f'',
+ ]
+ mock_resolve.side_effect = ["https://cdn.example.com/e1.m3u8", "https://cdn.example.com/e2.m3u8"]
+ first = self.spider.playerContent("在线播放", "vod/play/1001-1-1.html", {})
+ second = self.spider.playerContent("在线播放", "vod/play/1001-1-2.html", {})
+ self.assertEqual(first["url"], "https://cdn.example.com/e1.m3u8")
+ self.assertEqual(second["url"], "https://cdn.example.com/e2.m3u8")
+
+ @patch.object(Spider, "_resolve_player_url", return_value="")
+ @patch.object(Spider, "_get_html", return_value="missing player")
+ def test_player_content_falls_back_to_parse_when_player_missing(self, mock_html, mock_resolve):
+ result = self.spider.playerContent("在线播放", "vod/play/1001-1-1.html", {})
+ self.assertEqual(result["parse"], 1)
+ self.assertEqual(result["url"], "https://www.lmm85.com/vod/play/1001-1-1.html")
+
if __name__ == "__main__":
unittest.main()
diff --git a/py/路漫漫.py b/py/路漫漫.py
index 0a1c69e..8944cb1 100644
--- a/py/路漫漫.py
+++ b/py/路漫漫.py
@@ -1,7 +1,12 @@
# coding=utf-8
+import base64
+import json
import re
import sys
-from urllib.parse import quote, urljoin
+from urllib.parse import quote, unquote, urljoin
+
+from Crypto.Cipher import AES
+from Crypto.Util.Padding import unpad
from base.spider import Spider as BaseSpider
@@ -61,6 +66,10 @@ class Spider(BaseSpider):
matched = re.search(r"(/vod/play/[^?#]+\.html)", raw)
return matched.group(1).lstrip("/") if matched else raw.lstrip("/")
+ def _decode_play_id(self, play_id):
+ raw = str(play_id or "").strip().lstrip("/")
+ return self._abs_url(raw)
+
def _clean_text(self, text):
return re.sub(r"\s+", " ", str(text or "")).strip()
@@ -184,3 +193,60 @@ class Spider(BaseSpider):
"vod_play_url": "$$$".join(urls for _, urls in groups),
}
return {"list": [vod]}
+
+ def _decode_player_url(self, raw_url, encrypt):
+ value = str(raw_url or "").strip()
+ mode = str(encrypt or "0").strip()
+ if mode == "1":
+ return unquote(value)
+ if mode == "2":
+ try:
+ return unquote(base64.b64decode(value).decode("utf-8"))
+ except Exception:
+ return ""
+ return value
+
+ def _extract_player_data(self, html):
+ matched = re.search(r"player_[a-z0-9_]+\s*=\s*(\{[\s\S]*?\})\s*;?", str(html or ""), re.I)
+ if not matched:
+ return {}
+ try:
+ return json.loads(matched.group(1))
+ except Exception:
+ return {}
+
+ def _decrypt_token(self, token):
+ try:
+ payload = base64.b64decode(str(token or ""))
+ cipher = AES.new(b"ejjooopppqqqrwww", AES.MODE_CBC, b"1348987635684651")
+ value = unpad(cipher.decrypt(payload), AES.block_size)
+ return value.decode("utf-8")
+ except Exception:
+ return ""
+
+ def _is_media_url(self, url):
+ return bool(re.search(r"\.(m3u8|mp4|flv|avi|mkv|ts)(?:[?#]|$)", str(url or ""), re.I))
+
+ def _build_player_headers(self, referer):
+ return {
+ "User-Agent": self.mobile_ua,
+ "Referer": referer,
+ "Origin": self.host,
+ }
+
+ def _resolve_player_url(self, player, play_page_url):
+ media = self._decode_player_url(player.get("url", ""), player.get("encrypt", "0"))
+ if self._is_media_url(media):
+ return media.split("&", 1)[0]
+ return ""
+
+ def playerContent(self, flag, id, vipFlags):
+ play_page_url = self._decode_play_id(id)
+ if self._is_media_url(play_page_url):
+ return {"parse": 0, "url": play_page_url, "header": self._build_player_headers(self.host + "/")}
+ body = self._get_html(play_page_url, headers={"Referer": self.host + "/"})
+ player = self._extract_player_data(body)
+ real_url = self._resolve_player_url(player, play_page_url)
+ if real_url:
+ return {"parse": 0, "url": real_url, "header": self._build_player_headers(play_page_url)}
+ return {"parse": 1, "url": play_page_url, "header": self._build_player_headers(self.host + "/")}