feat: scaffold wanou aggregate spider

py/tests/test_玩偶聚合.py

@@ -0,0 +1,76 @@
+import base64
+import json
+import unittest
+from importlib.machinery import SourceFileLoader
+from pathlib import Path
+from unittest.mock import patch
+
+
+ROOT = Path(__file__).resolve().parents[1]
+MODULE = SourceFileLoader("wanou_aggregate_spider", str(ROOT / "玩偶聚合.py")).load_module()
+Spider = MODULE.Spider
+
+
+class TestWanouAggregateSpider(unittest.TestCase):
+    def setUp(self):
+        Spider._instance = None
+        self.spider = Spider()
+        self.spider.init()
+
+    def test_home_content_exposes_site_classes_and_category_filter(self):
+        content = self.spider.homeContent(False)
+        self.assertEqual(
+            [item["type_id"] for item in content["class"][:3]],
+            ["site_wanou", "site_muou", "site_labi"],
+        )
+        self.assertEqual(content["class"][0]["type_name"], "玩偶")
+        self.assertEqual(content["filters"]["site_wanou"][0]["key"], "categoryId")
+        self.assertEqual(content["filters"]["site_wanou"][0]["value"][1], {"n": "电影", "v": "1"})
+
+    @patch.object(
+        Spider,
+        "_load_local_filter_groups",
+        return_value=[
+            {
+                "key": "year",
+                "name": "年份",
+                "init": "",
+                "value": [{"n": "全部", "v": ""}, {"n": "2025", "v": "2025"}],
+            }
+        ],
+    )
+    def test_home_content_appends_local_filter_groups_after_category_filter(self, mock_load_local_filter_groups):
+        content = self.spider.homeContent(False)
+        self.assertEqual(
+            [item["key"] for item in content["filters"]["site_wanou"][:2]],
+            ["categoryId", "year"],
+        )
+
+    def test_encode_and_decode_site_vod_id_round_trip(self):
+        vod_id = self.spider._encode_site_vod_id("wanou", "/voddetail/12345.html")
+        self.assertEqual(vod_id, "site:wanou:/voddetail/12345.html")
+        self.assertEqual(
+            self.spider._decode_site_vod_id(vod_id),
+            {"site": "wanou", "path": "/voddetail/12345.html"},
+        )
+
+    def test_encode_and_decode_aggregate_vod_id_round_trip(self):
+        payload = [
+            {"site": "wanou", "path": "/voddetail/1.html", "name": "繁花", "year": "2024"},
+            {"site": "muou", "path": "/voddetail/2.html", "name": "繁花", "year": "2024"},
+        ]
+        vod_id = self.spider._encode_aggregate_vod_id(payload)
+        self.assertTrue(vod_id.startswith("agg:"))
+        decoded = self.spider._decode_aggregate_vod_id(vod_id)
+        self.assertEqual(decoded, payload)
+
+    def test_normalize_title_removes_spaces_punctuation_and_resolution_tags(self):
+        self.assertEqual(
+            self.spider._normalize_title(" 繁花 4K.HDR-玩偶 "),
+            "繁花",
+        )
+
+    def test_is_same_title_rejects_year_conflict(self):
+        left = {"vod_name": "繁花", "vod_year": "2024"}
+        right = {"vod_name": "繁花", "vod_year": "2023"}
+        self.assertFalse(self.spider._is_same_title(left, right))

py/玩偶聚合.py

@@ -0,0 +1,118 @@
+# coding=utf-8
+import base64
+import json
+import os
+import re
+import sys
+
+from base.spider import Spider as BaseSpider
+
+sys.path.append("..")
+
+
+class Spider(BaseSpider):
+    def __init__(self):
+        self.name = "玩偶聚合"
+        self.filter_root = os.path.join(os.path.dirname(__file__), "../筛选")
+        self.headers = {
+            "User-Agent": (
+                "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
+                "AppleWebKit/537.36 (KHTML, like Gecko) "
+                "Chrome/136.0.0.0 Safari/537.36"
+            )
+        }
+        self.site_priority = [
+            "wanou",
+            "muou",
+            "labi",
+            "zhizhen",
+            "erxiao",
+            "huban",
+            "kuaiying",
+            "shandian",
+            "ouge",
+        ]
+        self.sites = [
+            {
+                "id": "wanou",
+                "name": "玩偶",
+                "domains": ["https://www.wogg.net"],
+                "filter_files": ["wogg.json"],
+                "default_categories": [("1", "电影"), ("2", "电视剧"), ("3", "动漫"), ("4", "综艺")],
+            },
+            {
+                "id": "muou",
+                "name": "木偶",
+                "domains": ["https://www.muou.site"],
+                "filter_files": ["mogg.json"],
+                "default_categories": [("1", "电影"), ("2", "电视剧"), ("3", "动漫"), ("29", "综艺")],
+            },
+            {
+                "id": "labi",
+                "name": "蜡笔",
+                "domains": ["http://xiaocge.fun"],
+                "filter_files": ["labi.json"],
+                "default_categories": [("1", "电影"), ("2", "电视剧"), ("3", "动漫"), ("4", "综艺")],
+            },
+        ]
+
+    def init(self, extend=""):
+        return None
+
+    def getName(self):
+        return self.name
+
+    def homeVideoContent(self):
+        return {"list": []}
+
+    def _load_local_filter_groups(self, site):
+        return []
+
+    def _build_site_filters(self, site):
+        groups = [
+            {
+                "key": "categoryId",
+                "name": "分类",
+                "init": site["default_categories"][0][0],
+                "value": [{"n": "全部", "v": ""}]
+                + [{"n": name, "v": cid} for cid, name in site["default_categories"]],
+            }
+        ]
+        groups.extend(self._load_local_filter_groups(site))
+        return groups
+
+    def homeContent(self, filter):
+        classes = [{"type_id": f"site_{site['id']}", "type_name": site["name"]} for site in self.sites]
+        filters = {f"site_{site['id']}": self._build_site_filters(site) for site in self.sites}
+        return {"class": classes, "filters": filters}
+
+    def _encode_site_vod_id(self, site_id, path):
+        return f"site:{site_id}:{path}"
+
+    def _decode_site_vod_id(self, value):
+        prefix, site_id, path = str(value).split(":", 2)
+        if prefix != "site":
+            raise ValueError("not site vod id")
+        return {"site": site_id, "path": path}
+
+    def _encode_aggregate_vod_id(self, payload):
+        raw = json.dumps(payload, ensure_ascii=False, separators=(",", ":")).encode("utf-8")
+        return "agg:" + base64.urlsafe_b64encode(raw).decode("utf-8").rstrip("=")
+
+    def _decode_aggregate_vod_id(self, value):
+        encoded = str(value)[4:]
+        padded = encoded + "=" * (-len(encoded) % 4)
+        return json.loads(base64.urlsafe_b64decode(padded.encode("utf-8")).decode("utf-8"))
+
+    def _normalize_title(self, value):
+        text = str(value or "").lower()
+        text = re.sub(r"(4k|hdr|2160p|1080p|720p|玩偶|木偶|蜡笔)", "", text, flags=re.I)
+        text = re.sub(r"[\s\-_.·,，。!！?:：()（）\[\]]+", "", text)
+        return text
+
+    def _is_same_title(self, left, right):
+        left_year = str(left.get("vod_year") or "").strip()
+        right_year = str(right.get("vod_year") or "").strip()
+        if left_year and right_year and left_year != right_year:
+            return False
+        return self._normalize_title(left.get("vod_name")) == self._normalize_title(right.get("vod_name"))