diff --git a/py/tests/test_步步影视.py b/py/tests/test_步步影视.py new file mode 100644 index 0000000..4e6dfd5 --- /dev/null +++ b/py/tests/test_步步影视.py @@ -0,0 +1,232 @@ +import unittest +from importlib.machinery import SourceFileLoader +from pathlib import Path +from unittest.mock import patch + + +ROOT = Path(__file__).resolve().parents[1] +MODULE = SourceFileLoader("bbys_spider", str(ROOT / "步步影视.py")).load_module() +Spider = MODULE.Spider + + +class TestBuBuYingShiSpider(unittest.TestCase): + def setUp(self): + Spider._instance = None + self.spider = Spider() + self.spider.init() + + 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"]], + ["1", "2", "3", "4"], + ) + self.assertEqual( + [item["key"] for item in content["filters"]["1"]], + ["class", "area", "year", "by"], + ) + + @patch("bbys_spider.time.time", return_value=1713523200) + @patch("bbys_spider.random.randint", side_effect=[8, 37]) + def test_build_app_headers_includes_signed_fields(self, mock_randint, mock_time): + headers = self.spider._get_app_headers() + self.assertEqual(headers["x-aid"], "com.sunshine.tv") + self.assertEqual(headers["x-ave"], "4") + self.assertEqual(headers["x-time"], "1713523200") + self.assertEqual(headers["x-nonc"], "837") + self.assertEqual(len(headers["x-sign"]), 64) + + @patch("bbys_spider.time.time", return_value=1713523200) + @patch("bbys_spider.random.randint", side_effect=[8, 37]) + def test_build_web_headers_adds_web_specific_fields(self, mock_randint, mock_time): + headers = self.spider._get_web_headers() + self.assertEqual(headers["web-sign"], "f65f3a83d6d9ad6f") + self.assertEqual(headers["X-Client"], "8f3d2a1c7b6e5d4c9a0b1f2e3d4c5b6a") + + def test_generate_years_for_movie_starts_with_current_year_and_includes_ranges(self): + years = self.spider._generate_years("电影") + self.assertEqual(years[0], {"n": "全部", "v": ""}) + self.assertEqual(years[1]["n"], str(self.spider.current_year)) + self.assertIn({"n": "2015-2011", "v": "2015-2011"}, years) + + def test_convert_json_to_vods_maps_common_fields(self): + items = self.spider._convert_json_to_vods( + [ + { + "vod_id": 11, + "vod_name": "首页片", + "vod_pic": "https://img.example/a.jpg", + "vod_duration": "更新至2集", + "vod_class": ["剧情", "悬疑"], + "vod_year": 2026, + "vod_area": "大陆", + } + ] + ) + self.assertEqual(items[0]["vod_id"], "11") + self.assertEqual(items[0]["type_name"], "剧情/悬疑") + self.assertEqual(items[0]["vod_year"], "2026") + + @patch.object(Spider, "_request_json") + def test_home_video_content_reads_home_categories(self, mock_request_json): + mock_request_json.return_value = { + "data": { + "categories": [ + { + "type_id": "1", + "type_name": "电影", + "videos": [{"vod_id": 7, "vod_name": "首页影片", "vod_pic": "/a.jpg"}], + } + ] + } + } + result = self.spider.homeVideoContent() + self.assertEqual(result["list"][0]["vod_id"], "7") + self.assertEqual(result["list"][0]["vod_name"], "首页影片") + + @patch.object(Spider, "_request_json") + @patch.object(Spider, "_get_class_list") + def test_home_video_content_falls_back_to_main_categories(self, mock_get_class_list, mock_request_json): + mock_request_json.return_value = {"data": {"categories": []}} + mock_get_class_list.side_effect = [ + {"list": [{"vod_id": "1", "vod_name": "电影片"}]}, + {"list": [{"vod_id": "2", "vod_name": "剧集片"}]}, + {"list": []}, + {"list": []}, + ] + result = self.spider.homeVideoContent() + self.assertEqual([item["vod_id"] for item in result["list"]], ["1", "2"]) + + @patch.object(Spider, "_request_json") + def test_category_content_builds_expected_web_request_and_page_shape(self, mock_request_json): + mock_request_json.return_value = { + "data": [{"vod_id": 9, "vod_name": "分类片", "vod_pic": "/cate.jpg", "vod_remarks": "HD"}] + } + result = self.spider.categoryContent("电影", "2", False, {"area": "大陆", "year": "2025", "by": "score"}) + kwargs = mock_request_json.call_args.kwargs + self.assertEqual(kwargs["headers"]["web-sign"], "f65f3a83d6d9ad6f") + self.assertEqual(kwargs["params"]["type_name"], "电影") + self.assertEqual(kwargs["params"]["page"], 2) + self.assertEqual(kwargs["params"]["sort"], "score") + self.assertEqual(result["page"], 2) + self.assertNotIn("pagecount", result) + self.assertEqual(result["list"][0]["vod_id"], "9") + + @patch.object(Spider, "_get_class_list") + def test_category_content_unknown_type_falls_back_to_hot_aggregate(self, mock_get_class_list): + mock_get_class_list.side_effect = [ + {"list": [{"vod_id": "1", "vod_name": "电影片"}]}, + {"list": [{"vod_id": "2", "vod_name": "剧集片"}]}, + {"list": [{"vod_id": "3", "vod_name": "综艺片"}]}, + {"list": [{"vod_id": "4", "vod_name": "动漫片"}]}, + ] + result = self.spider.categoryContent("步步影视", "1", False, {}) + self.assertEqual([item["vod_id"] for item in result["list"]], ["1", "2", "3", "4"]) + + @patch.object(Spider, "_request_json") + def test_search_content_skips_blank_keyword_and_builds_app_request(self, mock_request_json): + blank = self.spider.searchContent("", False, "1") + self.assertEqual(blank, {"page": 1, "limit": 0, "total": 0, "list": []}) + mock_request_json.assert_not_called() + + mock_request_json.return_value = { + "data": [{"vod_id": 13, "vod_name": "搜索片", "vod_pic": "/search.jpg"}] + } + result = self.spider.searchContent("繁花", False, "3") + kwargs = mock_request_json.call_args.kwargs + self.assertEqual(kwargs["params"]["wd"], "繁花") + self.assertEqual(kwargs["params"]["page"], 3) + self.assertEqual(kwargs["params"]["limit"], 15) + self.assertEqual(result["list"][0]["vod_id"], "13") + + def test_clean_html_content_strips_tags_and_keeps_line_breaks(self): + text = self.spider._clean_html_content("
第一段
第二段
第三行
一段简介
", + "vod_class": "剧情", + "vod_play_from": "线路一", + "vod_play_url": "第1集$/play/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") + + 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") + self.assertEqual( + self.spider._extract_decode_url({"data": {"url": "https://b.example/m3u8"}}), + "https://b.example/m3u8", + ) + self.assertEqual(self.spider._extract_decode_url({"url": "https://c.example/m3u8"}), "https://c.example/m3u8") + + @patch.object(Spider, "_request_json") + def test_player_content_decodes_encoded_play_id(self, mock_request_json): + mock_request_json.return_value = {"data": {"url": "https://cdn.example/real.m3u8"}} + result = self.spider.playerContent("线路一", "线路一@1@/play/1", {}) + kwargs = mock_request_json.call_args.kwargs + self.assertEqual(kwargs["params"]["url"], "/play/1") + self.assertEqual(kwargs["params"]["vodFrom"], "线路一") + self.assertEqual(result["parse"], 0) + self.assertEqual(result["playUrl"], "") + self.assertEqual(result["url"], "https://cdn.example/real.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"}} + result = self.spider.playerContent("线路一", "线路一@1@/play/1", {}) + self.assertEqual(result["jx"], 1) + + def test_player_content_passthrough_for_direct_url(self): + result = self.spider.playerContent("直连", "https://cdn.example/direct.m3u8", {}) + self.assertEqual(result["url"], "https://cdn.example/direct.m3u8") + self.assertEqual(result["jx"], 0) + + def test_home_content_keeps_reference_filter_values(self): + movie_filters = self.spider.homeContent(False)["filters"]["1"] + self.assertIn({"n": "动作", "v": "动作"}, movie_filters[0]["value"]) + self.assertIn({"n": "大陆", "v": "大陆"}, movie_filters[1]["value"]) + self.assertIn({"n": "评分", "v": "score"}, movie_filters[3]["value"]) + + @patch.object(Spider, "_request_json") + def test_detail_content_returns_empty_list_when_payload_missing(self, mock_request_json): + mock_request_json.return_value = {"data": []} + self.assertEqual(self.spider.detailContent(["404"]), {"list": []}) + + @patch.object(Spider, "_request_json") + def test_player_content_falls_back_to_raw_input_when_decode_fails(self, mock_request_json): + mock_request_json.return_value = {} + result = self.spider.playerContent("线路一", "线路一@1@/play/404", {}) + self.assertEqual(result["url"], "/play/404") + + +if __name__ == "__main__": + unittest.main() diff --git a/py/步步影视.py b/py/步步影视.py new file mode 100644 index 0000000..950ebd9 --- /dev/null +++ b/py/步步影视.py @@ -0,0 +1,488 @@ +# coding=utf-8 +import hashlib +import random +import re +import sys +import time + +from base.spider import Spider as BaseSpider + +sys.path.append("..") + + +class Spider(BaseSpider): + def __init__(self): + self.name = "步步影视" + self.host = "https://bbys.app" + self.pkg = "com.sunshine.tv" + self.ver = "4" + self.finger = "SF-C3B2B41F6EFFFF9869176CF68F6790E8F07506FC88632C94B4F5F0430D5498CA" + self.sk = "SK-thanks" + self.web_sign = "f65f3a83d6d9ad6f" + self.x_client = "8f3d2a1c7b6e5d4c9a0b1f2e3d4c5b6a" + self.current_year = time.localtime().tm_year + self.headers = { + "User-Agent": "okhttp/4.12.0", + "Accept": "application/json", + } + self.classes = [ + {"type_id": "1", "type_name": "电影"}, + {"type_id": "2", "type_name": "剧集"}, + {"type_id": "3", "type_name": "动漫"}, + {"type_id": "4", "type_name": "综艺"}, + ] + self.filters = { + "1": [ + { + "key": "class", + "name": "类型", + "value": [ + {"n": "全部", "v": ""}, + {"n": "动作", "v": "动作"}, + {"n": "喜剧", "v": "喜剧"}, + {"n": "爱情", "v": "爱情"}, + {"n": "科幻", "v": "科幻"}, + {"n": "恐怖", "v": "恐怖"}, + {"n": "悬疑", "v": "悬疑"}, + {"n": "犯罪", "v": "犯罪"}, + {"n": "战争", "v": "战争"}, + {"n": "动画", "v": "动画"}, + {"n": "冒险", "v": "冒险"}, + {"n": "历史", "v": "历史"}, + {"n": "灾难", "v": "灾难"}, + {"n": "纪录", "v": "纪录"}, + {"n": "剧情", "v": "剧情"}, + ], + }, + { + "key": "area", + "name": "地区", + "value": [ + {"n": "全部", "v": ""}, + {"n": "大陆", "v": "大陆"}, + {"n": "香港", "v": "香港"}, + {"n": "台湾", "v": "台湾"}, + {"n": "美国", "v": "美国"}, + {"n": "日本", "v": "日本"}, + {"n": "韩国", "v": "韩国"}, + {"n": "泰国", "v": "泰国"}, + {"n": "印度", "v": "印度"}, + {"n": "英国", "v": "英国"}, + {"n": "法国", "v": "法国"}, + {"n": "德国", "v": "德国"}, + {"n": "加拿大", "v": "加拿大"}, + {"n": "西班牙", "v": "西班牙"}, + {"n": "意大利", "v": "意大利"}, + {"n": "澳大利亚", "v": "澳大利亚"}, + ], + }, + {"key": "year", "name": "年份", "value": self._generate_years("电影")}, + { + "key": "by", + "name": "排序", + "value": [ + {"n": "最热", "v": "hits"}, + {"n": "最新", "v": "time"}, + {"n": "评分", "v": "score"}, + ], + }, + ], + "2": [ + { + "key": "class", + "name": "类型", + "value": [ + {"n": "全部", "v": ""}, + {"n": "爱情", "v": "爱情"}, + {"n": "古装", "v": "古装"}, + {"n": "武侠", "v": "武侠"}, + {"n": "历史", "v": "历史"}, + {"n": "家庭", "v": "家庭"}, + {"n": "喜剧", "v": "喜剧"}, + {"n": "悬疑", "v": "悬疑"}, + {"n": "犯罪", "v": "犯罪"}, + {"n": "战争", "v": "战争"}, + {"n": "奇幻", "v": "奇幻"}, + {"n": "科幻", "v": "科幻"}, + {"n": "恐怖", "v": "恐怖"}, + ], + }, + { + "key": "area", + "name": "地区", + "value": [ + {"n": "全部", "v": ""}, + {"n": "大陆", "v": "大陆"}, + {"n": "香港", "v": "香港"}, + {"n": "台湾", "v": "台湾"}, + {"n": "美国", "v": "美国"}, + {"n": "日本", "v": "日本"}, + {"n": "韩国", "v": "韩国"}, + {"n": "泰国", "v": "泰国"}, + {"n": "英国", "v": "英国"}, + ], + }, + {"key": "year", "name": "年份", "value": self._generate_years("剧集")}, + { + "key": "by", + "name": "排序", + "value": [ + {"n": "最热", "v": "hits"}, + {"n": "最新", "v": "time"}, + {"n": "评分", "v": "score"}, + ], + }, + ], + "3": [ + { + "key": "class", + "name": "类型", + "value": [ + {"n": "全部", "v": ""}, + {"n": "冒险", "v": "冒险"}, + {"n": "奇幻", "v": "奇幻"}, + {"n": "科幻", "v": "科幻"}, + {"n": "武侠", "v": "武侠"}, + {"n": "悬疑", "v": "悬疑"}, + ], + }, + { + "key": "area", + "name": "地区", + "value": [ + {"n": "全部", "v": ""}, + {"n": "大陆", "v": "大陆"}, + {"n": "日本", "v": "日本"}, + {"n": "欧美", "v": "欧美"}, + ], + }, + {"key": "year", "name": "年份", "value": self._generate_years("动漫")}, + { + "key": "by", + "name": "排序", + "value": [ + {"n": "最热", "v": "hits"}, + {"n": "最新", "v": "time"}, + {"n": "评分", "v": "score"}, + ], + }, + ], + "4": [ + { + "key": "class", + "name": "类型", + "value": [ + {"n": "全部", "v": ""}, + {"n": "真人秀", "v": "真人秀"}, + {"n": "音乐", "v": "音乐"}, + {"n": "脱口秀", "v": "脱口秀"}, + {"n": "歌舞", "v": "歌舞"}, + {"n": "爱情", "v": "爱情"}, + ], + }, + { + "key": "area", + "name": "地区", + "value": [ + {"n": "全部", "v": ""}, + {"n": "大陆", "v": "大陆"}, + {"n": "香港", "v": "香港"}, + {"n": "台湾", "v": "台湾"}, + {"n": "美国", "v": "美国"}, + {"n": "日本", "v": "日本"}, + {"n": "韩国", "v": "韩国"}, + ], + }, + {"key": "year", "name": "年份", "value": self._generate_years("综艺")}, + { + "key": "by", + "name": "排序", + "value": [ + {"n": "最热", "v": "hits"}, + {"n": "最新", "v": "time"}, + {"n": "评分", "v": "score"}, + ], + }, + ], + } + + def init(self, extend=""): + return None + + def getName(self): + return self.name + + def homeContent(self, filter): + return {"class": self.classes, "filters": self.filters} + + def homeVideoContent(self): + payload = self._request_json("/api.php/web/index/home", headers=self._get_web_headers()) + categories = ((payload.get("data") or {}).get("categories") or []) + videos = [] + for category in categories: + videos.extend(self._convert_json_to_vods(category.get("videos") or [])) + if not videos: + for type_name in ("电影", "剧集", "综艺", "动漫"): + videos.extend(self._get_class_list(type_name, 1, {"by": "hits"}).get("list", [])) + return {"list": videos} + + def categoryContent(self, tid, pg, filter, extend): + type_name = self._map_type_name(tid) + if type_name not in ("电影", "剧集", "动漫", "综艺"): + videos = [] + for name in ("电影", "剧集", "综艺", "动漫"): + videos.extend(self._get_class_list(name, 1, {"by": "hits"}).get("list", [])) + return {"page": 1, "limit": len(videos), "total": len(videos), "list": videos} + return self._get_class_list(type_name, int(pg), extend or {}) + + def detailContent(self, ids): + vod_id = self._stringify(ids[0] if isinstance(ids, list) and ids else ids).strip() + if not vod_id: + return {"list": []} + payload = self._request_json( + "/api.php/web/vod/get_detail", + headers=self._get_web_headers(), + params={"vod_id": vod_id}, + ) + detail = payload.get("data") + if isinstance(detail, list): + 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", "")) + return { + "list": [ + { + "vod_id": self._stringify(detail.get("vod_id") or vod_id), + "vod_name": self._stringify(detail.get("vod_name")), + "vod_pic": self._stringify(detail.get("vod_pic")), + "vod_remarks": self._stringify(detail.get("vod_remarks") or detail.get("vod_duration")), + "vod_year": self._stringify(detail.get("vod_year")), + "vod_area": self._stringify(detail.get("vod_area")), + "vod_actor": self._stringify(detail.get("vod_actor")), + "vod_director": self._stringify(detail.get("vod_director")), + "vod_content": self._clean_html_content(detail.get("vod_content")), + "type_name": self._stringify(detail.get("vod_class") or detail.get("type_name")), + "vod_play_from": play_data["vod_play_from"], + "vod_play_url": play_data["vod_play_url"], + } + ] + } + + def searchContent(self, key, quick, pg="1"): + keyword = self._stringify(key).strip() + page = int(pg) + if not keyword: + return {"page": page, "limit": 0, "total": 0, "list": []} + payload = self._request_json( + "/api.php/app/search/index", + headers=self._get_app_headers(), + params={"wd": keyword, "page": page, "limit": 15}, + ) + items = self._convert_json_to_vods(payload.get("data") or []) + return {"page": page, "limit": len(items), "total": len(items), "list": items} + + def playerContent(self, flag, id, vipFlags): + raw = self._stringify(id).strip() + if not raw: + return {"parse": 0, "playUrl": "", "url": "", "jx": 0} + + play_from = "" + play_url = raw + need_decode = False + matched = re.match(r"^([^@]+)@([01])@([\s\S]+)$", raw) + if matched: + play_from = matched.group(1) + need_decode = matched.group(2) == "1" + play_url = matched.group(3) + elif not raw.startswith(("http://", "https://")): + need_decode = True + + final_url = play_url + if need_decode: + payload = self._request_json( + "/api.php/app/decode/url/", + headers=self._get_app_headers(), + params={"url": play_url, "vodFrom": play_from}, + ) + decoded = self._extract_decode_url(payload) + if decoded: + final_url = decoded + + need_jx = 1 if re.search(r"(iqiyi\.com|v\.qq\.com|youku\.com|mgtv\.com|bilibili\.com)", final_url, re.I) else 0 + return { + "parse": 0, + "playUrl": "", + "url": final_url, + "jx": need_jx, + "header": {"User-Agent": "okhttp/4.12.0"}, + } + + def _stringify(self, value): + return "" if value is None else str(value) + + def _join_text(self, value): + if isinstance(value, list): + return "/".join([self._stringify(item) for item in value if self._stringify(item)]) + return self._stringify(value) + + def _clean_html_content(self, content): + if not content: + return "" + text = str(content).replace("", "").replace("
", "\n") + text = re.sub(r"