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

feat: add jinpai spider

Browse Source
This commit is contained in:
Harold
2026-04-24 16:33:34 +08:00
parent 153589549d
commit 60f54818a8
2 changed files with 454 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
py/tests/test_金牌.py
+214
View File
@@ -0,0 +1,214 @@
import hashlib
import unittest
from importlib.machinery import SourceFileLoader
from pathlib import Path
from unittest.mock import patch
ROOT = Path(__file__).resolve().parents[1]
MODULE = SourceFileLoader("jinpai_spider", str(ROOT / "金牌.py")).load_module()
Spider = MODULE.Spider
class TestJinPaiSpider(unittest.TestCase):
def setUp(self):
Spider._instance = None
self.spider = Spider()
self.spider.init()
@patch("jinpai_spider.time.time", return_value=1713952212.345)
def test_signed_headers_generates_expected_hashes(self, _mock_time):
headers = self.spider._signed_headers({"pageNum": "1", "type1": "2"})
expected_t = "1713952212345"
expected_obj = "pageNum=1&type1=2&key=cb808529bae6b6be45ecfab29a4889bc&t=1713952212345"
expected_sign = hashlib.sha1(
hashlib.md5(expected_obj.encode("utf-8")).hexdigest().encode("utf-8")
).hexdigest()
self.assertEqual(headers["t"], expected_t)
self.assertEqual(headers["sign"], expected_sign)
self.assertEqual(headers["Referer"], "https://m.jiabaide.cn/")
@patch.object(Spider, "_fetch_json")
def test_home_content_returns_class_filters_and_hot_list(self, mock_fetch_json):
mock_fetch_json.side_effect = [
{"data": [{"typeId": 1, "typeName": "电影"}, {"typeId": 2, "typeName": "剧集"}]},
{
"data": {
"1": {
"typeList": [{"itemText": "动作", "itemValue": "dongzuo"}],
"plotList": [{"itemText": "剧情"}],
"districtList": [{"itemText": "中国香港"}],
"languageList": [{"itemText": "粤语"}],
"yearList": [{"itemText": "2026"}],
}
}
},
{
"data": [
{
"vodId": 9,
"vodName": "热播片",
"vodPic": "https://img/9.jpg",
"vodRemarks": "更新至10集",
"vodDoubanScore": "8.6",
"vodPubdate": "2026-01-01",
"typeId": 1,
"typeName": "电影",
}
]
},
]
result = self.spider.homeContent(False)
self.assertEqual(
result["class"],
[{"type_id": "1", "type_name": "电影"}, {"type_id": "2", "type_name": "剧集"}],
)
self.assertEqual(result["filters"]["1"][0]["key"], "type")
self.assertEqual(result["filters"]["1"][0]["value"][0], {"n": "全部", "v": ""})
self.assertEqual(result["filters"]["1"][0]["value"][1], {"n": "动作", "v": "dongzuo"})
self.assertEqual(result["filters"]["1"][-1]["key"], "by")
self.assertEqual(result["list"][0]["vod_remarks"], "更新至10集_8.6")
self.assertEqual(result["list"][0]["vod_year"], "2026")
def test_home_video_content_returns_empty_list(self):
self.assertEqual(self.spider.homeVideoContent(), {"list": []})
@patch.object(Spider, "_fetch_json")
def test_category_content_maps_filters_to_api_payload(self, mock_fetch_json):
mock_fetch_json.return_value = {
"data": {
"list": [
{
"vodId": 11,
"vodName": "分类片",
"vodPic": "https://img/11.jpg",
"vodRemarks": "完结",
"vodDoubanScore": "7.1",
"vodPubdate": "2025-09-09",
"typeId": 1,
"typeName": "电影",
}
]
}
}
result = self.spider.categoryContent(
"1",
"2",
False,
{"area": "中国香港", "lang": "粤语", "year": "2025", "by": "4", "type": "dongzuo", "class": "剧情"},
)
self.assertEqual(
mock_fetch_json.call_args.args,
(
"/api/mw-movie/anonymous/video/list",
{
"area": "中国香港",
"lang": "粤语",
"pageNum": "2",
"pageSize": "30",
"sort": "4",
"sortBy": "1",
"type": "dongzuo",
"type1": "1",
"v_class": "剧情",
"year": "2025",
},
),
)
self.assertEqual(result["page"], 2)
self.assertEqual(result["limit"], 30)
self.assertEqual(result["list"][0]["vod_id"], "11")
self.assertNotIn("pagecount", result)
def test_search_content_returns_empty_for_blank_keyword(self):
self.assertEqual(self.spider.searchContent("", False, "1"), {"page": 1, "limit": 30, "total": 0, "list": []})
@patch.object(Spider, "_fetch_json")
def test_search_content_maps_keyword_and_page(self, mock_fetch_json):
mock_fetch_json.return_value = {
"data": {
"list": [
{
"vodId": 21,
"vodName": "搜索片",
"vodPic": "https://img/21.jpg",
"vodRemarks": "更新",
"vodDoubanScore": "8.0",
"vodPubdate": "2024-01-01",
"typeId": 2,
"typeName": "剧集",
}
]
}
}
result = self.spider.searchContent("繁花", False, "3")
self.assertEqual(
mock_fetch_json.call_args.args,
(
"/api/mw-movie/anonymous/video/searchByWordPageable",
{"keyword": "繁花", "pageNum": "3", "pageSize": "30"},
),
)
self.assertEqual(result["page"], 3)
self.assertEqual(result["list"][0]["vod_name"], "搜索片")
self.assertEqual(result["list"][0]["vod_year"], "2024")
@patch.object(Spider, "_fetch_json")
def test_detail_content_maps_metadata_and_episode_ids(self, mock_fetch_json):
mock_fetch_json.return_value = {
"data": {
"vodId": 88,
"vodName": "详情片",
"vodPic": "https://img/88.jpg",
"vodClass": "动作",
"vodRemarks": "更新至12集",
"vodYear": "2026",
"vodArea": "中国香港",
"vodLang": "粤语",
"vodDirector": "导演甲",
"vodActor": "演员甲/演员乙",
"vodContent": "一段简介",
"episodeList": [{"name": "第1集", "nid": 9001}, {"name": "第2集", "nid": 9002}],
}
}
result = self.spider.detailContent(["88"])
vod = result["list"][0]
self.assertEqual(mock_fetch_json.call_args.args, ("/api/mw-movie/anonymous/video/detail", {"id": "88"}))
self.assertEqual(vod["vod_id"], "88")
self.assertEqual(vod["vod_name"], "详情片")
self.assertEqual(vod["vod_play_from"], "金牌线路")
self.assertEqual(vod["vod_play_url"], "第1集$88@9001#第2集$88@9002")
def test_detail_content_returns_empty_for_blank_id(self):
self.assertEqual(self.spider.detailContent([""]), {"list": []})
@patch.object(Spider, "_fetch_json")
def test_player_content_returns_first_episode_url_with_headers(self, mock_fetch_json):
mock_fetch_json.return_value = {
"data": {
"list": [
{"name": "1080P", "url": "https://cdn.example/1080.m3u8"},
{"name": "720P", "url": "https://cdn.example/720.m3u8"},
]
}
}
result = self.spider.playerContent("金牌线路", "88@9001", {})
self.assertEqual(
mock_fetch_json.call_args.args,
("/api/mw-movie/anonymous/v2/video/episode/url", {"clientType": "3", "id": "88", "nid": "9001"}),
)
self.assertEqual(
result,
{
"parse": 0,
"playUrl": "",
"url": "https://cdn.example/1080.m3u8",
"header": {"User-Agent": self.spider.mobile_ua},
},
)
def test_player_content_rejects_broken_play_id(self):
self.assertEqual(
self.spider.playerContent("金牌线路", "broken", {}),
{"parse": 0, "playUrl": "", "url": "", "header": {}},
)
py/金牌.py
+240
View File
@@ -0,0 +1,240 @@
# coding=utf-8
import hashlib
import json
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://m.jiabaide.cn"
self.app_key = "cb808529bae6b6be45ecfab29a4889bc"
self.mobile_ua = (
"Mozilla/5.0 (Linux; Android 11; Pixel 5) "
"AppleWebKit/537.36 (KHTML, like Gecko) "
"Chrome/90.0.4430.91 Mobile Safari/537.36"
)
self.headers = {
"User-Agent": self.mobile_ua,
"Referer": self.host + "/",
}
self.class_name_map = {
"typeList": ("type", "类型"),
"plotList": ("class", "剧情"),
"districtList": ("area", "地区"),
"languageList": ("lang", "语言"),
"yearList": ("year", "年份"),
"serialList": ("by", "排序"),
}
self.sort_values = [
{"n": "最近更新", "v": "1"},
{"n": "添加时间", "v": "2"},
{"n": "人气高低", "v": "3"},
{"n": "评分高低", "v": "4"},
]
def init(self, extend=""):
return None
def getName(self):
return self.name
def homeVideoContent(self):
return {"list": []}
def _obj_to_form(self, obj=None):
pairs = []
for key, value in (obj or {}).items():
if value in ("", None):
continue
pairs.append(f"{key}={value}")
return "&".join(pairs)
def _signed_headers(self, params=None):
current = str(int(time.time() * 1000))
sign_obj = dict(params or {})
sign_obj["key"] = self.app_key
sign_obj["t"] = current
obj_str = self._obj_to_form(sign_obj)
md5_value = hashlib.md5(obj_str.encode("utf-8")).hexdigest()
sign = hashlib.sha1(md5_value.encode("utf-8")).hexdigest()
headers = dict(self.headers)
headers["t"] = current
headers["sign"] = sign
return headers
def _fetch_json(self, path, params=None):
target = self.host + str(path)
query = self._obj_to_form(params or {})
if query:
target = target + "?" + query
response = self.fetch(target, headers=self._signed_headers(params or {}), timeout=10, verify=False)
if response.status_code != 200:
return {}
try:
data = json.loads(response.text or "{}")
except Exception:
return {}
if str(data.get("code", "200")) not in ("200", "0", ""):
return {}
return data
def _map_vod(self, item):
item = item or {}
pubdate = str(item.get("vodPubdate", "")).strip()
year = pubdate.split("-")[0] if pubdate else ""
remarks = [str(item.get("vodRemarks", "")).strip(), str(item.get("vodDoubanScore", "")).strip()]
return {
"vod_id": str(item.get("vodId", "")),
"vod_name": str(item.get("vodName", "")),
"vod_pic": str(item.get("vodPic", "")),
"vod_remarks": "_".join([value for value in remarks if value]),
"vod_year": year,
"type_id": str(item.get("typeId", "")),
"type_name": str(item.get("typeName", "")),
}
def _build_filters(self):
classes = []
filters = {}
type_res = self._fetch_json("/api/mw-movie/anonymous/get/filer/type")
for item in type_res.get("data", []):
classes.append(
{
"type_id": str(item.get("typeId", "")),
"type_name": str(item.get("typeName", "")),
}
)
filter_res = self._fetch_json("/api/mw-movie/anonymous/v1/get/filer/list")
filter_data = filter_res.get("data", {})
for cls in classes:
tid = cls["type_id"]
filters[tid] = []
current = (filter_data or {}).get(tid, {})
for raw_key, (key, name) in self.class_name_map.items():
values = [{"n": "全部", "v": ""}]
if raw_key == "serialList":
values.extend(self.sort_values)
else:
for item in current.get(raw_key, []):
label = str(item.get("itemText", "")).strip()
if not label:
continue
value_key = "itemValue" if raw_key == "typeList" else "itemText"
values.append({"n": label, "v": str(item.get(value_key, "")).strip()})
if len(values) > 1:
filters[tid].append({"key": key, "name": name, "value": values})
return classes, filters
def _page_result(self, items, pg, limit=30):
page = max(int(str(pg or 1)), 1)
return {"page": page, "limit": limit, "total": page * limit + len(items), "list": items}
def homeContent(self, filter):
classes, filters = self._build_filters()
home = self._fetch_json("/api/mw-movie/anonymous/home/hotSearch")
return {
"class": classes,
"filters": filters,
"list": [self._map_vod(item) for item in home.get("data", [])],
}
def categoryContent(self, tid, pg, filter, extend):
page = max(int(str(pg or 1)), 1)
values = dict(extend or {})
payload = {
"area": str(values.get("area", "")),
"lang": str(values.get("lang", "")),
"pageNum": str(page),
"pageSize": "30",
"sort": str(values.get("by", "1")),
"sortBy": "1",
"type": str(values.get("type", "")),
"type1": str(tid),
"v_class": str(values.get("class", "")),
"year": str(values.get("year", "")),
}
result = self._fetch_json("/api/mw-movie/anonymous/video/list", payload)
items = [self._map_vod(item) for item in ((result.get("data") or {}).get("list") or [])]
return self._page_result(items, page, 30)
def searchContent(self, key, quick, pg="1"):
page = max(int(str(pg or 1)), 1)
keyword = str(key or "").strip()
if not keyword:
return {"page": page, "limit": 30, "total": 0, "list": []}
result = self._fetch_json(
"/api/mw-movie/anonymous/video/searchByWordPageable",
{"keyword": keyword, "pageNum": str(page), "pageSize": "30"},
)
items = [self._map_vod(item) for item in ((result.get("data") or {}).get("list") or [])]
return self._page_result(items, page, 30)
def _build_play_id(self, vod_id, nid):
return f"{str(vod_id).strip()}@{str(nid).strip()}"
def _parse_play_id(self, value):
parts = str(value or "").split("@", 1)
if len(parts) != 2:
return {"id": "", "nid": ""}
sid = parts[0].strip()
nid = parts[1].strip()
if not sid or not nid:
return {"id": "", "nid": ""}
return {"id": sid, "nid": nid}
def detailContent(self, ids):
vod_id = str(ids[0] if isinstance(ids, list) and ids else ids or "").strip()
if not vod_id:
return {"list": []}
result = self._fetch_json("/api/mw-movie/anonymous/video/detail", {"id": vod_id})
data = result.get("data") or {}
if not data:
return {"list": []}
episodes = []
for item in data.get("episodeList", []):
name = str(item.get("name", "")).strip()
nid = str(item.get("nid", "")).strip()
if not name or not nid:
continue
episodes.append(f"{name}${self._build_play_id(data.get('vodId', vod_id), nid)}")
vod = {
"vod_id": str(data.get("vodId", vod_id)),
"vod_name": str(data.get("vodName", "")),
"vod_pic": str(data.get("vodPic", "")),
"type_name": str(data.get("vodClass", "")),
"vod_remarks": str(data.get("vodRemarks", "")),
"vod_year": str(data.get("vodYear", "")),
"vod_area": str(data.get("vodArea", "")),
"vod_lang": str(data.get("vodLang", "")),
"vod_director": str(data.get("vodDirector", "")),
"vod_actor": str(data.get("vodActor", "")),
"vod_content": str(data.get("vodContent", "")),
"vod_play_from": "金牌线路" if episodes else "",
"vod_play_url": "#".join(episodes),
}
return {"list": [vod]}
def playerContent(self, flag, id, vipFlags):
parsed = self._parse_play_id(id)
if not parsed["id"] or not parsed["nid"]:
return {"parse": 0, "playUrl": "", "url": "", "header": {}}
result = self._fetch_json(
"/api/mw-movie/anonymous/v2/video/episode/url",
{"clientType": "3", "id": parsed["id"], "nid": parsed["nid"]},
)
for item in ((result.get("data") or {}).get("list") or []):
url = str(item.get("url", "")).strip()
if url:
return {
"parse": 0,
"playUrl": "",
"url": url,
"header": {"User-Agent": self.mobile_ua},
}
return {"parse": 0, "playUrl": "", "url": "", "header": {}}