星星短剧
This commit is contained in:
Harold
@@ -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="<html>blocked</html>")
|
||||
self.assertEqual(self.spider._get_json("http://example.com/api"), {})
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
+180
@@ -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 {}
|
||||
Reference in New Issue
Block a user