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

feat: add wooyun list and search flows

Browse Source
This commit is contained in:
Harold
2026-04-19 13:33:46 +08:00
parent fac29bb064
commit 6fb92a7790
2 changed files with 169 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
py/tests/test_乌云影视.py
+85
View File
@@ -1,6 +1,7 @@
import unittest
from importlib.machinery import SourceFileLoader
from pathlib import Path
from unittest.mock import patch
ROOT = Path(__file__).resolve().parents[1]
@@ -108,6 +109,90 @@ class TestWooyunSpider(unittest.TestCase):
self.assertEqual(body["sortCode"], "hot")
self.assertEqual(body["topCode"], "movie")
def test_extract_home_list_deduplicates_media_resources(self):
home_blocks = {
"records": [
{
"mediaResources": [
{"id": 1, "title": "影片A", "posterUrl": "/a.jpg"},
{"id": 1, "title": "影片A", "posterUrl": "/a.jpg"},
]
},
{
"mediaResources": [
{"id": 2, "title": "影片B", "posterUrlS3": "https://img.example/b.jpg"},
]
},
]
}
items = self.spider._extract_home_list(home_blocks)
self.assertEqual([item["id"] for item in items], ["1", "2"])
def test_map_vod_picks_expected_fields(self):
item = {
"id": 7,
"title": "示例片",
"posterUrl": "/poster.jpg",
"mediaType": {"code": "movie", "name": "电影"},
"episodeStatus": "更新至4集",
"releaseYear": "2026",
"rating": "8.8",
"actors": ["甲", "乙"],
"directors": ["丙"],
}
vod = self.spider._map_vod(item)
self.assertEqual(vod["vod_id"], "7")
self.assertEqual(vod["vod_pic"], "/poster.jpg")
self.assertEqual(vod["vod_actor"], "甲/乙")
self.assertEqual(vod["vod_director"], "丙")
@patch.object(Spider, "_request_json")
def test_home_video_content_reads_home_blocks(self, mock_request_json):
mock_request_json.return_value = {
"records": [
{"mediaResources": [{"id": 1, "title": "首页片", "posterUrl": "/a.jpg"}]}
]
}
result = self.spider.homeVideoContent()
self.assertEqual(result["list"][0]["vod_id"], "1")
self.assertEqual(result["list"][0]["vod_name"], "首页片")
@patch.object(Spider, "_request_json")
def test_category_content_posts_search_body_and_returns_page_data(self, mock_request_json):
mock_request_json.return_value = {
"records": [{"id": 9, "title": "分类片", "posterUrl": "/cate.jpg"}],
"total": 30,
"pages": 2,
}
result = self.spider.categoryContent("movie", "2", False, {"sort": "latest"})
kwargs = mock_request_json.call_args.kwargs
self.assertEqual(kwargs["method"], "POST")
self.assertEqual(kwargs["data"]["pageIndex"], "2")
self.assertEqual(kwargs["data"]["sortCode"], "latest")
self.assertEqual(result["page"], 2)
self.assertEqual(result["pagecount"], 2)
self.assertEqual(result["limit"], 24)
self.assertEqual(result["total"], 30)
@patch.object(Spider, "_request_json")
def test_search_content_returns_empty_result_for_blank_keyword(self, mock_request_json):
result = self.spider.searchContent("", False, "1")
self.assertEqual(result, {"page": 1, "pagecount": 0, "total": 0, "list": []})
mock_request_json.assert_not_called()
@patch.object(Spider, "_request_json")
def test_search_content_posts_keyword_body(self, mock_request_json):
mock_request_json.return_value = {
"records": [{"id": 15, "title": "搜索片", "posterUrl": "/search.jpg"}],
"total": 1,
"pages": 1,
}
result = self.spider.searchContent("繁花", False, "1")
body = mock_request_json.call_args.kwargs["data"]
self.assertEqual(body["searchKey"], "繁花")
self.assertEqual(body["topCode"], "")
self.assertEqual(result["list"][0]["vod_id"], "15")
if __name__ == "__main__":
unittest.main()
py/乌云影视.py
+84
View File
@@ -1,5 +1,6 @@
# coding=utf-8
import json
import math
import sys
from base.spider import Spider as BaseSpider
@@ -184,7 +185,90 @@ class Spider(BaseSpider):
"topCode": self._map_top_code(category_id),
}
def _join_text(self, value):
return "/".join([self._stringify(item) for item in self._ensure_list(value) if self._stringify(item)])
def _pick_poster(self, item):
return self._stringify(
item.get("posterUrlS3") or item.get("posterUrl") or item.get("thumbUrlS3") or item.get("thumbUrl")
)
def _extract_home_list(self, home_blocks):
if isinstance(home_blocks, dict):
blocks = self._ensure_list(home_blocks.get("records"))
else:
blocks = self._ensure_list(home_blocks)
seen = set()
items = []
for block in blocks:
for item in self._ensure_list(block.get("mediaResources")):
vod_id = self._stringify(item.get("id"))
if not vod_id or vod_id in seen:
continue
seen.add(vod_id)
copied = dict(item)
copied["id"] = vod_id
items.append(copied)
return items
def _map_vod(self, item):
media_type = item.get("mediaType") or {}
return {
"vod_id": self._stringify(item.get("id")),
"vod_name": self._stringify(item.get("title")),
"vod_pic": self._pick_poster(item),
"type_id": self._stringify(media_type.get("code")),
"type_name": self._stringify(media_type.get("name")),
"vod_remarks": self._stringify(item.get("episodeStatus") or item.get("remark")),
"vod_year": self._stringify(item.get("releaseYear")),
"vod_douban_score": self._stringify(item.get("rating")),
"vod_actor": self._join_text(item.get("actors")),
"vod_director": self._join_text(item.get("directors")),
}
def homeContent(self, filter):
menu = self._request_json("/movie/category/menu")
classes = self._build_classes(menu)
return {"class": classes, "filters": self._build_filters(menu, classes)}
def homeVideoContent(self):
data = self._request_json(f"/movie/media/home/custom/classify/1/3?limit={self.home_limit}")
items = self._extract_home_list(data)[: self.home_limit]
return {"list": [self._map_vod(item) for item in items]}
def categoryContent(self, tid, pg, filter, extend):
page = int(pg)
data = self._request_json(
"/movie/media/search",
method="POST",
data=self._build_category_body(self._stringify(tid or "movie"), page, extend),
headers={"Content-Type": "application/json"},
) or {}
records = [self._map_vod(item) for item in self._ensure_list(data.get("records"))]
total = int(data.get("total") or 0)
pagecount = int(data.get("pages") or (math.ceil(total / self.page_size) if total else (page if records else 0)))
return {"page": page, "pagecount": pagecount, "limit": self.page_size, "total": total, "list": records}
def searchContent(self, key, quick, pg="1"):
page = int(pg)
keyword = self._stringify(key).strip()
if not keyword:
return {"page": page, "pagecount": 0, "total": 0, "list": []}
data = self._request_json(
"/movie/media/search",
method="POST",
data={
"menuCodeList": [],
"pageIndex": self._stringify(page),
"pageSize": self.page_size,
"searchKey": keyword,
"sortCode": "",
"topCode": "",
},
headers={"Content-Type": "application/json"},
) or {}
records = [self._map_vod(item) for item in self._ensure_list(data.get("records"))]
total = int(data.get("total") or 0)
pagecount = int(data.get("pages") or (math.ceil(total / self.page_size) if total else (page if records else 0)))
return {"page": page, "pagecount": pagecount, "total": total, "list": records}