diff --git a/py/tests/test_修罗.py b/py/tests/test_修罗.py new file mode 100644 index 0000000..e44e641 --- /dev/null +++ b/py/tests/test_修罗.py @@ -0,0 +1,315 @@ +import gzip +import json +import unittest +from importlib.machinery import SourceFileLoader +from pathlib import Path +from unittest.mock import Mock, patch + + +ROOT = Path(__file__).resolve().parents[1] +MODULE = SourceFileLoader("xiuluo_spider", str(ROOT / "修罗.py")).load_module() +Spider = MODULE.Spider + + +def build_proxy_payload(text): + return (b"x" * 3354) + gzip.compress(text.encode("utf-8")) + + +class TestXiuLuoSpider(unittest.TestCase): + def setUp(self): + Spider._instance = None + self.spider = Spider() + self.spider.init() + + def test_home_content_returns_classes_and_filters(self): + result = self.spider.homeContent(False) + self.assertEqual( + result["class"], + [ + {"type_id": "movie", "type_name": "电影"}, + {"type_id": "tv", "type_name": "电视剧"}, + {"type_id": "zongyi", "type_name": "综艺"}, + {"type_id": "duanju", "type_name": "短剧"}, + ], + ) + self.assertEqual(result["filters"]["movie"][0]["key"], "genre") + self.assertEqual(result["filters"]["movie"][0]["value"][0], {"n": "不限", "v": "all"}) + self.assertEqual(result["filters"]["movie"][-1]["key"], "order") + self.assertEqual(result["filters"]["zongyi"][0]["key"], "area") + self.assertEqual(result["list"], []) + + def test_category_content_builds_filter_url_and_parses_cards(self): + html = """ +
+ + +
+ """ + response = Mock(status_code=200, text=html) + with patch.object(self.spider, "fetch", return_value=response) as mock_fetch: + result = self.spider.categoryContent( + "movie", + "2", + False, + {"genre": "dongzuo", "area": "中国大陆", "year": "2025", "order": "1"}, + ) + + self.assertEqual( + mock_fetch.call_args.args[0], + "https://www.xlys02.com/s/dongzuo/2?type=0&area=%E4%B8%AD%E5%9B%BD%E5%A4%A7%E9%99%86&year=2025&order=1", + ) + self.assertEqual(result["page"], 2) + self.assertEqual(result["list"][0]["vod_id"], "movie/demo-a") + self.assertEqual(result["list"][0]["vod_pic"], "https://www.xlys02.com/img/a.jpg") + self.assertEqual(result["list"][1]["vod_pic"], "https://img.example/b.jpg") + self.assertEqual(result["list"][1]["vod_remarks"], "更新至第3集") + + def test_search_content_filters_target_site_and_extracts_vod_items(self): + mock_json = { + "data": [ + { + "website": "哔滴", + "url": "https://foo.example//vod-detail-id-88413-src-1-num-1.htm", + "text": "第二十条", + "icon": "https://img.example/poster.jpg", + }, + { + "website": "其他站", + "url": "https://foo.example//vod-detail-id-1-src-1-num-1.htm", + "text": "应被过滤", + "icon": "https://img.example/skip.jpg", + }, + ] + } + response = Mock(status_code=200) + response.json.return_value = mock_json + with patch.object(self.spider, "fetch", return_value=response) as mock_fetch: + result = self.spider.searchContent("第二十条", False, "1") + + self.assertEqual( + result["list"], + [ + { + "vod_id": "/vod-detail-id-88413-src-1-num-1.htm", + "vod_name": "第二十条", + "vod_pic": "https://img.example/poster.jpg", + "vod_remarks": "哔滴", + } + ], + ) + called_url = mock_fetch.call_args.args[0] + self.assertIn("https://www.ymck.pro/API/v2.php?q=第二十条&size=50", called_url) + + def test_detail_content_builds_single_vod_shell_from_search_id(self): + html = """ + +

测试影片

+ +
一段剧情简介
+
+ 正片 + 备用 +
+ + """ + response = Mock(status_code=200, text=html) + with patch.object(self.spider, "fetch", return_value=response) as mock_fetch: + result = self.spider.detailContent(["movie/test-alpha"]) + + self.assertEqual(mock_fetch.call_args.args[0], "https://www.xlys02.com/movie/test-alpha.htm") + self.assertEqual( + result["list"], + [ + { + "vod_id": "movie/test-alpha", + "vod_name": "测试影片", + "vod_pic": "https://www.xlys02.com/poster.jpg", + "type_name": "", + "vod_class": "", + "vod_year": "", + "vod_area": "", + "vod_lang": "", + "vod_remarks": "", + "vod_actor": "", + "vod_director": "", + "vod_douban_score": "", + "vod_douban_id": "", + "vod_content": "一段剧情简介", + "vod_play_from": "修罗直连", + "vod_play_url": "正片$play/alpha-1#备用$play/alpha-2", + } + ], + ) + + def test_detail_content_skips_watch_history_heading_when_extracting_title(self): + html = """ + +

观看历史

+
+

真正片名

+ +
详情简介
+
正片
+
+ + """ + response = Mock(status_code=200, text=html) + with patch.object(self.spider, "fetch", return_value=response): + result = self.spider.detailContent(["movie/right-title"]) + + self.assertEqual(result["list"][0]["vod_name"], "真正片名") + + def test_detail_content_parses_labeled_metadata_block(self): + html = """ + +

八千里路云和月

+ +
大时代里的群像故事
+
+

别名:八千里路雲和月 / 厨子与将军

+

导演:张永新

+

主演:王阳万茜

+

类型:剧情历史战争

+

制片国家/地区:[中国大陆]

+

语言:汉语普通话

+

豆瓣链接:33371937

+

摘要:40集全.HD1080P4K.国语中字

+
+
第1集
+ + """ + response = Mock(status_code=200, text=html) + with patch.object(self.spider, "fetch", return_value=response): + result = self.spider.detailContent(["movie/8klyhy"]) + + vod = result["list"][0] + self.assertEqual(vod["vod_name"], "八千里路云和月") + self.assertEqual(vod["type_name"], "剧情 / 历史 / 战争") + self.assertEqual(vod["vod_class"], "剧情 / 历史 / 战争") + self.assertEqual(vod["vod_area"], "中国大陆") + self.assertEqual(vod["vod_lang"], "汉语普通话") + self.assertEqual(vod["vod_director"], "张永新") + self.assertEqual(vod["vod_actor"], "王阳 / 万茜") + self.assertEqual(vod["vod_douban_id"], "33371937") + self.assertEqual(vod["vod_remarks"], "40集全.HD1080P4K.国语中字") + + @patch.object(Spider, "_current_millis", return_value=1713952212345, create=True) + def test_player_content_requests_lines_and_returns_first_allowed_direct_url(self, _mock_time): + play_html = '' + play_response = Mock(status_code=200, text=play_html) + lines_response = Mock(status_code=200) + lines_response.json.return_value = { + "code": 0, + "data": { + "url3": ( + "https://media.example.com/direct.mp4," + "https://vip.ffzy/video/index.m3u8," + "https://p3-tt.byteimg.com/byte.mp4" + ) + } + } + with patch.object(self.spider, "fetch", side_effect=[play_response, lines_response]) as mock_fetch: + result = self.spider.playerContent("修罗直连", "play/alpha-1", {}) + + self.assertEqual( + mock_fetch.call_args_list[1].args[0], + "https://www.xlys02.com/lines?t=1713952212345&sg=4322A578A3DB6BD0D1936C4F381C9925E2388F66F2DDE5D783B65E2379BE5B56&pid=9527", + ) + self.assertEqual(mock_fetch.call_args_list[1].kwargs["headers"]["X-Requested-With"], "XMLHttpRequest") + self.assertEqual(result["parse"], 0) + self.assertEqual(result["url"], "https://media.example.com/direct.mp4") + self.assertEqual(result["header"]["Referer"], "https://www.xlys02.com") + + @patch.object(Spider, "_current_millis", return_value=1713952212345, create=True) + def test_player_content_uses_tos_fallback_when_no_allowed_url3(self, _mock_time): + play_html = '' + play_response = Mock(status_code=200, text=play_html) + lines_response = Mock(status_code=200) + lines_response.json.return_value = { + "code": 0, + "data": { + "url3": "https://vip.ffzy/video/index.m3u8,https://p3-tt.byteimg.com/byte.mp4", + "tos": True, + } + } + tos_response = Mock(status_code=200) + tos_response.json.return_value = {"url": "https://media.example.com/from-tos.mp4"} + with patch.object(self.spider, "fetch", side_effect=[play_response, lines_response]): + with patch.object(self.spider, "post", return_value=tos_response) as mock_post: + result = self.spider.playerContent("修罗直连", "play/alpha-1", {}) + + self.assertEqual(mock_post.call_args.args[0], "https://www.xlys02.com/god/9527?type=1") + self.assertEqual( + mock_post.call_args.kwargs["data"], + "t=1713952212345&sg=4322A578A3DB6BD0D1936C4F381C9925E2388F66F2DDE5D783B65E2379BE5B56&verifyCode=888", + ) + self.assertEqual(result["parse"], 0) + self.assertEqual(result["url"], "https://media.example.com/from-tos.mp4") + + def test_player_content_returns_proxy_url_for_real_m3u8(self): + result = self.spider.playerContent("修罗", "https://cdn.example.com/demo/index.m3u8", {}) + self.assertEqual(result["parse"], 0) + self.assertEqual(result["url"], "http://127.0.0.1:9978/proxy?do=py&type=m3u8&url=https%3A%2F%2Fcdn.example.com%2Fdemo%2Findex.m3u8") + self.assertIn("Mozilla/5.0", result["header"]["User-Agent"]) + + def test_player_content_returns_direct_url_for_mp4(self): + probe = Mock() + probe.status_code = 206 + probe.headers = {"Content-Type": "video/mp4"} + with patch.object(self.spider, "_range_probe", return_value=probe): + result = self.spider.playerContent("修罗", "https://cdn.example.com/demo/video.mp4", {}) + self.assertEqual(result["parse"], 0) + self.assertEqual(result["url"], "https://cdn.example.com/demo/video.mp4") + + def test_local_proxy_decodes_rewrites_and_returns_m3u8(self): + body = build_proxy_payload( + "#EXTM3U\n" + "#EXTINF:10,\n" + "seg-1.ts\n" + "#EXTINF:10,\n" + "/seg-2.ts\n" + "#EXT-X-DISCONTINUITY\n" + "https://up.example.com/seg-3.ts\n" + "#EXT-X-ENDLIST\n" + ) + response = Mock(status_code=200, content=body) + response.headers = {} + with patch.object(self.spider, "fetch", return_value=response): + status, content_type, data = self.spider.localProxy( + { + "type": "m3u8", + "url": "https://cdn.example.com/path/master.m3u8", + } + ) + + text = data.decode("utf-8") + self.assertEqual(status, 200) + self.assertEqual(content_type, "application/vnd.apple.mpegurl") + self.assertIn("http://127.0.0.1:9978/proxy?do=py&type=m3u8&url=https%3A%2F%2Fcdn.example.com%2Fpath%2Fseg-1.ts", text) + self.assertIn("http://127.0.0.1:9978/proxy?do=py&type=m3u8&url=https%3A%2F%2Fcdn.example.com%2Fseg-2.ts", text) + self.assertIn("http://127.0.0.1:9978/proxy?do=py&type=m3u8&url=https%3A%2F%2Fup.example.com%2Fseg-3.ts", text) + + def test_local_proxy_rejects_unsupported_type(self): + self.assertIsNone(self.spider.localProxy({"type": "other", "url": "https://cdn.example.com/a"})) + + def test_aes_ecb_hex_matches_reference_cipher(self): + self.assertEqual( + self.spider._aes_ecb_hex("1234567890abcdef", "hello"), + "3ae1bf1edbd01914c379e7a9e993bf59", + ) + + +if __name__ == "__main__": + unittest.main() diff --git a/py/修罗.py b/py/修罗.py new file mode 100644 index 0000000..073502f --- /dev/null +++ b/py/修罗.py @@ -0,0 +1,560 @@ +# coding=utf-8 +import gzip +import hashlib +import json +import os +import re +import sys +import time +from urllib.parse import quote, urljoin + +from Crypto.Cipher import AES +from Crypto.Util.Padding import pad + +CURRENT_DIR = os.path.dirname(os.path.abspath(__file__)) +if CURRENT_DIR not in sys.path: + sys.path.append(CURRENT_DIR) +if os.path.dirname(CURRENT_DIR) not in sys.path: + sys.path.append(os.path.dirname(CURRENT_DIR)) + +from base.spider import Spider as BaseSpider + + +class Spider(BaseSpider): + BASE_SITE = "https://v.xlys.ltd.ua/" + HOST = "v.xlys.ltd.ua" + SEARCH_API = "https://www.ymck.pro/API/v2.php?q=" + SEARCH_SIZE = "&size=50" + TARGET_WEBSITE = "哔滴" + UA = ( + "Mozilla/5.0 (X11; Linux aarch64) AppleWebKit/537.36 " + "(KHTML, like Gecko) Chrome/140.0.0.0 Safari/537.36" + ) + LOCAL_PROXY_BASE = "http://127.0.0.1:9978/proxy?do=py" + + def __init__(self): + self.name = "修罗" + self.host = "https://www.xlys02.com" + self.origin = self.BASE_SITE + self.url_re = re.compile(r"https?://[^/]+(?P/.*?\.htm)", re.I) + self.classes = [ + {"type_id": "movie", "type_name": "电影"}, + {"type_id": "tv", "type_name": "电视剧"}, + {"type_id": "zongyi", "type_name": "综艺"}, + {"type_id": "duanju", "type_name": "短剧"}, + ] + common_genres = [ + {"n": "不限", "v": "all"}, + {"n": "动作", "v": "dongzuo"}, + {"n": "爱情", "v": "aiqing"}, + {"n": "喜剧", "v": "xiju"}, + {"n": "科幻", "v": "kehuan"}, + {"n": "恐怖", "v": "kongbu"}, + {"n": "战争", "v": "zhanzheng"}, + {"n": "武侠", "v": "wuxia"}, + {"n": "魔幻", "v": "mohuan"}, + {"n": "剧情", "v": "juqing"}, + {"n": "动画", "v": "donghua"}, + {"n": "惊悚", "v": "jingsong"}, + {"n": "灾难", "v": "zainan"}, + {"n": "悬疑", "v": "xuanyi"}, + {"n": "警匪", "v": "jingfei"}, + {"n": "文艺", "v": "wenyi"}, + {"n": "青春", "v": "qingchun"}, + {"n": "冒险", "v": "maoxian"}, + {"n": "犯罪", "v": "fanzui"}, + {"n": "记录", "v": "jilu"}, + {"n": "古装", "v": "guzhuang"}, + {"n": "奇幻", "v": "奇幻"}, + ] + common_areas = [ + {"n": "不限", "v": ""}, + {"n": "中国大陆", "v": "中国大陆"}, + {"n": "中国香港", "v": "中国香港"}, + {"n": "美国", "v": "美国"}, + {"n": "日本", "v": "日本"}, + {"n": "韩国", "v": "韩国"}, + {"n": "法国", "v": "法国"}, + {"n": "印度", "v": "印度"}, + {"n": "德国", "v": "德国"}, + {"n": "西班牙", "v": "西班牙"}, + {"n": "意大利", "v": "意大利"}, + {"n": "澳大利亚", "v": "澳大利亚"}, + {"n": "比利时", "v": "比利时"}, + {"n": "瑞典", "v": "瑞典"}, + {"n": "荷兰", "v": "荷兰"}, + {"n": "丹麦", "v": "丹麦"}, + {"n": "加拿大", "v": "加拿大"}, + {"n": "俄罗斯", "v": "俄罗斯"}, + ] + years = [{"n": "不限", "v": ""}] + [{"n": str(year), "v": str(year)} for year in range(2026, 2001, -1)] + order_values = [{"n": "更新时间", "v": "0"}, {"n": "豆瓣评分", "v": "1"}] + self.filters = { + "movie": [ + {"key": "genre", "name": "类型", "init": "all", "value": list(common_genres)}, + {"key": "area", "name": "地区", "init": "", "value": list(common_areas)}, + {"key": "year", "name": "年份", "init": "", "value": list(years)}, + {"key": "order", "name": "排序", "init": "0", "value": list(order_values)}, + ], + "tv": [ + {"key": "genre", "name": "类型", "init": "all", "value": list(common_genres)}, + {"key": "area", "name": "地区", "init": "", "value": list(common_areas)}, + {"key": "year", "name": "年份", "init": "", "value": list(years)}, + {"key": "order", "name": "排序", "init": "0", "value": list(order_values)}, + ], + "zongyi": [ + {"key": "area", "name": "地区", "init": "", "value": list(common_areas)}, + {"key": "year", "name": "年份", "init": "", "value": list(years)}, + {"key": "order", "name": "排序", "init": "0", "value": list(order_values)}, + ], + "duanju": [ + {"key": "year", "name": "年份", "init": "", "value": [{"n": "不限", "v": ""}] + [{"n": str(year), "v": str(year)} for year in range(2026, 2022, -1)]}, + {"key": "order", "name": "排序", "init": "0", "value": list(order_values)}, + ], + } + self._detail_title_blacklist = {"观看历史", "猜你喜欢", "相关推荐", "相关影视", "热门推荐", "为你推荐"} + + def init(self, extend=""): + return None + + def getName(self): + return self.name + + def homeContent(self, filter): + return {"class": list(self.classes), "filters": self.filters, "list": []} + + def homeVideoContent(self): + return {"list": []} + + def categoryContent(self, tid, pg, filter, extend): + page = max(int(str(pg or "1")), 1) + type_id = self._stringify(tid).strip() + if type_id not in self.filters: + return {"page": page, "limit": 0, "total": 0, "list": []} + values = extend or {} + if "genre" not in values: + if type_id == "movie": + values["genre"] = "all" + if type_id == "tv": + values["genre"] = "all" + url = self._build_category_url(type_id, page, values) + response = self.fetch(url, headers=self.build_headers(), timeout=15) + if response.status_code != 200: + return {"page": page, "limit": 0, "total": 0, "list": []} + items = self._parse_category_cards(response.text or "") + return {"page": page, "limit": len(items), "total": page * 30 + len(items), "list": items} + + def build_headers(self, extra=None): + headers = { + "User-Agent": self.UA, + "Origin": self.origin, + "Referer": self.origin, + } + if isinstance(extra, dict): + headers.update(extra) + return headers + + def _site_headers(self, referer=None, extra=None): + headers = {"User-Agent": self.UA, "Referer": referer or (self.host + "/")} + if isinstance(extra, dict): + headers.update(extra) + return headers + + def _stringify(self, value): + return "" if value is None else str(value) + + def _build_url(self, value): + raw = self._stringify(value).strip() + if not raw: + return "" + if raw.startswith(("http://", "https://")): + return raw + if raw.startswith("//"): + return "https:" + raw + return urljoin(self.host + "/", raw) + + def _extract_vod_path(self, value): + raw = self._stringify(value).strip() + if not raw: + return "" + if raw.startswith("/"): + return "/" + raw.lstrip("/") + matched = self.url_re.search(raw) + if matched: + return "/" + matched.group("path").lstrip("/") + generic = re.search(r"(/[^?#\s]+?\.htm)", raw, re.I) + if generic: + return "/" + generic.group(1).lstrip("/") + return "" + + def _extract_site_id(self, href): + raw = self._stringify(href).strip() + if not raw: + return "" + if raw.startswith(("http://", "https://")): + matched = re.search(r"^https?://[^/]+/(.+?)(?:\.htm|[?#]|$)", raw, re.I) + return matched.group(1).lstrip("/") if matched else "" + cleaned = raw.split(".htm", 1)[0] + return cleaned.lstrip("/") + + def _normalize_detail_id(self, value): + raw = self._stringify(value).strip() + if not raw: + return "" + path = self._extract_vod_path(raw) + if path: + return self._extract_site_id(path) + return self._extract_site_id(raw) + + def _clean_text(self, value): + return re.sub(r"\s+", " ", self._stringify(value).replace("\xa0", " ")).strip() + + def _extract_detail_title(self, root): + if root is None: + return "" + candidate_xpaths = [ + "//*[contains(concat(' ', normalize-space(@class), ' '), ' detail ')]//*[self::h1 or self::h2][1]//text()", + "//*[contains(concat(' ', normalize-space(@class), ' '), ' col ')]//*[self::h1 or self::h2][1]//text()", + "//h1[1]//text()", + "//h2//text()", + ] + for xpath in candidate_xpaths: + nodes = root.xpath(xpath) + if not nodes: + continue + texts = nodes if isinstance(nodes, list) else [nodes] + for text in texts: + title = self._clean_text(text) + if title and title not in self._detail_title_blacklist: + return title + return "" + + def _extract_labeled_meta(self, root): + info = {} + if root is None: + return info + for node in root.xpath("//p[strong[1]]|//li[strong[1]]|//div[strong[1]]"): + label = self._clean_text("".join(node.xpath("./strong[1]//text()"))).rstrip("::") + if not label: + continue + + link_texts = [] + for text in node.xpath(".//a//text()"): + value = self._clean_text(text) + if value and value not in link_texts: + link_texts.append(value) + + raw_text = self._clean_text("".join(node.xpath(".//text()"))) + raw_text = re.sub(r"^\s*%s\s*[::]?\s*" % re.escape(label), "", raw_text).strip() + + value = "" + if label in ("导演", "主演", "演员", "编剧", "类型"): + value = " / ".join(link_texts) if link_texts else raw_text + else: + value = raw_text + + if label in ("制片国家/地区", "地区"): + value = value.strip("[]") + + hrefs = [self._stringify(item).strip() for item in node.xpath(".//a/@href") if self._stringify(item).strip()] + info[label] = {"value": value, "hrefs": hrefs, "links": link_texts} + return info + + def _is_media_url(self, value): + return re.search(r"^https?://.+\.(?:m3u8|mp4|flv|mkv|ts)(?:[?#].*)?$", self._stringify(value), re.I) is not None + + def _page_result(self, items, pg): + page = max(int(str(pg or "1")), 1) + return {"page": page, "limit": len(items), "total": (page - 1) * 50 + len(items), "list": items} + + def _build_category_url(self, tid, page, values): + genre = self._stringify((values or {}).get("genre")).strip() or tid + url = f"{self.host}/s/{genre}/{page}" + params = [] + if tid not in ("zongyi", "duanju"): + params.append("type=1" if tid == "tv" else "type=0") + area = self._stringify((values or {}).get("area")).strip() + year = self._stringify((values or {}).get("year")).strip() + order = self._stringify((values or {}).get("order")).strip() + if area: + params.append(f"area={quote(area)}") + if year: + params.append(f"year={year}") + if order: + params.append(f"order={order}") + return url + ("?" + "&".join(params) if params else "") + + def _parse_category_cards(self, html): + root = self.html(html) + if root is None: + return [] + items = [] + for card in root.xpath("//*[contains(concat(' ', normalize-space(@class), ' '), ' card ') and contains(concat(' ', normalize-space(@class), ' '), ' card-link ')]"): + href = self._stringify((card.xpath(".//a[@href][1]/@href") or [""])[0]).strip() + vod_id = self._extract_site_id(href) + if not vod_id: + continue + title = self._stringify("".join(card.xpath(".//*[contains(concat(' ', normalize-space(@class), ' '), ' card-title ')]//text()"))).strip() + pic = self._stringify((card.xpath(".//img[1]/@src") or [""])[0]).strip() + remarks = self._stringify("".join(card.xpath(".//*[contains(concat(' ', normalize-space(@class), ' '), ' text-muted ')]//text()"))).strip() + items.append( + { + "vod_id": vod_id, + "vod_name": title, + "vod_pic": self._build_url(pic), + "vod_remarks": remarks, + } + ) + return items + + def searchContent(self, key, quick, pg="1"): + keyword = self._stringify(key).strip() + if not keyword: + return self._page_result([], pg) + url = f"{self.SEARCH_API}{keyword}{self.SEARCH_SIZE}" + response = self.fetch(url, headers=self.build_headers(), timeout=15) + if response.status_code != 200: + return self._page_result([], pg) + try: + payload = response.json() + except Exception: + return self._page_result([], pg) + + items = [] + for row in payload.get("data", []) or []: + if self._stringify((row or {}).get("website")).strip() != self.TARGET_WEBSITE: + continue + vod_id = self._extract_vod_path((row or {}).get("url")) + if not vod_id: + continue + items.append( + { + "vod_id": vod_id, + "vod_name": self._stringify((row or {}).get("text")).strip(), + "vod_pic": self._stringify((row or {}).get("icon")).strip(), + "vod_remarks": self._stringify((row or {}).get("website")).strip(), + } + ) + return self._page_result(items, pg) + + def detailContent(self, ids): + vod_id = self._normalize_detail_id(ids[0] if ids else "") + if not vod_id: + return {"list": []} + detail_url = f"{self.host}/{vod_id}.htm" + response = self.fetch(detail_url, headers=self._site_headers(detail_url), timeout=15) + if response.status_code != 200: + return {"list": []} + root = self.html(response.text or "") + if root is None: + return {"list": []} + play_urls = [] + for anchor in root.xpath("//*[@id='play-list']//a[@href]"): + name = self._stringify("".join(anchor.xpath(".//text()"))).strip() + href = self._stringify((anchor.xpath("./@href") or [""])[0]).strip() + play_id = self._extract_site_id(href) + if name and play_id: + play_urls.append(f"{name}${play_id}") + meta = self._extract_labeled_meta(root) + type_name = self._stringify((meta.get("类型") or {}).get("value")).strip() + douban_meta = meta.get("豆瓣链接") or meta.get("豆瓣ID") or {} + douban_text = self._stringify(douban_meta.get("value")).strip() + douban_id = "" + for href in douban_meta.get("hrefs") or []: + matched = re.search(r"douban\.com/subject/(\d+)", href, re.I) + if matched: + douban_id = matched.group(1) + break + if not douban_id: + matched = re.search(r"\b(\d{5,})\b", douban_text) + if matched: + douban_id = matched.group(1) + score_meta = meta.get("豆瓣评分") or meta.get("豆瓣") or meta.get("评分") or {} + score_text = self._stringify(score_meta.get("value")).strip() + score_match = re.search(r"(\d+(?:\.\d+)?)", score_text) + vod = { + "vod_id": vod_id, + "vod_name": self._extract_detail_title(root), + "vod_pic": self._build_url((root.xpath("//img[contains(concat(' ', normalize-space(@class), ' '), ' cover ')]/@src") or [""])[0]), + "type_name": type_name, + "vod_class": type_name, + "vod_year": self._stringify((meta.get("年代") or meta.get("年份") or {}).get("value")).strip(), + "vod_area": self._stringify((meta.get("制片国家/地区") or meta.get("地区") or {}).get("value")).strip(), + "vod_lang": self._stringify((meta.get("语言") or {}).get("value")).strip(), + "vod_remarks": self._stringify((meta.get("摘要") or {}).get("value")).strip(), + "vod_actor": self._stringify((meta.get("主演") or meta.get("演员") or {}).get("value")).strip(), + "vod_director": self._stringify((meta.get("导演") or {}).get("value")).strip(), + "vod_douban_score": score_match.group(1) if score_match else "", + "vod_douban_id": douban_id, + "vod_content": self._clean_text("".join(root.xpath("//*[@id='synopsis'][1]//text()"))), + "vod_play_from": "修罗直连" if play_urls else "", + "vod_play_url": "#".join(play_urls), + } + return {"list": [vod]} + + def _current_millis(self): + return int(time.time() * 1000) + + def _load_json(self, response): + if response is None: + return {} + try: + return response.json() + except Exception: + try: + return json.loads(self._stringify(getattr(response, "text", "")) or "{}") + except Exception: + return {} + + def _build_play_page_url(self, value): + raw = self._stringify(value).strip() + if not raw: + return "" + if self._is_media_url(raw): + return raw + if raw.startswith(("http://", "https://")): + return raw + return f"{self.host}/{self._extract_site_id(raw)}.htm" + + def _extract_pid(self, html): + matched = re.search(r"var\s+pid\s*=\s*(\d+)\s*;", self._stringify(html)) + return self._stringify(matched.group(1) if matched else "").strip() + + def _build_lines_signature(self, pid, millis): + payload = f"{pid}-{millis}" + key = hashlib.md5(payload.encode("utf-8")).hexdigest()[:16] + return self._aes_ecb_hex(key, payload).upper() + + def _pick_direct_urls(self, data): + urls = [] + for item in self._stringify((data or {}).get("url3")).split(","): + url = self._stringify(item).strip() + if not url: + continue + lower = url.lower() + if ".m3u8" in lower or "byteimg" in lower: + continue + urls.append(url) + return urls + + def _request_tos_url(self, pid, millis, sg, play_page_url): + target = f"{self.host}/god/{pid}?type=1" + response = self.post( + target, + data=f"t={millis}&sg={sg}&verifyCode=888", + headers=self._site_headers(play_page_url), + timeout=15, + ) + data = self._load_json(response) + url = self._stringify((data or {}).get("url")).strip() + if not url: + return "" + lower = url.lower() + if ".m3u8" in lower or "byteimg" in lower: + return "" + return url + + def _range_probe(self, url): + return self.fetch( + url, + headers=self.build_headers({"Range": "bytes=0-0"}), + timeout=20, + stream=True, + ) + + def _make_proxy_url(self, real_url): + return f"{self.LOCAL_PROXY_BASE}&type=m3u8&url={quote(self._stringify(real_url).strip(), safe='')}" + + def _aes_ecb_hex(self, key, data): + cipher = AES.new(str(key).encode("utf-8"), AES.MODE_ECB) + return cipher.encrypt(pad(self._stringify(data).encode("utf-8"), 16)).hex() + + def _decode_m3u8_payload(self, payload): + return gzip.decompress(payload[3354:]).decode("utf-8") + + def _rewrite_m3u8(self, text, real_url): + lines = [] + for raw_line in self._stringify(text).splitlines(): + line = raw_line.strip() + if not line or line.startswith("#"): + lines.append(raw_line) + continue + target = line if line.startswith(("http://", "https://")) else urljoin(real_url, line) + lines.append(self._make_proxy_url(target)) + return "\n".join(lines) + + def playerContent(self, flag, id, vipFlags): + raw = self._stringify(id).strip() + if not raw: + return {"parse": 1, "jx": 1, "playUrl": "", "url": "", "header": self.build_headers()} + + if not self._is_media_url(raw): + play_page_url = self._build_play_page_url(raw) + response = self.fetch(play_page_url, headers=self._site_headers(play_page_url), timeout=15) + html = response.text if response.status_code == 200 else "" + pid = self._extract_pid(html) + if not pid: + return {"parse": 1, "jx": 1, "playUrl": "", "url": play_page_url, "header": self._site_headers(play_page_url)} + + millis = self._current_millis() + sg = self._build_lines_signature(pid, millis) + lines_url = f"{self.host}/lines?t={millis}&sg={sg}&pid={pid}" + lines_response = self.fetch( + lines_url, + headers=self._site_headers(play_page_url, {"X-Requested-With": "XMLHttpRequest"}), + timeout=15, + ) + payload = self._load_json(lines_response) + data = (payload or {}).get("data") or {} + if int((payload or {}).get("code", 0)) == 0: + direct_urls = self._pick_direct_urls(data) + if direct_urls: + return { + "parse": 0, + "jx": 0, + "playUrl": "", + "url": direct_urls[0], + "header": {"User-Agent": self.UA, "Referer": self.host}, + } + if data.get("tos"): + tos_url = self._request_tos_url(pid, millis, sg, play_page_url) + if tos_url: + return { + "parse": 0, + "jx": 0, + "playUrl": "", + "url": tos_url, + "header": {"User-Agent": self.UA, "Referer": self.host}, + } + return {"parse": 1, "jx": 1, "playUrl": "", "url": play_page_url, "header": self._site_headers(play_page_url)} + + if re.search(r"\.mp4(?:[?#]|$)", raw, re.I): + try: + response = self._range_probe(raw) + if "video/mp4" in self._stringify(response.headers.get("Content-Type")).lower(): + return {"parse": 0, "jx": 0, "playUrl": "", "url": raw, "header": self.build_headers()} + except Exception: + return {"parse": 0, "jx": 0, "playUrl": "", "url": raw, "header": self.build_headers()} + + return {"parse": 0, "jx": 0, "playUrl": "", "url": self._make_proxy_url(raw), "header": self.build_headers()} + + def localProxy(self, params): + if not isinstance(params, dict): + return None + if self._stringify(params.get("type")).strip().lower() != "m3u8": + return None + real_url = self._stringify(params.get("url")).strip() + if not real_url: + return None + + response = self.fetch(real_url, headers=self.build_headers(), timeout=20) + if response.status_code != 200: + return [response.status_code, "text/plain", b""] + rewritten = self._rewrite_m3u8(self._decode_m3u8_payload(response.content), real_url) + return [200, "application/vnd.apple.mpegurl", rewritten.encode("utf-8")] + + def isVideoFormat(self, url): + return self._is_media_url(url) + + def manualVideoCheck(self): + return False