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

Merge branch 'siwan-spider'

Browse Source
This commit is contained in:
Harold
2026-04-24 18:22:25 +08:00
parent b89b4a9e79 434b30711d
commit e7ef9b7f25
2 changed files with 500 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
py/tests/test_四万影视.py
+205
View File
@@ -0,0 +1,205 @@
import json
import unittest
from importlib.machinery import SourceFileLoader
from pathlib import Path
from types import SimpleNamespace
from unittest.mock import patch
ROOT = Path(__file__).resolve().parents[1]
MODULE = SourceFileLoader("siwan_spider", str(ROOT / "四万影视.py")).load_module()
Spider = MODULE.Spider
class TestSiWanSpider(unittest.TestCase):
def setUp(self):
Spider._instance = None
self.spider = Spider()
self.spider.init()
def _response(self, status_code=200, payload=None, text=None):
body = text if text is not None else json.dumps(payload or {}, ensure_ascii=False)
return SimpleNamespace(status_code=status_code, text=body)
def test_home_content_exposes_expected_classes_and_filters(self):
content = self.spider.homeContent(False)
self.assertEqual(
[item["type_id"] for item in content["class"]],
["20", "30", "39", "45", "32"],
)
self.assertEqual(
[item["key"] for item in content["filters"]["20"]],
["subType", "year", "sort"],
)
self.assertEqual(content["filters"]["20"][0]["init"], "20")
self.assertEqual(content["filters"]["20"][1]["value"][0], {"n": "全部", "v": ""})
self.assertEqual(content["filters"]["20"][2]["value"][0], {"n": "时间", "v": "time"})
def test_home_video_content_returns_empty_list(self):
self.assertEqual(self.spider.homeVideoContent(), {"list": []})
@patch.object(Spider, "fetch")
def test_api_get_requests_maccms_endpoint_with_browser_headers(self, mock_fetch):
mock_fetch.return_value = self._response(payload={"list": []})
data = self.spider._api_get({"ac": "detail", "pg": 1})
self.assertEqual(data, {"list": []})
self.assertEqual(mock_fetch.call_args.args[0], "https://40000.me/api/maccms")
self.assertEqual(mock_fetch.call_args.kwargs["params"], {"ac": "detail", "pg": 1})
self.assertEqual(mock_fetch.call_args.kwargs["headers"]["Referer"], "https://40000.me/")
@patch.object(Spider, "fetch")
def test_api_get_rejects_non_200_and_non_object_body(self, mock_fetch):
mock_fetch.return_value = self._response(status_code=500, payload={"msg": "bad"})
with self.assertRaises(ValueError):
self.spider._api_get({"ac": "detail"})
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",
)
@patch.object(Spider, "_api_get")
def test_category_content_returns_empty_payload_on_api_error(self, mock_api_get):
mock_api_get.side_effect = ValueError("boom")
self.assertEqual(
self.spider.categoryContent("20", "1", False, {}),
{"page": 1, "limit": 0, "total": 0, "list": []},
)
@patch.object(Spider, "_api_get")
def test_detail_content_returns_empty_list_on_api_error(self, mock_api_get):
mock_api_get.side_effect = ValueError("boom")
self.assertEqual(self.spider.detailContent(["9"]), {"list": []})
def test_player_content_returns_direct_url_for_http_play_id(self):
result = self.spider.playerContent("量子", "https://cdn.example/test.m3u8", {})
self.assertEqual(result["parse"], 0)
self.assertEqual(result["jx"], 0)
self.assertEqual(result["url"], "https://cdn.example/test.m3u8")
self.assertEqual(result["header"]["Referer"], "https://40000.me/")
def test_player_content_marks_non_http_play_id_for_parse(self):
result = self.spider.playerContent("量子", "play-2", {})
self.assertEqual(result["parse"], 1)
self.assertEqual(result["jx"], 1)
self.assertEqual(result["url"], "play-2")
def test_player_content_returns_empty_parse_payload_for_blank_id(self):
self.assertEqual(
self.spider.playerContent("量子", "", {}),
{
"parse": 1,
"jx": 1,
"url": "",
"header": {
"User-Agent": self.spider.headers["User-Agent"],
"Referer": "https://40000.me/",
},
},
)
py/四万影视.py
+295
View File
@@ -0,0 +1,295 @@
# coding=utf-8
import json
import sys
from base.spider import Spider as BaseSpider
sys.path.append("..")
class Spider(BaseSpider):
def __init__(self):
self.name = "四万影视"
self.host = "https://40000.me"
self.api = self.host + "/api/maccms"
self.fallback_pic = self.host + "/public/favicon.png"
self.headers = {
"User-Agent": (
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
"AppleWebKit/537.36 (KHTML, like Gecko) "
"Chrome/146.0.0.0 Safari/537.36 Edg/146.0.0.0"
),
"Referer": self.host + "/",
"Accept": "*/*",
}
self.classes = [
{"type_id": "20", "type_name": "电影"},
{"type_id": "30", "type_name": "电视剧"},
{"type_id": "39", "type_name": "动漫"},
{"type_id": "45", "type_name": "综艺"},
{"type_id": "32", "type_name": "欧美"},
]
self.filter_type_options = {
"20": [
{"n": "全部", "v": "20"},
{"n": "动作片", "v": "21"},
{"n": "喜剧片", "v": "22"},
{"n": "恐怖片", "v": "23"},
{"n": "科幻片", "v": "24"},
{"n": "爱情片", "v": "25"},
{"n": "剧情片", "v": "26"},
{"n": "战争片", "v": "27"},
{"n": "纪录片", "v": "28"},
{"n": "理论片", "v": "29"},
{"n": "预告片", "v": "52"},
{"n": "电影解说", "v": "51"},
],
"30": [
{"n": "全部", "v": "30"},
{"n": "国产剧", "v": "31"},
{"n": "欧美剧", "v": "32"},
{"n": "香港剧", "v": "33"},
{"n": "韩国剧", "v": "34"},
{"n": "台湾剧", "v": "35"},
{"n": "日本剧", "v": "36"},
{"n": "海外剧", "v": "37"},
{"n": "泰国剧", "v": "38"},
{"n": "短剧大全", "v": "58"},
],
"39": [
{"n": "全部", "v": "39"},
{"n": "国产动漫", "v": "40"},
{"n": "日韩动漫", "v": "41"},
{"n": "欧美动漫", "v": "42"},
{"n": "港台动漫", "v": "43"},
{"n": "海外动漫", "v": "44"},
{"n": "动画片", "v": "50"},
],
"45": [
{"n": "全部", "v": "45"},
{"n": "大陆综艺", "v": "46"},
{"n": "港台综艺", "v": "47"},
{"n": "日韩综艺", "v": "48"},
{"n": "欧美综艺", "v": "49"},
],
"32": [
{"n": "全部", "v": "32"},
{"n": "欧美剧", "v": "32"},
{"n": "欧美动漫", "v": "42"},
{"n": "欧美综艺", "v": "49"},
{"n": "海外剧", "v": "37"},
],
}
self.type_name_map = {item["type_id"]: item["type_name"] for item in self.classes}
self.subtype_parent_map = {
"20": "20",
"21": "20",
"22": "20",
"23": "20",
"24": "20",
"25": "20",
"26": "20",
"27": "20",
"28": "20",
"29": "20",
"51": "20",
"52": "20",
"30": "30",
"31": "30",
"32": "32",
"33": "30",
"34": "30",
"35": "30",
"36": "30",
"37": "32",
"38": "30",
"58": "30",
"39": "39",
"40": "39",
"41": "39",
"42": "32",
"43": "39",
"44": "39",
"50": "39",
"45": "45",
"46": "45",
"47": "45",
"48": "45",
"49": "32",
}
self.filters = self._build_filters()
def init(self, extend=""):
return None
def getName(self):
return self.name
def _build_filters(self):
years = [{"n": "全部", "v": ""}]
for year in range(2026, 1999, -1):
years.append({"n": str(year), "v": str(year)})
sort_values = [
{"n": "时间", "v": "time"},
{"n": "人气", "v": "hits"},
{"n": "评分", "v": "score"},
{"n": "点赞", "v": "up"},
]
filters = {}
for item in self.classes:
type_id = item["type_id"]
filters[type_id] = [
{"key": "subType", "name": "分类", "init": type_id, "value": self.filter_type_options[type_id]},
{"key": "year", "name": "年代", "init": "", "value": years},
{"key": "sort", "name": "排序", "init": "time", "value": list(sort_values)},
]
return filters
def homeContent(self, filter):
return {"class": self.classes, "filters": self.filters}
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,
params=params,
headers=dict(self.headers),
timeout=10,
)
if response.status_code != 200:
raise ValueError("api request failed")
data = json.loads(response.text or "{}")
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"],
}
]
}
def playerContent(self, flag, id, vipFlags):
play_id = str(id or "").strip()
header = {
"User-Agent": self.headers["User-Agent"],
"Referer": self.host + "/",
}
if not play_id:
return {"parse": 1, "jx": 1, "url": "", "header": header}
if play_id.startswith("http://") or play_id.startswith("https://"):
return {"parse": 0, "jx": 0, "url": play_id, "header": header}
return {"parse": 1, "jx": 1, "url": play_id, "header": header}