From 197eec9a61dd0c40ff7c51fabec964c6f66e8fc4 Mon Sep 17 00:00:00 2001 From: Harold <8866033@gmail.com> Date: Sat, 18 Apr 2026 11:59:46 +0800 Subject: [PATCH] feat: add dbku category and search flow --- py/tests/test_dbku.py | 57 ++++++++++++++++++++++++++++++++++ py/独播库.py | 71 +++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 128 insertions(+) diff --git a/py/tests/test_dbku.py b/py/tests/test_dbku.py index 7512f8b..0da31c3 100644 --- a/py/tests/test_dbku.py +++ b/py/tests/test_dbku.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] @@ -38,6 +39,62 @@ class TestDBKUSpider(unittest.TestCase): ], ) + @patch.object(Spider, "fetch") + def test_request_html_uses_dbku_headers(self, mock_fetch): + class FakeResponse: + def __init__(self, text): + self.text = text + self.status_code = 200 + self.encoding = "utf-8" + + mock_fetch.return_value = FakeResponse("ok") + html = self.spider._request_html("/vodtype/1--------1---.html", expect_xpath="//body") + self.assertIn("ok", html) + called_headers = mock_fetch.call_args.kwargs["headers"] + self.assertEqual(called_headers["Referer"], "https://www.dbku.tv") + self.assertEqual(called_headers["Origin"], "https://www.dbku.tv") + + @patch.object(Spider, "_request_html") + def test_category_content_builds_page_result(self, mock_request_html): + mock_request_html.return_value = """ +
+ + HD +
+ """ + result = self.spider.categoryContent("movie", "2", False, {}) + self.assertEqual(result["page"], 2) + self.assertEqual(result["list"][0]["vod_name"], "分类影片") + self.assertEqual(result["list"][0]["vod_pic"], "https://www.dbku.tv/cover.jpg") + + def test_parse_search_cards_prefers_search_list_container(self): + html = """ +
+
+ +
+
+
+ +
+ """ + results = self.spider._parse_search_cards(html) + self.assertEqual(len(results), 1) + self.assertEqual(results[0]["vod_name"], "搜索命中") + + @patch.object(Spider, "_request_html") + def test_search_content_reuses_search_parser(self, mock_request_html): + mock_request_html.return_value = """ +
+
+ +
+
+ """ + result = self.spider.searchContent("繁花", False, "1") + self.assertEqual(result["list"][0]["vod_id"], "https://www.dbku.tv/voddetail/321.html") + self.assertEqual(result["list"][0]["vod_name"], "搜索影片") + if __name__ == "__main__": unittest.main() diff --git a/py/独播库.py b/py/独播库.py index 5223e36..52935ec 100644 --- a/py/独播库.py +++ b/py/独播库.py @@ -1,5 +1,8 @@ # coding=utf-8 import sys +from urllib.parse import quote + +from lxml import etree from base.spider import Spider as BaseSpider @@ -25,6 +28,14 @@ class Spider(BaseSpider): {"type_name": "港剧", "type_id": "hk"}, {"type_name": "陆剧", "type_id": "luju"}, ] + self.category_paths = { + "index": "/vodtype/2--------{pg}---.html", + "movie": "/vodtype/1--------{pg}---.html", + "variety": "/vodtype/3--------{pg}---.html", + "anime": "/vodtype/4--------{pg}---.html", + "hk": "/vodtype/20--------{pg}---.html", + "luju": "/vodtype/13--------{pg}---.html", + } def init(self, extend=""): return None @@ -89,3 +100,63 @@ class Spider(BaseSpider): } ) return results + + def _request_html(self, path_or_url, expect_xpath=None, referer=None): + target = path_or_url if path_or_url.startswith("http") else self._build_url(path_or_url) + headers = dict(self.headers) + headers["Referer"] = referer or self.host + headers["Origin"] = self.host + response = self.fetch(target, headers=headers, timeout=10) + if response.status_code != 200: + return "" + html = response.text or "" + if expect_xpath: + root = self.html(html) + if root is None or not root.xpath(expect_xpath): + return "" + return html + + def _parse_cards_from_nodes(self, nodes): + results = [] + seen = set() + for card in nodes: + snippet = self._parse_list_cards(etree.tostring(card, encoding="unicode")) + for item in snippet: + if item["vod_id"] in seen: + continue + seen.add(item["vod_id"]) + results.append(item) + return results + + def _parse_search_cards(self, html): + root = self.html(html) + if root is None: + return [] + search_list = root.xpath("//*[@id='searchList']") + if search_list: + cards = search_list[0].xpath(".//*[contains(@class,'myui-vodlist__box')]") + parsed = self._parse_cards_from_nodes(cards) + if parsed: + return parsed + return self._parse_list_cards(html) + + def _page_result(self, items, pg): + page = int(pg) + pagecount = page + 1 if items else page + return { + "list": items, + "page": page, + "pagecount": pagecount, + "limit": len(items), + "total": pagecount * max(len(items), 1), + } + + def categoryContent(self, tid, pg, filter, extend): + path = self.category_paths.get(tid, self.category_paths["index"]).format(pg=pg) + html = self._request_html(path, expect_xpath="//*[contains(@class,'myui-vodlist__box')]") + return self._page_result(self._parse_list_cards(html), pg) + + def searchContent(self, key, quick, pg="1"): + path = "/vodsearch/-------------.html?wd={0}&submit=".format(quote(key)) + html = self._request_html(path, expect_xpath="//*[@id='searchList']|//*[contains(@class,'myui-vodlist__box')]") + return self._page_result(self._parse_search_cards(html), pg)