feat: add 四万影视 content parsing

This commit is contained in:
Harold
2026-04-24 18:19:06 +08:00
parent c17dd94b2d
commit 11ba774ecb
2 changed files with 224 additions and 0 deletions
+107
View File
@@ -56,3 +56,110 @@ class TestSiWanSpider(unittest.TestCase):
mock_fetch.return_value = self._response(text="[]")
with self.assertRaises(ValueError):
self.spider._api_get({"ac": "detail"})
def test_normalize_vod_applies_type_name_actor_cleanup_and_fallback_cover(self):
self.assertEqual(
self.spider._normalize_vod(
{
"vod_id": 7,
"vod_name": "示例影片",
"vod_pic": "",
"type_id": "20",
"vod_remarks": "HD",
"vod_year": "2025",
"vod_actor": "O'Brien , 张三",
}
),
{
"vod_id": "7",
"vod_name": "示例影片",
"vod_pic": "https://40000.me/public/favicon.png",
"vod_remarks": "HD",
"vod_year": "2025",
"type_name": "电影",
"vod_actor": "O'Brien, 张三",
},
)
@patch.object(Spider, "_api_get")
def test_category_content_maps_subtype_year_sort_and_page(self, mock_api_get):
mock_api_get.return_value = {
"page": 2,
"total": 55,
"list": [
{"vod_id": 100, "vod_name": "分类影片", "vod_pic": "/p.jpg", "type_id": "21", "vod_remarks": "更新中"}
],
}
result = self.spider.categoryContent("20", "2", False, {"subType": "21", "year": "2025", "sort": "hits"})
self.assertEqual(
mock_api_get.call_args.args[0],
{"ac": "detail", "t": "21", "pg": 2, "h": "2025", "by": "hits"},
)
self.assertEqual(result["page"], 2)
self.assertEqual(result["total"], 55)
self.assertEqual(result["limit"], 1)
self.assertEqual(result["list"][0]["vod_id"], "100")
self.assertNotIn("pagecount", result)
@patch.object(Spider, "_api_get")
def test_search_content_maps_keyword_and_page(self, mock_api_get):
mock_api_get.return_value = {
"page": 3,
"total": 10,
"list": [
{"vod_id": 200, "vod_name": "搜索结果", "vod_pic": "", "type_id": "30", "vod_remarks": "完结"}
],
}
result = self.spider.searchContent("繁花", False, "3")
self.assertEqual(mock_api_get.call_args.args[0], {"ac": "detail", "wd": "繁花", "pg": 3})
self.assertEqual(result["page"], 3)
self.assertEqual(result["list"][0]["type_name"], "电视剧")
def test_search_content_returns_empty_list_for_blank_keyword(self):
self.assertEqual(self.spider.searchContent("", False, "1"), {"page": 1, "limit": 0, "total": 0, "list": []})
def test_parse_play_groups_builds_repo_style_play_fields(self):
self.assertEqual(
self.spider._parse_play_groups(
{
"vod_play_from": "量子$$$",
"vod_play_url": "第1集$https://cdn.example/1.m3u8#第2集$abc$$$solo",
}
),
{
"vod_play_from": "量子$$$线路2",
"vod_play_url": "第1集$https://cdn.example/1.m3u8#第2集$abc$$$第1集$solo",
},
)
@patch.object(Spider, "_api_get")
def test_detail_content_builds_detail_and_play_fields(self, mock_api_get):
mock_api_get.return_value = {
"list": [
{
"vod_id": 300,
"vod_name": "详情影片",
"vod_pic": "/poster.jpg",
"type_id": "30",
"type_name": "",
"vod_year": "2024",
"vod_area": "大陆",
"vod_director": "导演甲",
"vod_actor": "A'B , 演员乙",
"vod_blurb": "一段简介",
"vod_remarks": "更新至2集",
"vod_play_from": "量子$$$非凡",
"vod_play_url": "第1集$https://cdn.example/1.m3u8#第2集$play-2$$$正片$play-3",
}
]
}
result = self.spider.detailContent(["300"])
vod = result["list"][0]
self.assertEqual(vod["vod_id"], "300")
self.assertEqual(vod["type_name"], "电视剧")
self.assertEqual(vod["vod_actor"], "A'B, 演员乙")
self.assertEqual(vod["vod_play_from"], "量子$$$非凡")
self.assertEqual(
vod["vod_play_url"],
"第1集$https://cdn.example/1.m3u8#第2集$play-2$$$正片$play-3",
)
+117
View File
@@ -151,6 +151,60 @@ class Spider(BaseSpider):
def homeVideoContent(self):
return {"list": []}
def _type_name_by_id(self, type_id):
raw = str(type_id or "")
parent = self.subtype_parent_map.get(raw, raw)
return self.type_name_map.get(parent, "")
def _normalize_actor(self, value):
return str(value or "").replace("'", "'").replace(" , ", ", ")
def _normalize_vod(self, item):
return {
"vod_id": str(item.get("vod_id", "")),
"vod_name": str(item.get("vod_name", "")),
"vod_pic": str(item.get("vod_pic") or self.fallback_pic),
"vod_remarks": str(item.get("vod_remarks", "")),
"vod_year": str(item.get("vod_year", "")),
"type_name": str(item.get("type_name") or self._type_name_by_id(item.get("type_id"))),
"vod_actor": self._normalize_actor(item.get("vod_actor", "")),
}
def _page_result(self, items, page, total=0):
return {
"page": int(page),
"limit": len(items),
"total": int(total) if total else len(items),
"list": items,
}
def _parse_play_groups(self, item):
from_list = str(item.get("vod_play_from", "")).split("$$$")
url_list = str(item.get("vod_play_url", "")).split("$$$")
final_from = []
final_url = []
for index, raw_urls in enumerate(url_list):
raw_urls = str(raw_urls or "").strip()
if not raw_urls:
continue
line_name = str(from_list[index] if index < len(from_list) else "").strip() or f"线路{index + 1}"
episodes = []
for episode_index, part in enumerate(raw_urls.split("#")):
chunks = str(part or "").split("$", 1)
if len(chunks) > 1:
title = chunks[0].strip()
play_id = chunks[1].strip()
else:
title = ""
play_id = chunks[0].strip() if chunks else ""
title = title or f"{episode_index + 1}"
if play_id:
episodes.append(f"{title}${play_id}")
if episodes:
final_from.append(line_name)
final_url.append("#".join(episodes))
return {"vod_play_from": "$$$".join(final_from), "vod_play_url": "$$$".join(final_url)}
def _api_get(self, params):
response = self.fetch(
self.api,
@@ -164,3 +218,66 @@ class Spider(BaseSpider):
if not isinstance(data, dict):
raise ValueError("api response is not object")
return data
def categoryContent(self, tid, pg, filter, extend):
page = max(int(pg), 1)
values = dict(extend or {})
params = {
"ac": "detail",
"t": str(values.get("subType") or tid),
"pg": page,
"by": str(values.get("sort") or "time"),
}
year = str(values.get("year") or "").strip()
if year:
params["h"] = year
try:
data = self._api_get(params)
except ValueError:
return {"page": page, "limit": 0, "total": 0, "list": []}
items = [self._normalize_vod(item) for item in data.get("list", [])]
return self._page_result(items, data.get("page", page), data.get("total", 0))
def searchContent(self, key, quick, pg="1"):
page = max(int(pg), 1)
keyword = str(key or "").strip()
if not keyword:
return {"page": page, "limit": 0, "total": 0, "list": []}
try:
data = self._api_get({"ac": "detail", "wd": keyword, "pg": page})
except ValueError:
return {"page": page, "limit": 0, "total": 0, "list": []}
items = [self._normalize_vod(item) for item in data.get("list", [])]
return self._page_result(items, data.get("page", page), data.get("total", 0))
def detailContent(self, ids):
vod_id = str(ids[0] if isinstance(ids, list) and ids else ids or "").strip()
if not vod_id:
return {"list": []}
try:
data = self._api_get({"ac": "detail", "ids": vod_id})
except ValueError:
return {"list": []}
item = (data.get("list") or [None])[0]
if not item:
return {"list": []}
normalized = self._normalize_vod(item)
play_data = self._parse_play_groups(item)
return {
"list": [
{
"vod_id": normalized["vod_id"],
"vod_name": normalized["vod_name"],
"vod_pic": normalized["vod_pic"],
"type_name": normalized["type_name"],
"vod_year": str(item.get("vod_year", "")),
"vod_area": str(item.get("vod_area", "")),
"vod_director": str(item.get("vod_director", "")),
"vod_actor": normalized["vod_actor"],
"vod_content": str(item.get("vod_blurb") or item.get("vod_content") or ""),
"vod_remarks": str(item.get("vod_remarks", "")),
"vod_play_from": play_data["vod_play_from"],
"vod_play_url": play_data["vod_play_url"],
}
]
}