feat: add ddys detail parsing

This commit is contained in:
Harold
2026-04-19 16:43:15 +08:00
parent 9b68b6ba43
commit 53571a890a
2 changed files with 136 additions and 0 deletions
+45
View File
@@ -140,6 +140,51 @@ class TestDDYSSpider(unittest.TestCase):
self.assertEqual(result["list"][0]["vod_id"], "https://ddys.io/anime/result/")
self.assertEqual(result["pagecount"], 2)
def test_parse_detail_page_merges_direct_and_pan_sources(self):
html = """
<html>
<h1 class="text-xl md:text-3xl">低端示例<span class="block">DDYS Example</span></h1>
<img alt="低端示例" src="/poster-detail.jpg" />
<div class="text-xs md:text-sm text-gray-600 dark:text-gray-400">2025 · 日本 · 动画</div>
<div class="text-xs md:text-sm text-gray-700 dark:text-gray-300">导演:<span>导演甲</span></div>
<div class="text-xs md:text-sm text-gray-700 dark:text-gray-300">主演:<span>主演乙</span></div>
<div class="prose"><p>第一段简介。</p><p>第二段简介。</p></div>
<button onclick="switchSource(1, '第1集$/play/ep1#第2集$/play/ep2', 'mp4')">DDYS</button>
<div class="download-type-content" id="download-type-quark">
<button onclick="trackAndOpenResource(atob('aHR0cHM6Ly9wYW4ucXVhcmsuY24vcy9hYmMxMjM='))">夸克查看</button>
</div>
<div class="download-type-content" id="download-type-baidu">
<button onclick="trackAndOpenResource(atob('aHR0cHM6Ly9wYW4uYmFpZHUuY29tL3MvZGVtbw=='))">百度查看</button>
</div>
</html>
"""
vod = self.spider._parse_detail_page(html, "https://ddys.io/anime/demo/")
self.assertEqual(vod["vod_name"], "低端示例")
self.assertEqual(vod["vod_pic"], "https://ddys.io/poster-detail.jpg")
self.assertEqual(vod["vod_year"], "2025")
self.assertEqual(vod["vod_area"], "日本")
self.assertEqual(vod["vod_class"], "动画")
self.assertEqual(vod["vod_director"], "导演甲")
self.assertEqual(vod["vod_actor"], "主演乙")
self.assertEqual(vod["vod_content"], "第一段简介。\n第二段简介。")
self.assertEqual(vod["vod_play_from"], "DDYS$$$quark$$$baidu")
self.assertIn("第2集$/play/ep2", vod["vod_play_url"])
self.assertIn("夸克查看$https://pan.quark.cn/s/abc123", vod["vod_play_url"])
self.assertIn("百度查看$https://pan.baidu.com/s/demo", vod["vod_play_url"])
@patch.object(Spider, "_request_html")
def test_detail_content_reads_detail_page_and_returns_single_vod(self, mock_request_html):
mock_request_html.return_value = """
<h1 class="text-xl md:text-3xl">详情标题</h1>
<button onclick="switchSource(1, '/play/detail-demo', 'mp4')">直连</button>
"""
result = self.spider.detailContent(["https://ddys.io/movie/demo/"])
self.assertEqual(mock_request_html.call_args.args[0], "https://ddys.io/movie/demo/")
self.assertEqual(result["list"][0]["vod_id"], "https://ddys.io/movie/demo/")
self.assertEqual(result["list"][0]["vod_name"], "详情标题")
self.assertEqual(result["list"][0]["vod_play_from"], "直连")
self.assertEqual(result["list"][0]["vod_play_url"], "全集$/play/detail-demo")
if __name__ == "__main__":
unittest.main()
+91
View File
@@ -1,4 +1,5 @@
# coding=utf-8
import base64
import json
import re
import sys
@@ -176,3 +177,93 @@ class Spider(BaseSpider):
if not items:
items = self._parse_movie_cards(html)
return {"page": page, "pagecount": page + 1 if items else page, "total": len(items), "list": items}
def _extract_switch_sources(self, html):
root = self.html(html)
if root is None:
return []
groups = []
for button in root.xpath("//button[contains(@onclick,'switchSource')]"):
onclick = (button.xpath("./@onclick") or [""])[0]
matched = re.search(r"switchSource\(\d+,\s*'([^']*)',\s*'[^']*'\)", onclick, re.S)
if not matched:
continue
name = self._clean_text("".join(button.xpath(".//text()"))) or self.name
payload = self._clean_text(matched.group(1))
if not payload:
continue
if "$" not in payload:
payload = f"全集${payload}"
groups.append({"from": name, "urls": payload})
return groups
def _extract_pan_sources(self, html):
root = self.html(html)
if root is None:
return []
groups = []
for panel in root.xpath("//*[contains(@class,'download-type-content')]"):
panel_id = ((panel.xpath("./@id") or [""])[0]).replace("download-type-", "").strip()
if panel_id not in ("quark", "xunlei", "baidu"):
continue
entries = []
for button in panel.xpath(".//button[@onclick]"):
onclick = (button.xpath("./@onclick") or [""])[0]
matched = re.search(r"atob\('([^']+)'\)", onclick)
if not matched:
continue
try:
link = base64.b64decode(matched.group(1)).decode("utf-8").strip()
except Exception:
continue
title = self._clean_text("".join(button.xpath(".//text()"))) or panel_id
if link:
entries.append(f"{title}${link}")
if entries:
groups.append({"from": panel_id, "urls": "#".join(dict.fromkeys(entries))})
return groups
def _parse_detail_page(self, html, vod_id):
root = self.html(html)
if root is None:
return {"vod_id": vod_id, "vod_name": "", "vod_play_from": "", "vod_play_url": ""}
title = self._clean_text("".join(root.xpath("//h1[1]/text()")))
pic = ((root.xpath("//img[@alt][1]/@src") or [""])[0]).strip()
meta = self._clean_text("".join(root.xpath("(//*[contains(@class,'text-gray-600')])[1]//text()")))
parts = [self._clean_text(item) for item in meta.split("·")] if meta else []
director = ""
actor = ""
for text in root.xpath("//*[contains(@class,'text-gray-700')]"):
joined = self._clean_text("".join(text.xpath(".//text()")))
if joined.startswith("导演:"):
director = joined.split("", 1)[-1].strip()
if joined.startswith("主演:"):
actor = joined.split("", 1)[-1].strip()
content = "\n".join(
[self._clean_text("".join(node.xpath(".//text()"))) for node in root.xpath("//*[contains(@class,'prose')]//p")]
).strip()
play_groups = self._extract_switch_sources(html) + self._extract_pan_sources(html)
return {
"vod_id": vod_id,
"vod_name": title,
"vod_pic": self._build_url(pic),
"vod_content": content,
"vod_remarks": " · ".join([item for item in parts if item]),
"vod_year": parts[0] if len(parts) > 0 else "",
"vod_area": parts[1] if len(parts) > 1 else "",
"vod_class": parts[2] if len(parts) > 2 else "",
"vod_director": director,
"vod_actor": actor,
"vod_play_from": "$$$".join([item["from"] for item in play_groups]),
"vod_play_url": "$$$".join([item["urls"] for item in play_groups]),
}
def detailContent(self, ids):
result = {"list": []}
for raw_id in ids:
vod_id = self._stringify(raw_id).strip()
if not vod_id:
continue
vod = self._parse_detail_page(self._request_html(vod_id), vod_id)
result["list"].append(vod)
return result