From 8eb9cc6f237e23ffcd7a0c73bbd621ff091f8be1 Mon Sep 17 00:00:00 2001 From: Harold <8866033@gmail.com> Date: Mon, 20 Apr 2026 09:32:49 +0800 Subject: [PATCH] feat: add letu list and search parsing --- py/tests/test_乐兔.py | 60 +++++++++++++++++++++++++++++++++ py/乐兔.py | 78 ++++++++++++++++++++++++++++++++++++++++++- 2 files changed, 137 insertions(+), 1 deletion(-) diff --git a/py/tests/test_乐兔.py b/py/tests/test_乐兔.py index 49b229f..e2f4593 100644 --- a/py/tests/test_乐兔.py +++ b/py/tests/test_乐兔.py @@ -1,6 +1,7 @@ import unittest from importlib.machinery import SourceFileLoader from pathlib import Path +from unittest.mock import patch ROOT = Path(__file__).resolve().parents[1] @@ -30,6 +31,65 @@ class TestLeTuSpider(unittest.TestCase): 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") + def test_parse_cards_extracts_compact_vod_ids(self): + html = """ +
+
+ + +
更新至1集
+
+
+ """ + self.assertEqual( + self.spider._parse_cards(html), + [ + { + "vod_id": "detail/demo", + "vod_name": "示例影片", + "vod_pic": "https://www.letu.me/cover.jpg", + "vod_remarks": "更新至1集", + } + ], + ) + + @patch.object(Spider, "_request_html") + def test_category_content_uses_type_page_url(self, mock_request_html): + 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) + self.assertEqual(result["limit"], 1) + self.assertNotIn("pagecount", result) + self.assertEqual(result["list"][0]["vod_id"], "detail/cat-demo") + + @patch.object(Spider, "_request_html") + def test_search_content_uses_search_url_and_parses_cards(self, mock_request_html): + mock_request_html.return_value = """ +
+
+ 搜索影片 + +
全集
+
+
+ """ + result = self.spider.searchContent("繁花", False, "1") + self.assertEqual( + 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.assertNotIn("pagecount", result) + if __name__ == "__main__": unittest.main() diff --git a/py/乐兔.py b/py/乐兔.py index 34f7135..de7ac1b 100644 --- a/py/乐兔.py +++ b/py/乐兔.py @@ -1,7 +1,7 @@ # coding=utf-8 import re import sys -from urllib.parse import urljoin +from urllib.parse import quote, urljoin from base.spider import Spider as BaseSpider @@ -58,3 +58,79 @@ class Spider(BaseSpider): 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 "" + + def _clean_text(self, text): + return re.sub(r"\s+", " ", str(text or "")).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) + 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 node in root.xpath("//*[contains(@class,'grid') and contains(@class,'container_list')]//*[contains(@class,'s6')]"): + href = "".join(node.xpath(".//a[1]/@href")).strip() + vod_id = self._encode_vod_id(href) + if not vod_id or vod_id in seen: + continue + name = self._clean_text("".join(node.xpath(".//a[1]/@title")) or "".join(node.xpath(".//a[1]//text()"))) + if not name: + continue + pic = self._clean_text( + "".join(node.xpath(".//*[contains(@class,'large')][1]/@data-src | .//*[contains(@class,'large')][1]/@src")) + ) + remark = self._clean_text("".join(node.xpath(".//*[contains(@class,'small-text')][1]//text()"))) + seen.add(vod_id) + items.append( + { + "vod_id": vod_id, + "vod_name": name, + "vod_pic": self._build_url(pic), + "vod_remarks": remark, + } + ) + return items + + def _page_result(self, items, pg): + page = int(pg) + return {"page": page, "limit": len(items), "total": len(items), "list": items} + + def categoryContent(self, tid, pg, filter, extend): + html = self._request_html(self._build_url(f"/type/{tid}-{int(pg)}.html")) + return self._page_result(self._parse_cards(html), pg) + + def searchContent(self, key, quick, pg="1"): + keyword = self._clean_text(key) + if not keyword: + return self._page_result([], pg) + html = self._request_html(self._build_url(f"/vodsearch/-------------.html?wd={quote(keyword)}")) + root = self.html(html) + if root is None: + return self._page_result([], pg) + items = [] + for node in root.xpath("//*[contains(@class,'result-list')]//*[contains(@class,'result-item')]"): + href = "".join(node.xpath(".//a[1]/@href")).strip() + vod_id = self._encode_vod_id(href) + name = self._clean_text("".join(node.xpath(".//a[1]//text()"))) + if not vod_id or not name: + continue + pic = self._clean_text( + "".join(node.xpath(".//*[contains(@class,'large')][1]/@data-src | .//*[contains(@class,'large')][1]/@src")) + ) + remark = self._clean_text("".join(node.xpath(".//*[contains(@class,'small-text')][1]//text()"))) + items.append( + { + "vod_id": vod_id, + "vod_name": name, + "vod_pic": self._build_url(pic), + "vod_remarks": remark, + } + ) + return self._page_result(items, pg)