yeyinly/tvboxzt
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

feat: add wanou aggregate category and failover helpers

Browse Source
This commit is contained in:
Harold
2026-04-20 14:07:39 +08:00
parent c9ef53592b
commit cd12177a3b
2 changed files with 179 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
py/tests/test_玩偶聚合.py
+74
View File
@@ -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="<html><body>ok</body></html>")
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 = """
<div class="module-item">
<div class="module-item-pic">
<a href="/voddetail/123.html"></a>
<img data-src="/poster.jpg" alt="示例影片" />
</div>
<div class="module-item-text">HD</div>
</div>
"""
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 = """
<div class="module-item">
<div class="module-item-pic">
<a href="/voddetail/456.html"></a>
<img data-src="/cate.jpg" alt="分类影片" />
</div>
<div class="module-item-text">更新至10集</div>
</div>
"""
result = self.spider.categoryContent("site_wanou", "2", False, {})
self.assertEqual(result["page"], 2)
self.assertEqual(result["list"][0]["vod_name"], "分类影片")
py/玩偶聚合.py
+105
View File
@@ -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)}