樱花动漫

This commit is contained in:
Harold
2026-04-20 11:46:43 +08:00
parent bb3d89405b
commit af29ab7e46
2 changed files with 519 additions and 0 deletions
+236
View File
@@ -0,0 +1,236 @@
import unittest
from importlib.machinery import SourceFileLoader
from pathlib import Path
from unittest.mock import patch
ROOT = Path(__file__).resolve().parents[1]
MODULE = SourceFileLoader("yinghua_spider", str(ROOT / "樱花动漫.py")).load_module()
Spider = MODULE.Spider
SAMPLE_DETAIL_HTML = """
<html><body>
<div class="detail"><h2>进击的巨人</h2></div>
<div class="cover"><img data-original="https://pic.example.com/zr.jpg" /></div>
<div class="item"><span>状态:</span><em>已完结</em></div>
<div class="item"><span>年份:</span>2023</div>
<div class="item"><span>地区:</span>日本</div>
<div class="item"><span>类型:</span>热血</div>
<div class="item"><span>主演:</span>梶裕贵</div>
<li class="blurb"><span>简介</span>讲述巨人的故事</li>
<div class="module-tab-item">高清</div>
<div class="module-tab-item">ikun</div>
<div class="module-play-list">
<a href="/play/12345-1-1/">第01集</a>
<a href="/play/12345-1-2/">第02集</a>
<a href="/play/12345-1-3/">第03集</a>
</div>
<div class="module-play-list">
<a href="/play/12345-2-1/">第01集</a>
<a href="/play/12345-2-2/">第02集</a>
</div>
</body></html>
"""
SAMPLE_LIST_HTML = """
<html><body>
<ul>
<li><a href="/detail/123/" title="进击的巨人"><img data-original="https://pic/1.jpg" /></a><p>已完结</p></li>
<li><a href="/detail/456/" title="鬼灭之刃"><img data-original="https://pic/2.jpg" /></a><p>更新中</p></li>
<li><a href="/other/" title="忽略项"></a></li>
</ul>
</body></html>
"""
SAMPLE_SEARCH_HTML = """
<html><body>
找到 <em>5</em> 条结果
<ul>
<li><a class="cover" href="/detail/123/" title="进击的巨人" data-original="https://pic/1.jpg">
<div class="item"><span>状态:</span>已完结</div>
</a></li>
<li><a class="cover" href="/detail/456/" title="鬼灭之刃" data-original="https://pic/2.jpg">
<div class="item"><span>状态:</span>更新中</div>
</a></li>
</ul>
</body></html>
"""
SAMPLE_PLAY_HTML = """
<html><body>
<script>
var art = new Artplayer({
url: 'https://cdn.example.com/video/12345.m3u8?key=abc',
});
</script>
</body></html>
"""
SAMPLE_CATEGORY_HTML = """
<html><body>
<ul>
<li><a href="/detail/123/" title="进击的巨人"><img data-original="https://pic/1.jpg" /></a><p>已完结</p></li>
</ul>
<div class="pagination">
<a href="/type/riman/3/">3</a>
<a href="/type/riman/5/">5</a>
</div>
</body></html>
"""
class TestYingHuaSpider(unittest.TestCase):
def setUp(self):
Spider._instance = None
self.spider = Spider()
self.spider.init()
def test_home_content_classes(self):
content = self.spider.homeContent(False)
ids = [c["type_id"] for c in content["class"]]
self.assertEqual(len(content["class"]), 4)
self.assertIn("guoman", ids)
self.assertIn("riman", ids)
self.assertIn("oman", ids)
self.assertIn("dmfilm", ids)
def test_get_name(self):
self.assertEqual(self.spider.getName(), "樱花动漫")
@patch.object(Spider, "_get_html")
def test_home_video_content(self, mock_html):
mock_html.return_value = SAMPLE_LIST_HTML
result = self.spider.homeVideoContent()
self.assertEqual(len(result["list"]), 2)
self.assertEqual(result["list"][0]["vod_name"], "进击的巨人")
self.assertEqual(result["list"][0]["vod_remarks"], "已完结")
self.assertEqual(result["list"][1]["vod_name"], "鬼灭之刃")
@patch.object(Spider, "_get_html")
def test_home_video_dedup(self, mock_html):
mock_html.return_value = """
<ul>
<li><a href="/detail/123/" title="A"><img src="p.jpg" /></a></li>
<li><a href="/detail/123/" title="A"><img src="p.jpg" /></a></li>
</ul>
"""
result = self.spider.homeVideoContent()
self.assertEqual(len(result["list"]), 1)
@patch.object(Spider, "_get_html")
def test_category_content(self, mock_html):
mock_html.return_value = SAMPLE_CATEGORY_HTML
result = self.spider.categoryContent("riman", "1", False, {})
self.assertEqual(len(result["list"]), 1)
self.assertEqual(result["list"][0]["vod_name"], "进击的巨人")
self.assertEqual(result["pagecount"], 5)
@patch.object(Spider, "_get_html")
def test_category_page2_url(self, mock_html):
mock_html.return_value = SAMPLE_LIST_HTML
self.spider.categoryContent("riman", "2", False, {})
mock_html.assert_called_with("https://www.dmvvv.com/type/riman/2/")
@patch.object(Spider, "_get_html")
def test_detail_content(self, mock_html):
mock_html.return_value = SAMPLE_DETAIL_HTML
result = self.spider.detailContent(["/detail/12345/"])
vod = result["list"][0]
self.assertEqual(vod["vod_name"], "进击的巨人")
self.assertEqual(vod["vod_pic"], "https://pic.example.com/zr.jpg")
self.assertEqual(vod["vod_remarks"], "已完结")
self.assertEqual(vod["vod_year"], "2023")
self.assertEqual(vod["vod_area"], "日本")
self.assertEqual(vod["vod_content"], "讲述巨人的故事")
self.assertIn("高清", vod["vod_play_from"])
self.assertIn("ikun", vod["vod_play_from"])
self.assertIn("第01集$/play/12345-1-1/", vod["vod_play_url"])
self.assertIn("第02集$/play/12345-1-2/", vod["vod_play_url"])
@patch.object(Spider, "_get_html")
def test_detail_multi_source_grouping(self, mock_html):
mock_html.return_value = SAMPLE_DETAIL_HTML
result = self.spider.detailContent(["/detail/12345/"])
vod = result["list"][0]
from_list = vod["vod_play_from"].split("$$$")
url_groups = vod["vod_play_url"].split("$$$")
self.assertEqual(len(from_list), 2)
self.assertEqual(len(url_groups), 2)
# 高清 source (sourceIdx=1)
self.assertEqual(from_list[0], "高清")
self.assertIn("第01集$/play/12345-1-1/", url_groups[0])
self.assertIn("第03集$/play/12345-1-3/", url_groups[0])
# ikun source (sourceIdx=2)
self.assertEqual(from_list[1], "ikun")
self.assertIn("第01集$/play/12345-2-1/", url_groups[1])
self.assertIn("第02集$/play/12345-2-2/", url_groups[1])
# ensure no cross-contamination
self.assertNotIn("12345-2-", url_groups[0])
self.assertNotIn("12345-1-", url_groups[1])
@patch.object(Spider, "_get_html")
def test_detail_title_from_title_tag(self, mock_html):
mock_html.return_value = "<html><body><title>鬼灭之刃 - 樱花动漫</title></body></html>"
result = self.spider.detailContent(["/detail/999/"])
vod = result["list"][0]
self.assertEqual(vod["vod_name"], "鬼灭之刃")
@patch.object(Spider, "_get_html")
def test_detail_no_episodes(self, mock_html):
mock_html.return_value = """
<html><body>
<div class="detail"><h2>测试</h2></div>
</body></html>
"""
result = self.spider.detailContent(["/detail/1/"])
vod = result["list"][0]
self.assertEqual(vod["vod_play_from"], "default")
self.assertEqual(vod["vod_play_url"], "")
@patch.object(Spider, "_get_html")
def test_search_content(self, mock_html):
mock_html.return_value = SAMPLE_SEARCH_HTML
result = self.spider.searchContent("巨人", False, "1")
self.assertEqual(len(result["list"]), 2)
self.assertEqual(result["list"][0]["vod_name"], "进击的巨人")
self.assertEqual(result["pagecount"], 1)
def test_search_empty_keyword(self):
result = self.spider.searchContent("", False, "1")
self.assertEqual(result["list"], [])
@patch.object(Spider, "_get_html")
def test_search_quick_mode(self, mock_html):
mock_html.return_value = SAMPLE_SEARCH_HTML
result = self.spider.searchContent("巨人", True, "1")
self.assertEqual(len(result["list"]), 2)
@patch.object(Spider, "_get_html")
def test_player_content_artplayer(self, mock_html):
mock_html.return_value = SAMPLE_PLAY_HTML
result = self.spider.playerContent("高清", "/play/12345-1-1/", {})
self.assertEqual(result["parse"], 0)
self.assertEqual(result["url"], "https://cdn.example.com/video/12345.m3u8?key=abc")
self.assertIn("Referer", result["header"])
@patch.object(Spider, "_get_html")
def test_player_content_m3u8_fallback(self, mock_html):
mock_html.return_value = '<html><script>var src = "https://cdn.example.com/stream.m3u8";</script></html>'
result = self.spider.playerContent("高清", "/play/12345-1-1/", {})
self.assertEqual(result["url"], "https://cdn.example.com/stream.m3u8")
@patch.object(Spider, "_get_html")
def test_player_content_no_match(self, mock_html):
mock_html.return_value = "<html><body>no video</body></html>"
result = self.spider.playerContent("高清", "/play/12345-1-1/", {})
self.assertEqual(result["url"], "https://www.dmvvv.com/play/12345-1-1/")
@patch.object(Spider, "_get_html")
def test_detail_full_url(self, mock_html):
mock_html.return_value = SAMPLE_DETAIL_HTML
self.spider.detailContent(["https://www.dmvvv.com/detail/12345/"])
mock_html.assert_called_with("https://www.dmvvv.com/detail/12345/")
if __name__ == "__main__":
unittest.main()
+283
View File
@@ -0,0 +1,283 @@
# coding=utf-8
import re
import sys
from urllib.parse import quote
from base.spider import Spider as BaseSpider
sys.path.append("..")
class Spider(BaseSpider):
def __init__(self):
self.name = "樱花动漫"
self.host = "https://www.dmvvv.com"
self.headers = {
"User-Agent": (
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "
"(KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36"
),
"Referer": "https://www.dmvvv.com/",
"Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
"Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8",
}
self.classes = [
{"type_id": "guoman", "type_name": "国产动漫"},
{"type_id": "riman", "type_name": "日本动漫"},
{"type_id": "oman", "type_name": "欧美动漫"},
{"type_id": "dmfilm", "type_name": "动漫电影"},
]
def init(self, extend=""):
return None
def getName(self):
return self.name
def homeContent(self, filter):
return {"class": self.classes}
def homeVideoContent(self):
html = self._get_html(self.host + "/")
items = self._parse_list(html)
seen = set()
unique = []
for item in items:
if item["vod_id"] not in seen:
seen.add(item["vod_id"])
unique.append(item)
return {"list": unique[:20]}
def categoryContent(self, tid, pg, filter, extend):
page = int(pg)
url = f"{self.host}/type/{tid}/" if page <= 1 else f"{self.host}/type/{tid}/{page}/"
html = self._get_html(url)
items = self._parse_list(html)
pagecount = self._parse_page_count(html, tid)
if pagecount <= page and len(items) >= 36:
pagecount = page + 1
return {
"list": items,
"page": page,
"pagecount": pagecount,
"total": pagecount * len(items) if items else 0,
}
def detailContent(self, ids):
raw_id = ids[0] if isinstance(ids, list) else ids
url = raw_id if raw_id.startswith("http") else self.host + raw_id
html = self._get_html(url)
vod = self._parse_detail(html, raw_id)
return {"list": [vod] if vod else []}
def searchContent(self, key, quick, pg="1"):
page = int(pg)
keyword = key.strip()
if not keyword:
return {"list": [], "page": page, "pagecount": 0, "total": 0}
encoded = quote(keyword)
url = (
f"{self.host}/search/?wd={encoded}"
if page <= 1
else f"{self.host}/search/?wd={encoded}&pageno={page}"
)
html = self._get_html(url)
items = self._parse_search_list(html)
pagecount = self._parse_search_page_count(html, page, len(items))
return {
"list": items[:10] if quick else items,
"page": page,
"pagecount": pagecount,
"total": len(items),
}
def playerContent(self, flag, id, vipFlags):
url = id if id.startswith("http") else self.host + id
html = self._get_html(url)
play_url = self._extract_play_url(html, url)
return {
"parse": 0,
"url": play_url,
"header": {
"User-Agent": self.headers["User-Agent"],
"Referer": self.host + "/",
},
}
# --- helpers ---
def _s(self, val):
return "" if val is None else str(val).strip()
def _get_html(self, url):
rsp = self.fetch(url, headers=self.headers, timeout=15, verify=False)
if rsp.status_code != 200:
return ""
return rsp.text or ""
def _parse_list(self, html):
root = self.html(html)
if root is None:
return []
items = []
for li in root.xpath("//li"):
link = li.xpath(".//a[@href]")
if not link:
continue
href = self._s(link[0].get("href", ""))
title = self._s(link[0].get("title", ""))
if not href or "/detail/" not in href or not title:
continue
pic = self._s((li.xpath(".//img/@data-original") or li.xpath(".//img/@src") or [""])[0])
remarks = self._s((li.xpath(".//p//text()") or [""])[0])
items.append({
"vod_id": href,
"vod_name": title,
"vod_pic": pic,
"vod_remarks": remarks,
})
return items
def _parse_search_list(self, html):
items = []
pattern = re.compile(r'<a class="cover" href="(/detail/\d+/)"', re.S)
for block in re.finditer(r"<li>\s*<a class=\"cover\".*?</li>", html, re.S):
chunk = block.group(0)
href_m = pattern.search(chunk)
if not href_m:
continue
title_m = re.search(r'title="([^"]+)"', chunk)
cover_m = re.search(r'data-original="([^"]+)"', chunk)
remark_m = re.search(r'<div class="item"><span>状态:</span>([^<]*)', chunk)
if not title_m:
continue
items.append({
"vod_id": href_m.group(1),
"vod_name": title_m.group(1).strip(),
"vod_pic": cover_m.group(1).strip() if cover_m else "",
"vod_remarks": remark_m.group(1).strip() if remark_m else "",
})
return items
def _parse_page_count(self, html, tid):
max_page = 1
for m in re.finditer(r"/type/[^/]+/(\d+)/", html):
max_page = max(max_page, int(m.group(1)))
for m in re.finditer(r"[?&]page(?:no)?=(\d+)", html):
max_page = max(max_page, int(m.group(1)))
return max_page
def _parse_search_page_count(self, html, current_page, result_count):
total_m = re.search(r"找到\s*<em>(\d+)</em>", html)
if total_m:
return max(1, -(-int(total_m.group(1)) // 12))
max_page = current_page
for m in re.finditer(r"pageno=(\d+)", html):
max_page = max(max_page, int(m.group(1)))
if max_page == current_page and result_count >= 12:
return current_page + 1
return max(max_page, 1)
def _parse_detail(self, html, raw_id):
root = self.html(html)
if root is None:
return None
title = ""
title_m = re.search(r'<div class="detail">.*?<h2>([^<]+)</h2>', html, re.S)
if title_m:
title = title_m.group(1).strip()
if not title:
title_m2 = re.search(r"<title>([^<]+)", html)
if title_m2:
title = title_m2.group(1).split("-")[0].strip()
cover = ""
cover_m = re.search(r'<div class="cover">\s*<img[^>]+data-original="([^"]+)"', html, re.S)
if cover_m:
cover = cover_m.group(1)
def get_info(label, use_em=True):
pat = rf'<span>{label}:</span><em>([^<]+)</em>' if use_em else rf'<span>{label}:</span>([^<]+)'
m = re.search(pat, html)
return m.group(1).strip() if m else ""
vod_remarks = get_info("状态", True)
vod_year = get_info("年份", False)
vod_area = get_info("地区", False)
vod_type = get_info("类型", False)
vod_actor = get_info("主演", False)
vod_content = ""
desc_m = re.search(r'class="blurb"[^>]*>.*?<span>[^<]+</span>(.*?)</li>', html, re.S)
if desc_m:
vod_content = re.sub(r"<[^>]+>", "", desc_m.group(1)).strip()
play_data = self._parse_play_sources(html, raw_id)
return {
"vod_id": raw_id,
"vod_name": title,
"vod_pic": cover,
"vod_year": vod_year,
"vod_area": vod_area,
"vod_type": vod_type,
"vod_actor": vod_actor,
"vod_remarks": vod_remarks,
"vod_content": vod_content,
"vod_play_from": play_data["vod_play_from"],
"vod_play_url": play_data["vod_play_url"],
}
def _parse_play_sources(self, html, raw_id):
root = self.html(html)
if root is None:
return {"vod_play_from": "default", "vod_play_url": ""}
source_names = ["高清", "ikun", "非凡", "量子"]
# collect tab names in order
tabs = []
for idx, tab in enumerate(root.xpath("//*[contains(@class,'module-tab-item') or contains(@class,'tab-item')]")):
name = self._s(tab.xpath("string(.)")).strip()
if not name:
name = source_names[idx] if idx < len(source_names) else f"线路{idx+1}"
tabs.append(name)
# group episodes by source index from URL: /play/{vid}-{sourceIdx}-{epIdx}/
grouped = {}
all_links = root.xpath("//*[contains(@class,'module-play-list')]//a[@href] | //*[contains(@class,'playlist')]//a[@href]")
# fallback: try all links under .scroll-content or .module-list
if not all_links:
all_links = root.xpath("//*[contains(@class,'scroll-content') or contains(@class,'module-list')]//a[@href]")
for a in all_links:
ep_name = self._s(a.xpath("string(.)")).strip()
ep_url = self._s(a.get("href", ""))
if not ep_name or not ep_url or ep_url.startswith(("javascript:", "#")):
continue
m = re.search(r"/play/\d+-(\d+)-\d+/", ep_url)
if m:
src_idx = int(m.group(1)) - 1
else:
src_idx = 0
if src_idx not in grouped:
grouped[src_idx] = []
grouped[src_idx].append(f"{ep_name}${ep_url}")
if not grouped:
return {"vod_play_from": "default", "vod_play_url": ""}
# build output ordered by source index
play_from = []
play_url = []
for src_idx in sorted(grouped.keys()):
name = tabs[src_idx] if src_idx < len(tabs) else (source_names[src_idx] if src_idx < len(source_names) else f"线路{src_idx+1}")
play_from.append(name)
play_url.append("#".join(grouped[src_idx]))
return {
"vod_play_from": "$$$".join(play_from),
"vod_play_url": "$$$".join(play_url),
}
def _extract_play_url(self, html, fallback_url):
m = re.search(r"url:\s*'(https?://[^']+)'", html)
if m:
return m.group(1)
m = re.search(r"(https?://[^\s'\"<>]+\.m3u8(?:\?[^\s'\"<>]*)?)", html)
if m:
return m.group(1)
return fallback_url