From 5efe589685377a11d8ccdb0ab8a13b1e99bb7688 Mon Sep 17 00:00:00 2001 From: Harold <8866033@gmail.com> Date: Mon, 20 Apr 2026 10:33:33 +0800 Subject: [PATCH] =?UTF-8?q?=E7=9F=AD=E5=89=A7=E7=BD=91?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- py/tests/test_乐兔.py | 60 ++++++----- py/乐兔.py | 24 +++-- py/短剧网.py | 238 ++++++++++++++++++++++++++++++++++++++++++ 3 files changed, 286 insertions(+), 36 deletions(-) create mode 100644 py/短剧网.py diff --git a/py/tests/test_乐兔.py b/py/tests/test_乐兔.py index f96ffdd..5ea5050 100644 --- a/py/tests/test_乐兔.py +++ b/py/tests/test_乐兔.py @@ -28,16 +28,16 @@ class TestLeTuSpider(unittest.TestCase): self.assertEqual(self.spider.homeVideoContent(), {"list": []}) def test_encode_and_decode_detail_and_play_ids(self): - self.assertEqual(self.spider._encode_vod_id("/detail/demo.html"), "detail/demo") - self.assertEqual(self.spider._decode_vod_id("detail/demo"), "https://www.letu.me/detail/demo.html") - self.assertEqual(self.spider._encode_play_id("/play/123-1-1.html"), "play/123-1-1") - self.assertEqual(self.spider._decode_play_id("play/123-1-1"), "https://www.letu.me/play/123-1-1.html") + self.assertEqual(self.spider._encode_vod_id("/vods/305741.html"), "vods/305741") + self.assertEqual(self.spider._decode_vod_id("vods/305741"), "https://www.letu.me/vods/305741.html") + self.assertEqual(self.spider._encode_play_id("/vod/305741-1-1.html"), "vod/305741-1-1") + self.assertEqual(self.spider._decode_play_id("vod/305741-1-1"), "https://www.letu.me/vod/305741-1-1.html") def test_parse_cards_extracts_compact_vod_ids(self): html = """
- +
更新至1集
@@ -47,7 +47,7 @@ class TestLeTuSpider(unittest.TestCase): self.spider._parse_cards(html), [ { - "vod_id": "detail/demo", + "vod_id": "vods/305741", "vod_name": "示例影片", "vod_pic": "https://www.letu.me/cover.jpg", "vod_remarks": "更新至1集", @@ -60,25 +60,31 @@ class TestLeTuSpider(unittest.TestCase): mock_request_html.return_value = """
- +
HD
""" - result = self.spider.categoryContent("2", "3", False, {}) - self.assertEqual(mock_request_html.call_args.args[0], "https://www.letu.me/type/2-3.html") - self.assertEqual(result["page"], 3) + result = self.spider.categoryContent("2", "5", False, {}) + self.assertEqual(mock_request_html.call_args.args[0], "https://www.letu.me/type/2-5.html") + self.assertEqual(result["page"], 5) self.assertEqual(result["limit"], 1) self.assertNotIn("pagecount", result) - self.assertEqual(result["list"][0]["vod_id"], "detail/cat-demo") + self.assertEqual(result["list"][0]["vod_id"], "vods/90001") + + @patch.object(Spider, "_request_html") + def test_category_content_uses_plain_type_path_for_first_page(self, mock_request_html): + mock_request_html.return_value = "" + self.spider.categoryContent("1", "1", False, {}) + self.assertEqual(mock_request_html.call_args.args[0], "https://www.letu.me/type/1.html") @patch.object(Spider, "_request_html") def test_search_content_uses_search_url_and_parses_cards(self, mock_request_html): mock_request_html.return_value = """
- 搜索影片 + 搜索影片
全集
@@ -89,7 +95,7 @@ class TestLeTuSpider(unittest.TestCase): mock_request_html.call_args.args[0], "https://www.letu.me/vodsearch/-------------.html?wd=%E7%B9%81%E8%8A%B1", ) - self.assertEqual(result["list"][0]["vod_id"], "detail/search-demo") + self.assertEqual(result["list"][0]["vod_id"], "vods/70001") self.assertNotIn("pagecount", result) def test_parse_detail_page_extracts_metadata_and_playlists(self): @@ -108,16 +114,16 @@ class TestLeTuSpider(unittest.TestCase): 线路B
- 第1集 - 第2集 + 第1集 + 第2集
- 正片 + 正片
""" - result = self.spider._parse_detail_page(html, "detail/demo") + result = self.spider._parse_detail_page(html, "vods/305741") vod = result["list"][0] - self.assertEqual(vod["vod_id"], "detail/demo") + self.assertEqual(vod["vod_id"], "vods/305741") self.assertEqual(vod["vod_name"], "详情标题") self.assertEqual(vod["vod_pic"], "https://www.letu.me/poster.jpg") self.assertEqual(vod["type_name"], "电影") @@ -128,19 +134,19 @@ class TestLeTuSpider(unittest.TestCase): self.assertEqual(vod["vod_play_from"], "线路A$$$线路B") self.assertEqual( vod["vod_play_url"], - "第1集$play/123-1-1#第2集$play/123-1-2$$$正片$play/123-2-1", + "第1集$vod/305741-1-1#第2集$vod/305741-1-2$$$正片$vod/305741-2-1", ) @patch.object(Spider, "_request_html") def test_detail_content_decodes_compact_vod_id(self, mock_request_html): mock_request_html.return_value = "

详情标题

" - self.spider.detailContent(["detail/demo"]) - self.assertEqual(mock_request_html.call_args.args[0], "https://www.letu.me/detail/demo.html") + self.spider.detailContent(["vods/305741"]) + self.assertEqual(mock_request_html.call_args.args[0], "https://www.letu.me/vods/305741.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", {}) + result = self.spider.playerContent("线路A", "vod/305741-1-1", {}) self.assertEqual(result["parse"], 0) self.assertEqual(result["jx"], 0) self.assertEqual(result["url"], "https://video.example/direct.m3u8") @@ -149,7 +155,7 @@ class TestLeTuSpider(unittest.TestCase): 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", {}) + result = self.spider.playerContent("线路A", "vod/305741-1-1", {}) self.assertEqual(result["parse"], 0) self.assertEqual(result["url"], "https://video.example/rose.m3u8") @@ -164,18 +170,18 @@ class TestLeTuSpider(unittest.TestCase): '', '', ] - encrypt_1 = self.spider.playerContent("线路A", "play/123-1-1", {}) - encrypt_2 = self.spider.playerContent("线路A", "play/123-1-2", {}) + encrypt_1 = self.spider.playerContent("线路A", "vod/305741-1-1", {}) + encrypt_2 = self.spider.playerContent("线路A", "vod/305741-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 = "" - result = self.spider.playerContent("线路A", "play/123-1-1", {}) + result = self.spider.playerContent("线路A", "vod/305741-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") + self.assertEqual(result["url"], "https://www.letu.me/vod/305741-1-1.html") if __name__ == "__main__": diff --git a/py/乐兔.py b/py/乐兔.py index 905c441..9cf1fec 100644 --- a/py/乐兔.py +++ b/py/乐兔.py @@ -46,20 +46,20 @@ class Spider(BaseSpider): return urljoin(self.host + "/", str(path or "").strip()) def _encode_vod_id(self, href): - matched = re.search(r"/detail/([^/?#]+)\.html", self._build_url(href)) - return f"detail/{matched.group(1)}" if matched else "" + matched = re.search(r"/(vods|vod|detail)/([^/?#]+)\.html", self._build_url(href)) + return f"{matched.group(1)}/{matched.group(2)}" if matched else "" def _decode_vod_id(self, vod_id): - matched = re.search(r"^detail/([^/?#]+)$", str(vod_id or "").strip()) - return self._build_url(f"/detail/{matched.group(1)}.html") if matched else "" + matched = re.search(r"^(vods|vod|detail)/([^/?#]+)$", str(vod_id or "").strip()) + return self._build_url(f"/{matched.group(1)}/{matched.group(2)}.html") if matched else "" def _encode_play_id(self, href): - matched = re.search(r"/play/([^/?#]+)\.html", self._build_url(href)) - return f"play/{matched.group(1)}" if matched else "" + matched = re.search(r"/(vod|vodplay|play)/([^/?#]+)\.html", self._build_url(href)) + return f"{matched.group(1)}/{matched.group(2)}" if matched else "" def _decode_play_id(self, play_id): - matched = re.search(r"^play/([^/?#]+)$", str(play_id or "").strip()) - return self._build_url(f"/play/{matched.group(1)}.html") if matched else "" + matched = re.search(r"^(vod|vodplay|play)/([^/?#]+)$", str(play_id or "").strip()) + return self._build_url(f"/{matched.group(1)}/{matched.group(2)}.html") if matched else "" def _clean_text(self, text): return re.sub(r"\s+", " ", str(text or "")).strip() @@ -104,8 +104,14 @@ class Spider(BaseSpider): page = int(pg) return {"page": page, "limit": len(items), "total": len(items), "list": items} + def _build_category_url(self, tid, pg): + page = int(pg) + if page <= 1: + return self._build_url(f"/type/{tid}.html") + return self._build_url(f"/type/{tid}-{page}.html") + def categoryContent(self, tid, pg, filter, extend): - html = self._request_html(self._build_url(f"/type/{tid}-{int(pg)}.html")) + html = self._request_html(self._build_category_url(tid, pg)) return self._page_result(self._parse_cards(html), pg) def searchContent(self, key, quick, pg="1"): diff --git a/py/短剧网.py b/py/短剧网.py new file mode 100644 index 0000000..5eed18c --- /dev/null +++ b/py/短剧网.py @@ -0,0 +1,238 @@ +# coding=utf-8 +import re +import sys +from urllib.parse import quote + +from base.spider import Spider as BaseSpider + +sys.path.append("..") + + +class Spider(BaseSpider): + def __init__(self): + self.name = "短剧网" + self.host = "https://sm3.cc" + self.headers = { + "User-Agent": ( + "Mozilla/5.0 (iPhone; CPU iPhone OS 16_0 like Mac OS X) " + "AppleWebKit/605.1.15 (KHTML, like Gecko) Mobile/15E148" + ), + "Referer": self.host + "/", + } + self.classes = [ + {"type_id": "1", "type_name": "短剧大全"}, + {"type_id": "2", "type_name": "更新短剧"}, + ] + + def init(self, extend=""): + return None + + def getName(self): + return self.name + + def homeContent(self, filter): + return {"class": self.classes} + + def homeVideoContent(self): + return {"list": []} + + def _stringify(self, value): + return "" if value is None else str(value) + + def _build_url(self, path): + raw = self._stringify(path).strip() + if not raw: + return "" + if raw.startswith(("http://", "https://")): + return raw + if raw.startswith("//"): + return "https:" + raw + if raw.startswith("/"): + return self.host + raw + return self.host + "/" + raw + + def _clean_text(self, text): + return re.sub(r"\s+", " ", str(text or "").replace("\xa0", " ")).strip() + + def _request_html(self, path_or_url): + target = path_or_url if str(path_or_url).startswith("http") else self._build_url(path_or_url) + response = self.fetch(target, headers=dict(self.headers), timeout=10, 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 card in root.xpath("//li[contains(@class,'col-6')]"): + anchors = card.xpath(".//h3[contains(@class,'f-14')]//a[@href]") + if not anchors: + continue + anchor = anchors[0] + name = self._clean_text("".join(anchor.xpath(".//text()"))) + href = self._stringify((anchor.xpath("./@href") or [""])[0]).strip() + title_attr = self._stringify((anchor.xpath("./@title") or [""])[0]).strip() + if not name or not href: + continue + url = self._build_url(href) + if url in seen: + continue + seen.add(url) + pic = self._build_url( + (card.xpath(".//img[contains(@class,'lazy')]/@data-original") or [""])[0] + ) + remarks = "" + if title_attr: + m = re.search(r"（(.+?)）", title_attr) + if m: + remarks = m.group(1) + items.append( + { + "vod_id": url, + "vod_name": name, + "vod_pic": pic, + "vod_remarks": remarks, + } + ) + return items + + def _page_result(self, items, pg): + page = int(pg) + return { + "list": items, + "page": page, + "pagecount": page + 1 if items else page, + "limit": len(items) or 20, + "total": (page + 1 if items else page) * (len(items) or 20), + } + + def categoryContent(self, tid, pg, filter, extend): + page = int(pg) + url = f"{self.host}/?cate={tid}&page={page}" + items = self._parse_cards(self._request_html(url)) + return self._page_result(items, pg) + + def searchContent(self, key, quick, pg="1"): + page = int(pg) + keyword = self._clean_text(key) + if not keyword: + return self._page_result([], pg) + url = f"{self.host}/search.php?q={quote(keyword)}&page={page}" + items = self._parse_cards(self._request_html(url)) + return self._page_result(items, pg) + + def _is_netdisk_url(self, value): + text = self._stringify(value).strip().lower() + return any( + token in text + for token in [ + "drive.uc.cn", + "pan.quark.cn", + "pan.baidu.com", + "pan.xunlei.com", + "alipan.com", + "aliyundrive.com", + ] + ) + + def _normalize_disk_name_from_url(self, value): + text = self._stringify(value).strip().lower() + if "pan.baidu.com" in text: + return "baidu" + if "pan.quark.cn" in text: + return "quark" + if "drive.uc.cn" in text: + return "uc" + if "alipan.com" in text or "aliyundrive.com" in text: + return "aliyun" + if "pan.xunlei.com" in text: + return "xunlei" + return "" + + def _disk_priority(self, name): + order = {"baidu": 1, "quark": 2, "uc": 3, "aliyun": 4, "xunlei": 5} + return order.get(name, 999) + + def _extract_share_links(self, html): + root = self.html(html) + if root is None: + return [] + links = [] + seen = set() + for anchor in root.xpath("//*[contains(@class,'content')]//a[@href]"): + href = self._build_url((anchor.xpath("./@href") or [""])[0]) + if not href or href in seen: + continue + seen.add(href) + links.append(href) + return links + + def _build_play_from_links(self, share_links): + if not share_links: + return {"vod_play_from": "短剧网", "vod_play_url": ""} + grouped = {} + order_seen = [] + for link in share_links: + disk_name = self._normalize_disk_name_from_url(link) + if not disk_name: + continue + if disk_name not in grouped: + grouped[disk_name] = [] + order_seen.append(disk_name) + title = disk_name + urls_in_group = {item.split("$", 1)[1] for item in grouped[disk_name]} + if link not in urls_in_group: + grouped[disk_name].append(f"{title}${link}") + if not grouped: + fallback = "#".join( + f"{i + 1}$push://{url}" for i, url in enumerate(share_links) + ) + return {"vod_play_from": "短剧网", "vod_play_url": fallback} + names = sorted(order_seen, key=self._disk_priority) + return { + "vod_play_from": "$$$".join(names), + "vod_play_url": "$$$".join("#".join(grouped[n]) for n in names), + } + + def detailContent(self, ids): + result = {"list": []} + for raw_id in ids: + url = self._build_url(raw_id) + if not url: + continue + html = self._request_html(url) + root = self.html(html) + if root is None: + continue + title = self._clean_text("".join(root.xpath("//h1[1]//text()"))) + if not title: + title_text = self._clean_text("".join(root.xpath("//title[1]//text()"))) + title = title_text.split("(")[0].strip() or "短剧" + pic = self._build_url( + (root.xpath("(//img[contains(@class,'lazy')]/@data-original)[1]") or [""])[0] + or (root.xpath("(//img/@src)[1]") or [""])[0] + ) + share_links = self._extract_share_links(html) + play_data = self._build_play_from_links(share_links) + result["list"].append( + { + "vod_id": url, + "vod_name": title, + "vod_pic": pic, + "vod_content": "此为推送网盘规则", + "vod_play_from": play_data["vod_play_from"], + "vod_play_url": play_data["vod_play_url"], + } + ) + return result + + def playerContent(self, flag, id, vipFlags): + raw = self._stringify(id).strip() + if raw.startswith("push://"): + return {"parse": 0, "url": raw[7:]} + if self._is_netdisk_url(raw): + return {"parse": 0, "url": raw} + return {"parse": 0, "url": raw}