diff --git a/py/tests/test_奕搜.py b/py/tests/test_奕搜.py
new file mode 100644
index 0000000..05f0369
--- /dev/null
+++ b/py/tests/test_奕搜.py
@@ -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 = """
+
+ """
+ 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 = """
+
+ """
+ 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 = """
+
+ """
+ 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 = """
+
+ 示例详情 [更12] [2025]
+
+
+ 这是一段足够长的剧情简介,用于测试详情页内容提取逻辑。
+ 夸克
+ 百度
+ 提取码:a1b2
+
+ """
+ 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()
diff --git a/py/tests/test_玩偶聚合.py b/py/tests/test_玩偶聚合.py
index dfcd071..5f5d9d0 100644
--- a/py/tests/test_玩偶聚合.py
+++ b/py/tests/test_玩偶聚合.py
@@ -22,7 +22,7 @@ class TestWanouAggregateSpider(unittest.TestCase):
content = self.spider.homeContent(False)
self.assertEqual(
[item["type_id"] for item in content["class"][:3]],
- ["site_wanou", "site_muou", "site_labi"],
+ ["site_wanou", "site_zhizhen", "site_shandian"],
)
self.assertEqual(content["class"][0]["type_name"], "玩偶")
self.assertEqual(content["filters"]["site_wanou"][0]["key"], "categoryId")
diff --git a/py/奕搜.py b/py/奕搜.py
new file mode 100644
index 0000000..e07cd76
--- /dev/null
+++ b/py/奕搜.py
@@ -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}