From a33bbcda20d7a5ca981ac81a74153d999d20e0cf Mon Sep 17 00:00:00 2001 From: Harold <8866033@gmail.com> Date: Sat, 25 Apr 2026 12:46:01 +0800 Subject: [PATCH] =?UTF-8?q?=E6=98=9F=E6=98=9F=E7=9F=AD=E5=89=A7?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- py/tests/test_星星短剧.py | 144 ++++++++++++++++++++++++++++++ py/星星短剧.py | 180 ++++++++++++++++++++++++++++++++++++++ 2 files changed, 324 insertions(+) create mode 100644 py/tests/test_星星短剧.py create mode 100644 py/星星短剧.py diff --git a/py/tests/test_星星短剧.py b/py/tests/test_星星短剧.py new file mode 100644 index 0000000..96b7a67 --- /dev/null +++ b/py/tests/test_星星短剧.py @@ -0,0 +1,144 @@ +import unittest +from importlib.machinery import SourceFileLoader +from pathlib import Path +from unittest.mock import MagicMock, patch + + +ROOT = Path(__file__).resolve().parents[1] +MODULE = SourceFileLoader("xingxingduanju_spider", str(ROOT / "星星短剧.py")).load_module() +Spider = MODULE.Spider + + +class TestXingXingDuanJuSpider(unittest.TestCase): + def setUp(self): + Spider._instance = None + self.spider = Spider() + self.spider.init() + + def test_home_content_exposes_fixed_categories(self): + content = self.spider.homeContent(False) + self.assertEqual(content["class"][0]["type_id"], "1287") + self.assertEqual(content["class"][-1]["type_name"], "古代") + self.assertEqual(content["filters"], {}) + + @patch.object(Spider, "_get_json") + def test_home_video_content_uses_default_category(self, mock_json): + mock_json.return_value = { + "data": { + "datalist": [ + { + "id": 1001, + "name": "闪婚后成了大佬心尖宠", + "icon": "http://img/1.jpg", + "introduction": "剧情简介", + "heat": 291.6, + } + ], + "page": {"totalCount": 34}, + } + } + result = self.spider.homeVideoContent() + self.assertEqual(result["list"][0]["vod_id"], "1001@@闪婚后成了大佬心尖宠@@剧情简介") + self.assertEqual(result["list"][0]["vod_name"], "闪婚后成了大佬心尖宠") + self.assertEqual(result["list"][0]["vod_remarks"], "291.6万播放") + self.assertIn("resourceId=1287", mock_json.call_args.args[0]) + self.assertIn("pageNum=1", mock_json.call_args.args[0]) + + @patch.object(Spider, "_get_json") + def test_category_content_builds_query_and_maps_items(self, mock_json): + mock_json.return_value = { + "data": { + "datalist": [ + { + "id": 2002, + "name": "宸王的复仇王妃", + "icon": "http://img/2.jpg", + "introduction": "王妃复仇记", + "heat": 398.3, + } + ], + "page": {"totalCount": 34}, + } + } + result = self.spider.categoryContent("1291", "2", False, {}) + self.assertEqual(result["page"], 2) + self.assertEqual(result["limit"], 1) + self.assertEqual(result["total"], 34) + self.assertEqual(result["list"][0]["vod_id"], "2002@@宸王的复仇王妃@@王妃复仇记") + self.assertNotIn("pagecount", result) + self.assertIn("resourceId=1291", mock_json.call_args.args[0]) + self.assertIn("pageNum=2", mock_json.call_args.args[0]) + self.assertIn("pageSize=10", mock_json.call_args.args[0]) + + @patch.object(Spider, "_get_json") + def test_detail_content_builds_single_playlist(self, mock_json): + mock_json.return_value = { + "data": [ + { + "chapterName": "第1集", + "shortPlayList": [ + { + "chapterShortPlayVoList": [ + {"shortPlayUrl": "http://video/1.m3u8"} + ] + } + ], + }, + { + "chapterName": "第2集", + "shortPlayList": [ + { + "chapterShortPlayVoList": [ + {"shortPlayUrl": "http://video/2.m3u8"} + ] + } + ], + }, + ] + } + result = self.spider.detailContent(["422599939194881@@穿越诡计爱上你@@一段简介"]) + vod = result["list"][0] + self.assertEqual(vod["vod_id"], "422599939194881@@穿越诡计爱上你@@一段简介") + self.assertEqual(vod["vod_name"], "穿越诡计爱上你") + self.assertEqual(vod["vod_content"], "一段简介") + self.assertEqual(vod["vod_play_from"], "星星短剧") + self.assertIn("第1集$http://video/1.m3u8", vod["vod_play_url"]) + self.assertIn("第2集$http://video/2.m3u8", vod["vod_play_url"]) + self.assertIn("bookId=422599939194881", mock_json.call_args.args[0]) + + @patch.object(Spider, "_get_json") + def test_detail_content_returns_empty_when_id_missing(self, mock_json): + self.assertEqual(self.spider.detailContent([""]), {"list": []}) + mock_json.assert_not_called() + + def test_search_content_returns_empty(self): + self.assertEqual( + self.spider.searchContent("任意关键词", False, "3"), + {"page": 3, "limit": 0, "total": 0, "list": []}, + ) + + def test_player_content_returns_direct_url(self): + result = self.spider.playerContent("星星短剧", "http://video/play.m3u8", {}) + self.assertEqual(result["parse"], 0) + self.assertEqual(result["url"], "http://video/play.m3u8") + self.assertIn("Mozilla/5.0", result["header"]["User-Agent"]) + + def test_player_content_upgrades_known_cdn_to_https(self): + result = self.spider.playerContent( + "星星短剧", + "http://img.novel.wsljf.xyz/shortPlay-mp4/demo/output.m3u8", + {}, + ) + self.assertEqual( + result["url"], + "https://img.novel.wsljf.xyz/shortPlay-mp4/demo/output.m3u8", + ) + + @patch.object(Spider, "fetch") + def test_get_json_returns_empty_on_invalid_json(self, mock_fetch): + mock_fetch.return_value = MagicMock(status_code=200, text="blocked") + self.assertEqual(self.spider._get_json("http://example.com/api"), {}) + + +if __name__ == "__main__": + unittest.main() diff --git a/py/星星短剧.py b/py/星星短剧.py new file mode 100644 index 0000000..8f42ce5 --- /dev/null +++ b/py/星星短剧.py @@ -0,0 +1,180 @@ +# coding=utf-8 +import json +import sys +from urllib.parse import urlencode + +from base.spider import Spider as BaseSpider + +sys.path.append("..") + + +HOST = "http://read.api.duodutek.com" +UA = ( + "Mozilla/5.0 (Windows NT 6.1; WOW64) " + "AppleWebKit/537.36 (KHTML, like Gecko) " + "Chrome/50.0.2661.87 Safari/537.36" +) +COMMON_PARAMS = { + "productId": "2a8c14d1-72e7-498b-af23-381028eb47c0", + "vestId": "2be070e0-c824-4d0e-a67a-8f688890cadb", + "channel": "oppo19", + "osType": "android", + "version": "20", + "token": "202509271001001446030204698626", +} +CLASS_LIST = [ + {"type_id": "1287", "type_name": "甜宠"}, + {"type_id": "1288", "type_name": "逆袭"}, + {"type_id": "1289", "type_name": "热血"}, + {"type_id": "1290", "type_name": "现代"}, + {"type_id": "1291", "type_name": "古代"}, +] + + +class Spider(BaseSpider): + def __init__(self): + self.name = "星星短剧" + self.host = HOST + self.headers = {"User-Agent": UA} + self.classes = CLASS_LIST + + def init(self, extend=""): + return None + + def getName(self): + return self.name + + def homeContent(self, filter): + return {"class": self.classes, "filters": {}} + + def homeVideoContent(self): + return {"list": self._list_category_items("1287", 1)[:12]} + + def categoryContent(self, tid, pg, filter, extend): + page = max(1, int(pg)) + payload = self._get_json( + self._build_url( + "/novel-api/app/pageModel/getResourceById", + {"resourceId": str(tid), "pageNum": str(page), "pageSize": "10"}, + ) + ) + data = payload.get("data") if isinstance(payload, dict) else {} + items = [self._map_vod(item) for item in (data.get("datalist") or [])] + total = self._coerce_int( + data.get("total") + or data.get("totalCount") + or (data.get("page") or {}).get("totalCount") + or payload.get("total") + ) + return {"page": page, "limit": len(items), "total": total, "list": items} + + def detailContent(self, ids): + result = {"list": []} + for raw_id in ids: + vod_id = str(raw_id or "").strip() + if not vod_id: + continue + book_id, book_name, intro = self._split_vod_id(vod_id) + payload = self._get_json( + self._build_url( + "/novel-api/basedata/book/getChapterList", + {"bookId": book_id}, + ) + ) + chapters = payload.get("data") if isinstance(payload, dict) else [] + episodes = [] + for index, chapter in enumerate(chapters or [], start=1): + play_url = self._extract_play_url(chapter) + if not play_url: + continue + chapter_name = str(chapter.get("chapterName") or f"第{index}集").strip() + episodes.append(f"{chapter_name}${play_url}") + result["list"].append( + { + "vod_id": vod_id, + "vod_name": book_name, + "vod_content": intro, + "vod_remarks": f"共{len(episodes)}集" if episodes else "", + "vod_play_from": "星星短剧", + "vod_play_url": "#".join(episodes), + } + ) + return result + + def searchContent(self, key, quick, pg="1"): + page = max(1, int(pg)) + return {"page": page, "limit": 0, "total": 0, "list": []} + + def playerContent(self, flag, id, vipFlags): + play_url = self._normalize_play_url(id) + if not play_url: + return {"parse": 0, "jx": 0, "url": "", "header": {}} + return { + "parse": 0, + "jx": 0, + "url": play_url, + "header": {"User-Agent": UA}, + } + + def _list_category_items(self, tid, page): + payload = self._get_json( + self._build_url( + "/novel-api/app/pageModel/getResourceById", + {"resourceId": str(tid), "pageNum": str(page), "pageSize": "10"}, + ) + ) + data = payload.get("data") if isinstance(payload, dict) else {} + return [self._map_vod(item) for item in (data.get("datalist") or [])] + + def _map_vod(self, item): + book_id = str(item.get("id") or "").strip() + name = str(item.get("name") or "").strip() + intro = str(item.get("introduction") or "").strip() + heat = str(item.get("heat") or "").strip() + return { + "vod_id": f"{book_id}@@{name}@@{intro}", + "vod_name": name, + "vod_pic": str(item.get("icon") or "").strip(), + "vod_remarks": f"{heat}万播放" if heat else "", + } + + def _split_vod_id(self, vod_id): + parts = vod_id.split("@@", 2) + if len(parts) == 3: + return parts[0], parts[1], parts[2] + if len(parts) == 2: + return parts[0], parts[1], "" + return vod_id, "", "" + + def _extract_play_url(self, chapter): + for play_group in chapter.get("shortPlayList") or []: + for play in play_group.get("chapterShortPlayVoList") or []: + play_url = self._normalize_play_url(play.get("shortPlayUrl")) + if play_url: + return play_url + return "" + + def _normalize_play_url(self, value): + play_url = str(value or "").strip() + if play_url.startswith("http://img.novel.wsljf.xyz/"): + return "https://" + play_url[len("http://"):] + return play_url + + def _build_url(self, path, params=None): + query = urlencode({**COMMON_PARAMS, **(params or {})}) + return f"{self.host}{path}?{query}" + + def _coerce_int(self, value): + try: + return int(value) + except (TypeError, ValueError): + return 0 + + def _get_json(self, url): + response = self.fetch(url, headers=dict(self.headers), timeout=10, verify=False) + if response.status_code != 200: + return {} + try: + return json.loads(response.text or "{}") + except (json.JSONDecodeError, TypeError, ValueError): + return {}