diff --git a/py/tests/test_玩偶聚合.py b/py/tests/test_玩偶聚合.py index b5b7d8d..25fe08e 100644 --- a/py/tests/test_玩偶聚合.py +++ b/py/tests/test_玩偶聚合.py @@ -3,6 +3,7 @@ import json import unittest from importlib.machinery import SourceFileLoader from pathlib import Path +from types import SimpleNamespace from unittest.mock import patch @@ -74,3 +75,76 @@ class TestWanouAggregateSpider(unittest.TestCase): left = {"vod_name": "繁花", "vod_year": "2024"} right = {"vod_name": "繁花", "vod_year": "2023"} self.assertFalse(self.spider._is_same_title(left, right)) + + def test_build_category_url_uses_selected_category_and_filters(self): + site = { + "id": "wanou", + "domains": ["https://www.wogg.net"], + "category_url": "/vodshow/{categoryId}--------{page}---.html", + "category_url_with_filters": "/vodshow/{categoryId}-{area}-{by}-{class}-----{page}---{year}.html", + } + url = self.spider._build_category_url( + site, + "1", + "2", + {"categoryId": "1", "area": "香港", "by": "score", "class": "动作", "year": "2025"}, + ) + self.assertEqual( + url, + "https://www.wogg.net/vodshow/1-%E9%A6%99%E6%B8%AF-score-%E5%8A%A8%E4%BD%9C-----2---2025.html", + ) + + @patch.object(Spider, "fetch") + def test_request_with_failover_tries_next_domain_when_first_fails(self, mock_fetch): + def fake_fetch(url, headers=None, timeout=10): + if url.startswith("https://bad.example"): + raise RuntimeError("boom") + return SimpleNamespace(status_code=200, text="ok") + + mock_fetch.side_effect = fake_fetch + site = {"domains": ["https://bad.example", "https://good.example"]} + html = self.spider._request_with_failover(site, "/vodshow/1--------1---.html") + self.assertIn("ok", html) + self.assertEqual(site["domains"][0], "https://good.example") + + def test_parse_cards_extracts_short_site_vod_id_title_cover_and_remarks(self): + site = {"id": "wanou", "domains": ["https://www.wogg.net"], "list_xpath": "//*[contains(@class,'module-item')]"} + html = """ +
+
+ + 示例影片 +
+
HD
+
+ """ + cards = self.spider._parse_cards(site, html) + self.assertEqual( + cards, + [ + { + "vod_id": "site:wanou:/voddetail/123.html", + "vod_name": "示例影片", + "vod_pic": "https://www.wogg.net/poster.jpg", + "vod_remarks": "HD", + "vod_year": "", + "_site": "wanou", + "_detail_path": "/voddetail/123.html", + } + ], + ) + + @patch.object(Spider, "_request_with_failover") + def test_category_content_uses_default_category_when_extend_missing(self, mock_request_with_failover): + mock_request_with_failover.return_value = """ +
+
+ + 分类影片 +
+
更新至10集
+
+ """ + result = self.spider.categoryContent("site_wanou", "2", False, {}) + self.assertEqual(result["page"], 2) + self.assertEqual(result["list"][0]["vod_name"], "分类影片") diff --git a/py/玩偶聚合.py b/py/玩偶聚合.py index febd6e9..ccbfb0c 100644 --- a/py/玩偶聚合.py +++ b/py/玩偶聚合.py @@ -4,6 +4,7 @@ import json import os import re import sys +from urllib.parse import quote from base.spider import Spider as BaseSpider @@ -38,6 +39,9 @@ class Spider(BaseSpider): "name": "玩偶", "domains": ["https://www.wogg.net"], "filter_files": ["wogg.json"], + "list_xpath": "//*[contains(@class,'module-item')]", + "category_url": "/vodshow/{categoryId}--------{page}---.html", + "category_url_with_filters": "/vodshow/{categoryId}-{area}-{by}-{class}-----{page}---{year}.html", "default_categories": [("1", "电影"), ("2", "电视剧"), ("3", "动漫"), ("4", "综艺")], }, { @@ -45,6 +49,8 @@ class Spider(BaseSpider): "name": "木偶", "domains": ["https://www.muou.site"], "filter_files": ["mogg.json"], + "list_xpath": "//*[contains(@class,'module-item')]", + "category_url": "/vodshow/{categoryId}--------{page}---.html", "default_categories": [("1", "电影"), ("2", "电视剧"), ("3", "动漫"), ("29", "综艺")], }, { @@ -52,6 +58,8 @@ class Spider(BaseSpider): "name": "蜡笔", "domains": ["http://xiaocge.fun"], "filter_files": ["labi.json"], + "list_xpath": "//*[contains(@class,'module-item')]", + "category_url": "/vodshow/{categoryId}--------{page}---.html", "default_categories": [("1", "电影"), ("2", "电视剧"), ("3", "动漫"), ("4", "综艺")], }, ] @@ -116,3 +124,100 @@ class Spider(BaseSpider): 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")) + + def _get_site(self, site_id): + for site in self.sites: + if site["id"] == site_id: + return site + return None + + def _build_absolute_url(self, base, path): + raw = str(path or "").strip() + if not raw: + return "" + if raw.startswith(("http://", "https://")): + return raw + if raw.startswith("//"): + return "https:" + raw + return str(base).rstrip("/") + "/" + raw.lstrip("/") + + def _build_category_url(self, site, category_id, pg, extend): + values = dict(extend or {}) + values.setdefault("categoryId", category_id) + values.setdefault("area", "") + values.setdefault("by", values.get("sort", "")) + values.setdefault("class", "") + values.setdefault("year", "") + if site.get("category_url_with_filters") and any(values.get(key) for key in ("area", "by", "class", "year")): + path = site["category_url_with_filters"].format( + **{ + "categoryId": values["categoryId"], + "area": quote(str(values["area"])), + "by": quote(str(values["by"])), + "class": quote(str(values["class"])), + "page": int(pg), + "year": quote(str(values["year"])), + } + ) + else: + path = site["category_url"].format(categoryId=values["categoryId"], page=int(pg)) + return self._build_absolute_url(site["domains"][0], path) + + def _request_with_failover(self, site, path_or_url, referer=None): + last_error = None + for index, domain in enumerate(list(site["domains"])): + target = path_or_url if str(path_or_url).startswith("http") else self._build_absolute_url(domain, path_or_url) + try: + headers = dict(self.headers) + headers["Referer"] = referer or self._build_absolute_url(domain, "/") + response = self.fetch(target, headers=headers, timeout=10) + if response.status_code == 200 and response.text: + if index > 0: + site["domains"].insert(0, site["domains"].pop(index)) + return response.text + except Exception as exc: + last_error = exc + raise RuntimeError(str(last_error or "all domains failed")) + + def _parse_cards(self, site, html): + root = self.html(html) + if root is None: + return [] + + items = [] + seen = set() + for card in root.xpath(site["list_xpath"]): + href = ((card.xpath(".//a[@href][1]/@href") or [""])[0]).strip() + title = ( + ((card.xpath(".//img[@alt][1]/@alt") or [""])[0]).strip() + or ((card.xpath(".//a[@title][1]/@title") or [""])[0]).strip() + ) + pic = ( + ((card.xpath(".//img[@data-src][1]/@data-src") or [""])[0]).strip() + or ((card.xpath(".//img[@src][1]/@src") or [""])[0]).strip() + ) + remarks = "".join(card.xpath(".//*[contains(@class,'module-item-text')][1]//text()")).strip() + if not href or not title or href in seen: + continue + seen.add(href) + items.append( + { + "vod_id": self._encode_site_vod_id(site["id"], href), + "vod_name": title, + "vod_pic": self._build_absolute_url(site["domains"][0], pic), + "vod_remarks": remarks, + "vod_year": "", + "_site": site["id"], + "_detail_path": href, + } + ) + return items + + def categoryContent(self, tid, pg, filter, extend): + site_id = str(tid).replace("site_", "", 1) + site = self._get_site(site_id) + values = extend if isinstance(extend, dict) else {} + category_id = values.get("categoryId") or site["default_categories"][0][0] + html = self._request_with_failover(site, self._build_category_url(site, category_id, pg, values)) + items = self._parse_cards(site, html) + return {"list": items, "page": int(pg), "limit": len(items), "total": int(pg) * 20 + len(items)}