yeyinly/tvboxzt
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

feat: add zhizhen spider and aggregate support

Browse Source
This commit is contained in:
Harold
2026-04-20 15:00:35 +08:00
parent 71ed5778fe
commit c0b51be9dc
8 changed files with 1891 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
py/docs/superpowers/plans/2026-04-20-wanou-aggregate-zhizhen.md
+289
View File
@@ -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 = """
<div class="module-search-item">
<a class="video-serial" href="/index.php/vod/detail/id/789.html" title="至臻影片">抢先版</a>
<div class="module-item-pic"><img data-src="/search.jpg" alt="至臻影片" /></div>
</div>
"""
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 = """
<div class="module-item">
<div class="module-item-pic">
<a href="/index.php/vod/detail/id/456.html"></a>
<img data-src="/cate.jpg" alt="至臻分类片" />
</div>
<div class="module-item-text">HD</div>
</div>
"""
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/<id>/page/<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:<path>`、`zhizhen` 站点 ID 和 `http://www.miqk.cc` 域名，命名保持一致。
py/docs/superpowers/plans/2026-04-20-zhizhen-spider.md
+571
View File
@@ -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 = """
<div id="main">
<div class="module-item">
<div class="module-item-pic">
<a href="/index.php/vod/detail/id/123.html"></a>
<img data-src="/poster.jpg" alt="示例影片" />
</div>
<div class="module-item-text">HD</div>
<div class="module-item-caption"><span>2025</span></div>
</div>
</div>
"""
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 = """
<div id="main">
<div class="module-item">
<div class="module-item-pic">
<a href="/index.php/vod/detail/id/456.html"></a>
<img data-src="/cate.jpg" alt="分类影片" />
</div>
<div class="module-item-text">更新至10集</div>
<div class="module-item-caption"><span>2024</span></div>
</div>
</div>
"""
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 = """
<div class="module-search-item">
<a class="video-serial" href="/index.php/vod/detail/id/789.html" title="搜索影片">抢先版</a>
<div class="module-item-pic">
<img data-src="/search.jpg" alt="搜索影片" />
</div>
</div>
"""
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 = """
<div class="page-title">示例剧</div>
<div class="mobile-play"><img class="lazyload" data-src="/poster.jpg" /></div>
<div class="video-info-itemtitle">年代</div><div><a>2024</a></div>
<div class="video-info-itemtitle">导演</div><div><a>导演甲</a></div>
<div class="video-info-itemtitle">主演</div><div><a>演员甲</a><a>演员乙</a></div>
<div class="video-info-itemtitle">剧情</div><div><p>一段剧情简介</p></div>
<div class="module-row-info">
<p>https://pan.quark.cn/s/q1</p>
<p>https://pan.baidu.com/s/b1</p>
</div>
"""
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 = """
<div class="page-title">示例剧</div>
<div class="mobile-play"><img class="lazyload" data-src="/poster.jpg" /></div>
<div class="video-info-itemtitle">年代</div><div><a>2024</a></div>
<div class="video-info-itemtitle">导演</div><div><a>导演甲</a></div>
<div class="video-info-itemtitle">主演</div><div><a>演员甲</a></div>
<div class="video-info-itemtitle">剧情</div><div><p>一段剧情简介</p></div>
<div class="module-row-info">
<p>https://pan.quark.cn/s/q1</p>
<p>https://pan.baidu.com/s/b1</p>
</div>
"""
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` 的命名在各任务中保持一致。
py/docs/superpowers/specs/2026-04-20-wanou-aggregate-zhizhen-design.md
+170
View File
@@ -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/<categoryId>/page/<pg>.html`
- 按现有通用列表解析逻辑输出站内条目
返回结构保持与现有聚合站一致：
- `vod_id` 使用 `site:zhizhen:<detail_path>`
- 保留 `_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/<keyword>.html`
- 断言结果被编码为 `site:zhizhen:<path>`
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` 通过
- 聚合层返回结构不引入新的字段格式变化
py/docs/superpowers/specs/2026-04-20-zhizhen-spider-design.md
+291
View File
@@ -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` 覆盖核心行为
- 针对 `至臻` 模块的单测全部通过
- 返回结构与当前仓库同类盘站蜘蛛保持一致
py/tests/test_玩偶聚合.py
+99
View File
@@ -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 = """
<div class="module-search-item">
<a class="video-serial" href="/index.php/vod/detail/id/789.html" title="至臻影片">抢先版</a>
<div class="module-item-pic"><img data-src="/search.jpg" alt="至臻影片" /></div>
</div>
"""
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 = """
<div class="module-item">
<div class="module-item-pic">
<a href="/index.php/vod/detail/id/456.html"></a>
<img data-src="/cate.jpg" alt="至臻分类片" />
</div>
<div class="module-item-text">HD</div>
</div>
"""
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")
py/tests/test_至臻.py
+199
View File
@@ -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 = """
<div id="main">
<div class="module-item">
<div class="module-item-pic">
<a href="/index.php/vod/detail/id/123.html"></a>
<img data-src="/poster.jpg" alt="示例影片" />
</div>
<div class="module-item-text">HD</div>
<div class="module-item-caption"><span>2025</span></div>
</div>
</div>
"""
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 = """
<div id="main">
<div class="module-item">
<div class="module-item-pic">
<a href="/index.php/vod/detail/id/456.html"></a>
<img data-src="/cate.jpg" alt="分类影片" />
</div>
<div class="module-item-text">更新至10集</div>
<div class="module-item-caption"><span>2024</span></div>
</div>
</div>
"""
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 = """
<div class="module-search-item">
<a class="video-serial" href="/index.php/vod/detail/id/789.html" title="搜索影片">抢先版</a>
<div class="module-item-pic">
<img data-src="/search.jpg" alt="搜索影片" />
</div>
</div>
"""
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 = """
<div class="page-title">示例剧</div>
<div class="mobile-play"><img class="lazyload" data-src="/poster.jpg" /></div>
<div class="video-info-itemtitle">年代</div><div><a>2024</a></div>
<div class="video-info-itemtitle">导演</div><div><a>导演甲</a></div>
<div class="video-info-itemtitle">主演</div><div><a>演员甲</a><a>演员乙</a></div>
<div class="video-info-itemtitle">剧情</div><div><p>一段剧情简介</p></div>
<div class="module-row-info">
<p>https://pan.quark.cn/s/q1</p>
<p>https://pan.baidu.com/s/b1</p>
</div>
"""
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 = """
<div class="page-title">示例剧</div>
<div class="mobile-play"><img class="lazyload" data-src="/poster.jpg" /></div>
<div class="video-info-itemtitle">年代</div><div><a>2024</a></div>
<div class="video-info-itemtitle">导演</div><div><a>导演甲</a></div>
<div class="video-info-itemtitle">主演</div><div><a>演员甲</a></div>
<div class="video-info-itemtitle">剧情</div><div><p>一段剧情简介</p></div>
<div class="module-row-info">
<p>https://pan.quark.cn/s/q1</p>
<p>https://pan.baidu.com/s/b1</p>
</div>
"""
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()
py/玩偶聚合.py
+20
View File
@@ -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=""):
py/至臻.py
+252
View File
@@ -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": ""}