From 4dc3be3b146ad081e40995ad7476092c1876782d Mon Sep 17 00:00:00 2001 From: Harold <8866033@gmail.com> Date: Sat, 18 Apr 2026 12:55:43 +0800 Subject: [PATCH] =?UTF-8?q?=E7=8B=AC=E6=92=AD=E5=BA=93.py?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../plans/2026-04-18-dbku-spider.md | 672 ++++++++++++++++++ py/tests/test_czzy.py | 228 ------ 2 files changed, 672 insertions(+), 228 deletions(-) create mode 100644 py/docs/superpowers/plans/2026-04-18-dbku-spider.md delete mode 100644 py/tests/test_czzy.py diff --git a/py/docs/superpowers/plans/2026-04-18-dbku-spider.md b/py/docs/superpowers/plans/2026-04-18-dbku-spider.md new file mode 100644 index 0000000..c8db55d --- /dev/null +++ b/py/docs/superpowers/plans/2026-04-18-dbku-spider.md @@ -0,0 +1,672 @@ +# 独播库 Spider Implementation Plan + +> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. + +**Goal:** 在当前 Python 仓库中新增一个符合 `base.spider.Spider` 接口的独播库爬虫，支持分类、详情、搜索和站内播放解析。 + +**Architecture:** 以单文件站点脚本 `独播库.py` 承担独播库站点逻辑，内部拆分为 URL 归一化、卡片解析、详情解析和 `player_data` 解码几组辅助方法。测试沿用当前仓库 `unittest + SourceFileLoader + mock` 风格，优先覆盖纯解析函数和高层方法的 mock 网络流程，不依赖真实站点网络。 + +**Tech Stack:** Python 3, `requests`, `lxml`, `unittest`, `unittest.mock`, `base64`, `json`, `urllib.parse` + +--- + +## File Structure + +- Create: `独播库.py` + - 独播库站点实现，继承 `base.spider.Spider` + - 暴露 `init`、`homeContent`、`homeVideoContent`、`categoryContent`、`detailContent`、`searchContent`、`playerContent` + - 私有方法负责 URL 归一化、列表卡片解析、搜索结果解析、详情解析、`player_data` 提取与解码 +- Create: `tests/test_dbku.py` + - 用 `SourceFileLoader` 加载 `独播库.py` + - 用 HTML 片段与 mock response 测试分类、搜索、详情和播放器解析 +- Modify: `docs/superpowers/plans/2026-04-18-dbku-spider.md` + - 仅用于勾选执行状态 + +### Task 1: Scaffold Spider And List Card Parsing + +**Files:** +- Create: `tests/test_dbku.py` +- Create: `独播库.py` + +- [ ] **Step 1: Write the failing test** + +```python +import unittest +from importlib.machinery import SourceFileLoader +from pathlib import Path + + +ROOT = Path(__file__).resolve().parents[1] +MODULE = SourceFileLoader("dbku_spider", str(ROOT / "独播库.py")).load_module() +Spider = MODULE.Spider + + +class TestDBKUSpider(unittest.TestCase): + def setUp(self): + self.spider = Spider() + self.spider.init() + + def test_home_content_exposes_expected_categories(self): + content = self.spider.homeContent(False) + class_ids = [item["type_id"] for item in content["class"]] + self.assertEqual(class_ids, ["index", "movie", "variety", "anime", "hk", "luju"]) + + def test_parse_list_cards_extracts_detail_url_title_cover_and_description(self): + html = """ +
+ + 更新至10集 +
+ """ + cards = self.spider._parse_list_cards(html) + self.assertEqual( + cards, + [{ + "vod_id": "https://www.dbku.tv/voddetail/123.html", + "vod_name": "示例影片", + "vod_pic": "https://img.example/dbku.jpg", + "vod_remarks": "更新至10集", + }], + ) + + +if __name__ == "__main__": + unittest.main() +``` + +- [ ] **Step 2: Run test to verify it fails** + +Run: `python -m unittest tests.test_dbku.TestDBKUSpider -v` +Expected: FAIL with `FileNotFoundError` for `独播库.py` or missing methods. + +- [ ] **Step 3: Write minimal implementation** + +```python +# coding=utf-8 +import sys +from urllib.parse import urljoin + +from base.spider import Spider as BaseSpider + +sys.path.append("..") + + +class Spider(BaseSpider): + def __init__(self): + self.name = "独播库" + self.host = "https://www.dbku.tv" + self.headers = { + "User-Agent": ( + "Mozilla/5.0 (Windows NT 10.0; Win64; x64) " + "AppleWebKit/537.36 (KHTML, like Gecko) " + "Chrome/124.0.0.0 Safari/537.36" + ) + } + self.categories = [ + {"type_name": "连续剧", "type_id": "index"}, + {"type_name": "电影", "type_id": "movie"}, + {"type_name": "综艺", "type_id": "variety"}, + {"type_name": "动漫", "type_id": "anime"}, + {"type_name": "港剧", "type_id": "hk"}, + {"type_name": "陆剧", "type_id": "luju"}, + ] + + def init(self, extend=""): + return None + + def getName(self): + return self.name + + def homeContent(self, filter): + return {"class": self.categories} + + def homeVideoContent(self): + return {"list": []} + + def _build_url(self, href): + raw = str(href or "").strip() + if not raw: + return "" + if raw.startswith("http://") or raw.startswith("https://"): + return raw + if raw.startswith("//"): + return "https:" + raw + if raw.startswith("/"): + return self.host + raw + return self.host + "/" + raw + + def _parse_list_cards(self, html): + root = self.html(html) + results = [] + if root is None: + return results + + cards = root.xpath("//*[contains(@class,'myui-vodlist__box')]") + seen = set() + for card in cards: + href = "" + title = "" + pic = "" + for anchor in card.xpath(".//a[@href]"): + raw_href = (anchor.xpath("./@href") or [""])[0].strip() + if "/voddetail/" in raw_href: + href = self._build_url(raw_href) + title = ( + (anchor.xpath("./@title") or [""])[0].strip() + or "".join(anchor.xpath(".//text()")).strip() + ) + pic = ( + (anchor.xpath("./@data-original") or [""])[0].strip() + or (anchor.xpath("./@src") or [""])[0].strip() + ) + break + if not href or href in seen or not title: + continue + remarks = "".join(card.xpath(".//*[contains(@class,'pic-text')][1]//text()")).strip() + seen.add(href) + results.append({ + "vod_id": href, + "vod_name": title, + "vod_pic": self._build_url(pic), + "vod_remarks": remarks, + }) + return results +``` + +- [ ] **Step 4: Run test to verify it passes** + +Run: `python -m unittest tests.test_dbku.TestDBKUSpider -v` +Expected: PASS for the two new tests. + +- [ ] **Step 5: Commit** + +```bash +git add tests/test_dbku.py 独播库.py +git commit -m "feat: scaffold dbku spider parsing" +``` + +### Task 2: Add Category And Search Flows + +**Files:** +- Modify: `tests/test_dbku.py` +- Modify: `独播库.py` + +- [ ] **Step 1: Write the failing test** + +```python +from unittest.mock import patch + + +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"], "搜索影片") +``` + +- [ ] **Step 2: Run test to verify it fails** + +Run: `python -m unittest tests.test_dbku.TestDBKUSpider.test_request_html_uses_dbku_headers tests.test_dbku.TestDBKUSpider.test_category_content_builds_page_result tests.test_dbku.TestDBKUSpider.test_parse_search_cards_prefers_search_list_container tests.test_dbku.TestDBKUSpider.test_search_content_reuses_search_parser -v` +Expected: FAIL with missing `_request_html`, `categoryContent`, `searchContent`, or `_parse_search_cards`. + +- [ ] **Step 3: Write minimal implementation** + +```python +from urllib.parse import quote +from lxml import etree + + +class Spider(BaseSpider): + def __init__(self): + self.name = "独播库" + self.host = "https://www.dbku.tv" + self.headers = { + "User-Agent": ( + "Mozilla/5.0 (Windows NT 10.0; Win64; x64) " + "AppleWebKit/537.36 (KHTML, like Gecko) " + "Chrome/124.0.0.0 Safari/537.36" + ) + } + self.categories = [ + {"type_name": "连续剧", "type_id": "index"}, + {"type_name": "电影", "type_id": "movie"}, + {"type_name": "综艺", "type_id": "variety"}, + {"type_name": "动漫", "type_id": "anime"}, + {"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 _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_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 _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 _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) +``` + +- [ ] **Step 4: Run test to verify it passes** + +Run: `python -m unittest tests.test_dbku.TestDBKUSpider.test_request_html_uses_dbku_headers tests.test_dbku.TestDBKUSpider.test_category_content_builds_page_result tests.test_dbku.TestDBKUSpider.test_parse_search_cards_prefers_search_list_container tests.test_dbku.TestDBKUSpider.test_search_content_reuses_search_parser -v` +Expected: PASS for all four tests. + +- [ ] **Step 5: Commit** + +```bash +git add tests/test_dbku.py 独播库.py +git commit -m "feat: add dbku category and search flow" +``` + +### Task 3: Add Detail Parsing And Episode List Extraction + +**Files:** +- Modify: `tests/test_dbku.py` +- Modify: `独播库.py` + +- [ ] **Step 1: Write the failing test** + +```python +class TestDBKUSpider(unittest.TestCase): + def test_parse_detail_page_extracts_meta_and_episodes(self): + html = """ +
+ +
+
+

独播剧

+

年份：2025

+

地区：大陆

+

导演：张三

+

主演：李四

+
+ 一段剧情简介 + 第1集 + 第2集 + """ + result = self.spider._parse_detail_page(html, "https://www.dbku.tv/voddetail/100.html") + vod = result["list"][0] + self.assertEqual(vod["vod_name"], "独播剧") + self.assertEqual(vod["vod_year"], "2025") + self.assertEqual(vod["vod_play_from"], "独播库") + self.assertIn("第1集$https://www.dbku.tv/vodplay/100-1-1.html", vod["vod_play_url"]) + + @patch.object(Spider, "_request_html") + def test_detail_content_reads_from_vod_id_url(self, mock_request_html): + mock_request_html.return_value = """ +

详情影片

+ 第1集 + """ + result = self.spider.detailContent(["https://www.dbku.tv/voddetail/200.html"]) + self.assertEqual(result["list"][0]["vod_id"], "https://www.dbku.tv/voddetail/200.html") + self.assertEqual(result["list"][0]["vod_name"], "详情影片") +``` + +- [ ] **Step 2: Run test to verify it fails** + +Run: `python -m unittest tests.test_dbku.TestDBKUSpider.test_parse_detail_page_extracts_meta_and_episodes tests.test_dbku.TestDBKUSpider.test_detail_content_reads_from_vod_id_url -v` +Expected: FAIL with missing `_parse_detail_page` or `detailContent`. + +- [ ] **Step 3: Write minimal implementation** + +```python +import re + + +class Spider(BaseSpider): + def _extract_text_by_prefix(self, html, prefixes): + texts = re.findall(r">([^<>]+)<", html) + for text in texts: + clean = text.strip() + for prefix in prefixes: + if clean.startswith(prefix): + return clean.split("：", 1)[-1].strip() + return "" + + def _parse_detail_page(self, html, vod_id): + root = self.html(html) + title = ((root.xpath("//*[contains(@class,'title')][1]//text()") or [""])[0]).strip() + pic = ( + (root.xpath("//*[contains(@class,'myui-content__thumb')]//img/@data-original") or [""])[0].strip() + or (root.xpath("//*[contains(@class,'myui-content__thumb')]//img/@src") or [""])[0].strip() + ) + content = "".join(root.xpath("//*[contains(@class,'data')][1]//text()")).strip() + + episodes = [] + seen = set() + for href, label in re.findall(r']+href=["\\\']([^"\\\']*/vodplay/\\d+-\\d+-\\d+\\.html[^"\\\']*)["\\\'][^>]*>([\\s\\S]*?)', html, re.I): + url = self._build_url(href) + name = re.sub(r"<[^>]*>", "", label).strip() + if not url or not name or "立即播放" in name or url in seen: + continue + seen.add(url) + episodes.append(f"{name}${url}") + + vod = { + "vod_id": vod_id, + "vod_name": title, + "vod_pic": self._build_url(pic), + "vod_year": self._extract_text_by_prefix(html, ["年份："]), + "vod_area": self._extract_text_by_prefix(html, ["地区："]), + "vod_actor": self._extract_text_by_prefix(html, ["主演："]), + "vod_director": self._extract_text_by_prefix(html, ["导演："]), + "vod_content": content, + "vod_play_from": "独播库", + "vod_play_url": "#".join(episodes), + } + return {"list": [vod]} + + def detailContent(self, ids): + vod_id = ids[0] + html = self._request_html(vod_id, expect_xpath="//*[contains(@class,'title')]|//a[contains(@href,'/vodplay/')]") + return self._parse_detail_page(html, vod_id) +``` + +- [ ] **Step 4: Run test to verify it passes** + +Run: `python -m unittest tests.test_dbku.TestDBKUSpider.test_parse_detail_page_extracts_meta_and_episodes tests.test_dbku.TestDBKUSpider.test_detail_content_reads_from_vod_id_url -v` +Expected: PASS for both detail tests. + +- [ ] **Step 5: Commit** + +```bash +git add tests/test_dbku.py 独播库.py +git commit -m "feat: add dbku detail parsing" +``` + +### Task 4: Add player_data Extraction And Encrypt Decoders + +**Files:** +- Modify: `tests/test_dbku.py` +- Modify: `独播库.py` + +- [ ] **Step 1: Write the failing test** + +```python +import base64 + + +class TestDBKUSpider(unittest.TestCase): + def test_parse_player_data_reads_json_block(self): + html = '' + data = self.spider._parse_player_data(html) + self.assertEqual(data["url"], "https://video.example/a.m3u8") + + def test_decode_play_url_by_encrypt_supports_modes_0_1_2_3(self): + self.assertEqual( + self.spider._decode_play_url_by_encrypt("https://video.example/raw.m3u8", 0), + "https://video.example/raw.m3u8", + ) + self.assertEqual( + self.spider._decode_play_url_by_encrypt("https%3A//video.example/escape.m3u8", 1), + "https://video.example/escape.m3u8", + ) + mode2 = base64.b64encode("https://video.example/base64.m3u8".encode("utf-8")).decode("utf-8") + self.assertEqual( + self.spider._decode_play_url_by_encrypt(mode2, 2), + "https://video.example/base64.m3u8", + ) + mode3_raw = "ABCDEFGHhttps://video.example/trimmed.m3u8HGFEDCBA" + mode3 = base64.b64encode(mode3_raw.encode("utf-8")).decode("utf-8") + self.assertEqual( + self.spider._decode_play_url_by_encrypt("12345678" + mode3, 3), + "https://video.example/trimmed.m3u8", + ) +``` + +- [ ] **Step 2: Run test to verify it fails** + +Run: `python -m unittest tests.test_dbku.TestDBKUSpider.test_parse_player_data_reads_json_block tests.test_dbku.TestDBKUSpider.test_decode_play_url_by_encrypt_supports_modes_0_1_2_3 -v` +Expected: FAIL with missing `_parse_player_data` or `_decode_play_url_by_encrypt`. + +- [ ] **Step 3: Write minimal implementation** + +```python +import base64 +import json +from urllib.parse import unquote + + +class Spider(BaseSpider): + def _parse_player_data(self, html): + matched = re.search(r"var\\s+player_data\\s*=\\s*(\\{[\\s\\S]*?\\})\\s*", html, re.I) + if not matched: + matched = re.search(r"var\\s+player_[^=]*\\s*=\\s*(\\{[\\s\\S]*?\\})\\s*", html, re.I) + if not matched: + return None + try: + return json.loads(matched.group(1)) + except Exception: + return None + + def _decode_play_url_by_encrypt(self, value, encrypt): + raw = str(value or "") + mode = int(encrypt or 0) + if not raw: + return "" + try: + if mode == 1: + return unquote(raw) + if mode == 2: + decoded = base64.b64decode(raw).decode("utf-8") + return unquote(decoded) + if mode == 3: + text = raw[8:] if len(raw) > 16 else raw + text = base64.b64decode(text).decode("utf-8") + if len(text) > 16: + text = text[8:-8] + return text + return raw + except Exception: + return raw +``` + +- [ ] **Step 4: Run test to verify it passes** + +Run: `python -m unittest tests.test_dbku.TestDBKUSpider.test_parse_player_data_reads_json_block tests.test_dbku.TestDBKUSpider.test_decode_play_url_by_encrypt_supports_modes_0_1_2_3 -v` +Expected: PASS for both decoder tests. + +- [ ] **Step 5: Commit** + +```bash +git add tests/test_dbku.py 独播库.py +git commit -m "feat: add dbku player decoders" +``` + +### Task 5: Wire High-Level playerContent And Full Regression + +**Files:** +- Modify: `tests/test_dbku.py` +- Modify: `独播库.py` + +- [ ] **Step 1: Write the failing test** + +```python +class TestDBKUSpider(unittest.TestCase): + @patch.object(Spider, "_request_html") + def test_player_content_returns_decoded_direct_url(self, mock_request_html): + mock_request_html.return_value = """ + + """ + result = self.spider.playerContent("独播库", "https://www.dbku.tv/vodplay/100-1-1.html", {}) + self.assertEqual(result["parse"], 0) + self.assertEqual(result["url"], "https://video.example/final.m3u8") + self.assertEqual(result["header"]["Referer"], "https://www.dbku.tv/vodplay/100-1-1.html") + + @patch.object(Spider, "_request_html") + def test_player_content_follows_internal_jump(self, mock_request_html): + mock_request_html.side_effect = [ + '', + '', + ] + result = self.spider.playerContent("独播库", "https://www.dbku.tv/vodplay/100-1-1.html", {}) + self.assertEqual(result["parse"], 0) + self.assertEqual(result["url"], "https://video.example/jump-final.m3u8") +``` + +- [ ] **Step 2: Run test to verify it fails** + +Run: `python -m unittest tests.test_dbku.TestDBKUSpider.test_player_content_returns_decoded_direct_url tests.test_dbku.TestDBKUSpider.test_player_content_follows_internal_jump -v` +Expected: FAIL with missing `playerContent` or incorrect return payload. + +- [ ] **Step 3: Write minimal implementation** + +```python +class Spider(BaseSpider): + def _build_player_headers(self, referer): + return { + "User-Agent": self.headers["User-Agent"], + "Referer": referer, + } + + def playerContent(self, flag, id, vipFlags): + current = id + for _ in range(3): + html = self._request_html(current, referer=self.host) + data = self._parse_player_data(html) + if not data: + return {"parse": 0, "playUrl": "", "url": ""} + decoded = self._decode_play_url_by_encrypt(data.get("url", ""), data.get("encrypt", 0)) + play_url = self._build_url(decoded) + if not play_url: + return {"parse": 0, "playUrl": "", "url": ""} + if self.host in play_url and "/vodplay/" in play_url and play_url != current: + current = play_url + continue + return { + "parse": 0, + "playUrl": "", + "url": play_url, + "header": self._build_player_headers(current), + } + return {"parse": 0, "playUrl": "", "url": ""} +``` + +- [ ] **Step 4: Run test to verify it passes** + +Run: `python -m unittest tests.test_dbku.TestDBKUSpider.test_player_content_returns_decoded_direct_url tests.test_dbku.TestDBKUSpider.test_player_content_follows_internal_jump -v` +Expected: PASS for both player tests. + +- [ ] **Step 5: Run the full regression suite** + +Run: `python -m unittest tests.test_dbku -v` +Expected: PASS for the complete `tests/test_dbku.py` suite. + +- [ ] **Step 6: Commit** + +```bash +git add tests/test_dbku.py 独播库.py +git commit -m "feat: finish dbku spider implementation" +``` diff --git a/py/tests/test_czzy.py b/py/tests/test_czzy.py deleted file mode 100644 index 7a66558..0000000 --- a/py/tests/test_czzy.py +++ /dev/null @@ -1,228 +0,0 @@ -import unittest -from base64 import b64encode -from importlib.machinery import SourceFileLoader -from pathlib import Path -from unittest.mock import patch - -try: - from Crypto.Cipher import AES - from Crypto.Util.Padding import pad -except ModuleNotFoundError: - from Cryptodome.Cipher import AES - from Cryptodome.Util.Padding import pad - - -ROOT = Path(__file__).resolve().parents[1] -MODULE = SourceFileLoader("czzy_spider", str(ROOT / "厂长资源.py")).load_module() -Spider = MODULE.Spider - - -class TestCZZYSpider(unittest.TestCase): - def setUp(self): - self.spider = Spider() - self.spider.init() - - def test_default_host_prefers_czzy89(self): - self.assertEqual(self.spider.hosts[0], "https://www.czzy89.com") - self.assertEqual(self.spider.current_host, "https://www.czzy89.com") - - def test_home_content_exposes_expected_categories(self): - content = self.spider.homeContent(False) - class_ids = [item["type_id"] for item in content["class"]] - self.assertEqual(class_ids[:3], ["movie", "tv", "anime"]) - self.assertIn("cn_drama", class_ids) - - def test_parse_media_cards_extracts_basic_fields(self): - html = """ - - """ - cards = self.spider._parse_media_cards(html, "https://www.cz01.org") - self.assertEqual( - cards, - [ - { - "vod_id": "/movie/abc.html", - "vod_name": "测试影片", - "vod_pic": "https://img.example/cover.jpg", - "vod_remarks": "更新至10集", - } - ], - ) - - @patch.object(Spider, "fetch") - def test_request_html_falls_back_to_second_host(self, mock_fetch): - class FakeResponse: - def __init__(self, text, status_code=200): - self.text = text - self.status_code = status_code - self.encoding = "utf-8" - - mock_fetch.side_effect = [ - Exception("first host down"), - FakeResponse("ok"), - ] - - html, host = self.spider._request_html("/movie_bt/page/1", expect_xpath="//body") - self.assertIn("ok", html) - self.assertEqual(host, "https://www.cz01.org") - self.assertEqual(self.spider.current_host, "https://www.cz01.org") - - @patch.object(Spider, "_request_html") - def test_category_content_builds_media_list(self, mock_request_html): - mock_request_html.return_value = ( - """ - - """, - "https://www.czzy89.com", - ) - - 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.czzy89.com/cover.jpg") - - @patch.object(Spider, "_request_html") - def test_search_content_reuses_media_card_parser(self, mock_request_html): - mock_request_html.return_value = ( - """ -
-
  • - - -
    • -
    - """, - "https://www.czzy89.com", - ) - - result = self.spider.searchContent("繁花", False, "1") - self.assertEqual(result["list"][0]["vod_id"], "/movie/search-hit.html") - self.assertEqual(result["list"][0]["vod_name"], "搜索影片") - - def test_parse_detail_page_splits_direct_and_pan_sources(self): - html = """ -
    -

    繁花

    - -
    -
  • 年份：2024
    • -
  • 地区：中国大陆
    • -
  • 导演：王家卫
    • -
  • 主演：胡歌
    • -
    -
    一段剧情简介
    -
    -
    - 第1集 - 第2集 -
    -
    - 阿里云盘 -
    - """ - - detail = self.spider._parse_detail_page(html, "https://www.czzy89.com", "/movie/fanhua.html") - vod = detail["list"][0] - self.assertEqual(vod["vod_name"], "繁花") - self.assertEqual(vod["vod_year"], "2024") - self.assertEqual(vod["vod_play_from"], "厂长资源$$$网盘资源") - self.assertIn("第1集$https://www.czzy89.com/v_play/1.html", vod["vod_play_url"]) - self.assertIn("阿里云盘$https://www.alipan.com/s/demo", vod["vod_play_url"]) - - @patch.object(Spider, "_request_html") - def test_detail_content_reads_from_vod_id_path(self, mock_request_html): - mock_request_html.return_value = ( - """ -

    示例影片

    -
    立即播放
    - """, - "https://www.czzy89.com", - ) - - result = self.spider.detailContent(["/movie/example.html"]) - self.assertEqual(result["list"][0]["vod_id"], "/movie/example.html") - self.assertEqual(result["list"][0]["vod_name"], "示例影片") - - def test_extract_iframe_src(self): - html = '' - self.assertEqual( - self.spider._extract_iframe_src(html, "https://www.czzy89.com"), - "https://www.czzy89.com/player-v2/test", - ) - - def test_extract_player_url_prefers_mysvg_then_art_url(self): - self.assertEqual( - self.spider._extract_player_url_from_iframe("var mysvg='https://video.example/a.m3u8';"), - "https://video.example/a.m3u8", - ) - self.assertEqual( - self.spider._extract_player_url_from_iframe("art.url='https://video.example/b.m3u8';"), - "https://video.example/b.m3u8", - ) - - def test_extract_player_url_decodes_data_payload(self): - original = "https://video.example/data.m3u8" - middle = len(original) // 2 - obfuscated = original[:middle] + "ABCDEFG" + original[middle:] - encoded = "".join("{:02x}".format(ord(ch)) for ch in obfuscated)[::-1] - html = 'var config = {"data":"%s"};' % encoded - self.assertEqual(self.spider._extract_player_url_from_iframe(html), original) - - def test_extract_player_url_decrypts_player_payload(self): - payload = '{"url":"https://video.example/encrypted.m3u8"}' - key = b"VFBTzdujpR9FWBhe" - iv = b"1234567890abcdef" - cipher = AES.new(key, AES.MODE_CBC, iv) - encrypted = b64encode(cipher.encrypt(pad(payload.encode("utf-8"), AES.block_size))).decode("utf-8") - html = 'var player="%s";var rand="%s";' % (encrypted, iv.decode("utf-8")) - self.assertEqual( - self.spider._extract_player_url_from_iframe(html), - "https://video.example/encrypted.m3u8", - ) - - def test_extract_player_url_supports_wp_nonce_fallback(self): - html = """ - - """ - self.assertEqual( - self.spider._extract_player_url_from_iframe(html), - "https://video.example/wp.m3u8", - ) - - @patch.object(Spider, "_request_html") - def test_player_content_returns_pan_link_directly(self, mock_request_html): - result = self.spider.playerContent("网盘资源", "https://www.alipan.com/s/demo", {}) - self.assertEqual(result["parse"], 0) - self.assertEqual(result["url"], "https://www.alipan.com/s/demo") - mock_request_html.assert_not_called() - - @patch.object(Spider, "_request_html") - def test_player_content_resolves_iframe_chain(self, mock_request_html): - mock_request_html.side_effect = [ - ('', "https://www.czzy89.com"), - ("var mysvg='https://media.example/final.m3u8';", "https://www.czzy89.com"), - ] - - result = self.spider.playerContent("厂长资源", "https://www.czzy89.com/v_play/1.html", {}) - self.assertEqual(result["parse"], 0) - self.assertEqual(result["url"], "https://media.example/final.m3u8") - self.assertIn("Referer", result["header"]) - - -if __name__ == "__main__": - unittest.main()