Merge branch 'yisou-spider'

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 = """
+        <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()

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")

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}