feat: add lumman player decode

2026-04-24 17:34:37 +08:00
parent 535fcaea72
commit f4f65c1243
2 changed files with 109 additions and 1 deletions
@@ -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 = [
+            '<script>var player_aaaa={"url":"https%3A%2F%2Fcdn.example.com%2Fe1.m3u8","encrypt":"1","from":"line"};</script>',
+            f'<script>var player_aaaa={{"url":"{encoded}","encrypt":"2","from":"line"}};</script>',
+        ]
+        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="<html><body>missing player</body></html>")
+    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()
@@ -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 + "/")}