From c0b51be9dc1235b1336e6b5d4c49378ba76d8f9a Mon Sep 17 00:00:00 2001 From: Harold <8866033@gmail.com> Date: Mon, 20 Apr 2026 15:00:35 +0800 Subject: [PATCH] feat: add zhizhen spider and aggregate support --- .../2026-04-20-wanou-aggregate-zhizhen.md | 289 +++++++++ .../plans/2026-04-20-zhizhen-spider.md | 571 ++++++++++++++++++ ...26-04-20-wanou-aggregate-zhizhen-design.md | 170 ++++++ .../specs/2026-04-20-zhizhen-spider-design.md | 291 +++++++++ py/tests/test_玩偶聚合.py | 99 +++ py/tests/test_至臻.py | 199 ++++++ py/玩偶聚合.py | 20 + py/至臻.py | 252 ++++++++ 8 files changed, 1891 insertions(+) create mode 100644 py/docs/superpowers/plans/2026-04-20-wanou-aggregate-zhizhen.md create mode 100644 py/docs/superpowers/plans/2026-04-20-zhizhen-spider.md create mode 100644 py/docs/superpowers/specs/2026-04-20-wanou-aggregate-zhizhen-design.md create mode 100644 py/docs/superpowers/specs/2026-04-20-zhizhen-spider-design.md create mode 100644 py/tests/test_至臻.py create mode 100644 py/至臻.py diff --git a/py/docs/superpowers/plans/2026-04-20-wanou-aggregate-zhizhen.md b/py/docs/superpowers/plans/2026-04-20-wanou-aggregate-zhizhen.md new file mode 100644 index 0000000..450b206 --- /dev/null +++ b/py/docs/superpowers/plans/2026-04-20-wanou-aggregate-zhizhen.md @@ -0,0 +1,289 @@ +# 玩偶聚合接入至臻 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:** 在现有 `玩偶聚合` 蜘蛛中正式接入 `至臻` 站点,使其参与首页展示、分类访问、搜索聚合和详情网盘线路合并。 + +**Architecture:** 继续沿用 `py/玩偶聚合.py` 的配置驱动结构,不新增共享基类,只补 `zhizhen` 站点配置并扩充测试覆盖。实现顺序遵循 TDD:先写失败测试锁定首页、搜索 URL 和详情合并行为,再做最小配置改动直至 `tests.test_玩偶聚合` 转绿。 + +**Tech Stack:** Python 3, `unittest`, `unittest.mock`, `base.spider.Spider`, `re`, `json`, `base64` + +--- + +## File Structure + +- Modify: `py/玩偶聚合.py` + - 在 `self.sites` 中新增 `zhizhen` 站点配置 + - 复用现有聚合逻辑,不改编码格式和公共 helper 接口 +- Modify: `py/tests/test_玩偶聚合.py` + - 新增 `site_zhizhen` 首页暴露测试 + - 新增 `zhizhen` 搜索 URL 与结果编码测试 + - 新增聚合详情合并 `至臻` 网盘线路测试 + +### Task 1: Lock HomeContent And Search Expectations For 至臻 + +**Files:** +- Modify: `py/tests/test_玩偶聚合.py` +- Modify: `py/玩偶聚合.py` + +- [ ] **Step 1: Write the failing test** + +```python + def test_home_content_exposes_zhizhen_site_and_categories(self): + content = self.spider.homeContent(False) + type_ids = [item["type_id"] for item in content["class"]] + self.assertIn("site_zhizhen", type_ids) + self.assertEqual( + content["filters"]["site_zhizhen"][0]["value"][1:], + [ + {"n": "电影", "v": "1"}, + {"n": "剧集", "v": "2"}, + {"n": "动漫", "v": "3"}, + {"n": "综艺", "v": "4"}, + {"n": "短剧", "v": "5"}, + {"n": "老剧", "v": "24"}, + {"n": "严选", "v": "26"}, + ], + ) + + @patch.object(Spider, "_request_with_failover") + def test_fetch_site_search_builds_zhizhen_search_url_and_parses_results(self, mock_request_with_failover): + mock_request_with_failover.return_value = """ +
+ 抢先版 +
至臻影片
+
+ """ + site = self.spider._get_site("zhizhen") + results = self.spider._fetch_site_search(site, "繁花", 1) + self.assertEqual( + mock_request_with_failover.call_args.args[1], + "/index.php/vod/search/page/1/wd/%E7%B9%81%E8%8A%B1.html", + ) + self.assertEqual( + results[0], + { + "vod_id": "site:zhizhen:/index.php/vod/detail/id/789.html", + "vod_name": "至臻影片", + "vod_pic": "http://www.miqk.cc/search.jpg", + "vod_remarks": "", + "vod_year": "", + "_site": "zhizhen", + "_detail_path": "/index.php/vod/detail/id/789.html", + }, + ) +``` + +- [ ] **Step 2: Run test to verify it fails** + +Run from `py/`: `python -m unittest tests.test_玩偶聚合.TestWanouAggregateSpider.test_home_content_exposes_zhizhen_site_and_categories tests.test_玩偶聚合.TestWanouAggregateSpider.test_fetch_site_search_builds_zhizhen_search_url_and_parses_results -v` +Expected: FAIL because `site_zhizhen` is absent and `_get_site("zhizhen")` returns `None`. + +- [ ] **Step 3: Write minimal implementation** + +```python + { + "id": "zhizhen", + "name": "至臻", + "domains": ["http://www.miqk.cc"], + "filter_files": [], + "list_xpath": "//*[contains(@class,'module-item')]", + "search_xpath": "//*[contains(@class,'module-search-item')]", + "detail_pan_xpath": "//*[contains(@class,'module-row-info')]//p", + "category_url": "/index.php/vod/show/id/{categoryId}/page/{page}.html", + "search_url": "/index.php/vod/search/page/{page}/wd/{keyword}.html", + "default_categories": [ + ("1", "电影"), + ("2", "剧集"), + ("3", "动漫"), + ("4", "综艺"), + ("5", "短剧"), + ("24", "老剧"), + ("26", "严选"), + ], + }, +``` + +- [ ] **Step 4: Run test to verify it passes** + +Run from `py/`: `python -m unittest tests.test_玩偶聚合.TestWanouAggregateSpider.test_home_content_exposes_zhizhen_site_and_categories tests.test_玩偶聚合.TestWanouAggregateSpider.test_fetch_site_search_builds_zhizhen_search_url_and_parses_results -v` +Expected: PASS with `site_zhizhen` visible and search URL/path matching the `miqk` rule. + +- [ ] **Step 5: Commit** + +```bash +git add py/玩偶聚合.py py/tests/test_玩偶聚合.py +git commit -m "feat: add zhizhen aggregate site config" +``` + +### Task 2: Lock Detail Merge Behavior For 至臻 Lines + +**Files:** +- Modify: `py/tests/test_玩偶聚合.py` +- Modify: `py/玩偶聚合.py` + +- [ ] **Step 1: Write the failing test** + +```python + @patch.object(Spider, "_fetch_site_detail") + def test_detail_content_for_aggregate_id_merges_zhizhen_pan_lines(self, mock_fetch_site_detail): + mock_fetch_site_detail.side_effect = [ + { + "vod_name": "繁花", + "vod_pic": "https://img.example/w.jpg", + "vod_year": "2024", + "vod_director": "导演甲", + "vod_actor": "演员甲", + "vod_content": "玩偶简介", + "pan_urls": ["https://pan.baidu.com/s/b1"], + "_site_name": "玩偶", + }, + { + "vod_name": "繁花", + "vod_pic": "http://www.miqk.cc/poster.jpg", + "vod_year": "2024", + "vod_director": "导演乙", + "vod_actor": "演员乙", + "vod_content": "至臻简介", + "pan_urls": ["https://pan.quark.cn/s/z1", "https://pan.baidu.com/s/b1"], + "_site_name": "至臻", + }, + ] + payload = [ + {"site": "wanou", "path": "/voddetail/1.html", "name": "繁花", "year": "2024"}, + {"site": "zhizhen", "path": "/index.php/vod/detail/id/2.html", "name": "繁花", "year": "2024"}, + ] + result = self.spider.detailContent([self.spider._encode_aggregate_vod_id(payload)]) + vod = result["list"][0] + self.assertEqual(vod["vod_name"], "繁花") + self.assertEqual(vod["vod_play_from"], "baidu#玩偶$$$quark#至臻") + self.assertEqual( + vod["vod_play_url"], + "百度资源$https://pan.baidu.com/s/b1$$$夸克资源$https://pan.quark.cn/s/z1", + ) +``` + +- [ ] **Step 2: Run test to verify it fails** + +Run from `py/`: `python -m unittest tests.test_玩偶聚合.TestWanouAggregateSpider.test_detail_content_for_aggregate_id_merges_zhizhen_pan_lines -v` +Expected: FAIL before Task 1 is implemented because `zhizhen` site lookup is missing or detail merge path cannot resolve it. + +- [ ] **Step 3: Write minimal implementation** + +```python + self.sites = [ + # existing wanou / muou / labi configs... + { + "id": "zhizhen", + "name": "至臻", + "domains": ["http://www.miqk.cc"], + "filter_files": [], + "list_xpath": "//*[contains(@class,'module-item')]", + "search_xpath": "//*[contains(@class,'module-search-item')]", + "detail_pan_xpath": "//*[contains(@class,'module-row-info')]//p", + "category_url": "/index.php/vod/show/id/{categoryId}/page/{page}.html", + "search_url": "/index.php/vod/search/page/{page}/wd/{keyword}.html", + "default_categories": [ + ("1", "电影"), + ("2", "剧集"), + ("3", "动漫"), + ("4", "综艺"), + ("5", "短剧"), + ("24", "老剧"), + ("26", "严选"), + ], + }, + ] +``` + +这一步不需要再改 `detailContent` 逻辑,只要确保 `zhizhen` 站点能被 `_get_site` 和 `_fetch_site_detail` 走通即可。 + +- [ ] **Step 4: Run test to verify it passes** + +Run from `py/`: `python -m unittest tests.test_玩偶聚合.TestWanouAggregateSpider.test_detail_content_for_aggregate_id_merges_zhizhen_pan_lines -v` +Expected: PASS and duplicate百度链接被去重,只保留 `至臻` 的夸克线路增量。 + +- [ ] **Step 5: Commit** + +```bash +git add py/玩偶聚合.py py/tests/test_玩偶聚合.py +git commit -m "test: cover zhizhen aggregate detail merge" +``` + +### Task 3: Run Full 玩偶聚合 Verification + +**Files:** +- Modify: `py/tests/test_玩偶聚合.py` +- Modify: `py/玩偶聚合.py` + +- [ ] **Step 1: Write the failing test** + +```python + @patch.object(Spider, "_request_with_failover") + def test_category_content_builds_zhizhen_category_url(self, mock_request_with_failover): + mock_request_with_failover.return_value = """ +
+
+ + 至臻分类片 +
+
HD
+
+ """ + result = self.spider.categoryContent("site_zhizhen", "2", False, {"categoryId": "24"}) + self.assertEqual( + mock_request_with_failover.call_args.args[1], + "http://www.miqk.cc/index.php/vod/show/id/24/page/2.html", + ) + self.assertEqual(result["list"][0]["vod_id"], "site:zhizhen:/index.php/vod/detail/id/456.html") +``` + +- [ ] **Step 2: Run test to verify it fails** + +Run from `py/`: `python -m unittest tests.test_玩偶聚合.TestWanouAggregateSpider.test_category_content_builds_zhizhen_category_url -v` +Expected: FAIL before the new config is in place because `site_zhizhen` cannot resolve. + +- [ ] **Step 3: Write minimal implementation** + +```python + { + "id": "zhizhen", + "name": "至臻", + "domains": ["http://www.miqk.cc"], + "filter_files": [], + "list_xpath": "//*[contains(@class,'module-item')]", + "search_xpath": "//*[contains(@class,'module-search-item')]", + "detail_pan_xpath": "//*[contains(@class,'module-row-info')]//p", + "category_url": "/index.php/vod/show/id/{categoryId}/page/{page}.html", + "search_url": "/index.php/vod/search/page/{page}/wd/{keyword}.html", + "default_categories": [ + ("1", "电影"), + ("2", "剧集"), + ("3", "动漫"), + ("4", "综艺"), + ("5", "短剧"), + ("24", "老剧"), + ("26", "严选"), + ], + }, +``` + +这一步依旧是复用 Task 1 的配置,目标是用完整模块测试证明没有遗漏分类 URL 规则。 + +- [ ] **Step 4: Run test to verify it passes** + +Run from `py/`: `python -m unittest tests.test_玩偶聚合.TestWanouAggregateSpider.test_category_content_builds_zhizhen_category_url -v` +Expected: PASS and category URL 使用 `/index.php/vod/show/id//page/.html`。 + +- [ ] **Step 5: Commit** + +```bash +git add py/玩偶聚合.py py/tests/test_玩偶聚合.py +git commit -m "test: verify zhizhen aggregate category url" +``` + +## Self-Review + +- Spec coverage: 首页暴露、搜索 URL、分类 URL、详情合并都对应了独立任务,没有遗漏 `至臻` 在聚合层的关键入口。 +- Placeholder scan: 已给出精确文件路径、测试名、命令、站点配置代码和期望输出,没有保留 TBD 或泛化描述。 +- Type consistency: 全程使用 `site_zhizhen`、`site:zhizhen:`、`zhizhen` 站点 ID 和 `http://www.miqk.cc` 域名,命名保持一致。 diff --git a/py/docs/superpowers/plans/2026-04-20-zhizhen-spider.md b/py/docs/superpowers/plans/2026-04-20-zhizhen-spider.md new file mode 100644 index 0000000..47cb02f --- /dev/null +++ b/py/docs/superpowers/plans/2026-04-20-zhizhen-spider.md @@ -0,0 +1,571 @@ +# 至臻 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 仓库中新增一个独立单站 `至臻` 蜘蛛,支持固定分类、分类列表、搜索、详情页网盘线路整理和网盘分享链接透传。 + +**Architecture:** 采用单文件站点脚本 `py/至臻.py` 承担全部站点逻辑,内部拆分为 URL 组装、文本清洗、请求封装、卡片解析、详情字段提取、网盘类型识别和线路拼接几个 helper。测试沿用现有 `unittest + SourceFileLoader + mock` 风格,先写失败测试锁定分类、解析和透传行为,再实现最小代码直到模块测试转绿。 + +**Tech Stack:** Python 3, `unittest`, `unittest.mock`, `re`, `sys`, `urllib.parse`, `base.spider.Spider` + +--- + +## File Structure + +- Create: `py/至臻.py` + - 实现 `Spider` 类和站点全部逻辑 + - 暴露 `init`、`getName`、`homeContent`、`homeVideoContent`、`categoryContent`、`searchContent`、`detailContent`、`playerContent` + - 私有方法负责 URL 拼装、文本清洗、请求、卡片解析、详情字段提取、网盘识别和线路组装 +- Create: `py/tests/test_至臻.py` + - 使用 `SourceFileLoader` 加载 `py/至臻.py` + - 通过内联 HTML 与 `mock` 覆盖分类、搜索、详情、线路排序和透传 + +### Task 1: Scaffold Spider And Pan Detection + +**Files:** +- Create: `py/tests/test_至臻.py` +- Create: `py/至臻.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("zhizhen_spider", str(ROOT / "至臻.py")).load_module() +Spider = MODULE.Spider + + +class TestZhiZhenSpider(unittest.TestCase): + def setUp(self): + Spider._instance = None + self.spider = Spider() + self.spider.init() + + def test_home_content_exposes_all_categories(self): + content = self.spider.homeContent(False) + self.assertEqual( + [(item["type_id"], item["type_name"]) for item in content["class"]], + [ + ("1", "至臻电影"), + ("2", "至臻剧集"), + ("3", "至臻动漫"), + ("4", "至臻综艺"), + ("5", "至臻短剧"), + ("24", "至臻老剧"), + ("26", "至臻严选"), + ], + ) + + def test_home_video_content_returns_empty_list(self): + self.assertEqual(self.spider.homeVideoContent(), {"list": []}) + + def test_build_url_and_detect_pan_type(self): + self.assertEqual( + self.spider._build_url("/index.php/vod/detail/id/1.html"), + "http://www.miqk.cc/index.php/vod/detail/id/1.html", + ) + self.assertEqual( + self.spider._detect_pan_type("https://pan.baidu.com/s/demo"), + ("baidu", "百度资源"), + ) + self.assertEqual( + self.spider._detect_pan_type("https://example.com/video"), + ("", ""), + ) +``` + +- [ ] **Step 2: Run test to verify it fails** + +Run from `py/`: `python -m unittest tests.test_至臻.TestZhiZhenSpider -v` +Expected: FAIL with `FileNotFoundError` for `至臻.py` or missing `Spider` attributes. + +- [ ] **Step 3: Write minimal implementation** + +```python +# coding=utf-8 +import re +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 = "http://www.miqk.cc" + self.headers = { + "User-Agent": ( + "Mozilla/5.0 (Windows NT 10.0; Win64; x64) " + "AppleWebKit/537.36 (KHTML, like Gecko) " + "Chrome/136.0.0.0 Safari/537.36" + ), + "Referer": self.host + "/", + } + self.categories = [ + {"type_id": "1", "type_name": "至臻电影"}, + {"type_id": "2", "type_name": "至臻剧集"}, + {"type_id": "3", "type_name": "至臻动漫"}, + {"type_id": "4", "type_name": "至臻综艺"}, + {"type_id": "5", "type_name": "至臻短剧"}, + {"type_id": "24", "type_name": "至臻老剧"}, + {"type_id": "26", "type_name": "至臻严选"}, + ] + self.pan_patterns = [ + ("baidu", "百度资源", r"pan\.baidu\.com|yun\.baidu\.com"), + ("a139", "139资源", r"yun\.139\.com"), + ("a189", "天翼资源", r"cloud\.189\.cn"), + ("a123", "123资源", r"123684\.com|123865\.com|123912\.com|123pan\.com"), + ("a115", "115资源", r"115\.com"), + ("quark", "夸克资源", r"pan\.quark\.cn"), + ("xunlei", "迅雷资源", r"pan\.xunlei\.com"), + ("aliyun", "阿里资源", r"aliyundrive\.com|alipan\.com"), + ("uc", "UC资源", r"drive\.uc\.cn"), + ] + + 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, path): + return urljoin(self.host + "/", str(path or "").strip()) + + def _detect_pan_type(self, url): + raw = str(url or "").strip() + for pan_type, title, pattern in self.pan_patterns: + if re.search(pattern, raw, re.I): + return pan_type, title + return "", "" +``` + +- [ ] **Step 4: Run test to verify it passes** + +Run from `py/`: `python -m unittest tests.test_至臻.TestZhiZhenSpider -v` +Expected: PASS for the scaffold tests. + +- [ ] **Step 5: Commit** + +```bash +git add py/tests/test_至臻.py py/至臻.py +git commit -m "feat: scaffold zhizhen spider" +``` + +### Task 2: Add Category And Search Parsing + +**Files:** +- Modify: `py/tests/test_至臻.py` +- Modify: `py/至臻.py` + +- [ ] **Step 1: Write the failing test** + +```python +from unittest.mock import patch + + +class TestZhiZhenSpider(unittest.TestCase): + def test_parse_cards_extracts_short_path_ids(self): + html = """ +
+
+
+ + 示例影片 +
+
HD
+
2025
+
+
+ """ + self.assertEqual( + self.spider._parse_cards(html), + [ + { + "vod_id": "/index.php/vod/detail/id/123.html", + "vod_name": "示例影片", + "vod_pic": "http://www.miqk.cc/poster.jpg", + "vod_remarks": "HD", + "vod_year": "2025", + } + ], + ) + + @patch.object(Spider, "_request_html") + def test_category_content_builds_reference_url_and_returns_page_payload(self, mock_request_html): + mock_request_html.return_value = """ +
+
+
+ + 分类影片 +
+
更新至10集
+
2024
+
+
+ """ + result = self.spider.categoryContent("2", "3", False, {}) + self.assertEqual( + mock_request_html.call_args.args[0], + "http://www.miqk.cc/index.php/vod/show/id/2/page/3.html", + ) + self.assertEqual(result["page"], 3) + self.assertEqual(result["limit"], 1) + self.assertEqual(result["list"][0]["vod_name"], "分类影片") + self.assertNotIn("pagecount", result) + + @patch.object(Spider, "_request_html") + def test_search_content_builds_reference_search_url_and_parses_results(self, mock_request_html): + mock_request_html.return_value = """ +
+ 抢先版 +
+ 搜索影片 +
+
+ """ + result = self.spider.searchContent("繁花", False, "2") + self.assertEqual( + mock_request_html.call_args.args[0], + "http://www.miqk.cc/index.php/vod/search/page/2/wd/%E7%B9%81%E8%8A%B1.html", + ) + self.assertEqual( + result["list"][0], + { + "vod_id": "/index.php/vod/detail/id/789.html", + "vod_name": "搜索影片", + "vod_pic": "http://www.miqk.cc/search.jpg", + "vod_remarks": "抢先版", + }, + ) +``` + +- [ ] **Step 2: Run test to verify it fails** + +Run from `py/`: `python -m unittest tests.test_至臻.TestZhiZhenSpider -v` +Expected: FAIL with missing `_parse_cards` or `categoryContent`/`searchContent`. + +- [ ] **Step 3: Write minimal implementation** + +```python +from urllib.parse import quote + + + 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) + if response.status_code != 200: + return "" + return response.text or "" + + def _page_result(self, items, pg): + page = int(pg) + return {"page": page, "limit": len(items), "total": page * 20 + len(items), "list": items} + + def _parse_cards(self, html): + root = self.html(html) + if root is None: + return [] + items = [] + for node in root.xpath("//*[@id='main']//*[contains(@class,'module-item')]"): + href = "".join(node.xpath(".//*[contains(@class,'module-item-pic')]//a[1]/@href")).strip() + title = "".join(node.xpath(".//*[contains(@class,'module-item-pic')]//img[1]/@alt")).strip() + pic = ( + "".join(node.xpath(".//*[contains(@class,'module-item-pic')]//img[1]/@data-src")).strip() + or "".join(node.xpath(".//*[contains(@class,'module-item-pic')]//img[1]/@src")).strip() + ) + remarks = self._clean_text("".join(node.xpath(".//*[contains(@class,'module-item-text')][1]//text()"))) + year = self._clean_text("".join(node.xpath(".//*[contains(@class,'module-item-caption')][1]//span[1]//text()"))) + if not href or not title: + continue + items.append( + { + "vod_id": href, + "vod_name": title, + "vod_pic": self._build_url(pic), + "vod_remarks": remarks, + "vod_year": year, + } + ) + return items + + def categoryContent(self, tid, pg, filter, extend): + url = self._build_url(f"/index.php/vod/show/id/{tid}/page/{int(pg)}.html") + return self._page_result(self._parse_cards(self._request_html(url)), pg) + + def searchContent(self, key, quick, pg="1"): + keyword = self._clean_text(key) + page = int(pg) + if not keyword: + return {"page": page, "total": 0, "list": []} + url = self._build_url(f"/index.php/vod/search/page/{page}/wd/{quote(keyword)}.html") + root = self.html(self._request_html(url)) + if root is None: + return {"page": page, "total": 0, "list": []} + items = [] + for node in root.xpath("//*[contains(@class,'module-search-item')]"): + href = "".join(node.xpath(".//*[contains(@class,'video-serial')][1]/@href")).strip() + title = "".join(node.xpath(".//*[contains(@class,'video-serial')][1]/@title")).strip() + pic = ( + "".join(node.xpath(".//*[contains(@class,'module-item-pic')]//img[1]/@data-src")).strip() + or "".join(node.xpath(".//*[contains(@class,'module-item-pic')]//img[1]/@src")).strip() + ) + remarks = self._clean_text( + "".join(node.xpath(".//*[contains(@class,'video-serial')][1]//text()")) + or "".join(node.xpath(".//*[contains(@class,'module-item-text')][1]//text()")) + ) + if not href or not title: + continue + items.append( + { + "vod_id": href, + "vod_name": title, + "vod_pic": self._build_url(pic), + "vod_remarks": remarks, + } + ) + return {"page": page, "total": len(items), "list": items} +``` + +- [ ] **Step 4: Run test to verify it passes** + +Run from `py/`: `python -m unittest tests.test_至臻.TestZhiZhenSpider -v` +Expected: PASS for the list and search tests. + +- [ ] **Step 5: Commit** + +```bash +git add py/tests/test_至臻.py py/至臻.py +git commit -m "feat: add zhizhen list and search parsing" +``` + +### Task 3: Add Detail Parsing, Pan Line Building, And Player Passthrough + +**Files:** +- Modify: `py/tests/test_至臻.py` +- Modify: `py/至臻.py` + +- [ ] **Step 1: Write the failing test** + +```python +class TestZhiZhenSpider(unittest.TestCase): + def test_build_pan_lines_deduplicates_and_sorts_supported_links(self): + detail = { + "pan_urls": [ + "https://pan.quark.cn/s/q1", + "https://pan.baidu.com/s/b1", + "https://pan.baidu.com/s/b1", + "https://example.com/ignored", + ] + } + self.assertEqual( + self.spider._build_pan_lines(detail), + [ + ("baidu#至臻", "百度资源$https://pan.baidu.com/s/b1"), + ("quark#至臻", "夸克资源$https://pan.quark.cn/s/q1"), + ], + ) + + def test_parse_detail_page_extracts_meta_content_and_pan_urls(self): + html = """ +
示例剧
+
+
年代
2024
+
导演
导演甲
+
主演
演员甲演员乙
+
剧情

一段剧情简介

+
+

https://pan.quark.cn/s/q1

+

https://pan.baidu.com/s/b1

+
+ """ + detail = self.spider._parse_detail_page("/index.php/vod/detail/id/123.html", html) + self.assertEqual(detail["vod_name"], "示例剧") + self.assertEqual(detail["vod_pic"], "http://www.miqk.cc/poster.jpg") + self.assertEqual(detail["vod_year"], "2024") + self.assertEqual(detail["vod_director"], "导演甲") + self.assertEqual(detail["vod_actor"], "演员甲,演员乙") + self.assertEqual(detail["vod_content"], "一段剧情简介") + self.assertEqual(detail["pan_urls"], ["https://pan.quark.cn/s/q1", "https://pan.baidu.com/s/b1"]) + + @patch.object(Spider, "_request_html") + def test_detail_content_builds_pan_play_fields(self, mock_request_html): + mock_request_html.return_value = """ +
示例剧
+
+
年代
2024
+
导演
导演甲
+
主演
演员甲
+
剧情

一段剧情简介

+
+

https://pan.quark.cn/s/q1

+

https://pan.baidu.com/s/b1

+
+ """ + result = self.spider.detailContent(["/index.php/vod/detail/id/123.html"]) + vod = result["list"][0] + self.assertEqual(vod["vod_name"], "示例剧") + self.assertEqual(vod["vod_play_from"], "baidu#至臻$$$quark#至臻") + self.assertEqual( + vod["vod_play_url"], + "百度资源$https://pan.baidu.com/s/b1$$$夸克资源$https://pan.quark.cn/s/q1", + ) + + def test_player_content_passthroughs_supported_pan_urls(self): + self.assertEqual( + self.spider.playerContent("baidu#至臻", "https://pan.baidu.com/s/demo", {}), + {"parse": 0, "playUrl": "", "url": "https://pan.baidu.com/s/demo"}, + ) + + def test_player_content_rejects_non_pan_url(self): + self.assertEqual( + self.spider.playerContent("site", "/index.php/vod/play/id/1.html", {}), + {"parse": 0, "playUrl": "", "url": ""}, + ) +``` + +- [ ] **Step 2: Run test to verify it fails** + +Run from `py/`: `python -m unittest tests.test_至臻.TestZhiZhenSpider -v` +Expected: FAIL with missing detail parser, line builder, or player logic. + +- [ ] **Step 3: Write minimal implementation** + +```python + self.pan_priority = { + "baidu": 1, + "a139": 2, + "a189": 3, + "a123": 4, + "a115": 5, + "quark": 6, + "xunlei": 7, + "aliyun": 8, + "uc": 9, + } + + def _parse_detail_page(self, vod_id, html): + root = self.html(html) + if root is None: + return { + "vod_id": vod_id, + "vod_name": "", + "vod_pic": "", + "vod_year": "", + "vod_director": "", + "vod_actor": "", + "vod_content": "", + "pan_urls": [], + } + detail = { + "vod_id": vod_id, + "vod_name": self._clean_text("".join(root.xpath("//*[contains(@class,'page-title')][1]//text()"))), + "vod_pic": self._build_url( + "".join( + root.xpath( + "//*[contains(@class,'mobile-play')]//*[contains(@class,'lazyload')][1]/@data-src | " + "//*[contains(@class,'mobile-play')]//*[contains(@class,'lazyload')][1]/@src" + ) + ).strip() + ), + "vod_year": "", + "vod_director": "", + "vod_actor": "", + "vod_content": "", + "pan_urls": [], + } + for label_node in root.xpath("//*[contains(@class,'video-info-itemtitle')]"): + key = self._clean_text("".join(label_node.xpath(".//text()"))) + sibling = label_node.getnext() + if sibling is None: + continue + values = [self._clean_text(text) for text in sibling.xpath(".//a//text()")] + joined = ",".join([value for value in values if value]) + text_value = self._clean_text("".join(sibling.xpath(".//text()"))) + if "年代" in key: + detail["vod_year"] = joined or text_value + elif "导演" in key: + detail["vod_director"] = joined or text_value + elif "主演" in key: + detail["vod_actor"] = joined or text_value + elif "剧情" in key: + detail["vod_content"] = text_value + for node in root.xpath("//*[contains(@class,'module-row-info')]//p"): + text = self._clean_text("".join(node.xpath(".//text()"))) + if text: + detail["pan_urls"].append(text) + return detail + + def _build_pan_lines(self, detail): + lines = [] + seen = set() + for url in detail.get("pan_urls", []): + pan_type, title = self._detect_pan_type(url) + if not pan_type or url in seen: + continue + seen.add(url) + lines.append((self.pan_priority.get(pan_type, 999), f"{pan_type}#至臻", f"{title}${url}")) + lines.sort(key=lambda item: item[0]) + return [(item[1], item[2]) for item in lines] + + def detailContent(self, ids): + result = {"list": []} + for raw_id in ids: + vod_id = str(raw_id or "").strip() + detail = self._parse_detail_page(vod_id, self._request_html(self._build_url(vod_id))) + lines = self._build_pan_lines(detail) + result["list"].append( + { + "vod_id": vod_id, + "vod_name": detail["vod_name"], + "vod_pic": detail["vod_pic"], + "vod_year": detail["vod_year"], + "vod_director": detail["vod_director"], + "vod_actor": detail["vod_actor"], + "vod_content": detail["vod_content"], + "vod_play_from": "$$$".join([item[0] for item in lines]), + "vod_play_url": "$$$".join([item[1] for item in lines]), + } + ) + return result + + def playerContent(self, flag, id, vipFlags): + pan_type, _ = self._detect_pan_type(id) + if pan_type: + return {"parse": 0, "playUrl": "", "url": id} + return {"parse": 0, "playUrl": "", "url": ""} +``` + +- [ ] **Step 4: Run test to verify it passes** + +Run from `py/`: `python -m unittest tests.test_至臻.TestZhiZhenSpider -v` +Expected: PASS for the detail and player tests. + +- [ ] **Step 5: Commit** + +```bash +git add py/tests/test_至臻.py py/至臻.py +git commit -m "feat: add zhizhen detail and player support" +``` + +## Self-Review + +- Spec coverage: 覆盖了固定分类、列表、搜索、详情、网盘线路和播放透传,没有遗漏公共层改动。 +- Placeholder scan: 计划中的文件路径、测试名、命令和关键代码均已具体化,没有保留 TBD 或“类似前一项”。 +- Type consistency: `vod_id` 使用短路径,`vod_play_from` / `vod_play_url`、`_detect_pan_type`、`_build_pan_lines` 的命名在各任务中保持一致。 diff --git a/py/docs/superpowers/specs/2026-04-20-wanou-aggregate-zhizhen-design.md b/py/docs/superpowers/specs/2026-04-20-wanou-aggregate-zhizhen-design.md new file mode 100644 index 0000000..e4c324b --- /dev/null +++ b/py/docs/superpowers/specs/2026-04-20-wanou-aggregate-zhizhen-design.md @@ -0,0 +1,170 @@ +# 玩偶聚合接入至臻设计 + +**日期:** 2026-04-20 + +## 目标 + +在现有聚合蜘蛛 [玩偶聚合.py](/home/harold/workspace/tvbox-resources/py/玩偶聚合.py) 中正式接入 `至臻` 站点,使其参与: + +- 首页站点列表展示 +- 站点分类页抓取 +- 聚合搜索 +- 聚合详情页网盘线路合并 + +主域名固定为: + +- `http://www.miqk.cc` + +本次接入沿用已经确认的单站 `至臻` 解析边界: + +- 只抓列表、搜索、详情元数据和网盘链接 +- `playerContent` 只透传网盘分享链接 +- 不做站内直链播放解析 + +## 范围 + +本次改动包含: + +- 在 [玩偶聚合.py](/home/harold/workspace/tvbox-resources/py/玩偶聚合.py) 的 `self.sites` 中增加 `zhizhen` 配置 +- 让 `homeContent` 暴露 `site_zhizhen` +- 让 `categoryContent` 使用 `至臻` 的分类和 URL 模板 +- 让 `searchContent` 将 `至臻` 结果纳入聚合 +- 让 `detailContent` 能合并 `至臻` 的网盘链接 +- 在 [test_玩偶聚合.py](/home/harold/workspace/tvbox-resources/py/tests/test_玩偶聚合.py) 中补充对应测试 + +本次不包含: + +- 修改聚合 ID 编码格式 +- 抽取新的共享基类 +- 为 `至臻` 增加备用域名 +- 修改单站 [至臻.py](/home/harold/workspace/tvbox-resources/py/至臻.py) 的行为 + +## 现状 + +当前聚合蜘蛛已有: + +- `site_priority` 中的 `zhizhen` 优先级占位 +- 通用的列表、搜索、详情和网盘线路组装逻辑 +- 站点配置驱动的 URL 构建和 XPath 解析 + +当前缺口是: + +- `self.sites` 里还没有真正的 `zhizhen` 配置 +- 测试中也没有覆盖 `至臻` 在聚合层的展示、搜索和详情合并 + +这意味着聚合层逻辑本身基本够用,本次重点是补站点定义并用测试锁定行为。 + +## 方案选择 + +采用“最小配置接入 + 现有聚合逻辑复用”的方案。 + +具体做法: + +- 新增一条 `zhizhen` 站点配置 +- 复用现有 `module-item`、`module-search-item`、`module-row-info` 解析流程 +- 复用现有 `_fetch_site_search`、`_parse_detail_page`、`_build_pan_lines`、`detailContent` 合并逻辑 + +不采用“让聚合层调用单站 `至臻.py`”的方案,原因是: + +- 当前聚合蜘蛛的结构是配置驱动,不是子 Spider 组合 +- 强行调用单站 Spider 会把接口耦合变复杂 +- 本次站点结构与现有聚合模板兼容,没有必要新增调度层 + +## 站点配置设计 + +新增站点项字段: + +- `id`: `zhizhen` +- `name`: `至臻` +- `domains`: `["http://www.miqk.cc"]` +- `filter_files`: `[]` +- `list_xpath`: `//*[contains(@class,'module-item')]` +- `search_xpath`: `//*[contains(@class,'module-search-item')]` +- `detail_pan_xpath`: `//*[contains(@class,'module-row-info')]//p` +- `category_url`: `/index.php/vod/show/id/{categoryId}/page/{page}.html` +- `search_url`: `/index.php/vod/search/page/{page}/wd/{keyword}.html` +- `default_categories`: `[("1","电影"),("2","剧集"),("3","动漫"),("4","综艺"),("5","短剧"),("24","老剧"),("26","严选")]` + +这里不配置 `category_url_with_filters`,因为用户给出的 `至臻` 参考实现只确认了基础分类翻页 URL,没有额外筛选规则。 + +## 行为设计 + +### 首页 + +`homeContent` 应新增一项: + +- `type_id = site_zhizhen` +- `type_name = 至臻` + +`filters["site_zhizhen"]` 中的第一个筛选组应为 `categoryId`,值列表映射到 `至臻` 的 7 个默认分类。 + +### 分类页 + +`categoryContent("site_zhizhen", pg, ..., extend)` 应: + +- 从 `extend["categoryId"]` 读取分类 +- 若未提供,则默认使用 `1` +- 构建 `http://www.miqk.cc/index.php/vod/show/id//page/.html` +- 按现有通用列表解析逻辑输出站内条目 + +返回结构保持与现有聚合站一致: + +- `vod_id` 使用 `site:zhizhen:` +- 保留 `_site` 和 `_detail_path` +- 不返回 `pagecount` + +### 搜索 + +`searchContent` 不改接口,只需要确保: + +- `_fetch_site_search` 能对 `zhizhen` 使用其 `search_url` +- 结果进入 `_aggregate_search_results` +- 若同名同年命中多个站点,仍按 `site_priority` 决定主信息来源 + +因为 `site_priority` 中 `zhizhen` 已排在 `labi` 后、`erxiao` 前,本次不调整站点优先级。 + +### 详情 + +聚合详情和单站详情都不改数据模型,只要 `zhizhen` 配置加入后能被现有流程消费。 + +重点保证: + +- `_fetch_site_detail` 能按 `site["domains"][0] + path` 请求 `至臻` 详情页 +- `_parse_detail_page` 可用通用 `.page-title` / `.mobile-play` / `.video-info-itemtitle` / `detail_pan_xpath` 提取字段 +- `detailContent` 合并 `至臻` 的网盘线路时遵守现有去重和排序规则 + +## 测试设计 + +本次至少新增以下测试: + +1. 首页暴露 `site_zhizhen` + - 校验 `homeContent(False)["class"]` 中包含 `site_zhizhen` + - 校验 `filters["site_zhizhen"]` 的分类项包含 `1/2/3/4/5/24/26` + +2. `至臻` 搜索 URL 构造与解析 + - 构造一个 `zhizhen` 站点配置 + - mock `_request_with_failover` + - 断言 `_fetch_site_search` 请求 `http://www.miqk.cc/index.php/vod/search/page/1/wd/.html` + - 断言结果被编码为 `site:zhizhen:` + +3. 聚合详情合并 `至臻` 网盘线路 + - mock `_fetch_site_detail`,让一个聚合 payload 同时包含例如 `wanou` 和 `zhizhen` + - 断言 `vod_play_from` / `vod_play_url` 中含 `#至臻` 的线路 + - 断言重复链接仍会按现有规则去重 + +必要时再补一个分类页测试,锁定 `site_zhizhen` 的分类 URL 模板。 + +## 风险 + +- `至臻` 的分类 URL 与现有 `/vodshow/...` 模板不同,若直接沿用旧模板会导致站点在聚合层无法访问分类页 +- `default_categories` 若错误复用其他站的 `29` 或 `21` 等 ID,会让筛选项和真实站点不一致 +- 若测试只验证首页展示、不验证搜索和详情,则很容易出现“站点名显示了,但实际不可用”的假集成 + +## 验收标准 + +满足以下条件即可认为完成: + +- [玩偶聚合.py](/home/harold/workspace/tvbox-resources/py/玩偶聚合.py) 的 `self.sites` 中新增 `zhizhen` 配置 +- [test_玩偶聚合.py](/home/harold/workspace/tvbox-resources/py/tests/test_玩偶聚合.py) 覆盖 `至臻` 的首页、搜索或详情至少三类行为 +- `python -m unittest tests.test_玩偶聚合 -v` 通过 +- 聚合层返回结构不引入新的字段格式变化 diff --git a/py/docs/superpowers/specs/2026-04-20-zhizhen-spider-design.md b/py/docs/superpowers/specs/2026-04-20-zhizhen-spider-design.md new file mode 100644 index 0000000..0ebc6b2 --- /dev/null +++ b/py/docs/superpowers/specs/2026-04-20-zhizhen-spider-design.md @@ -0,0 +1,291 @@ +# 至臻 Python 爬虫设计 + +**日期:** 2026-04-20 + +## 目标 + +在当前 Python Spider 仓库中新增一个独立单站蜘蛛 `至臻.py`,参考用户提供的 JS 版本行为,实现符合 `base.spider.Spider` 接口的网盘资源站适配。 + +本次实现需要覆盖: + +- 固定 7 个分类 +- 分类列表 +- 搜索 +- 详情页元数据解析 +- 网盘链接整理 +- 播放透传 +- 对应 `unittest` + +## 范围 + +本次实现包含: + +- 新增独立蜘蛛文件 `py/至臻.py` +- 新增测试文件 `py/tests/test_至臻.py` +- 单域名站点适配:`http://www.miqk.cc` +- 固定 7 个分类,分类 ID 与参考 JS 保持一致 +- 分类页和搜索页卡片解析 +- 详情页元数据与网盘链接提取 +- 按网盘类型输出 `vod_play_from` 和 `vod_play_url` +- `playerContent` 对已识别网盘分享链接直接透传 + +本次实现不包含: + +- 聚合多站 +- 站内直链播放解析 +- 本地筛选配置文件 +- 验证码、浏览器执行或复杂反爬绕过 +- 修改 `base/` 公共层 + +## 方案选择 + +采用“单站单文件 + 少量 helper + 单测”的仓库现有模式,而不是直接保存用户提供的 JS 代码。 + +原因: + +- 用户已确认以独立单站爬虫交付 +- 当前仓库已存在多个相同结构的单文件 Spider +- 私有 helper 能把 URL 组装、文本清洗、卡片解析、详情提取和网盘识别拆开,便于后续修站 +- 解析逻辑可以通过静态 HTML 单测稳定覆盖,不依赖真实网络 + +不采用“提前抽公共盘站基类”的方案,因为本次目标是尽快落一个站点,过早抽象会扩大改动面。 + +## 接口设计 + +### `homeContent` + +返回固定 7 个分类: + +- `1 -> 至臻电影` +- `2 -> 至臻剧集` +- `3 -> 至臻动漫` +- `4 -> 至臻综艺` +- `5 -> 至臻短剧` +- `24 -> 至臻老剧` +- `26 -> 至臻严选` + +不返回筛选项。 + +### `homeVideoContent` + +返回空列表: + +- `{"list": []}` + +### `categoryContent` + +分类页 URL 规则: + +- `/index.php/vod/show/id/{tid}/page/{page}.html` + +解析 `#main .module-item` 卡片并输出: + +- `vod_id` +- `vod_name` +- `vod_pic` +- `vod_remarks` +- `vod_year` + +分页返回字段: + +- `page` +- `limit` +- `total` +- `list` + +不返回 `pagecount`。 + +### `searchContent` + +搜索 URL 规则: + +- `/index.php/vod/search/page/{page}/wd/{keyword}.html` + +空关键词直接返回空列表。 + +搜索结果结构与分类列表保持一致,但 `vod_remarks` 优先取 `.video-serial` 文本。 + +### `detailContent` + +通过详情页提取: + +- `vod_id` +- `vod_name` +- `vod_pic` +- `vod_year` +- `vod_director` +- `vod_actor` +- `vod_content` +- `vod_play_from` +- `vod_play_url` + +详情页只整理网盘分享链接,不解析站内播放器。 + +### `playerContent` + +若 `id` 是支持的网盘分享链接,则返回透传结果: + +```python +{"parse": 0, "playUrl": "", "url": id} +``` + +若不是已识别网盘链接,则返回空 URL: + +```python +{"parse": 0, "playUrl": "", "url": ""} +``` + +## 模块边界 + +新蜘蛛内部拆分为以下职责: + +- 站点配置与固定分类 +- URL 组装 +- 文本清洗 +- HTML 请求封装 +- 列表卡片解析 +- 搜索结果解析 +- 详情页字段提取 +- 网盘类型识别 +- 网盘线路拼接 + +不新增公共基类,不抽共享模块。 + +## URL 与 ID 设计 + +详情页 `vod_id` 使用站内短路径,而不是完整 URL。 + +编码方式: + +- 详情链接 `/index.php/vod/detail/id/123.html` 对外直接保存为 `/index.php/vod/detail/id/123.html` + +原因: + +- 与用户给出的 JS 行为一致 +- 当前仓库已有多个蜘蛛直接使用站内短路径作为 `vod_id` +- 单站实现不需要额外编码层 + +详情请求时再基于主域拼成完整地址。 + +## 请求策略 + +主域固定为: + +- `http://www.miqk.cc` + +请求头包含固定 `User-Agent` 和首页 `Referer`。 + +异常处理策略: + +- 页面请求失败时返回空列表或空字段结果 +- 不向上抛出未处理异常 +- 不实现多域名切换 +- 不实现重试 + +## 列表与搜索解析 + +分类列表解析容器: + +- `#main .module-item` + +提取策略: + +- 链接:`.module-item-pic a[href]` +- 标题:`.module-item-pic img[alt]` +- 封面:`.module-item-pic img[data-src|src]` +- 备注:`.module-item-text` +- 年份:`.module-item-caption span:first-child` + +搜索结果解析容器: + +- `.module-search-item` + +提取策略: + +- 链接和标题优先来自 `.video-serial` +- 封面来自 `.module-item-pic img[data-src|src]` +- 备注优先取 `.video-serial` 文本,缺失时回退 `.module-item-text` + +列表和搜索都应忽略空标题或空链接项。 + +## 详情解析 + +详情页字段来源按用户提供的 JS 保持一致: + +- 标题:`.page-title` +- 封面:`.mobile-play .lazyload[data-src|src]` +- 标注区:`.video-info-itemtitle` 与其相邻节点 +- 网盘链接:`.module-row-info p` + +字段映射规则: + +- `年代` -> `vod_year` +- `导演` -> `vod_director` +- `主演` -> `vod_actor` +- `剧情` -> `vod_content` + +其中导演、主演优先拼接相邻区域中的链接文本;剧情提取文本内容并做空白清洗。 + +## 网盘线路整理 + +支持识别以下网盘: + +- 百度 +- 139 +- 天翼 +- 123 +- 115 +- 夸克 +- 迅雷 +- 阿里 +- UC + +排序优先级: + +1. 百度 +2. 139 +3. 天翼 +4. 123 +5. 115 +6. 夸克 +7. 迅雷 +8. 阿里 +9. UC + +输出规则: + +- `vod_play_from` 使用 `{pan_type}#至臻` 线路名拼接 +- `vod_play_url` 使用 `{标题}${分享链接}` 拼接 +- 不支持的链接忽略 +- 重复链接去重 + +## 测试策略 + +采用 `unittest` 和 `unittest.mock`,不依赖真实网络。 + +至少覆盖: + +- 固定分类和空首页 +- URL 构建与网盘识别 +- 分类卡片解析 +- 分类接口 URL 拼装 +- 搜索接口 URL 拼装和结果解析 +- 详情元数据提取 +- 网盘线路去重和排序 +- `detailContent` 最终输出 +- `playerContent` 对网盘链接透传和非网盘拒绝 + +## 风险与约束 + +- 站点 DOM 如果与用户提供的 JS 片段不一致,测试需要以当前仓库约定的静态夹具为准 +- 搜索关键词直接拼接到路径中,需做 URL 编码 +- 详情页相邻节点结构若出现空白文本节点,解析时需要回退到 XPath 文本合并,避免取值为空 + +## 验收标准 + +满足以下条件即可认为完成: + +- `py/至臻.py` 实现 `Spider` 所需接口 +- `py/tests/test_至臻.py` 覆盖核心行为 +- 针对 `至臻` 模块的单测全部通过 +- 返回结构与当前仓库同类盘站蜘蛛保持一致 diff --git a/py/tests/test_玩偶聚合.py b/py/tests/test_玩偶聚合.py index 51b6536..a542d57 100644 --- a/py/tests/test_玩偶聚合.py +++ b/py/tests/test_玩偶聚合.py @@ -28,6 +28,23 @@ class TestWanouAggregateSpider(unittest.TestCase): self.assertEqual(content["filters"]["site_wanou"][0]["key"], "categoryId") self.assertEqual(content["filters"]["site_wanou"][0]["value"][1], {"n": "电影", "v": "1"}) + def test_home_content_exposes_zhizhen_site_and_categories(self): + content = self.spider.homeContent(False) + type_ids = [item["type_id"] for item in content["class"]] + self.assertIn("site_zhizhen", type_ids) + self.assertEqual( + content["filters"]["site_zhizhen"][0]["value"][1:], + [ + {"n": "电影", "v": "1"}, + {"n": "剧集", "v": "2"}, + {"n": "动漫", "v": "3"}, + {"n": "综艺", "v": "4"}, + {"n": "短剧", "v": "5"}, + {"n": "老剧", "v": "24"}, + {"n": "严选", "v": "26"}, + ], + ) + @patch.object( Spider, "_load_local_filter_groups", @@ -328,6 +345,33 @@ class TestWanouAggregateSpider(unittest.TestCase): self.assertEqual(results[0]["vod_id"], "site:wanou:/voddetail/789.html") self.assertEqual(results[0]["vod_name"], "搜索影片") + @patch.object(Spider, "_request_with_failover") + def test_fetch_site_search_builds_zhizhen_search_url_and_parses_results(self, mock_request_with_failover): + mock_request_with_failover.return_value = """ +
+ 抢先版 +
至臻影片
+
+ """ + site = self.spider._get_site("zhizhen") + results = self.spider._fetch_site_search(site, "繁花", 1) + self.assertEqual( + mock_request_with_failover.call_args.args[1], + "/index.php/vod/search/page/1/wd/%E7%B9%81%E8%8A%B1.html", + ) + self.assertEqual( + results[0], + { + "vod_id": "site:zhizhen:/index.php/vod/detail/id/789.html", + "vod_name": "至臻影片", + "vod_pic": "http://www.miqk.cc/search.jpg", + "vod_remarks": "", + "vod_year": "", + "_site": "zhizhen", + "_detail_path": "/index.php/vod/detail/id/789.html", + }, + ) + @patch.object(Spider, "_fetch_site_search") def test_search_content_skips_site_errors(self, mock_fetch_site_search): mock_fetch_site_search.side_effect = [ @@ -355,3 +399,58 @@ class TestWanouAggregateSpider(unittest.TestCase): def test_search_content_returns_empty_list_for_blank_keyword(self): self.assertEqual(self.spider.searchContent("", False, "1"), {"page": 1, "total": 0, "list": []}) + + @patch.object(Spider, "_fetch_site_detail") + def test_detail_content_for_aggregate_id_merges_zhizhen_pan_lines(self, mock_fetch_site_detail): + mock_fetch_site_detail.side_effect = [ + { + "vod_name": "繁花", + "vod_pic": "https://img.example/w.jpg", + "vod_year": "2024", + "vod_director": "导演甲", + "vod_actor": "演员甲", + "vod_content": "玩偶简介", + "pan_urls": ["https://pan.baidu.com/s/b1"], + "_site_name": "玩偶", + }, + { + "vod_name": "繁花", + "vod_pic": "http://www.miqk.cc/poster.jpg", + "vod_year": "2024", + "vod_director": "导演乙", + "vod_actor": "演员乙", + "vod_content": "至臻简介", + "pan_urls": ["https://pan.quark.cn/s/z1", "https://pan.baidu.com/s/b1"], + "_site_name": "至臻", + }, + ] + payload = [ + {"site": "wanou", "path": "/voddetail/1.html", "name": "繁花", "year": "2024"}, + {"site": "zhizhen", "path": "/index.php/vod/detail/id/2.html", "name": "繁花", "year": "2024"}, + ] + result = self.spider.detailContent([self.spider._encode_aggregate_vod_id(payload)]) + vod = result["list"][0] + self.assertEqual(vod["vod_name"], "繁花") + self.assertEqual(vod["vod_play_from"], "baidu#玩偶$$$quark#至臻") + self.assertEqual( + vod["vod_play_url"], + "百度资源$https://pan.baidu.com/s/b1$$$夸克资源$https://pan.quark.cn/s/z1", + ) + + @patch.object(Spider, "_request_with_failover") + def test_category_content_builds_zhizhen_category_url(self, mock_request_with_failover): + mock_request_with_failover.return_value = """ +
+
+ + 至臻分类片 +
+
HD
+
+ """ + result = self.spider.categoryContent("site_zhizhen", "2", False, {"categoryId": "24"}) + self.assertEqual( + mock_request_with_failover.call_args.args[1], + "http://www.miqk.cc/index.php/vod/show/id/24/page/2.html", + ) + self.assertEqual(result["list"][0]["vod_id"], "site:zhizhen:/index.php/vod/detail/id/456.html") diff --git a/py/tests/test_至臻.py b/py/tests/test_至臻.py new file mode 100644 index 0000000..09628d2 --- /dev/null +++ b/py/tests/test_至臻.py @@ -0,0 +1,199 @@ +import unittest +from importlib.machinery import SourceFileLoader +from pathlib import Path +from unittest.mock import patch + + +ROOT = Path(__file__).resolve().parents[1] +MODULE = SourceFileLoader("zhizhen_spider", str(ROOT / "至臻.py")).load_module() +Spider = MODULE.Spider + + +class TestZhiZhenSpider(unittest.TestCase): + def setUp(self): + Spider._instance = None + self.spider = Spider() + self.spider.init() + + def test_home_content_exposes_all_categories(self): + content = self.spider.homeContent(False) + self.assertEqual( + [(item["type_id"], item["type_name"]) for item in content["class"]], + [ + ("1", "至臻电影"), + ("2", "至臻剧集"), + ("3", "至臻动漫"), + ("4", "至臻综艺"), + ("5", "至臻短剧"), + ("24", "至臻老剧"), + ("26", "至臻严选"), + ], + ) + + def test_home_video_content_returns_empty_list(self): + self.assertEqual(self.spider.homeVideoContent(), {"list": []}) + + def test_build_url_and_detect_pan_type(self): + self.assertEqual( + self.spider._build_url("/index.php/vod/detail/id/1.html"), + "http://www.miqk.cc/index.php/vod/detail/id/1.html", + ) + self.assertEqual(self.spider._detect_pan_type("https://pan.baidu.com/s/demo"), ("baidu", "百度资源")) + self.assertEqual(self.spider._detect_pan_type("https://pan.quark.cn/s/demo"), ("quark", "夸克资源")) + self.assertEqual(self.spider._detect_pan_type("https://example.com/video"), ("", "")) + + def test_parse_cards_extracts_short_path_ids(self): + html = """ +
+
+
+ + 示例影片 +
+
HD
+
2025
+
+
+ """ + self.assertEqual( + self.spider._parse_cards(html), + [ + { + "vod_id": "/index.php/vod/detail/id/123.html", + "vod_name": "示例影片", + "vod_pic": "http://www.miqk.cc/poster.jpg", + "vod_remarks": "HD", + "vod_year": "2025", + } + ], + ) + + @patch.object(Spider, "_request_html") + def test_category_content_builds_reference_url_and_returns_page_payload(self, mock_request_html): + mock_request_html.return_value = """ +
+
+
+ + 分类影片 +
+
更新至10集
+
2024
+
+
+ """ + result = self.spider.categoryContent("2", "3", False, {}) + self.assertEqual( + mock_request_html.call_args.args[0], + "http://www.miqk.cc/index.php/vod/show/id/2/page/3.html", + ) + self.assertEqual(result["page"], 3) + self.assertEqual(result["limit"], 1) + self.assertEqual(result["list"][0]["vod_name"], "分类影片") + self.assertNotIn("pagecount", result) + + @patch.object(Spider, "_request_html") + def test_search_content_builds_reference_search_url_and_parses_results(self, mock_request_html): + mock_request_html.return_value = """ +
+ 抢先版 +
+ 搜索影片 +
+
+ """ + result = self.spider.searchContent("繁花", False, "2") + self.assertEqual( + mock_request_html.call_args.args[0], + "http://www.miqk.cc/index.php/vod/search/page/2/wd/%E7%B9%81%E8%8A%B1.html", + ) + self.assertEqual( + result["list"][0], + { + "vod_id": "/index.php/vod/detail/id/789.html", + "vod_name": "搜索影片", + "vod_pic": "http://www.miqk.cc/search.jpg", + "vod_remarks": "抢先版", + }, + ) + + def test_search_content_returns_empty_list_for_blank_keyword(self): + self.assertEqual(self.spider.searchContent("", False, "1"), {"page": 1, "total": 0, "list": []}) + + def test_build_pan_lines_deduplicates_and_sorts_supported_links(self): + detail = { + "pan_urls": [ + "https://pan.quark.cn/s/q1", + "https://pan.baidu.com/s/b1", + "https://pan.baidu.com/s/b1", + "https://example.com/ignored", + ] + } + self.assertEqual( + self.spider._build_pan_lines(detail), + [ + ("baidu#至臻", "百度资源$https://pan.baidu.com/s/b1"), + ("quark#至臻", "夸克资源$https://pan.quark.cn/s/q1"), + ], + ) + + def test_parse_detail_page_extracts_meta_content_and_pan_urls(self): + html = """ +
示例剧
+
+
年代
+
导演
+
主演
+
剧情

一段剧情简介

+
+

https://pan.quark.cn/s/q1

+

https://pan.baidu.com/s/b1

+
+ """ + detail = self.spider._parse_detail_page("/index.php/vod/detail/id/123.html", html) + self.assertEqual(detail["vod_name"], "示例剧") + self.assertEqual(detail["vod_pic"], "http://www.miqk.cc/poster.jpg") + self.assertEqual(detail["vod_year"], "2024") + self.assertEqual(detail["vod_director"], "导演甲") + self.assertEqual(detail["vod_actor"], "演员甲,演员乙") + self.assertEqual(detail["vod_content"], "一段剧情简介") + self.assertEqual(detail["pan_urls"], ["https://pan.quark.cn/s/q1", "https://pan.baidu.com/s/b1"]) + + @patch.object(Spider, "_request_html") + def test_detail_content_builds_pan_play_fields(self, mock_request_html): + mock_request_html.return_value = """ +
示例剧
+
+
年代
+
导演
+
主演
+
剧情

一段剧情简介

+
+

https://pan.quark.cn/s/q1

+

https://pan.baidu.com/s/b1

+
+ """ + result = self.spider.detailContent(["/index.php/vod/detail/id/123.html"]) + vod = result["list"][0] + self.assertEqual(vod["vod_name"], "示例剧") + self.assertEqual(vod["vod_play_from"], "baidu#至臻$$$quark#至臻") + self.assertEqual( + vod["vod_play_url"], + "百度资源$https://pan.baidu.com/s/b1$$$夸克资源$https://pan.quark.cn/s/q1", + ) + + def test_player_content_passthroughs_supported_pan_urls(self): + self.assertEqual( + self.spider.playerContent("baidu#至臻", "https://pan.baidu.com/s/demo", {}), + {"parse": 0, "playUrl": "", "url": "https://pan.baidu.com/s/demo"}, + ) + + def test_player_content_rejects_non_pan_url(self): + self.assertEqual( + self.spider.playerContent("site", "/index.php/vod/play/id/1.html", {}), + {"parse": 0, "playUrl": "", "url": ""}, + ) + + +if __name__ == "__main__": + unittest.main() diff --git a/py/玩偶聚合.py b/py/玩偶聚合.py index a4732f5..9e9cb0f 100644 --- a/py/玩偶聚合.py +++ b/py/玩偶聚合.py @@ -82,6 +82,26 @@ class Spider(BaseSpider): "search_url": "/vodsearch/-------------.html?wd={keyword}&page={page}", "default_categories": [("1", "电影"), ("2", "电视剧"), ("3", "动漫"), ("4", "综艺")], }, + { + "id": "zhizhen", + "name": "至臻", + "domains": ["http://www.miqk.cc"], + "filter_files": [], + "list_xpath": "//*[contains(@class,'module-item')]", + "search_xpath": "//*[contains(@class,'module-search-item')]", + "detail_pan_xpath": "//*[contains(@class,'module-row-info')]//p", + "category_url": "/index.php/vod/show/id/{categoryId}/page/{page}.html", + "search_url": "/index.php/vod/search/page/{page}/wd/{keyword}.html", + "default_categories": [ + ("1", "电影"), + ("2", "剧集"), + ("3", "动漫"), + ("4", "综艺"), + ("5", "短剧"), + ("24", "老剧"), + ("26", "严选"), + ], + }, ] def init(self, extend=""): diff --git a/py/至臻.py b/py/至臻.py new file mode 100644 index 0000000..0d33569 --- /dev/null +++ b/py/至臻.py @@ -0,0 +1,252 @@ +# coding=utf-8 +import re +import sys +from urllib.parse import quote, urljoin + +from base.spider import Spider as BaseSpider + +sys.path.append("..") + + +class Spider(BaseSpider): + def __init__(self): + self.name = "至臻" + self.host = "http://www.miqk.cc" + self.headers = { + "User-Agent": ( + "Mozilla/5.0 (Windows NT 10.0; Win64; x64) " + "AppleWebKit/537.36 (KHTML, like Gecko) " + "Chrome/136.0.0.0 Safari/537.36" + ), + "Referer": self.host + "/", + } + self.categories = [ + {"type_id": "1", "type_name": "至臻电影"}, + {"type_id": "2", "type_name": "至臻剧集"}, + {"type_id": "3", "type_name": "至臻动漫"}, + {"type_id": "4", "type_name": "至臻综艺"}, + {"type_id": "5", "type_name": "至臻短剧"}, + {"type_id": "24", "type_name": "至臻老剧"}, + {"type_id": "26", "type_name": "至臻严选"}, + ] + self.pan_patterns = [ + ("baidu", "百度资源", r"pan\.baidu\.com|yun\.baidu\.com"), + ("a139", "139资源", r"yun\.139\.com"), + ("a189", "天翼资源", r"cloud\.189\.cn"), + ("a123", "123资源", r"123684\.com|123865\.com|123912\.com|123pan\.com"), + ("a115", "115资源", r"115\.com"), + ("quark", "夸克资源", r"pan\.quark\.cn"), + ("xunlei", "迅雷资源", r"pan\.xunlei\.com"), + ("aliyun", "阿里资源", r"aliyundrive\.com|alipan\.com"), + ("uc", "UC资源", r"drive\.uc\.cn"), + ] + self.pan_priority = { + "baidu": 1, + "a139": 2, + "a189": 3, + "a123": 4, + "a115": 5, + "quark": 6, + "xunlei": 7, + "aliyun": 8, + "uc": 9, + } + + 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, path): + return urljoin(self.host + "/", str(path or "").strip()) + + def _clean_text(self, text): + return re.sub(r"\s+", " ", str(text or "").replace("\xa0", " ")).strip() + + def _detect_pan_type(self, url): + raw = str(url or "").strip() + for pan_type, title, pattern in self.pan_patterns: + if re.search(pattern, raw, re.I): + return pan_type, title + return "", "" + + 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 _page_result(self, items, pg): + page = int(pg) + return {"page": page, "limit": len(items), "total": page * 20 + len(items), "list": items} + + def _parse_cards(self, html): + root = self.html(html) + if root is None: + return [] + + items = [] + seen = set() + for node in root.xpath("//*[@id='main']//*[contains(@class,'module-item')]"): + href = "".join(node.xpath(".//*[contains(@class,'module-item-pic')]//a[1]/@href")).strip() + title = "".join(node.xpath(".//*[contains(@class,'module-item-pic')]//img[1]/@alt")).strip() + pic = ( + "".join(node.xpath(".//*[contains(@class,'module-item-pic')]//img[1]/@data-src")).strip() + or "".join(node.xpath(".//*[contains(@class,'module-item-pic')]//img[1]/@src")).strip() + ) + remarks = self._clean_text("".join(node.xpath(".//*[contains(@class,'module-item-text')][1]//text()"))) + year = self._clean_text("".join(node.xpath(".//*[contains(@class,'module-item-caption')][1]//span[1]//text()"))) + if not href or not title or href in seen: + continue + seen.add(href) + items.append( + { + "vod_id": href, + "vod_name": title, + "vod_pic": self._build_url(pic), + "vod_remarks": remarks, + "vod_year": year, + } + ) + return items + + def categoryContent(self, tid, pg, filter, extend): + url = self._build_url(f"/index.php/vod/show/id/{tid}/page/{int(pg)}.html") + return self._page_result(self._parse_cards(self._request_html(url)), pg) + + def searchContent(self, key, quick, pg="1"): + keyword = self._clean_text(key) + page = int(pg) + if not keyword: + return {"page": page, "total": 0, "list": []} + + url = self._build_url(f"/index.php/vod/search/page/{page}/wd/{quote(keyword)}.html") + root = self.html(self._request_html(url)) + if root is None: + return {"page": page, "total": 0, "list": []} + + items = [] + for node in root.xpath("//*[contains(@class,'module-search-item')]"): + href = "".join(node.xpath(".//*[contains(@class,'video-serial')][1]/@href")).strip() + title = "".join(node.xpath(".//*[contains(@class,'video-serial')][1]/@title")).strip() + pic = ( + "".join(node.xpath(".//*[contains(@class,'module-item-pic')]//img[1]/@data-src")).strip() + or "".join(node.xpath(".//*[contains(@class,'module-item-pic')]//img[1]/@src")).strip() + ) + remarks = self._clean_text( + "".join(node.xpath(".//*[contains(@class,'video-serial')][1]//text()")) + or "".join(node.xpath(".//*[contains(@class,'module-item-text')][1]//text()")) + ) + if not href or not title: + continue + items.append( + { + "vod_id": href, + "vod_name": title, + "vod_pic": self._build_url(pic), + "vod_remarks": remarks, + } + ) + return {"page": page, "total": len(items), "list": items} + + def _parse_detail_page(self, vod_id, html): + root = self.html(html) + if root is None: + return { + "vod_id": vod_id, + "vod_name": "", + "vod_pic": "", + "vod_year": "", + "vod_director": "", + "vod_actor": "", + "vod_content": "", + "pan_urls": [], + } + + detail = { + "vod_id": vod_id, + "vod_name": self._clean_text("".join(root.xpath("//*[contains(@class,'page-title')][1]//text()"))), + "vod_pic": self._build_url( + "".join( + root.xpath( + "//*[contains(@class,'mobile-play')]//*[contains(@class,'lazyload')][1]/@data-src | " + "//*[contains(@class,'mobile-play')]//*[contains(@class,'lazyload')][1]/@src" + ) + ).strip() + ), + "vod_year": "", + "vod_director": "", + "vod_actor": "", + "vod_content": "", + "pan_urls": [], + } + + for label_node in root.xpath("//*[contains(@class,'video-info-itemtitle')]"): + key = self._clean_text("".join(label_node.xpath(".//text()"))) + sibling = label_node.getnext() + if sibling is None: + continue + values = [self._clean_text(text) for text in sibling.xpath(".//a//text()")] + joined = ",".join([value for value in values if value]) + text_value = self._clean_text("".join(sibling.xpath(".//text()"))) + if "年代" in key: + detail["vod_year"] = joined or text_value + elif "导演" in key: + detail["vod_director"] = joined or text_value + elif "主演" in key: + detail["vod_actor"] = joined or text_value + elif "剧情" in key: + detail["vod_content"] = text_value + + for node in root.xpath("//*[contains(@class,'module-row-info')]//p"): + text = self._clean_text("".join(node.xpath(".//text()"))) + if text: + detail["pan_urls"].append(text) + return detail + + def _build_pan_lines(self, detail): + lines = [] + seen = set() + for url in detail.get("pan_urls", []): + pan_type, title = self._detect_pan_type(url) + if not pan_type or url in seen: + continue + seen.add(url) + lines.append((self.pan_priority.get(pan_type, 999), f"{pan_type}#至臻", f"{title}${url}")) + lines.sort(key=lambda item: item[0]) + return [(item[1], item[2]) for item in lines] + + def detailContent(self, ids): + result = {"list": []} + for raw_id in ids: + vod_id = str(raw_id or "").strip() + detail = self._parse_detail_page(vod_id, self._request_html(self._build_url(vod_id))) + lines = self._build_pan_lines(detail) + result["list"].append( + { + "vod_id": vod_id, + "vod_name": detail["vod_name"], + "vod_pic": detail["vod_pic"], + "vod_year": detail["vod_year"], + "vod_director": detail["vod_director"], + "vod_actor": detail["vod_actor"], + "vod_content": detail["vod_content"], + "vod_play_from": "$$$".join([item[0] for item in lines]), + "vod_play_url": "$$$".join([item[1] for item in lines]), + } + ) + return result + + def playerContent(self, flag, id, vipFlags): + pan_type, _ = self._detect_pan_type(id) + if pan_type: + return {"parse": 0, "playUrl": "", "url": id} + return {"parse": 0, "playUrl": "", "url": ""}