feat: add yisou spider
This commit is contained in:
Harold
@@ -0,0 +1,165 @@
|
||||
import unittest
|
||||
from importlib.machinery import SourceFileLoader
|
||||
from pathlib import Path
|
||||
from urllib.parse import quote
|
||||
from unittest.mock import patch
|
||||
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[1]
|
||||
MODULE = SourceFileLoader("yisou_spider", str(ROOT / "奕搜.py")).load_module()
|
||||
Spider = MODULE.Spider
|
||||
|
||||
|
||||
class TestYiSouSpider(unittest.TestCase):
|
||||
def setUp(self):
|
||||
Spider._instance = None
|
||||
self.spider = Spider()
|
||||
self.spider.init()
|
||||
|
||||
def test_home_content_exposes_expected_categories(self):
|
||||
content = self.spider.homeContent(False)
|
||||
self.assertEqual(
|
||||
[item["type_id"] for item in content["class"]],
|
||||
["dy", "dsj", "zy", "dm", "jlp", "dj"],
|
||||
)
|
||||
|
||||
def test_home_video_content_returns_empty_list(self):
|
||||
self.assertEqual(self.spider.homeVideoContent(), {"list": []})
|
||||
|
||||
def test_extract_remark_from_title_prefers_update_score_then_year(self):
|
||||
self.assertEqual(self.spider._extract_remark_from_title("片名 [更12] [8.5分] [2025]"), "更新至12集")
|
||||
self.assertEqual(self.spider._extract_remark_from_title("片名 [8.5分] [2025]"), "评分:8.5")
|
||||
self.assertEqual(self.spider._extract_remark_from_title("片名 [2025]"), "首播:2025")
|
||||
|
||||
def test_clean_title_text_removes_bracket_tags(self):
|
||||
self.assertEqual(self.spider._clean_title_text("片名 [更12] [2025]"), "片名")
|
||||
|
||||
def test_parse_list_boxes_extracts_short_ids_and_remark(self):
|
||||
html = """
|
||||
<div class="list-boxes">
|
||||
<a class="text_title_p" href="/resource/demo-1">示例影片 [更12] [8.2分]</a>
|
||||
<div class="left_ly"><a href="/resource/fallback-1">备链</a></div>
|
||||
<img class="image_left" src="/cover.jpg" />
|
||||
<div class="list-actions"><span>列表备注</span></div>
|
||||
</div>
|
||||
"""
|
||||
self.assertEqual(
|
||||
self.spider._parse_list_boxes(html),
|
||||
[
|
||||
{
|
||||
"vod_id": "/resource/demo-1",
|
||||
"vod_name": "示例影片",
|
||||
"vod_pic": "https://ysso.cc/cover.jpg",
|
||||
"vod_remarks": "更新至12集",
|
||||
}
|
||||
],
|
||||
)
|
||||
|
||||
@patch.object(Spider, "_request_html")
|
||||
def test_category_content_uses_expected_url(self, mock_request_html):
|
||||
mock_request_html.return_value = """
|
||||
<div class="list-boxes">
|
||||
<a class="text_title_p" href="/resource/demo-2">分类影片 [2025]</a>
|
||||
<img class="image_left" src="/cat.jpg" />
|
||||
</div>
|
||||
"""
|
||||
result = self.spider.categoryContent("dy", "3", False, {})
|
||||
self.assertEqual(mock_request_html.call_args.args[0], "https://ysso.cc/dy.html?page=3")
|
||||
self.assertEqual(result["page"], 3)
|
||||
self.assertEqual(result["list"][0]["vod_id"], "/resource/demo-2")
|
||||
self.assertNotIn("pagecount", result)
|
||||
|
||||
@patch.object(Spider, "_request_html")
|
||||
def test_search_content_uses_encoded_keyword(self, mock_request_html):
|
||||
mock_request_html.return_value = """
|
||||
<div class="list-boxes">
|
||||
<a class="text_title_p" href="/resource/search-1">搜索影片 [8.9分]</a>
|
||||
<img class="image_left" src="/search.jpg" />
|
||||
</div>
|
||||
"""
|
||||
result = self.spider.searchContent("繁花", False, "2")
|
||||
self.assertEqual(
|
||||
mock_request_html.call_args.args[0],
|
||||
f"https://ysso.cc/search.html?keyword={quote('繁花')}&page=2",
|
||||
)
|
||||
self.assertEqual(result["page"], 2)
|
||||
self.assertEqual(result["list"][0]["vod_remarks"], "评分:8.9")
|
||||
|
||||
def test_search_content_empty_keyword_returns_empty_list(self):
|
||||
self.assertEqual(self.spider.searchContent("", False, "1"), {"page": 1, "total": 0, "list": []})
|
||||
|
||||
def test_identify_disk(self):
|
||||
self.assertEqual(self.spider._identify_disk("https://pan.baidu.com/s/demo"), "baidu")
|
||||
self.assertEqual(self.spider._identify_disk("https://pan.quark.cn/s/demo"), "quark")
|
||||
self.assertEqual(self.spider._identify_disk("https://drive.uc.cn/s/demo"), "uc")
|
||||
self.assertEqual(self.spider._identify_disk("https://www.alipan.com/s/demo"), "aliyun")
|
||||
self.assertEqual(self.spider._identify_disk("https://pan.xunlei.com/s/demo"), "xunlei")
|
||||
self.assertEqual(self.spider._identify_disk("https://example.com/file"), "")
|
||||
|
||||
def test_build_play_data_groups_and_sorts_share_links(self):
|
||||
play = self.spider._build_play_data(
|
||||
[
|
||||
"https://pan.quark.cn/s/quark1",
|
||||
"https://pan.baidu.com/s/baidu1",
|
||||
"https://pan.quark.cn/s/quark1",
|
||||
"https://pan.xunlei.com/s/xl1",
|
||||
]
|
||||
)
|
||||
self.assertEqual(play["vod_play_from"], "baidu$$$quark$$$xunlei")
|
||||
self.assertIn("baidu$https://pan.baidu.com/s/baidu1", play["vod_play_url"])
|
||||
self.assertIn("quark$https://pan.quark.cn/s/quark1", play["vod_play_url"])
|
||||
|
||||
def test_build_play_data_falls_back_to_push_urls_for_unknown_links(self):
|
||||
play = self.spider._build_play_data(
|
||||
["https://example.com/file1", "https://example.com/file2"]
|
||||
)
|
||||
self.assertEqual(play["vod_play_from"], "奕搜")
|
||||
self.assertEqual(
|
||||
play["vod_play_url"],
|
||||
"1$push://https://example.com/file1#2$push://https://example.com/file2",
|
||||
)
|
||||
|
||||
@patch.object(Spider, "_request_html")
|
||||
def test_detail_content_extracts_metadata_and_netdisk_links(self, mock_request_html):
|
||||
mock_request_html.return_value = """
|
||||
<html><body>
|
||||
<h1 class="articl_title">示例详情 [更12] [2025]</h1>
|
||||
<div class="tc-box article-box"><img src="/poster.jpg" /></div>
|
||||
<div id="info">
|
||||
<span><span class="pl">导演：</span><span class="attrs"><a>导演甲</a><a>编剧乙</a></span></span>
|
||||
<span><span class="pl">主演：</span><span class="attrs"><a>演员甲</a><a>演员乙</a></span></span>
|
||||
</div>
|
||||
<p style="color: rgb(51, 51, 51)">这是一段足够长的剧情简介，用于测试详情页内容提取逻辑。</p>
|
||||
<a target="_blank" href="https://pan.quark.cn/s/abc123">夸克</a>
|
||||
<a target="_blank" href="https://pan.baidu.com/s/demo456">百度</a>
|
||||
<div>提取码：a1b2</div>
|
||||
</body></html>
|
||||
"""
|
||||
result = self.spider.detailContent(["/resource/demo-1"])
|
||||
vod = result["list"][0]
|
||||
self.assertEqual(mock_request_html.call_args.args[0], "https://ysso.cc/resource/demo-1")
|
||||
self.assertEqual(vod["vod_id"], "/resource/demo-1")
|
||||
self.assertEqual(vod["vod_name"], "示例详情")
|
||||
self.assertEqual(vod["vod_pic"], "https://ysso.cc/poster.jpg")
|
||||
self.assertEqual(vod["vod_director"], "导演甲, 编剧乙")
|
||||
self.assertEqual(vod["vod_actor"], "演员甲, 演员乙")
|
||||
self.assertIn("剧情简介", vod["vod_content"])
|
||||
self.assertEqual(vod["vod_play_from"], "baidu$$$quark")
|
||||
self.assertIn("提取码: a1b2", vod["vod_remarks"])
|
||||
|
||||
def test_player_content_strips_push_prefix(self):
|
||||
result = self.spider.playerContent("奕搜", "push://https://pan.quark.cn/s/abc123", {})
|
||||
self.assertEqual(result, {"parse": 0, "url": "https://pan.quark.cn/s/abc123"})
|
||||
|
||||
def test_player_content_passthroughs_http_url(self):
|
||||
result = self.spider.playerContent("quark", "https://pan.quark.cn/s/abc123", {})
|
||||
self.assertEqual(result, {"parse": 0, "url": "https://pan.quark.cn/s/abc123"})
|
||||
|
||||
@patch.object(Spider, "_request_html")
|
||||
def test_detail_content_returns_empty_list_when_html_missing(self, mock_request_html):
|
||||
mock_request_html.return_value = ""
|
||||
self.assertEqual(self.spider.detailContent(["/resource/missing"]), {"list": []})
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -0,0 +1,240 @@
|
||||
# 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 = "https://ysso.cc"
|
||||
self.headers = {
|
||||
"User-Agent": (
|
||||
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
|
||||
"AppleWebKit/537.36 (KHTML, like Gecko) "
|
||||
"Chrome/145.0.0.0 Safari/537.36"
|
||||
),
|
||||
"Referer": self.host + "/",
|
||||
}
|
||||
self.classes = [
|
||||
{"type_id": "dy", "type_name": "电影"},
|
||||
{"type_id": "dsj", "type_name": "电视剧"},
|
||||
{"type_id": "zy", "type_name": "综艺"},
|
||||
{"type_id": "dm", "type_name": "动漫"},
|
||||
{"type_id": "jlp", "type_name": "纪录片"},
|
||||
{"type_id": "dj", "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):
|
||||
return {"list": []}
|
||||
|
||||
def _build_url(self, path):
|
||||
return urljoin(self.host + "/", str(path or "").strip())
|
||||
|
||||
def _extract_remark_from_title(self, title_text):
|
||||
parts = re.findall(r"\[(.*?)\]", str(title_text or ""))
|
||||
for part in parts:
|
||||
text = part.strip()
|
||||
if text.startswith("更"):
|
||||
match = re.search(r"更\s*([0-9]+)", text)
|
||||
return f"更新至{match.group(1)}集" if match else text
|
||||
for part in parts:
|
||||
text = part.strip()
|
||||
if text.endswith("分"):
|
||||
return f"评分:{text[:-1].strip()}"
|
||||
for part in parts:
|
||||
text = part.strip()
|
||||
if re.match(r"^\d{4}$", text):
|
||||
return f"首播:{text}"
|
||||
return ""
|
||||
|
||||
def _clean_title_text(self, title):
|
||||
return re.sub(r"\s*\[.*?\]", "", str(title or "")).strip()
|
||||
|
||||
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, verify=False)
|
||||
if response.status_code != 200:
|
||||
return ""
|
||||
return response.text or ""
|
||||
|
||||
def _parse_list_boxes(self, html):
|
||||
root = self.html(html)
|
||||
if root is None:
|
||||
return []
|
||||
items = []
|
||||
seen = set()
|
||||
for node in root.xpath("//*[contains(@class,'list-boxes')]"):
|
||||
title_href = "".join(node.xpath(".//a[contains(@class,'text_title_p')][1]/@href")).strip()
|
||||
fallback_href = "".join(node.xpath(".//*[contains(@class,'left_ly')]//a[1]/@href")).strip()
|
||||
vod_id = title_href or fallback_href
|
||||
title = self._clean_text("".join(node.xpath(".//a[contains(@class,'text_title_p')][1]//text()")))
|
||||
if not vod_id or not title or vod_id in seen:
|
||||
continue
|
||||
seen.add(vod_id)
|
||||
pic = self._build_url("".join(node.xpath(".//img[contains(@class,'image_left')][1]/@src")).strip())
|
||||
remark = self._extract_remark_from_title(title) or self._clean_text(
|
||||
"".join(node.xpath(".//*[contains(@class,'list-actions')]//span[1]//text()"))
|
||||
)
|
||||
items.append(
|
||||
{
|
||||
"vod_id": vod_id,
|
||||
"vod_name": self._clean_title_text(title),
|
||||
"vod_pic": pic,
|
||||
"vod_remarks": remark,
|
||||
}
|
||||
)
|
||||
return items
|
||||
|
||||
def categoryContent(self, tid, pg, filter, extend):
|
||||
page = int(pg)
|
||||
items = self._parse_list_boxes(self._request_html(self._build_url(f"/{tid}.html?page={page}")))
|
||||
return {"page": page, "limit": len(items) or 20, "total": page * (len(items) or 20), "list": items}
|
||||
|
||||
def searchContent(self, key, quick, pg="1"):
|
||||
page = int(pg)
|
||||
keyword = self._clean_text(key)
|
||||
if not keyword:
|
||||
return {"page": page, "total": 0, "list": []}
|
||||
url = self._build_url(f"/search.html?keyword={quote(keyword)}&page={page}")
|
||||
items = self._parse_list_boxes(self._request_html(url))
|
||||
return {"page": page, "limit": len(items) or 20, "total": len(items), "list": items}
|
||||
|
||||
def _identify_disk(self, value):
|
||||
text = str(value or "").strip().lower()
|
||||
if "pan.baidu.com" in text:
|
||||
return "baidu"
|
||||
if "pan.quark.cn" in text:
|
||||
return "quark"
|
||||
if "drive.uc.cn" in text:
|
||||
return "uc"
|
||||
if "alipan.com" in text or "aliyundrive.com" in text:
|
||||
return "aliyun"
|
||||
if "pan.xunlei.com" in text:
|
||||
return "xunlei"
|
||||
return ""
|
||||
|
||||
def _disk_priority(self, name):
|
||||
order = {"baidu": 1, "quark": 2, "uc": 3, "aliyun": 4, "xunlei": 5}
|
||||
return order.get(name, 999)
|
||||
|
||||
def _extract_share_links(self, root):
|
||||
links = []
|
||||
seen = set()
|
||||
for href in root.xpath("//a[@target='_blank']/@href"):
|
||||
url = str(href or "").strip()
|
||||
if not url.startswith(("http://", "https://")) or url in seen:
|
||||
continue
|
||||
seen.add(url)
|
||||
links.append(url)
|
||||
return links
|
||||
|
||||
def _build_play_data(self, share_links):
|
||||
unique_links = []
|
||||
seen = set()
|
||||
for link in share_links:
|
||||
if link and link not in seen:
|
||||
seen.add(link)
|
||||
unique_links.append(link)
|
||||
grouped = {}
|
||||
for link in unique_links:
|
||||
disk = self._identify_disk(link)
|
||||
if not disk:
|
||||
continue
|
||||
grouped.setdefault(disk, []).append(f"{disk}${link}")
|
||||
if grouped:
|
||||
names = sorted(grouped, key=self._disk_priority)
|
||||
return {
|
||||
"vod_play_from": "$$$".join(names),
|
||||
"vod_play_url": "$$$".join("#".join(grouped[name]) for name in names),
|
||||
}
|
||||
if unique_links:
|
||||
return {
|
||||
"vod_play_from": "奕搜",
|
||||
"vod_play_url": "#".join(
|
||||
f"{index + 1}$push://{link}" for index, link in enumerate(unique_links)
|
||||
),
|
||||
}
|
||||
return {"vod_play_from": "奕搜", "vod_play_url": ""}
|
||||
|
||||
def detailContent(self, ids):
|
||||
result = {"list": []}
|
||||
for raw_id in ids:
|
||||
vod_id = str(raw_id or "").strip()
|
||||
if not vod_id:
|
||||
continue
|
||||
html = self._request_html(self._build_url(vod_id))
|
||||
root = self.html(html)
|
||||
if root is None:
|
||||
continue
|
||||
title = self._clean_text("".join(root.xpath("//h1[contains(@class,'articl_title')][1]//text()")))
|
||||
director_items = []
|
||||
actor_items = []
|
||||
for span in root.xpath("//*[@id='info']/span"):
|
||||
key = self._clean_text(
|
||||
"".join(span.xpath(".//*[contains(@class,'pl')][1]//text()"))
|
||||
).replace("：", "").replace(":", "")
|
||||
names = [self._clean_text(text) for text in span.xpath(".//*[contains(@class,'attrs')]//a/text()")]
|
||||
names = [name for name in names if name]
|
||||
if "导演" in key or "编剧" in key:
|
||||
for name in names:
|
||||
if name not in director_items:
|
||||
director_items.append(name)
|
||||
if "主演" in key or "演员" in key:
|
||||
for name in names:
|
||||
if name not in actor_items:
|
||||
actor_items.append(name)
|
||||
content = ""
|
||||
for text in root.xpath("//p/text()"):
|
||||
cleaned = self._clean_text(text)
|
||||
if len(cleaned) > 20:
|
||||
content = cleaned
|
||||
break
|
||||
remarks = self._extract_remark_from_title(title)
|
||||
body_text = self._clean_text("".join(root.xpath("//body//text()")))
|
||||
code_match = re.search(r"提取码[：:]\s*([a-zA-Z0-9]{4})", body_text)
|
||||
if code_match:
|
||||
remarks = f"{remarks} 提取码: {code_match.group(1)}".strip()
|
||||
play = self._build_play_data(self._extract_share_links(root))
|
||||
result["list"].append(
|
||||
{
|
||||
"vod_id": vod_id,
|
||||
"vod_name": self._clean_title_text(title),
|
||||
"vod_pic": self._build_url(
|
||||
"".join(
|
||||
root.xpath(
|
||||
"(//*[contains(@class,'tc-box') and contains(@class,'article-box')]//img)[1]/@src"
|
||||
)
|
||||
).strip()
|
||||
),
|
||||
"vod_remarks": remarks,
|
||||
"vod_director": ", ".join(director_items),
|
||||
"vod_actor": ", ".join(actor_items),
|
||||
"vod_content": content,
|
||||
"vod_play_from": play["vod_play_from"],
|
||||
"vod_play_url": play["vod_play_url"],
|
||||
}
|
||||
)
|
||||
return result
|
||||
|
||||
def playerContent(self, flag, id, vipFlags):
|
||||
raw = str(id or "").strip()
|
||||
if raw.startswith("push://"):
|
||||
return {"parse": 0, "url": raw[7:]}
|
||||
return {"parse": 0, "url": raw}
|
||||
Reference in New Issue
Block a user