diff --git a/py/tests/test_路漫漫.py b/py/tests/test_路漫漫.py
new file mode 100644
index 0000000..c738906
--- /dev/null
+++ b/py/tests/test_路漫漫.py
@@ -0,0 +1,194 @@
+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
+
+SAMPLE_LIST_HTML = """
+
+
+
+
+"""
+
+SAMPLE_DETAIL_HTML = """
+
+进击的巨人
+
+状态：已完结
+地区：日本
+人类与巨人的战斗。
+在线播放
+云播
+
+
+
+"""
+
+
+class TestLuManManSpider(unittest.TestCase):
+ def setUp(self):
+ Spider._instance = None
+ self.spider = Spider()
+ self.spider.init()
+
+ def test_home_content_returns_only_animation_classes(self):
+ result = self.spider.homeContent(False)
+ self.assertEqual(
+ result["class"],
+ [
+ {"type_id": "6", "type_name": "日本动漫"},
+ {"type_id": "7", "type_name": "国产动漫"},
+ {"type_id": "8", "type_name": "欧美动漫"},
+ {"type_id": "3", "type_name": "日本动画电影"},
+ {"type_id": "4", "type_name": "国产动画电影"},
+ {"type_id": "5", "type_name": "欧美动画电影"},
+ ],
+ )
+
+ @patch.object(Spider, "_get_html")
+ def test_home_video_content_parses_cards(self, mock_html):
+ mock_html.return_value = SAMPLE_LIST_HTML
+ result = self.spider.homeVideoContent()
+ self.assertEqual(len(result["list"]), 2)
+ self.assertEqual(result["list"][0]["vod_id"], "vod/detail/1001.html")
+ self.assertEqual(result["list"][0]["vod_pic"], "https://www.lmm85.com/upload/1001.jpg")
+ self.assertEqual(result["list"][1]["vod_pic"], "https://img.example.com/1002.jpg")
+
+ @patch.object(Spider, "_get_html")
+ def test_category_content_builds_filtered_url(self, mock_html):
+ mock_html.return_value = SAMPLE_LIST_HTML
+ result = self.spider.categoryContent("6", "2", False, {"年代": "/year/2024", "排序": "/by/time"})
+ self.assertEqual(result["page"], 2)
+ self.assertEqual(len(result["list"]), 2)
+ mock_html.assert_called_with("https://www.lmm85.com/vod/show/id/6/year/2024/by/time/page/2.html")
+
+ @patch.object(Spider, "_get_html")
+ def test_search_content_uses_search_path(self, mock_html):
+ mock_html.return_value = SAMPLE_LIST_HTML
+ result = self.spider.searchContent("海贼", False, "3")
+ self.assertEqual(result["page"], 3)
+ self.assertEqual(result["list"][0]["vod_name"], "海贼王")
+ mock_html.assert_called_with("https://www.lmm85.com/vod/search/page/3/wd/%E6%B5%B7%E8%B4%BC.html")
+
+ @patch.object(Spider, "_get_html")
+ def test_detail_content_parses_meta_and_playlists(self, mock_html):
+ mock_html.return_value = SAMPLE_DETAIL_HTML
+ result = self.spider.detailContent(["vod/detail/1001.html"])
+ vod = result["list"][0]
+ self.assertEqual(vod["vod_name"], "进击的巨人")
+ self.assertEqual(vod["vod_pic"], "https://www.lmm85.com/upload/jjdr.jpg")
+ self.assertEqual(vod["vod_content"], "人类与巨人的战斗。")
+ self.assertEqual(vod["vod_remarks"], "状态：已完结 / 地区：日本")
+ self.assertEqual(vod["vod_play_from"], "在线播放$$$云播")
+ self.assertIn("第1集$vod/play/1001-1-1.html#第2集$vod/play/1001-1-2.html", vod["vod_play_url"])
+ self.assertIn("HD$vod/play/1001-2-1.html", vod["vod_play_url"])
+
+ @patch.object(Spider, "_get_html")
+ def test_detail_content_falls_back_to_direct_playlist_scan(self, mock_html):
+ mock_html.return_value = """
+
+ 测试
+
+
+ """
+ result = self.spider.detailContent(["vod/detail/1.html"])
+ vod = result["list"][0]
+ 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")
+
+ @patch.object(Spider, "_post_json")
+ @patch.object(Spider, "_get_html")
+ def test_resolve_player_url_requests_parser_api_with_decrypted_token(self, mock_html, mock_post_json):
+ player = {"url": "source-id", "encrypt": "0", "from": "lineA"}
+ mock_html.side_effect = [
+ 'document.querySelector("iframe").src = MacPlayer.Parse + MacPlayer.PlayUrl + "&type=m3u8";',
+ '',
+ ]
+ mock_post_json.return_value = {"url": "https://cdn.example.com/final.m3u8"}
+ with patch.object(self.spider, "_decrypt_token", return_value="decoded-token") as mock_decrypt:
+ result = self.spider._resolve_player_url(player, "https://www.lmm85.com/vod/play/1001-1-1.html")
+ self.assertEqual(result, "https://cdn.example.com/final.m3u8")
+ mock_decrypt.assert_called_once_with("YxF6M7yw8U7OQcW9t6Qx4w==")
+ mock_post_json.assert_called_once_with(
+ "https://www.lmm85.com/api.php",
+ {"vid": "vid-1", "t": "123", "token": "decoded-token", "act": "play", "play": "1"},
+ referer="https://www.lmm85.com",
+ )
+
+ @patch.object(Spider, "_post_json", return_value={})
+ @patch.object(Spider, "_get_html", return_value='document.querySelector("iframe").src = "";')
+ def test_resolve_player_url_returns_empty_when_chain_cannot_be_built(self, mock_html, mock_post_json):
+ result = self.spider._resolve_player_url(
+ {"url": "vid", "encrypt": "0", "from": "lineA"},
+ "https://www.lmm85.com/vod/play/1.html",
+ )
+ self.assertEqual(result, "")
+
+
+if __name__ == "__main__":
+ unittest.main()
diff --git a/py/路漫漫.py b/py/路漫漫.py
new file mode 100644
index 0000000..a20418d
--- /dev/null
+++ b/py/路漫漫.py
@@ -0,0 +1,297 @@
+# coding=utf-8
+import base64
+import json
+import re
+import sys
+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
+
+sys.path.append("..")
+
+
+class Spider(BaseSpider):
+ def __init__(self):
+ self.name = "路漫漫"
+ self.host = "https://www.lmm85.com"
+ self.mobile_ua = (
+ "Mozilla/5.0 (iPhone; CPU iPhone OS 14_0 like Mac OS X) "
+ "AppleWebKit/605.1.15 (KHTML, like Gecko) Version/14.0 "
+ "Mobile/15E148 Safari/604.1"
+ )
+ self.headers = {
+ "User-Agent": self.mobile_ua,
+ "Referer": "http://www.lmm50.com/",
+ "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
+ "Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8",
+ }
+ self.classes = [
+ {"type_id": "6", "type_name": "日本动漫"},
+ {"type_id": "7", "type_name": "国产动漫"},
+ {"type_id": "8", "type_name": "欧美动漫"},
+ {"type_id": "3", "type_name": "日本动画电影"},
+ {"type_id": "4", "type_name": "国产动画电影"},
+ {"type_id": "5", "type_name": "欧美动画电影"},
+ ]
+
+ def init(self, extend=""):
+ return None
+
+ def getName(self):
+ return self.name
+
+ def homeContent(self, filter):
+ return {"class": self.classes}
+
+ def _abs_url(self, value):
+ raw = str(value or "").strip()
+ if not raw:
+ return ""
+ return urljoin(self.host + "/", raw)
+
+ def _encode_vod_id(self, href):
+ raw = str(href or "").strip()
+ matched = re.search(r"(/vod/detail/[^?#]+\.html)", raw)
+ return matched.group(1).lstrip("/") if matched else raw.lstrip("/")
+
+ def _decode_vod_id(self, vod_id):
+ raw = str(vod_id or "").strip().lstrip("/")
+ return self._abs_url(raw)
+
+ def _encode_play_id(self, href):
+ raw = str(href or "").strip()
+ 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()
+
+ def _get_html(self, url, headers=None):
+ request_headers = dict(self.headers)
+ if headers:
+ request_headers.update(headers)
+ response = self.fetch(url, headers=request_headers, timeout=15, verify=False)
+ if response.status_code != 200:
+ return ""
+ return response.text or ""
+
+ def _parse_cards(self, html):
+ root = self.html(html)
+ if root is None:
+ return []
+ items = []
+ seen = set()
+ for box in root.xpath("//*[contains(@class,'video-img-box')]"):
+ href = self._clean_text((box.xpath(".//a[1]/@href") or [""])[0])
+ title = self._clean_text("".join(box.xpath(".//*[contains(@class,'title')][1]//text()")))
+ pic = self._clean_text(
+ (box.xpath(".//img[1]/@data-src") or box.xpath(".//img[1]/@src") or [""])[0]
+ )
+ remarks = self._clean_text("".join(box.xpath(".//*[contains(@class,'label')][1]//text()")))
+ vod_id = self._encode_vod_id(href)
+ if not vod_id or not title or vod_id in seen:
+ continue
+ seen.add(vod_id)
+ items.append(
+ {
+ "vod_id": vod_id,
+ "vod_name": title,
+ "vod_pic": self._abs_url(pic),
+ "vod_remarks": remarks,
+ }
+ )
+ return items
+
+ def homeVideoContent(self):
+ return {"list": self._parse_cards(self._get_html(self.host + "/"))[:20]}
+
+ def categoryContent(self, tid, pg, filter, extend):
+ page = int(pg)
+ extend = extend or {}
+ path = f"/vod/show/id/{tid}"
+ path += str(extend.get("年代", "") or "")
+ path += str(extend.get("排序", "") or "")
+ path += f"/page/{page}.html"
+ items = self._parse_cards(self._get_html(self.host + path))
+ return {"page": page, "total": len(items), "list": items}
+
+ def searchContent(self, key, quick, pg="1"):
+ page = int(pg)
+ keyword = self._clean_text(key)
+ if not keyword:
+ return {"page": page, "total": 0, "list": []}
+ url = f"{self.host}/vod/search/page/{page}/wd/{quote(keyword)}.html"
+ items = self._parse_cards(self._get_html(url))
+ return {"page": page, "total": len(items), "list": items[:10] if quick else items}
+
+ def _parse_play_groups(self, root):
+ groups = []
+ tabs = root.xpath("//*[contains(@class,'module-tab-item') and contains(@class,'tab-item')]")
+ for index, tab in enumerate(tabs):
+ name = self._clean_text("".join(tab.xpath(".//text()")))
+ target = self._clean_text((tab.xpath("./@href") or [""])[0])
+ if not name:
+ continue
+ playlist = []
+ if target.startswith("#"):
+ playlist = root.xpath(f"//*[@id='{target[1:]}']")
+ if not playlist:
+ playlist = root.xpath(f"(//*[contains(@class,'module-player-list')])[{index + 1}]")
+ if not playlist:
+ continue
+ episodes = []
+ for anchor in playlist[0].xpath(".//a[@href]"):
+ ep_name = self._clean_text("".join(anchor.xpath(".//text()")))
+ ep_id = self._encode_play_id((anchor.xpath("./@href") or [""])[0])
+ if ep_name and ep_id:
+ episodes.append(f"{ep_name}${ep_id}")
+ if episodes:
+ groups.append((name, "#".join(episodes)))
+ if groups:
+ return groups
+
+ fallback = []
+ for anchor in root.xpath("//*[contains(@class,'module-player-list')]//a[@href]"):
+ ep_name = self._clean_text("".join(anchor.xpath(".//text()")))
+ ep_id = self._encode_play_id((anchor.xpath("./@href") or [""])[0])
+ if ep_name and ep_id:
+ fallback.append(f"{ep_name}${ep_id}")
+ if fallback:
+ return [("播放列表", "#".join(fallback))]
+ return []
+
+ def detailContent(self, ids):
+ raw_id = ids[0] if isinstance(ids, list) else ids
+ html = self._get_html(self._decode_vod_id(raw_id))
+ root = self.html(html)
+ if root is None:
+ return {"list": []}
+ groups = self._parse_play_groups(root)
+ remarks = [
+ self._clean_text("".join(node.xpath(".//text()")))
+ for node in root.xpath("//*[contains(@class,'video-info-items')]")
+ ]
+ pic = (
+ root.xpath("//*[contains(@class,'module-item-pic')]//img[1]/@data-src")
+ or root.xpath("//*[contains(@class,'module-item-pic')]//img[1]/@src")
+ or [""]
+ )[0]
+ vod = {
+ "vod_id": str(raw_id),
+ "vod_name": self._clean_text("".join(root.xpath("//*[contains(@class,'page-title')][1]//text()"))),
+ "vod_pic": self._abs_url(pic),
+ "vod_content": self._clean_text("".join(root.xpath("//*[contains(@class,'video-info-content')][1]//text()"))),
+ "vod_remarks": " / ".join([value for value in remarks if value]),
+ "vod_play_from": "$$$".join(name for name, _ in groups),
+ "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 _post_json(self, url, data, referer=None):
+ headers = self._build_player_headers(referer or self.host)
+ response = self.post(url, data=data, headers=headers, timeout=10, verify=False)
+ if response.status_code != 200:
+ return {}
+ try:
+ return json.loads(response.text or "{}")
+ except Exception:
+ return {}
+
+ def _extract_js_src(self, js_text, video_url, play_page_url):
+ matched = re.search(r"\.src\s*=\s*(.*?);", str(js_text or ""))
+ if not matched:
+ return ""
+ raw = re.sub(r"[\+\s']", "", matched.group(1))
+ raw = raw.replace("MacPlayer.PlayUrl", video_url)
+ raw = raw.replace("window.location.href", play_page_url)
+ raw = raw.replace("MacPlayer.Parse", self.host + "/?url=")
+ return raw
+
+ 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]
+ source = self._clean_text(player.get("from", ""))
+ if not media or not source:
+ return ""
+ js_url = f"{self.host}/static/player/{source}.js"
+ js_text = self._get_html(js_url, headers={"Referer": play_page_url})
+ iframe_url = self._extract_js_src(js_text, media, play_page_url)
+ if not iframe_url or "type=" not in iframe_url:
+ return ""
+ iframe_html = self._get_html(iframe_url, headers={"Referer": self.host})
+ if not re.search(r'vid\s*=\s*".+?"', iframe_html):
+ return ""
+ api_path = self.regStr(r'post\("(.*?)"', iframe_html)
+ if not api_path:
+ return ""
+ post_url = urljoin(self.host + "/", api_path.lstrip("/"))
+ token = self.regStr(r'token\s*=\s*"(.*?)"', iframe_html)
+ payload = {
+ "vid": self.regStr(r'vid\s*=\s*"(.*?)"', iframe_html),
+ "t": self.regStr(r'var\s+t\s*=\s*"(.*?)"', iframe_html),
+ "token": self._decrypt_token(token),
+ "act": self.regStr(r'act\s*=\s*"(.*?)"', iframe_html),
+ "play": self.regStr(r'play\s*=\s*"(.*?)"', iframe_html),
+ }
+ data = self._post_json(post_url, payload, referer=self.host)
+ return str(data.get("url", "")).strip()
+ 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 + "/")}
+ 
+ 