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

feat: add juzi tv spider

Browse Source
This commit is contained in:
Harold
2026-04-19 09:20:04 +08:00
parent f52a2e3ed8
commit 22f8d1478f
3 changed files with 537 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files
py/tests/test_dbku.py
-18
View File
@@ -222,24 +222,6 @@ class TestDBKUSpider(unittest.TestCase):
self.assertEqual(result["url"], "https://video.example/final.m3u8")
self.assertEqual(result["header"]["Referer"], "https://www.dbku.tv/vodplay/100-1-1.html")
@patch.object(Spider, "log")
@patch.object(Spider, "_request_html")
def test_player_content_logs_resolved_player_fields(self, mock_request_html, mock_log):
mock_request_html.return_value = """
<script>
var player_data = {"url":"aHR0cHM6Ly92aWRlby5leGFtcGxlL2ZpbmFsLm0zdTg=","encrypt":"2"};
</script>
"""
self.spider.playerContent("独播库", "100-1-1", {})
mock_log.assert_called()
payload = mock_log.call_args.args[0]
self.assertEqual(payload["stage"], "playerContent")
self.assertEqual(payload["input_url"], "100-1-1")
self.assertEqual(payload["request_url"], "https://www.dbku.tv/vodplay/100-1-1.html")
self.assertEqual(payload["encrypt"], 2)
self.assertEqual(payload["decoded_url"], "https://video.example/final.m3u8")
self.assertEqual(payload["final_url"], "https://video.example/final.m3u8")
@patch.object(Spider, "_request_html")
def test_player_content_follows_internal_jump(self, mock_request_html):
mock_request_html.side_effect = [
py/tests/test_橘子TV.py
+285
View File
@@ -0,0 +1,285 @@
import unittest
from importlib.machinery import SourceFileLoader
from pathlib import Path
from unittest.mock import patch
ROOT = Path(__file__).resolve().parents[1]
MODULE = SourceFileLoader("juzi_spider", str(ROOT / "橘子TV.py")).load_module()
Spider = MODULE.Spider
class TestJuZiTVSpider(unittest.TestCase):
def setUp(self):
Spider._instance = None
self.spider = Spider()
@patch.object(Spider, "fetch")
def test_init_loads_api_host_from_json_config(self, mock_fetch):
class FakeResponse:
def __init__(self, text):
self.text = text
self.status_code = 200
self.encoding = "utf-8"
mock_fetch.return_value = FakeResponse('["https://api1.example.com","https://api2.example.com"]')
self.spider.init()
self.assertEqual(self.spider.api_host, "https://api1.example.com")
@patch.object(Spider, "fetch")
def test_init_loads_api_host_from_plain_text_config(self, mock_fetch):
class FakeResponse:
def __init__(self, text):
self.text = text
self.status_code = 200
self.encoding = "utf-8"
mock_fetch.return_value = FakeResponse("https://api3.example.com,\nhttps://api4.example.com")
self.spider.init()
self.assertEqual(self.spider.api_host, "https://api3.example.com")
@patch.object(Spider, "fetch")
def test_build_payload_adds_sign_and_common_fields(self, mock_fetch):
class FakeResponse:
def __init__(self, text):
self.text = text
self.status_code = 200
self.encoding = "utf-8"
mock_fetch.return_value = FakeResponse('["https://api1.example.com"]')
self.spider.init()
payload = self.spider._build_payload({"vodId": "123"})
self.assertEqual(payload["appId"], "fea23e11fc1241409682880e15fb2851")
self.assertEqual(payload["bundlerId"], "com.voraguzzee.ts")
self.assertEqual(payload["vodId"], "123")
self.assertIn("requestId", payload)
self.assertIn("sign", payload)
self.assertEqual(len(payload["sign"]), 32)
@patch.object(Spider, "_post_api")
@patch.object(Spider, "fetch")
def test_home_content_maps_channel_list(self, mock_fetch, mock_post_api):
class FakeResponse:
def __init__(self, text):
self.text = text
self.status_code = 200
self.encoding = "utf-8"
mock_fetch.return_value = FakeResponse('["https://api1.example.com"]')
mock_post_api.return_value = {
"data": {
"channeList": [
{"channelId": 1, "channelName": "电影"},
{"channelId": 2, "channelName": "剧集"},
]
}
}
self.spider.init()
result = self.spider.homeContent(False)
self.assertEqual(
result,
{
"class": [
{"type_id": "1", "type_name": "电影"},
{"type_id": "2", "type_name": "剧集"},
]
},
)
@patch.object(Spider, "_post_api")
@patch.object(Spider, "fetch")
def test_home_video_content_flattens_topic_vod_list(self, mock_fetch, mock_post_api):
class FakeResponse:
def __init__(self, text):
self.text = text
self.status_code = 200
self.encoding = "utf-8"
mock_fetch.return_value = FakeResponse('["https://api1.example.com"]')
mock_post_api.return_value = {
"data": {
"vodTopicList": [
{
"vodList": [
{
"vodId": 11,
"vodName": "推荐影片",
"coverImg": "https://img.example.com/a.jpg",
"remark": "更新至1集",
"flags": "2026 / 大陆",
}
]
}
]
}
}
self.spider.init()
result = self.spider.homeVideoContent()
self.assertEqual(result["list"][0]["vod_id"], "11")
self.assertEqual(result["list"][0]["vod_name"], "推荐影片")
self.assertEqual(result["list"][0]["vod_year"], "2026")
@patch.object(Spider, "_post_api")
@patch.object(Spider, "fetch")
def test_category_content_maps_items_and_page_fields(self, mock_fetch, mock_post_api):
class FakeResponse:
def __init__(self, text):
self.text = text
self.status_code = 200
self.encoding = "utf-8"
mock_fetch.return_value = FakeResponse('["https://api1.example.com"]')
mock_post_api.return_value = {
"data": {
"hasNext": 1,
"nextVal": "token-2",
"items": [
{
"vodId": 21,
"vodName": "分类影片",
"coverImg": "https://img.example.com/cate.jpg",
"remark": "完结",
"flags": "2025 / 美国",
"intro": "分类简介",
}
],
}
}
self.spider.init()
result = self.spider.categoryContent("7", "2", False, {})
self.assertEqual(result["page"], 2)
self.assertEqual(result["pagecount"], 3)
self.assertEqual(result["list"][0]["vod_id"], "21")
self.assertEqual(result["list"][0]["vod_content"], "分类简介")
payload = mock_post_api.call_args.args[1]
self.assertEqual(payload["nextCount"], 18)
self.assertEqual(payload["nextVal"], "")
self.assertEqual(payload["sortType"], "")
self.assertEqual(payload["queryValueJson"], '[{"filerName":"channelId","filerValue":"7"}]')
@patch.object(Spider, "_post_api")
@patch.object(Spider, "fetch")
def test_search_content_maps_items_without_next_page(self, mock_fetch, mock_post_api):
class FakeResponse:
def __init__(self, text):
self.text = text
self.status_code = 200
self.encoding = "utf-8"
mock_fetch.return_value = FakeResponse('["https://api1.example.com"]')
mock_post_api.return_value = {
"data": {
"hasNext": 0,
"items": [
{
"vodId": 31,
"vodName": "搜索影片",
"coverImg": "https://img.example.com/search.jpg",
"remark": "",
"score": "8.5",
"flags": "2024 / 日本",
"intro": "搜索简介",
}
],
}
}
self.spider.init()
result = self.spider.searchContent("繁花", False, "1")
self.assertEqual(result["page"], 1)
self.assertEqual(result["pagecount"], 1)
self.assertEqual(result["list"][0]["vod_id"], "31")
self.assertEqual(result["list"][0]["vod_remarks"], "评分：8.5")
payload = mock_post_api.call_args.args[1]
self.assertEqual(payload["keyword"], "繁花")
self.assertEqual(payload["nextVal"], "")
def test_parse_people_joins_worker_names(self):
self.assertEqual(
self.spider._parse_people(
[
{"vodWorkerId": 1, "vodWorkerName": "张三"},
{"vodWorkerId": 2, "vodWorkerName": "李四"},
]
),
"张三,李四",
)
@patch.object(Spider, "_post_api")
@patch.object(Spider, "fetch")
def test_detail_content_maps_meta_and_multi_line_playlists(self, mock_fetch, mock_post_api):
class FakeResponse:
def __init__(self, text):
self.text = text
self.status_code = 200
self.encoding = "utf-8"
mock_fetch.return_value = FakeResponse('["https://api1.example.com"]')
mock_post_api.return_value = {
"data": {
"vodId": 41,
"vodName": "详情影片",
"coverImg": "https://img.example.com/detail.jpg",
"updateRemark": "更新至3集",
"year": "2026",
"areaName": "大陆",
"intro": "详情简介",
"actorList": [{"vodWorkerName": "主演甲"}, {"vodWorkerName": "主演乙"}],
"directorList": [{"vodWorkerName": "导演甲"}],
"playerList": [
{
"playerName": "线路A",
"epList": [{"epName": "第1集", "epId": "ep-a-1"}, {"epName": "第2集", "epId": "ep-a-2"}],
},
{
"playerName": "线路B",
"epList": [{"epName": "正片", "epId": "ep-b-1"}],
},
],
}
}
self.spider.init()
result = self.spider.detailContent(["41"])
vod = result["list"][0]
self.assertEqual(vod["vod_id"], "41")
self.assertEqual(vod["vod_name"], "详情影片")
self.assertEqual(vod["vod_remarks"], "更新至3集")
self.assertEqual(vod["vod_year"], "2026")
self.assertEqual(vod["vod_area"], "大陆")
self.assertEqual(vod["vod_actor"], "主演甲,主演乙")
self.assertEqual(vod["vod_director"], "导演甲")
self.assertEqual(vod["vod_content"], "详情简介")
self.assertEqual(vod["vod_play_from"], "线路A$$$线路B")
self.assertEqual(vod["vod_play_url"], "第1集$ep-a-1#第2集$ep-a-2$$$正片$ep-b-1")
@patch.object(Spider, "_post_api")
@patch.object(Spider, "fetch")
def test_player_content_collects_resolution_urls(self, mock_fetch, mock_post_api):
class FakeResponse:
def __init__(self, text):
self.text = text
self.status_code = 200
self.encoding = "utf-8"
mock_fetch.return_value = FakeResponse('["https://api1.example.com"]')
mock_post_api.side_effect = [
{
"data": [
{"showName": "高清", "vodResolution": "1080p"},
{"showName": "标清", "vodResolution": "720p"},
]
},
{"data": {"playUrl": "https://cdn.example.com/1080.m3u8"}},
{"data": {"playUrl": "https://cdn.example.com/720.m3u8"}},
]
self.spider.init()
result = self.spider.playerContent("线路A", "ep-a-1", {})
self.assertEqual(result["parse"], 0)
self.assertEqual(
result["url"],
["高清", "https://cdn.example.com/1080.m3u8", "标清", "https://cdn.example.com/720.m3u8"],
)
self.assertEqual(result["header"]["User-Agent"], "ExoPlayer")
if __name__ == "__main__":
unittest.main()
py/橘子TV.py
+252
View File
@@ -0,0 +1,252 @@
# coding=utf-8
import hashlib
import json
import sys
import uuid
from base.spider import Spider as BaseSpider
sys.path.append("..")
class Spider(BaseSpider):
def __init__(self):
self.name = "橘子TV"
self.config_url = "https://gapi0725.5p8jcjc.com/config.json"
self.api_host = ""
self.headers = {
"User-Agent": "Dart/3.1 (dart:io)",
"Accept-Encoding": "gzip",
"content-type": "application/json; charset=utf-8",
}
self.app_id = "fea23e11fc1241409682880e15fb2851"
self.app_key = "f384b87cc9ef41e4842dda977bae2c7f"
self.udid = "bfc18c00-c866-46cb-8d7b-121c39b942d4"
self.bundler_id = "com.voraguzzee.ts"
self.source = "1003_default"
self.version = "1.0.1"
self.version_code = 1000
self.inited = False
def init(self, extend=""):
if self.inited:
return None
response = self.fetch(self.config_url, headers=self.headers, timeout=10, verify=False)
raw = response.text or ""
try:
hosts = json.loads(raw)
except Exception:
hosts = [item.strip() for item in raw.replace("\n", ",").split(",") if item.strip()]
for host in hosts:
if str(host).startswith("http"):
self.api_host = str(host).rstrip("/")
break
if not self.api_host:
raise ValueError("未能获取到有效的API地址")
self.inited = True
return None
def getName(self):
return self.name
def _dict_to_query(self, data):
parts = []
for key, value in data.items():
if value != "":
parts.append(f"{key}={value}")
return "&".join(parts)
def _sign_data(self, data):
sorted_data = {key: data[key] for key in sorted(data)}
query = self._dict_to_query(sorted_data)
sign_source = f"{query}&appKey={self.app_key}" if query else f"appKey={self.app_key}"
signed = dict(data)
signed["sign"] = hashlib.md5(sign_source.encode("utf-8")).hexdigest()
return signed
def _build_payload(self, extra=None):
payload = {
"appId": self.app_id,
"bundlerId": self.bundler_id,
"cus1tom": "cus3tom",
"deviceInfo": "xiaomi",
"osInfo": "15",
"otherParam": "1",
"patchNumber": 0,
"requestId": str(uuid.uuid4()),
"source": self.source,
"udid": self.udid,
"version": self.version,
"versionCode": self.version_code,
}
if extra:
payload.update(extra)
return self._sign_data(payload)
def _post_api(self, path, payload):
response = self.post(
f"{self.api_host}{path}",
json=payload,
headers=self.headers,
timeout=15,
verify=False,
)
if response.status_code != 200:
return {}
try:
return json.loads(response.text or "{}")
except Exception:
return {}
def _parse_year(self, flags):
text = str(flags or "")
if not text:
return ""
return text.split(" / ")[0].strip()
def homeContent(self, filter):
response = self._post_api("/v2/api/home/header", self._build_payload())
classes = []
for item in response.get("data", {}).get("channeList", []):
if item.get("channelId"):
classes.append(
{
"type_id": str(item.get("channelId")),
"type_name": item.get("channelName", ""),
}
)
return {"class": classes}
def homeVideoContent(self):
response = self._post_api("/v2/api/home/body", self._build_payload())
videos = []
for topic in response.get("data", {}).get("vodTopicList", []):
for item in topic.get("vodList", []):
videos.append(
{
"vod_id": str(item.get("vodId", "")),
"vod_name": item.get("vodName", ""),
"vod_pic": item.get("coverImg", ""),
"vod_remarks": item.get("remark") or f"评分：{item.get('score', '')}",
"vod_year": self._parse_year(item.get("flags")),
}
)
return {"list": videos}
def _map_video_item(self, item):
return {
"vod_id": str(item.get("vodId", "")),
"vod_name": item.get("vodName", ""),
"vod_pic": item.get("coverImg", ""),
"vod_remarks": item.get("remark") or f"评分：{item.get('score', '')}",
"vod_year": self._parse_year(item.get("flags")),
"vod_content": item.get("intro", ""),
}
def _page_result(self, items, pg, has_next):
page = int(pg)
pagecount = page + 1 if items and has_next else page
return {
"list": items,
"page": page,
"pagecount": pagecount,
"limit": len(items),
"total": pagecount * max(len(items), 1),
}
def categoryContent(self, tid, pg, filter, extend):
payload = self._build_payload(
{
"nextCount": 18,
"nextVal": "",
"queryValueJson": json.dumps(
[{"filerName": "channelId", "filerValue": str(tid)}],
ensure_ascii=False,
separators=(",", ":"),
),
"sortType": "",
}
)
response = self._post_api("/v1/api/search/queryNow", payload)
data = response.get("data", {})
items = [self._map_video_item(item) for item in data.get("items", []) if isinstance(item, dict)]
return self._page_result(items, pg, data.get("hasNext") == 1)
def searchContent(self, key, quick, pg="1"):
payload = self._build_payload({"keyword": key, "nextVal": ""})
response = self._post_api("/v1/api/search/search", payload)
data = response.get("data", {})
items = [self._map_video_item(item) for item in data.get("items", []) if isinstance(item, dict)]
return self._page_result(items, pg, data.get("hasNext") == 1)
def _parse_people(self, people):
names = []
for item in people or []:
if isinstance(item, dict) and item.get("vodWorkerName"):
names.append(item.get("vodWorkerName"))
return ",".join(names)
def detailContent(self, ids):
vod_id = ids[0]
response = self._post_api("/v2/api/vodInfo/index", self._build_payload({"vodId": vod_id}))
data = response.get("data", {})
play_from = []
play_urls = []
for player in data.get("playerList", []):
play_from.append(player.get("playerName", ""))
episodes = []
for episode in player.get("epList", []):
episodes.append(f"{episode.get('epName', '')}${episode.get('epId', '')}")
play_urls.append("#".join(episodes))
return {
"list": [
{
"vod_id": str(data.get("vodId", "")),
"vod_name": data.get("vodName", ""),
"vod_pic": data.get("coverImg", ""),
"vod_remarks": data.get("updateRemark", ""),
"vod_year": data.get("year", ""),
"vod_area": data.get("areaName", ""),
"vod_actor": self._parse_people(data.get("actorList")),
"vod_director": self._parse_people(data.get("directorList")),
"vod_content": data.get("intro", ""),
"vod_play_from": "$$$".join(play_from),
"vod_play_url": "$$$".join(play_urls),
}
]
}
def playerContent(self, flag, id, vipFlags):
response = self._post_api("/v2/api/vodInfo/epDetail", self._build_payload({"vodEpId": id}))
urls = []
for item in response.get("data", []):
show_name = item.get("showName")
resolution = item.get("vodResolution")
if not show_name or not resolution:
continue
play_response = self._post_api(
"/v2/api/vodInfo/playUrl",
self._build_payload({"epId": id, "vodResolution": resolution}),
)
play_url = play_response.get("data", {}).get("playUrl", "")
if str(play_url).startswith("http"):
urls.extend([show_name, play_url])
return {
"parse": 0,
"url": urls,
"header": {
"User-Agent": "ExoPlayer",
"Connection": "Keep-Alive",
"Accept-Encoding": "gzip",
},
}