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

fix: rename bubuyingshi spider

Browse Source
This commit is contained in:
Harold
2026-04-24 21:24:01 +08:00
parent 04c3e45a88
commit 61c7c2a49f
2 changed files with 78 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files
py/tests/test_步步影视.py → py/tests/test_布布影视.py
+29 -7
View File
@@ -5,7 +5,7 @@ from unittest.mock import patch
ROOT = Path(__file__).resolve().parents[1]
MODULE = SourceFileLoader("bbys_spider", str(ROOT / "步步影视.py")).load_module()
MODULE = SourceFileLoader("bbys_spider", str(ROOT / "布布影视.py")).load_module()
Spider = MODULE.Spider
@@ -153,7 +153,7 @@ class TestBuBuYingShiSpider(unittest.TestCase):
self.assertIn("正片$线路二@1@/movie/main", data["vod_play_url"])
@patch.object(Spider, "_request_json")
def test_detail_content_handles_array_payload_and_packs_play_urls(self, mock_request_json):
def test_detail_content_uses_vodplayer_from_codes_and_show_names(self, mock_request_json):
mock_request_json.return_value = {
"data": [
{
@@ -167,18 +167,19 @@ class TestBuBuYingShiSpider(unittest.TestCase):
"vod_director": "丙",
"vod_content": "<p>一段简介</p>",
"vod_class": "剧情",
"vod_play_from": "线路一",
"vod_play_url": "第1集$/play/1",
"vod_play_from": "rose",
"vod_play_url": "第1集$rose_token",
}
]
],
"vodplayer": [{"from": "rose", "show": "SE蓝光", "decode_status": "1"}],
}
result = self.spider.detailContent(["18"])
detail = result["list"][0]
self.assertEqual(detail["vod_id"], "18")
self.assertEqual(detail["vod_content"], "一段简介")
self.assertEqual(detail["type_name"], "剧情")
self.assertEqual(detail["vod_play_from"], "线路一(1)")
self.assertEqual(detail["vod_play_url"], "第1集$线路一@1@/play/1")
self.assertEqual(detail["vod_play_from"], "SE蓝光(1)")
self.assertEqual(detail["vod_play_url"], "第1集$rose@1@rose_token")
def test_extract_decode_url_accepts_multiple_shapes(self):
self.assertEqual(self.spider._extract_decode_url({"data": "https://a.example/m3u8"}), "https://a.example/m3u8")
@@ -199,6 +200,15 @@ class TestBuBuYingShiSpider(unittest.TestCase):
self.assertEqual(result["playUrl"], "")
self.assertEqual(result["url"], "https://cdn.example/real.m3u8")
@patch.object(Spider, "_request_json")
def test_player_content_parses_non_binary_decode_status(self, mock_request_json):
mock_request_json.return_value = {"data": {"url": "https://cdn.example/jd4k.m3u8"}}
result = self.spider.playerContent("JD蓝光", "JD4K@2@JD-a8fa0064", {})
kwargs = mock_request_json.call_args.kwargs
self.assertEqual(kwargs["params"]["url"], "JD-a8fa0064")
self.assertEqual(kwargs["params"]["vodFrom"], "JD4K")
self.assertEqual(result["url"], "https://cdn.example/jd4k.m3u8")
@patch.object(Spider, "_request_json")
def test_player_content_sets_jx_for_major_video_sites(self, mock_request_json):
mock_request_json.return_value = {"data": {"url": "https://v.qq.com/x/cover/demo.html"}}
@@ -227,6 +237,18 @@ class TestBuBuYingShiSpider(unittest.TestCase):
result = self.spider.playerContent("线路一", "线路一@1@/play/404", {})
self.assertEqual(result["url"], "/play/404")
@patch.object(Spider, "_request_json")
def test_player_content_marks_disabled_decode_line_as_unplayable(self, mock_request_json):
mock_request_json.return_value = {
"code": 0,
"msg": "线路暂时停用，请稍后再试",
"data": "JD-2eb205f0c280bdb5b93e14c25d5e72140",
}
result = self.spider.playerContent("JD蓝光", "JD4K@2@JD-2eb205f0c280bdb5b93e14c25d5e72140", {})
self.assertEqual(result["parse"], 1)
self.assertEqual(result["jx"], 0)
self.assertEqual(result["url"], "")
if __name__ == "__main__":
unittest.main()
py/步步影视.py → py/布布影视.py
+49 -7
View File
@@ -12,7 +12,7 @@ sys.path.append("..")
class Spider(BaseSpider):
def __init__(self):
self.name = "步步影视"
self.name = "布布影视"
self.host = "https://bbys.app"
self.pkg = "com.sunshine.tv"
self.ver = "4"
@@ -249,7 +249,11 @@ class Spider(BaseSpider):
detail = detail[0] if detail else None
if not detail:
return {"list": []}
play_data = self._build_play_data(detail.get("vod_play_from", ""), detail.get("vod_play_url", ""))
play_data = self._build_play_data(
detail.get("vod_play_from", ""),
detail.get("vod_play_url", ""),
payload.get("vodplayer") or [],
)
return {
"list": [
{
@@ -290,10 +294,10 @@ class Spider(BaseSpider):
play_from = ""
play_url = raw
need_decode = False
matched = re.match(r"^([^@]+)@([01])@([\s\S]+)$", raw)
matched = re.match(r"^([^@]+)@([^@]+)@([\s\S]+)$", raw)
if matched:
play_from = matched.group(1)
need_decode = matched.group(2) == "1"
need_decode = matched.group(2).strip() not in ("", "0")
play_url = matched.group(3)
elif not raw.startswith(("http://", "https://")):
need_decode = True
@@ -306,6 +310,14 @@ class Spider(BaseSpider):
params={"url": play_url, "vodFrom": play_from},
)
decoded = self._extract_decode_url(payload)
if self._is_disabled_decode_result(payload, play_url, decoded):
return {
"parse": 1,
"playUrl": "",
"url": "",
"jx": 0,
"header": {"User-Agent": "okhttp/4.12.0"},
}
if decoded:
final_url = decoded
@@ -339,9 +351,10 @@ class Spider(BaseSpider):
return str(value).split(sep)
return []
def _build_play_data(self, vod_play_from, vod_play_url):
def _build_play_data(self, vod_play_from, vod_play_url, vodplayer=None):
from_list = self._safe_split(vod_play_from, "$$$")
url_groups = self._safe_split(vod_play_url, "$$$")
player_meta = self._build_player_meta(vodplayer)
final_from = []
final_url = []
for index in range(max(len(from_list), len(url_groups))):
@@ -349,6 +362,8 @@ class Spider(BaseSpider):
group = (url_groups[index] if index < len(url_groups) else "").strip()
if not group:
continue
meta = player_meta.get(raw_from) or {}
display_from = self._stringify(meta.get("show")).strip() or raw_from
episodes = []
for item in [part.strip() for part in group.split("#") if part.strip()]:
if "$" in item:
@@ -358,12 +373,39 @@ class Spider(BaseSpider):
title = title.strip() or "播放"
raw_url = raw_url.strip()
if raw_url:
episodes.append(f"{title}${raw_from}@1@{raw_url}")
decode_status = self._stringify(meta.get("decode_status")).strip()
if not decode_status:
decode_status = "0" if raw_url.startswith(("http://", "https://")) else "1"
episodes.append(f"{title}${raw_from}@{decode_status}@{raw_url}")
if episodes:
final_from.append(f"{raw_from}({len(episodes)})")
final_from.append(f"{display_from}({len(episodes)})")
final_url.append("#".join(episodes))
return {"vod_play_from": "$$$".join(final_from), "vod_play_url": "$$$".join(final_url)}
def _build_player_meta(self, vodplayer):
meta = {}
for item in vodplayer or []:
if not isinstance(item, dict):
continue
source_from = self._stringify(item.get("from")).strip()
if not source_from:
continue
meta[source_from] = {
"show": self._stringify(item.get("show")).strip(),
"decode_status": self._stringify(item.get("decode_status")).strip(),
}
return meta
def _is_disabled_decode_result(self, payload, play_url, decoded):
if not isinstance(payload, dict):
return False
message = self._stringify(payload.get("msg")).strip()
if "停用" in message or "稍后再试" in message:
return True
if "解码失败" not in message:
return False
return bool(decoded) and decoded == play_url and not decoded.startswith(("http://", "https://", "/"))
def _generate_years(self, type_name):
years = [{"n": "全部", "v": ""}]
if type_name == "电影":