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

在线之家

Browse Source 
短剧优选
This commit is contained in:
Harold
2026-04-20 11:34:26 +08:00
parent d9b216d3dd
commit 9ba826e2c4
4 changed files with 1402 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files
py/tests/test_在线之家.py
+68
View File
@@ -1,6 +1,7 @@
import unittest
from importlib.machinery import SourceFileLoader
from pathlib import Path
from types import SimpleNamespace
from unittest.mock import patch
@@ -177,6 +178,19 @@ class TestZXZJSpider(unittest.TestCase):
"https://jx.zxzj.example/player?id=1",
)
def test_extract_iframe_url_reads_player_aaaa_target(self):
html = """
<script>
var player_aaaa = {
"url":"https://jx.zxzjys.com:9876/player-v2/common/demo/token"
};
</script>
"""
self.assertEqual(
self.spider._extract_iframe_url(html),
"https://jx.zxzjys.com:9876/player-v2/common/demo/token",
)
@patch.object(Spider, "_request_html")
def test_player_content_returns_decoded_direct_url_for_zxzj(self, mock_request_html):
mock_request_html.side_effect = [
@@ -200,6 +214,60 @@ class TestZXZJSpider(unittest.TestCase):
self.assertEqual(result["jx"], 1)
self.assertEqual(result["url"], "https://www.zxzjhd.com/vodplay/999-1-1.html")
@patch.object(Spider, "_request_html")
def test_player_content_falls_back_to_parser_page_when_new_line5_target_cannot_be_decoded(self, mock_request_html):
mock_request_html.side_effect = [
"""
<script>
var player_aaaa = {
"url":"https://jx.zxzjys.com:9876/player-v2/common/demo/token"
};
</script>
""",
"",
]
result = self.spider.playerContent("zxzj", "vodplay/4627-1-1.html", {})
self.assertEqual(result["parse"], 1)
self.assertEqual(result["jx"], 1)
self.assertEqual(result["url"], "https://jx.zxzjys.com:9876/player-v2/common/demo/token")
self.assertEqual(result["header"]["Referer"], "https://www.zxzjhd.com/vodplay/4627-1-1.html")
@patch.object(Spider, "fetch")
def test_player_content_uses_browser_like_headers_for_new_line5_parser_request(self, mock_fetch):
play_page = """
<script>
var player_aaaa = {
"url":"https://jx.zxzjys.com:9876/player-v2/common/demo/token"
};
</script>
"""
parser_page = """
<script>
var result_v2 = {
"data":"d6f636e256c607d6168756e2f67464544434241456469667f2f2a33707474786"
};
</script>
"""
def fake_fetch(url, headers=None, timeout=10):
if url == "https://www.zxzjhd.com/vodplay/4627-1-1.html":
return SimpleNamespace(status_code=200, text=play_page)
if url == "https://jx.zxzjys.com:9876/player-v2/common/demo/token":
self.assertEqual(headers["Referer"], "https://www.zxzjhd.com/vodplay/4627-1-1.html")
self.assertEqual(headers["Origin"], "https://www.zxzjhd.com")
self.assertEqual(headers["Sec-Fetch-Dest"], "iframe")
self.assertEqual(headers["Sec-Fetch-Mode"], "navigate")
self.assertEqual(headers["Sec-Fetch-Site"], "cross-site")
self.assertEqual(headers["Upgrade-Insecure-Requests"], "1")
return SimpleNamespace(status_code=200, text=parser_page)
raise AssertionError(url)
mock_fetch.side_effect = fake_fetch
result = self.spider.playerContent("zxzj", "vodplay/4627-1-1.html", {})
self.assertEqual(result["parse"], 0)
self.assertEqual(result["jx"], 0)
self.assertEqual(result["url"], "https://video.example.com")
def test_player_content_returns_direct_netdisk_link(self):
result = self.spider.playerContent("quark", "https://pan.quark.cn/s/demo", {})
self.assertEqual(
py/tests/test_短剧优选.py
+241
View File
@@ -0,0 +1,241 @@
import unittest
import json
from importlib.machinery import SourceFileLoader
from pathlib import Path
from unittest.mock import patch, MagicMock
ROOT = Path(__file__).resolve().parents[1]
MODULE = SourceFileLoader("duanjuyx_spider", str(ROOT / "短剧优选.py")).load_module()
Spider = MODULE.Spider
class TestDuanjuYouxuanSpider(unittest.TestCase):
def setUp(self):
Spider._instance = None
self.spider = Spider()
self.spider.init()
def test_home_content_has_active_platforms(self):
content = self.spider.homeContent(False)
ids = [c["type_id"] for c in content["class"]]
self.assertIn("七猫", ids)
self.assertIn("星芽", ids)
self.assertIn("西饭", ids)
self.assertIn("七星", ids)
self.assertIn("短剧网", ids)
def test_home_content_has_filters(self):
content = self.spider.homeContent(False)
self.assertIn("七猫", content["filters"])
self.assertIn("短剧网", content["filters"])
qm_filter = content["filters"]["七猫"][0]
self.assertEqual(qm_filter["key"], "area")
self.assertTrue(len(qm_filter["value"]) > 50)
def test_home_video_content_returns_empty(self):
self.assertEqual(self.spider.homeVideoContent(), {"list": []})
def test_player_tianquan(self):
result = self.spider.playerContent("甜圈短剧", "vid123", {})
self.assertEqual(result["parse"], 0)
self.assertIn("video_id=vid123", result["url"])
def test_player_xingya_passthrough(self):
result = self.spider.playerContent("星芽短剧", "https://example.com/video.m3u8", {})
self.assertEqual(result["url"], "https://example.com/video.m3u8")
def test_player_weiguan_selects_super(self):
play_setting = {"super": "s.mp4", "high": "h.mp4", "normal": "n.mp4"}
result = self.spider.playerContent("围观短剧", json.dumps(play_setting), {})
self.assertEqual(result["url"], "s.mp4")
def test_player_weiguan_falls_back_to_high(self):
play_setting = {"high": "h.mp4", "normal": "n.mp4"}
result = self.spider.playerContent("围观短剧", json.dumps(play_setting), {})
self.assertEqual(result["url"], "h.mp4")
def test_player_weiguan_invalid_json(self):
result = self.spider.playerContent("围观短剧", "not-json", {})
self.assertEqual(result["url"], "not-json")
def test_player_duanjuwang_passthrough(self):
result = self.spider.playerContent("短剧网", "push://pan.quark.cn/abc", {})
self.assertEqual(result["parse"], 0)
self.assertEqual(result["url"], "push://pan.quark.cn/abc")
@patch.object(Spider, "_get_text")
def test_player_juwang_extracts_m3u8(self, mock_text):
mock_text.return_value = 'var data = {"wwm3u8": "https:\\/\\/cdn.example.com\\/video.m3u8"};'
result = self.spider.playerContent("剧王短剧", "https://djw1.com/play/123", {})
self.assertEqual(result["url"], "https://cdn.example.com/video.m3u8")
self.assertEqual(result["parse"], 0)
@patch.object(Spider, "_get_json")
def test_category_tianquan(self, mock_json):
mock_json.return_value = {
"data": [
{"book_id": "b1", "title": "逆袭人生", "cover": "http://pic/1.jpg", "sub_title": "60集"},
{"book_id": "b2", "title": "霸总归来", "cover": "http://pic/2.jpg", "sub_title": "80集"},
]
}
result = self.spider.categoryContent("甜圈", "1", False, {"area": "逆袭"})
self.assertEqual(len(result["list"]), 2)
self.assertEqual(result["list"][0]["vod_id"], "甜圈@b1")
self.assertEqual(result["list"][0]["vod_name"], "逆袭人生")
self.assertEqual(result["pagecount"], 2)
@patch.object(Spider, "_post_json")
def test_category_weiguan(self, mock_json):
mock_json.return_value = {
"data": [
{"oneId": "wg1", "title": "都市情缘", "vertPoster": "http://pic/wg.jpg", "episodeCount": 50},
]
}
result = self.spider.categoryContent("围观", "1", False, {"area": "都市"})
self.assertEqual(len(result["list"]), 1)
self.assertEqual(result["list"][0]["vod_id"], "围观@wg1")
@patch.object(Spider, "_get_text")
def test_category_duanjuwang(self, mock_text):
mock_text.return_value = """
<li class="col-6">
<h3 class="f-14"><a href="/detail/123.html">短剧名（60集完结）</a></h3>
<img class="lazy" data-original="/cover.jpg" />
</li>
"""
result = self.spider.categoryContent("短剧网", "1", False, {"area": "1"})
self.assertEqual(len(result["list"]), 1)
self.assertEqual(result["list"][0]["vod_id"], "短剧网@https://sm3.cc/detail/123.html")
self.assertEqual(result["list"][0]["vod_name"], "短剧名（60集完结）")
self.assertEqual(result["list"][0]["vod_remarks"], "")
self.assertEqual(result["pagecount"], 2)
@patch.object(Spider, "_get_json")
def test_category_empty_result(self, mock_json):
mock_json.return_value = {"data": []}
result = self.spider.categoryContent("甜圈", "2", False, {"area": "逆袭"})
self.assertEqual(result["list"], [])
self.assertEqual(result["pagecount"], 2)
@patch.object(Spider, "_get_json")
def test_detail_tianquan(self, mock_json):
mock_json.return_value = {
"book_name": "甜剧1号", "book_pic": "http://pic/t.jpg", "desc": "剧情简介",
"duration": "60分钟", "author": "作者", "time": "2025-01-01",
"data": [
{"title": "第1集", "video_id": "v1"},
{"title": "第2集", "video_id": "v2"},
],
}
result = self.spider.detailContent(["甜圈@b1"])
vod = result["list"][0]
self.assertEqual(vod["vod_name"], "甜剧1号")
self.assertEqual(vod["vod_play_from"], "甜圈短剧")
self.assertIn("第1集$v1", vod["vod_play_url"])
self.assertIn("第2集$v2", vod["vod_play_url"])
@patch.object(Spider, "_get_json")
def test_detail_weiguan(self, mock_json):
mock_json.return_value = {
"data": [
{"title": "逆袭", "playOrder": 1, "playSetting": {"super": "s1.mp4"}, "vertPoster": "p.jpg", "collectionCount": 100},
{"title": "逆袭", "playOrder": 2, "playSetting": {"super": "s2.mp4"}, "vertPoster": "p.jpg", "collectionCount": 100},
]
}
result = self.spider.detailContent(["围观@wg1"])
vod = result["list"][0]
self.assertEqual(vod["vod_name"], "逆袭")
self.assertEqual(vod["vod_remarks"], "共2集")
self.assertIn("围观短剧", vod["vod_play_from"])
@patch.object(Spider, "_get_text")
def test_detail_juwang(self, mock_text):
mock_text.return_value = """
<html><body>
<h1>剧王短剧名</h1>
<img src="/poster.jpg" />
<div class="info-detail">剧情简介</div>
<div class="ep-list">
<a href="/play/1-1.html">第1集</a>
<a href="/play/1-2.html">第2集</a>
</div>
</body></html>
"""
result = self.spider.detailContent(["剧王@/show/123"])
vod = result["list"][0]
self.assertEqual(vod["vod_name"], "剧王短剧名")
self.assertEqual(vod["vod_content"], "剧情简介")
self.assertEqual(vod["vod_play_from"], "剧王短剧")
self.assertIn("第1集$/play/1-1.html", vod["vod_play_url"])
@patch.object(Spider, "_get_text")
def test_detail_duanjuwang(self, mock_text):
mock_text.return_value = """
<html><body>
<h1>网盘短剧名</h1>
<img class="lazy" data-original="/poster.jpg" />
<div class="content">
<a href="https://pan.quark.cn/s/abc123">夸克链接</a>
<a href="https://pan.baidu.com/s/xyz789">百度链接</a>
</div>
</body></html>
"""
result = self.spider.detailContent(["短剧网@https://sm3.cc/detail/123"])
vod = result["list"][0]
self.assertEqual(vod["vod_name"], "网盘短剧名")
self.assertEqual(vod["vod_content"], "此为推送网盘规则")
self.assertIn("quark", vod["vod_play_from"])
self.assertIn("baidu", vod["vod_play_from"])
self.assertIn("quark$https://pan.quark.cn/s/abc123", vod["vod_play_url"])
@patch.object(Spider, "_get_text")
def test_detail_duanjuwang_no_netdisk_links(self, mock_text):
mock_text.return_value = """
<html><body>
<h1>无链接短剧</h1>
<div class="content"><p>暂无链接</p></div>
</body></html>
"""
result = self.spider.detailContent(["短剧网@https://sm3.cc/detail/456"])
vod = result["list"][0]
self.assertEqual(vod["vod_name"], "无链接短剧")
self.assertEqual(vod["vod_play_from"], "短剧网")
self.assertEqual(vod["vod_play_url"], "")
@patch.object(Spider, "_get_json")
def test_detail_failure_returns_error_vod(self, mock_json):
mock_json.side_effect = Exception("network error")
result = self.spider.detailContent(["甜圈@missing"])
vod = result["list"][0]
self.assertIn("详情加载失败", vod["vod_name"])
@patch.object(Spider, "_aggregate_search")
def test_search_filters_by_keyword(self, mock_search):
mock_search.return_value = [
{"vod_id": "甜圈@1", "vod_name": "逆袭人生", "vod_pic": "", "vod_remarks": ""},
{"vod_id": "七猫@2", "vod_name": "霸道总裁", "vod_pic": "", "vod_remarks": ""},
]
result = self.spider.searchContent("逆袭", False, "1")
self.assertEqual(len(result["list"]), 1)
self.assertEqual(result["list"][0]["vod_name"], "逆袭人生")
@patch.object(Spider, "_aggregate_search")
def test_search_empty_keyword(self, mock_search):
result = self.spider.searchContent("", False, "1")
self.assertEqual(result["list"], [])
mock_search.assert_not_called()
def test_filter_defaults(self):
self.assertEqual(self.spider.filter_defaults["七猫"]["area"], "0")
self.assertEqual(self.spider.filter_defaults["围观"]["area"], "都市")
self.assertEqual(self.spider.filter_defaults["短剧网"]["area"], "1")
def test_identify_disk(self):
self.assertEqual(self.spider._identify_disk("https://pan.quark.cn/s/abc"), "quark")
self.assertEqual(self.spider._identify_disk("https://pan.baidu.com/s/xyz"), "baidu")
self.assertEqual(self.spider._identify_disk("https://drive.uc.cn/s/123"), "uc")
self.assertEqual(self.spider._identify_disk("https://example.com/video.mp4"), "")
if __name__ == "__main__":
unittest.main()
py/在线之家.py
+35 -6
View File
@@ -233,15 +233,29 @@ class Spider(BaseSpider):
return value
return value
def _request_html(self, path_or_url, referer=None):
def _request_html(self, path_or_url, referer=None, extra_headers=None):
target = path_or_url if str(path_or_url).startswith("http") else self._build_url(path_or_url)
headers = dict(self.headers)
headers["Referer"] = referer or self.headers["Referer"]
if isinstance(extra_headers, dict):
headers.update(extra_headers)
response = self.fetch(target, headers=headers, timeout=10)
if response.status_code != 200:
return ""
return self._fix_json_wrapped_html(response.text or "")
def _build_player_v2_headers(self):
return {
"Origin": self.host,
"Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,*/*;q=0.8",
"Accept-Language": "zh-CN,zh;q=0.9",
"Sec-Fetch-Dest": "iframe",
"Sec-Fetch-Mode": "navigate",
"Sec-Fetch-Site": "cross-site",
"Sec-Fetch-User": "?1",
"Upgrade-Insecure-Requests": "1",
}
def _extract_vod_id(self, href):
raw = str(href or "").strip()
matched = re.search(r"/?(voddetail/\d+\.html)", raw)
@@ -458,9 +472,11 @@ class Spider(BaseSpider):
return ""
def _extract_iframe_url(self, html):
matched = re.search(r'"url"\s*:\s*"(https:[^"]*?jx\.zxzj[^"]*?)"', str(html or ""))
matched = re.search(r'"url"\s*:\s*"(https:[^"]*?)"', str(html or ""))
if matched:
return matched.group(1).replace("\\/", "/")
value = matched.group(1).replace("\\/", "/")
if any(token in value for token in ["jx.zxzj", "player-v2", "/player", "/embed"]):
return value
matched = re.search(r"player_[a-z0-9_]+\s*=\s*(\{[\s\S]*?\})\s*;?", str(html or ""), re.I)
if not matched:
return ""
@@ -469,7 +485,7 @@ class Spider(BaseSpider):
except Exception:
return ""
value = str(payload.get("url") or "").replace("\\/", "/")
return value if "jx.zxzj" in value else ""
return value if any(token in value for token in ["jx.zxzj", "player-v2", "/player", "/embed"]) else ""
def _extract_result_v2_data(self, html):
matched = re.search(r"result_v2\s*=\s*(\{[\s\S]*?\})\s*;", str(html or ""))
@@ -493,11 +509,24 @@ class Spider(BaseSpider):
if not iframe_url:
return {"parse": 1, "jx": 1, "playUrl": "", "url": play_url, "header": self.headers}
iframe_html = self._request_html(iframe_url, referer=play_url)
iframe_headers = self._build_player_v2_headers() if "player-v2" in iframe_url else None
iframe_html = self._request_html(iframe_url, referer=play_url, extra_headers=iframe_headers)
encrypted = self._extract_result_v2_data(iframe_html)
final_url = self._decrypt_url(encrypted)
if not final_url:
return {"parse": 1, "jx": 1, "playUrl": "", "url": play_url, "header": self.headers}
fallback_url = iframe_url if "player-v2" in iframe_url else play_url
fallback_header = (
{"Referer": play_url, "User-Agent": self.headers["User-Agent"]}
if fallback_url == iframe_url
else self.headers
)
return {
"parse": 1,
"jx": 1,
"playUrl": "",
"url": fallback_url,
"header": fallback_header,
}
return {
"parse": 0,
"jx": 0,
py/短剧优选.py
+1058
View File
@@ -0,0 +1,1058 @@
# coding=utf-8
import hashlib
import json
import re
import sys
import time
from urllib.parse import quote, urlencode
from base.spider import Spider as BaseSpider
sys.path.append("..")
CHAR_MAP = {
"+": "P", "/": "X", "0": "M", "1": "U", "2": "l", "3": "E", "4": "r", "5": "Y",
"6": "W", "7": "b", "8": "d", "9": "J", "A": "9", "B": "s", "C": "a", "D": "I",
"E": "0", "F": "o", "G": "y", "H": "_", "I": "H", "J": "G", "K": "i", "L": "t",
"M": "g", "N": "N", "O": "A", "P": "8", "Q": "F", "R": "k", "S": "3", "T": "h",
"U": "f", "V": "R", "W": "q", "X": "C", "Y": "4", "Z": "p", "a": "m", "b": "B",
"c": "O", "d": "u", "e": "c", "f": "6", "g": "K", "h": "x", "i": "5", "j": "T",
"k": "-", "l": "2", "m": "z", "n": "S", "o": "Z", "p": "1", "q": "V", "r": "v",
"s": "j", "t": "Q", "u": "7", "v": "D", "w": "w", "x": "n", "y": "L", "z": "e",
}
QM_KEYS = "d3dGiJc651gSQ8w1"
XIFAN_SESSION = (
"eyJpbmZvIjp7InVpZCI6IiIsInJ0IjoiMTc0MDY1ODI5NCIsInVuIjoiT1BHXzFlZGQ5OTZhNjQ3"
"ZTQ1MjU4Nzc1MTE2YzFkNzViN2QwIiwiZnQiOiIxNzQwNjU4Mjk0In19"
)
XIFAN_FEEDS = (
"eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9."
"eyJ1dHlwIjowLCJidWlkIjoxNjMzOTY4MTI2MTQ4NjQxNTM2LCJhdWQiOiJkcmFtYSIsInZlciI6MiwicmF0IjoxNzQwNjU4Mjk0LC"
"J1bm0iOiJPUEdfMWVkZDk5NmE2NDdlNDUyNTg3NzUxMTY2YzFkNzViN2QwIiwiZXhwIjoxNzQxMjYzMDk0LCJkYyI6Imd6cXkifQ."
"JS3QY6ER0P2cQSxAE_OGKSMIWNAMsYUZ3mJTnEpf-Rc"
)
class Spider(BaseSpider):
def __init__(self):
self.name = "短剧优选"
self.xingya_headers = {}
self.default_headers = {
"User-Agent": "okhttp/3.12.11",
"Content-Type": "application/json; charset=utf-8",
}
self.platforms = {
"七猫": {
"host": "https://api-store.qmplaylet.com",
"url1": "/api/v1/playlet/index",
"url2": "https://api-read.qmplaylet.com/player/api/v1/playlet/info",
"search": "/api/v1/playlet/search",
},
"星芽": {
"host": "https://app.whjzjx.cn",
"url1": "/cloud/v2/theater/home_page?theater_class_id",
"url2": "/v2/theater_parent/detail",
"search": "/v3/search",
"login": "https://u.shytkjgs.com/user/v1/account/login",
},
"西饭": {
"host": "https://xifan-api-cn.youlishipin.com",
"url1": "/xifan/drama/portalPage",
"url2": "/xifan/drama/getDuanjuInfo",
"search": "/xifan/search/getSearchList",
},
"围观": {
"host": "https://api.drama.9ddm.com",
"url1": "/drama/home/shortVideoTags",
"url2": "/drama/home/shortVideoDetail",
"search": "/drama/home/search",
},
"剧王": {
"host": "https://djw1.com",
"url1": "/all/",
"search": "/search",
},
"七星": {
"host": "https://app.whjzjx.cn",
"url1": "/v1/theater/home_page?theater_class_id",
"url2": "/v2/theater_parent/detail",
"search": "/v3/search",
"login": "https://u.shytkjgs.com/user/v1/account/login",
},
"甜圈": {
"host": "https://mov.cenguigui.cn",
"url1": "/duanju/api.php?classname",
"url2": "/duanju/api.php?book_id",
"search": "/duanju/api.php?name",
},
"短剧网": {
"host": "https://sm3.cc",
},
}
self.classes = [
{"type_id": "七猫", "type_name": "七猫短剧"},
{"type_id": "星芽", "type_name": "星芽短剧"},
{"type_id": "西饭", "type_name": "西饭短剧"},
# {"type_id": "围观", "type_name": "围观短剧"},
# {"type_id": "剧王", "type_name": "剧王短剧"},
{"type_id": "七星", "type_name": "七星短剧"},
# {"type_id": "甜圈", "type_name": "甜圈短剧"},
{"type_id": "短剧网", "type_name": "短剧网"},
]
self.filters = {
"七猫": [self._filter_area("分类", [
("全部", "0"), ("新剧", "-1"), ("都市情感", "1273"), ("古装", "1272"),
("都市", "571"), ("玄幻仙侠", "1286"), ("青春校园", "1288"), ("年代", "572"),
("武侠", "371"), ("乡村", "590"), ("科幻", "594"), ("民国", "573"),
("奇幻", "570"), ("末世", "556"), ("二次元", "1289"), ("逆袭", "400"),
("复仇", "795"), ("重生", "784"), ("穿越", "373"), ("女性成长", "1294"),
("家庭", "670"), ("闪婚", "480"), ("强者回归", "402"), ("打脸虐渣", "716"),
("追妻火葬场", "715"), ("马甲", "558"), ("职场", "724"), ("高手下山", "1299"),
("穿书", "338"), ("系统", "787"), ("娱乐明星", "1295"), ("异能", "727"),
("致富", "492"), ("种田经商", "1291"), ("伦理", "1293"), ("社会话题", "1290"),
("脑洞", "526"), ("豪门总裁", "624"), ("萌宝", "356"), ("真假千金", "812"),
("战神", "527"), ("赘婿", "36"), ("神豪", "37"), ("小人物", "1296"),
("神医", "1269"), ("权谋", "790"), ("女帝", "617"), ("团宠", "545"),
("欢喜冤家", "464"), ("替身", "712"), ("银发", "1297"), ("兵王", "28"),
("虐恋", "16"), ("甜宠", "21"), ("搞笑", "793"), ("宅斗", "342"),
("宫斗", "343"), ("悬疑", "27"), ("商战", "723"), ("灵异", "1287"),
])],
"星芽": [self._filter_area("分类", [
("剧场", "1"), ("热播", "2"), ("新剧", "3"),
("阳光剧场", "5"), ("星选好剧", "7"), ("会员专享", "8"),
])],
"西饭": [self._filter_area("分类", [
("全部", ""), ("都市", "68@都市"), ("青春", "68@青春"),
("现代言情", "81@现代言情"), ("豪门", "81@豪门"),
("大女主", "80@大女主"), ("逆袭", "79@逆袭"),
("打脸虐渣", "79@打脸虐渣"), ("穿越", "81@穿越"),
])],
"围观": [self._filter_area("分类", [
("都市", "都市"), ("逆袭", "逆袭"), ("家庭", "家庭"),
("复仇", "复仇"), ("古装", "古装"), ("甜宠", "甜宠"),
("悬疑", "悬疑"), ("爱情", "爱情"), ("虐恋", "虐恋"),
("动漫", "动漫"), ("重生", "重生"), ("奇幻", "奇幻"),
("总裁", "总裁"), ("穿越", "穿越"), ("萌宝", "萌宝"),
("乡村", "乡村"), ("修仙", "修仙"), ("异能", "异能"),
("战神", "战神"), ("职场", "职场"), ("年代", "年代"),
("正能量", "正能量"), ("神豪", "神豪"), ("喜剧", "喜剧"),
("青春", "青春"), ("宫廷", "宫廷"), ("武侠", "武侠"),
("神医", "神医"), ("科幻", "科幻"), ("赘婿", "赘婿"),
])],
"剧王": [self._filter_area("分类", [
("逆袭", "/tag/逆袭/"), ("霸总", "/tag/霸总/"),
("战神", "/tag/战神/"), ("穿越", "/tag/穿越/"),
("重生", "/tag/重生/"), ("赘婿", "/tag/赘婿/"),
("甜宠", "/tag/甜宠/"), ("虐恋", "/tag/虐恋/"),
("古装", "/tag/古装/"), ("都市", "/tag/都市/"),
("神医", "/tag/神医/"), ("玄幻", "/tag/玄幻/"),
("修仙", "/tag/修仙/"), ("女频", "/tag/女频/"),
("男频", "/tag/男频/"), ("情感", "/tag/情感/"),
("复仇", "/tag/复仇/"), ("现代", "/tag/现代/"),
("萌宝", "/tag/萌宝/"), ("虐渣", "/tag/虐渣/"),
("系统", "/tag/系统/"), ("打脸", "/tag/打脸/"),
("奇幻", "/tag/奇幻/"), ("家庭", "/tag/家庭/"),
("亲情", "/tag/亲情/"), ("言情", "/tag/言情/"),
("年代", "/tag/年代/"), ("悬疑", "/tag/悬疑/"),
("豪门", "/tag/豪门/"),
])],
"七星": [self._filter_area("分类", [
("七星剧场", "1"), ("七星新剧", "3"), ("七星热播", "2"),
("七星星选", "7"), ("七星阳光", "5"),
])],
"甜圈": [self._filter_area("分类", [
("逆袭", "逆袭"), ("霸总", "霸总"), ("现代言情", "现代言情"),
("打脸虐渣", "打脸虐渣"), ("豪门恩怨", "豪门恩怨"),
("神豪", "神豪"), ("马甲", "马甲"), ("都市日常", "都市日常"),
("战神归来", "战神归来"), ("小人物", "小人物"),
("女性成长", "女性成长"), ("大女主", "大女主"),
("穿越", "穿越"), ("都市修仙", "都市修仙"),
("强者回归", "强者回归"), ("亲情", "亲情"), ("古装", "古装"),
("重生", "重生"), ("闪婚", "闪婚"), ("赘婿逆袭", "赘婿逆袭"),
("虐恋", "虐恋"), ("追妻", "追妻"), ("天下无敌", "天下无敌"),
("家庭伦理", "家庭伦理"), ("萌宝", "萌宝"), ("古风权谋", "古风权谋"),
("职场", "职场"), ("奇幻脑洞", "奇幻脑洞"), ("异能", "异能"),
("无敌神医", "无敌神医"), ("古风言情", "古风言情"),
("传承觉醒", "传承觉醒"), ("现言甜宠", "现言甜宠"),
("奇幻爱情", "奇幻爱情"), ("乡村", "乡村"),
("历史古代", "历史古代"), ("王妃", "王妃"),
("高手下山", "高手下山"), ("娱乐圈", "娱乐圈"),
("强强联合", "强强联合"), ("破镜重圆", "破镜重圆"),
("暗恋成真", "暗恋成真"), ("民国", "民国"),
("欢喜冤家", "欢喜冤家"), ("系统", "系统"),
("真假千金", "真假千金"), ("龙王", "龙王"), ("校园", "校园"),
("穿书", "穿书"), ("女帝", "女帝"), ("团宠", "团宠"),
("年代爱情", "年代爱情"), ("玄幻仙侠", "玄幻仙侠"),
("青梅竹马", "青梅竹马"), ("悬疑推理", "悬疑推理"),
("皇后", "皇后"), ("替身", "替身"), ("大叔", "大叔"),
("喜剧", "喜剧"), ("剧情", "剧情"),
])],
"短剧网": [self._filter_area("分类", [
("短剧大全", "1"), ("更新短剧", "2"),
])],
}
self.filter_defaults = {
"七猫": {"area": "0"}, "星芽": {"area": "1"}, "西饭": {"area": ""},
"围观": {"area": "都市"}, "剧王": {"area": "/tag/逆袭/"},
"七星": {"area": "1"}, "甜圈": {"area": "逆袭"},
"短剧网": {"area": "1"},
}
def init(self, extend=""):
self._init_xingya_token()
return None
def getName(self):
return self.name
def homeContent(self, filter):
return {"class": self.classes, "filters": self.filters}
def homeVideoContent(self):
return {"list": []}
def categoryContent(self, tid, pg, filter, extend):
page = int(pg)
ext = extend or {}
area = ext.get("area") or self.filter_defaults.get(tid, {}).get("area", "")
try:
items = self._fetch_category(tid, page, area)
except Exception as e:
self.log(f"categoryContent failed ({tid}): {e}")
items = []
pagecount = page + 1 if items else page
return {
"list": items,
"page": page,
"pagecount": pagecount,
"total": pagecount * len(items) if items else 0,
}
def detailContent(self, ids):
result = {"list": []}
for raw_id in ids:
parts = raw_id.split("@", 1)
platform = parts[0]
detail_id = parts[1] if len(parts) > 1 else ""
try:
vod = self._fetch_detail(platform, detail_id, raw_id)
except Exception as e:
self.log(f"detailContent failed ({platform}): {e}")
vod = {
"vod_id": raw_id,
"vod_name": f"{platform}：详情加载失败",
"vod_remarks": str(e),
}
result["list"].append(vod)
return result
def searchContent(self, key, quick, pg="1"):
page = int(pg)
keyword = key.strip()
if not keyword:
return {"list": [], "page": page, "pagecount": 0, "total": 0}
results = self._aggregate_search(keyword, page)
filtered = [
r for r in results
if keyword.lower() in (r.get("vod_name") or "").lower()
]
limit = 30
offset = (page - 1) * limit
sliced = filtered[offset:offset + limit]
total_pages = max(1, (len(filtered) + limit - 1) // limit) if filtered else 0
return {
"list": sliced,
"page": page,
"pagecount": total_pages,
"total": len(filtered),
}
def playerContent(self, flag, id, vipFlags):
if "甜圈" in flag:
return {"parse": 0, "jx": 0, "url": f"https://mov.cenguigui.cn/duanju/api.php?video_id={id}&type=mp4"}
if "星芽" in flag or "七猫" in flag or "西饭" in flag:
return {"parse": 0, "jx": 0, "url": id}
if "七星" in flag:
return {"parse": 0, "jx": 0, "url": id}
if "围观" in flag:
return self._play_weiguan(id)
if "剧王" in flag:
return self._play_juwang(id)
if "短剧网" in flag:
return {"parse": 0, "jx": 0, "url": id}
return {"parse": 0, "jx": 0, "url": id}
# --- helpers ---
@staticmethod
def _filter_area(name, values):
return {"key": "area", "name": name, "value": [{"n": n, "v": v} for n, v in values]}
def _s(self, val):
return "" if val is None else str(val)
def _clean(self, text):
return re.sub(r"\s+", " ", str(text or "").replace("\xa0", " ")).strip()
def _get_json(self, url, headers=None, params=None, timeout=6):
h = dict(self.default_headers)
if headers:
h.update(headers)
rsp = self.fetch(url, headers=h, params=params, timeout=timeout, verify=False)
if rsp.status_code != 200:
return {}
return rsp.json()
def _post_json(self, url, body=None, headers=None, timeout=6):
h = dict(self.default_headers)
if headers:
h.update(headers)
rsp = self.post(url, json=body, headers=h, timeout=timeout, verify=False)
if rsp.status_code != 200:
return {}
return rsp.json()
def _get_text(self, url, headers=None, timeout=6):
h = dict(self.default_headers)
if headers:
h.update(headers)
rsp = self.fetch(url, headers=h, timeout=timeout, verify=False)
if rsp.status_code != 200:
return ""
return rsp.text or ""
# --- 七猫 signing ---
@staticmethod
def _md5(text):
return hashlib.md5(text.encode("utf-8")).hexdigest()
def _qm_params_and_sign(self):
session_id = str(int(time.time() * 1000))
data = {
"static_score": "0.8", "uuid": "00000000-7fc7-08dc-0000-000000000000",
"device-id": "20250220125449b9b8cac84c2dd3d035c9052a2572f7dd0122edde3cc42a70",
"mac": "", "sourceuid": "aa7de295aad621a6", "refresh-type": "0",
"model": "22021211RC", "wlb-imei": "", "client-id": "aa7de295aad621a6",
"brand": "Redmi", "oaid": "", "oaid-no-cache": "", "sys-ver": "12",
"trusted-id": "", "phone-level": "H", "imei": "",
"wlb-uid": "aa7de295aad621a6", "session-id": session_id,
}
import base64
b64 = base64.b64encode(json.dumps(data, separators=(",", ":")).encode("utf-8")).decode("utf-8")
qm_params = "".join(CHAR_MAP.get(c, c) for c in b64)
sign_str = (
f"AUTHORIZATION=app-version=10001application-id=com.duoduo.read"
f"channel=unknownis-white=net-env=5platform=android"
f"qm-params={qm_params}reg={QM_KEYS}"
)
return qm_params, self._md5(sign_str)
def _qm_headers(self):
qm_params, sign = self._qm_params_and_sign()
return {
"net-env": "5", "reg": "", "channel": "unknown", "is-white": "",
"platform": "android", "application-id": "com.duoduo.read",
"authorization": "", "app-version": "10001",
"user-agent": "webviewversion/0", "qm-params": qm_params, "sign": sign,
}
# --- 星芽/七星 token ---
def _init_xingya_token(self):
if self.xingya_headers:
return
try:
plat = self.platforms["星芽"]
body = {"device": "24250683a3bdb3f118dff25ba4b1cba1a"}
h = {"User-Agent": "okhttp/4.10.0", "platform": "1", "Content-Type": "application/json"}
res = self._post_json(plat["login"], body=body, headers=h)
token = ""
data = res.get("data") or {}
if isinstance(data, dict):
token = data.get("token") or (data.get("data") or {}).get("token") or ""
if token:
self.xingya_headers = dict(self.default_headers)
self.xingya_headers["authorization"] = token
except Exception:
pass
# --- 西饭 URL builder ---
def _xifan_params(self, extra=None):
ts = int(time.time())
p = {
"quickEngineVersion": -1, "scene": "", "categoryVersion": 1,
"density": 1.5, "pageID": "page_theater", "version": 2001001,
"androidVersionCode": 28,
"requestId": f"{ts}aa498144140ef297",
"appId": "drama", "teenMode": False, "userBaseMode": False,
"session": XIFAN_SESSION, "feedssession": XIFAN_FEEDS,
}
if extra:
p.update(extra)
return p
# --- category per platform ---
def _fetch_category(self, tid, page, area):
plat = self.platforms.get(tid)
if not plat:
return []
handler = {
"七猫": self._cat_qimao,
"星芽": self._cat_xingya,
"西饭": self._cat_xifan,
"围观": self._cat_weiguan,
"剧王": self._cat_juwang,
"七星": self._cat_qixing,
"甜圈": self._cat_tianquan,
"短剧网": self._cat_duanjuwang,
}.get(tid)
if handler:
return handler(plat, page, area)
return []
def _cat_qimao(self, plat, page, area):
sign_str = f"operation=1playlet_privacy=1tag_id={area}{QM_KEYS}"
sign = self._md5(sign_str)
url = f"{plat['host']}{plat['url1']}?tag_id={area}&playlet_privacy=1&operation=1&sign={sign}"
res = self._get_json(url, headers=self._qm_headers())
items = []
for it in (res.get("data") or {}).get("list") or []:
items.append({
"vod_id": f"七猫@{quote(str(it.get('playlet_id', '')))}",
"vod_name": self._s(it.get("title")),
"vod_pic": self._s(it.get("image_link")),
"vod_remarks": f"{it.get('total_episode_num', 0)}集",
})
return items
def _cat_xingya(self, plat, page, area):
url = f"{plat['host']}{plat['url1']}={area}&type=1&class2_ids=0&page_num={page}&page_size=24"
res = self._get_json(url, headers=self.xingya_headers)
items = []
for it in (res.get("data") or {}).get("list") or []:
tid = it.get("theater", {}).get("id", "")
detail_url = f"{plat['host']}{plat['url2']}?theater_parent_id={tid}"
items.append({
"vod_id": f"星芽@{detail_url}",
"vod_name": self._s(it.get("theater", {}).get("title")),
"vod_pic": self._s(it.get("theater", {}).get("cover_url")),
"vod_remarks": f"{it.get('theater', {}).get('total', 0)}集",
})
return items
def _cat_xifan(self, plat, page, area):
type_id, type_name = (area.split("@", 1) + [""])[:2]
params = self._xifan_params({
"reqType": "aggregationPage",
"offset": (page - 1) * 24,
"categoryId": type_id,
"categoryNames": type_name,
})
url = f"{plat['host']}{plat['url1']}?{urlencode(params)}"
res = self._get_json(url)
items = []
for soup in (res.get("result") or {}).get("elements") or []:
for vod in soup.get("contents") or []:
dj = vod.get("duanjuVo") or {}
items.append({
"vod_id": f"西饭@{dj.get('duanjuId', '')}#{dj.get('source', '')}",
"vod_name": self._s(dj.get("title")),
"vod_pic": self._s(dj.get("coverImageUrl")),
"vod_remarks": f"{dj.get('total', 0)}集",
})
return items
def _cat_weiguan(self, plat, page, area):
body = {
"audience": "全部受众", "page": page, "pageSize": 30,
"searchWord": "", "subject": area or "全部主题",
}
res = self._post_json(f"{plat['host']}{plat['search']}", body=body)
items = []
for it in res.get("data") or []:
items.append({
"vod_id": f"围观@{it.get('oneId', '')}",
"vod_name": self._s(it.get("title")),
"vod_pic": self._s(it.get("vertPoster")),
"vod_remarks": f"集数:{it.get('episodeCount', 0)}",
})
return items
def _cat_juwang(self, plat, page, area):
category = area or "/all/"
if page == 1:
url = f"{plat['host']}{category}"
else:
url = f"{plat['host']}{category}page/{page}/"
html = self._get_text(url)
root = self.html(html)
if root is None:
return []
items = []
for li in root.xpath("//section[contains(@class,'container')]//li"):
name_el = li.xpath(".//img/@alt")
href_el = li.xpath(".//a[contains(@class,'image-line')]/@href")
src_el = li.xpath(".//img/@src")
remark_el = li.xpath(".//*[contains(@class,'remarks')]/text()")
name = self._s(name_el[0]) if name_el else ""
href = self._s(href_el[0]) if href_el else ""
pic = self._s(src_el[0]) if src_el else ""
remark = self._s(remark_el[0]).strip() if remark_el else ""
if href:
items.append({
"vod_id": f"剧王@{href}",
"vod_name": name,
"vod_pic": pic,
"vod_remarks": remark,
})
return items
def _cat_qixing(self, plat, page, area):
url = f"{plat['host']}{plat['url1']}={area}&page_num={page}&page_size=24"
res = self._get_json(url, headers=self.xingya_headers)
items = []
for it in (res.get("data") or {}).get("list") or []:
theater = it.get("theater") or {}
detail_url = f"{plat['host']}{plat['url2']}?theater_parent_id={theater.get('id', '')}"
items.append({
"vod_id": f"七星@{detail_url}",
"vod_name": self._s(theater.get("title")),
"vod_pic": self._s(theater.get("cover_url")),
"vod_remarks": f"{theater.get('total', 0)}集",
})
return items
def _cat_tianquan(self, plat, page, area):
url = f"{plat['host']}{plat['url1']}={area}&offset={page}"
res = self._get_json(url)
items = []
for it in res.get("data") or []:
items.append({
"vod_id": f"甜圈@{it.get('book_id', '')}",
"vod_name": self._s(it.get("title")),
"vod_pic": self._s(it.get("cover")),
"vod_remarks": self._s(it.get("sub_title")),
})
return items
def _cat_duanjuwang(self, plat, page, area):
host = plat["host"]
url = f"{host}/?cate={area or '1'}&page={page}"
html = self._get_text(url, headers={
"User-Agent": (
"Mozilla/5.0 (iPhone; CPU iPhone OS 16_0 like Mac OS X) "
"AppleWebKit/605.1.15 (KHTML, like Gecko) Mobile/15E148"
),
"Referer": host + "/",
})
root = self.html(html)
if root is None:
return []
items = []
seen = set()
for card in root.xpath("//li[contains(@class,'col-6')]"):
anchors = card.xpath(".//h3[contains(@class,'f-14')]//a[@href]")
if not anchors:
continue
anchor = anchors[0]
name = self._clean("".join(anchor.xpath(".//text()")))
href = self._s((anchor.xpath("./@href") or [""])[0]).strip()
if not name or not href:
continue
full_url = href if href.startswith("http") else host + href
if full_url in seen:
continue
seen.add(full_url)
pic = self._s((card.xpath(".//img[contains(@class,'lazy')]/@data-original") or [""])[0])
if pic and not pic.startswith(("http://", "https://", "//")):
pic = host + pic
elif pic.startswith("//"):
pic = "https:" + pic
items.append({
"vod_id": f"短剧网@{full_url}",
"vod_name": name,
"vod_pic": pic,
"vod_remarks": "",
})
return items
# --- detail per platform ---
def _fetch_detail(self, platform, detail_id, raw_id):
plat = self.platforms.get(platform)
if not plat:
return {"vod_id": raw_id, "vod_name": "未知平台"}
handler = {
"七猫": self._detail_qimao,
"星芽": self._detail_xingya,
"西饭": self._detail_xifan,
"围观": self._detail_weiguan,
"剧王": self._detail_juwang,
"七星": self._detail_qixing,
"甜圈": self._detail_tianquan,
"短剧网": self._detail_duanjuwang,
}.get(platform)
if handler:
return handler(plat, detail_id, raw_id)
return {"vod_id": raw_id, "vod_name": "未实现"}
def _detail_qimao(self, plat, detail_id, raw_id):
did = __import__("urllib.parse", fromlist=["unquote"]).unquote(detail_id)
sign = self._md5(f"playlet_id={did}{QM_KEYS}")
url = f"{plat['url2']}?playlet_id={did}&sign={sign}"
res = self._get_json(url, headers=self._qm_headers())
data = res.get("data") or {}
play_list = data.get("play_list") or []
play_urls = "#".join(
f"{it.get('sort', '')}${it.get('video_url', '')}" for it in play_list
)
return {
"vod_id": raw_id,
"vod_name": self._s(data.get("title")),
"vod_pic": self._s(data.get("image_link")),
"vod_remarks": f"{self._s(data.get('tags'))} {data.get('total_episode_num', 0)}集",
"vod_content": self._s(data.get("intro")),
"vod_play_from": "七猫短剧",
"vod_play_url": play_urls,
}
def _detail_xingya(self, plat, detail_id, raw_id):
url = detail_id
res = self._get_json(url, headers=self.xingya_headers)
data = res.get("data") or {}
theaters = data.get("theaters") or []
play_urls = "#".join(
f"{it.get('num', '')}${it.get('son_video_url', '')}" for it in theaters
)
return {
"vod_id": raw_id,
"vod_name": self._s(data.get("title")),
"vod_pic": self._s(data.get("cover_url")),
"vod_content": self._s(data.get("introduction")),
"vod_remarks": self._s(data.get("desc_tags")),
"vod_play_from": "星芽短剧",
"vod_play_url": play_urls,
}
def _detail_xifan(self, plat, detail_id, raw_id):
duanju_id, source = detail_id.split("#", 1) if "#" in detail_id else (detail_id, "")
params = self._xifan_params({
"duanjuId": duanju_id, "source": source, "openFrom": "homescreen",
"pageID": "page_inner_flow",
})
url = f"{plat['host']}{plat['url2']}?{urlencode(params)}"
res = self._get_json(url)
data = res.get("result") or {}
episodes = data.get("episodeList") or []
play_urls = "#".join(
f"{ep.get('index', '')}${ep.get('playUrl', '')}" for ep in episodes
)
total = data.get("total", 0)
status = data.get("updateStatus", "")
remark = f"{total}集 已完结" if status == "over" else f"更新{total}集"
return {
"vod_id": raw_id,
"vod_name": self._s(data.get("title")),
"vod_pic": self._s(data.get("coverImageUrl")),
"vod_content": self._s(data.get("desc")),
"vod_remarks": remark,
"vod_play_from": "西饭短剧",
"vod_play_url": play_urls or "暂无播放地址$0",
}
def _detail_weiguan(self, plat, detail_id, raw_id):
url = f"{plat['host']}{plat['url2']}?oneId={detail_id}&page=1&pageSize=1000"
res = self._get_json(url)
data = res.get("data") or []
if not data:
return {"vod_id": raw_id, "vod_name": "无数据"}
first = data[0]
play_urls = "#".join(
f"{ep.get('title', '')}第{ep.get('playOrder', '')}集${json.dumps(ep.get('playSetting', {}), ensure_ascii=False)}"
for ep in data
)
return {
"vod_id": raw_id,
"vod_name": self._s(first.get("title")),
"vod_pic": self._s(first.get("vertPoster")),
"vod_remarks": f"共{len(data)}集",
"vod_content": f"播放:{first.get('collectionCount', 0)}",
"vod_play_from": "围观短剧",
"vod_play_url": play_urls,
}
def _detail_juwang(self, plat, detail_id, raw_id):
url = detail_id if detail_id.startswith("http") else plat["host"] + detail_id
html_text = self._get_text(url)
root = self.html(html_text)
if root is None:
return {"vod_id": raw_id, "vod_name": "加载失败"}
title = self._s(root.xpath("string(//h1)")).strip()
pic = self._s((root.xpath("//img/@src") or [""])[0])
content = self._s(root.xpath("string(//*[contains(@class,'info-detail')])")).strip()
remarks = self._s(root.xpath("string(//*[contains(@class,'info-mark')])")).strip()
episodes = []
for a in root.xpath("//*[contains(@class,'ep-list')]//a[@href]"):
ep_name = self._s(a.xpath("string(.)")).strip()
ep_url = self._s((a.xpath("./@href") or [""])[0])
if ep_name and ep_url:
episodes.append(f"{ep_name}${ep_url}")
return {
"vod_id": raw_id,
"vod_name": title or "未知标题",
"vod_pic": pic,
"vod_content": content,
"vod_remarks": remarks,
"vod_play_from": "剧王短剧",
"vod_play_url": "#".join(episodes) or "暂无播放地址$0",
}
def _detail_qixing(self, plat, detail_id, raw_id):
res = self._get_json(detail_id, headers=self.xingya_headers)
data = res.get("data") or {}
theaters = data.get("theaters") or []
if theaters:
play_urls = "#".join(
f"{it.get('num', '')}${it.get('son_video_url', '')}" for it in theaters
)
elif data.get("video_url"):
play_urls = f"1${data['video_url']}"
else:
play_urls = "暂无播放地址$0"
tags = data.get("desc_tags") or []
return {
"vod_id": raw_id,
"vod_name": self._s(data.get("title")),
"vod_pic": self._s(data.get("cover_url")),
"vod_content": self._s(data.get("introduction")),
"vod_remarks": self._s(data.get("filing")),
"vod_area": ",".join(tags) if isinstance(tags, list) else self._s(tags),
"vod_play_from": "七星短剧",
"vod_play_url": play_urls,
}
def _detail_tianquan(self, plat, detail_id, raw_id):
url = f"{plat['host']}{plat['url2']}={detail_id}"
res = self._get_json(url)
episodes = res.get("data") or []
play_urls = "#".join(
f"{it.get('title', '')}${it.get('video_id', '')}" for it in episodes
)
return {
"vod_id": raw_id,
"vod_name": self._s(res.get("book_name")),
"vod_pic": self._s(res.get("book_pic")),
"vod_content": self._s(res.get("desc")),
"vod_remarks": self._s(res.get("duration")),
"vod_year": f"更新时间:{res.get('time', '')}",
"vod_actor": self._s(res.get("author")),
"vod_play_from": "甜圈短剧",
"vod_play_url": play_urls,
}
_DISK_TOKENS = [
"drive.uc.cn", "pan.quark.cn", "pan.baidu.com",
"pan.xunlei.com", "alipan.com", "aliyundrive.com",
]
_DISK_MAP = {
"pan.baidu.com": "baidu", "pan.quark.cn": "quark",
"drive.uc.cn": "uc", "alipan.com": "aliyun",
"aliyundrive.com": "aliyun", "pan.xunlei.com": "xunlei",
}
_DISK_PRIORITY = {"baidu": 1, "quark": 2, "uc": 3, "aliyun": 4, "xunlei": 5}
def _detail_duanjuwang(self, plat, detail_id, raw_id):
host = plat["host"]
headers = {
"User-Agent": (
"Mozilla/5.0 (iPhone; CPU iPhone OS 16_0 like Mac OS X) "
"AppleWebKit/605.1.15 (KHTML, like Gecko) Mobile/15E148"
),
"Referer": host + "/",
}
html_text = self._get_text(detail_id, headers=headers)
root = self.html(html_text)
if root is None:
return {"vod_id": raw_id, "vod_name": "加载失败"}
title = self._clean(root.xpath("string(//h1[1])"))
pic = self._s((root.xpath("(//img[contains(@class,'lazy')]/@data-original)[1]") or [""])[0])
if not pic:
pic = self._s((root.xpath("(//img/@src)[1]") or [""])[0])
if pic and not pic.startswith(("http://", "https://", "//")):
pic = host + pic
elif pic.startswith("//"):
pic = "https:" + pic
links = []
seen = set()
for a in root.xpath("//*[contains(@class,'content')]//a[@href]"):
href = self._s((a.xpath("./@href") or [""])[0])
if not href or href in seen:
continue
seen.add(href)
links.append(href)
play_data = self._build_disk_play(links)
return {
"vod_id": raw_id,
"vod_name": title or "短剧",
"vod_pic": pic,
"vod_content": "此为推送网盘规则",
"vod_play_from": play_data["vod_play_from"],
"vod_play_url": play_data["vod_play_url"],
}
def _build_disk_play(self, links):
if not links:
return {"vod_play_from": "短剧网", "vod_play_url": ""}
grouped = {}
order = []
for link in links:
disk = self._identify_disk(link)
if not disk:
continue
if disk not in grouped:
grouped[disk] = []
order.append(disk)
urls_in_group = {item.split("$", 1)[1] for item in grouped[disk]}
if link not in urls_in_group:
grouped[disk].append(f"{disk}${link}")
if not grouped:
fallback = "#".join(f"{i + 1}$push://{url}" for i, url in enumerate(links))
return {"vod_play_from": "短剧网", "vod_play_url": fallback}
names = sorted(order, key=lambda n: self._DISK_PRIORITY.get(n, 999))
return {
"vod_play_from": "$$$".join(names),
"vod_play_url": "$$$".join("#".join(grouped[n]) for n in names),
}
@classmethod
def _identify_disk(cls, url):
low = url.lower()
for token in cls._DISK_TOKENS:
if token in low:
return cls._DISK_MAP.get(token, "")
return ""
# --- search ---
def _aggregate_search(self, keyword, page):
from concurrent.futures import ThreadPoolExecutor, as_completed
results = []
with ThreadPoolExecutor(max_workers=7) as executor:
futures = {
executor.submit(self._search_platform, pid, keyword, page): pid
for pid in self.platforms
}
for future in as_completed(futures, timeout=8):
try:
items = future.result()
results.extend(items)
except Exception:
pass
return results
def _search_platform(self, platform_id, keyword, page):
plat = self.platforms[platform_id]
handler = {
"七猫": self._search_qimao,
"星芽": self._search_xingya,
"西饭": self._search_xifan,
"围观": self._search_weiguan,
"剧王": self._search_juwang,
"七星": self._search_qixing,
"甜圈": self._search_tianquan,
"短剧网": self._search_duanjuwang,
}.get(platform_id)
if handler:
return handler(plat, keyword, page)
return []
def _search_qimao(self, plat, keyword, page):
sign_str = f"operation=2playlet_privacy=1search_word={keyword}{QM_KEYS}"
sign = self._md5(sign_str)
url = f"{plat['host']}{plat['search']}?search_word={quote(keyword)}&playlet_privacy=1&operation=2&sign={sign}"
res = self._get_json(url, headers=self._qm_headers())
items = []
for it in (res.get("data") or {}).get("list") or []:
items.append({
"vod_id": f"七猫@{quote(str(it.get('playlet_id', '')))}",
"vod_name": self._s(it.get("title")),
"vod_pic": self._s(it.get("image_link")),
"vod_remarks": f"七猫短剧 | {it.get('total_episode_num', 0)}集",
})
return items
def _search_xingya(self, plat, keyword, page):
url = f"{plat['host']}{plat['search']}"
res = self._post_json(url, body={"text": keyword}, headers=self.xingya_headers)
items = []
for it in (res.get("data") or {}).get("theater", {}).get("search_data") or []:
detail_url = f"{plat['host']}{plat['url2']}?theater_parent_id={it.get('id', '')}"
items.append({
"vod_id": f"星芽@{detail_url}",
"vod_name": self._s(it.get("title")),
"vod_pic": self._s(it.get("cover_url")),
"vod_remarks": f"星芽短剧 | {it.get('total', 0)}集",
})
return items
def _search_xifan(self, plat, keyword, page):
params = self._xifan_params({
"reqType": "search",
"offset": (page - 1) * 30,
"keyword": keyword,
})
url = f"{plat['host']}{plat['search']}?{urlencode(params)}"
res = self._get_json(url)
items = []
for vod in (res.get("result") or {}).get("elements") or []:
dj = vod.get("duanjuVo") or {}
items.append({
"vod_id": f"西饭@{dj.get('duanjuId', '')}#{dj.get('source', '')}",
"vod_name": self._s(dj.get("title")),
"vod_pic": self._s(dj.get("coverImageUrl")),
"vod_remarks": f"西饭短剧 | {dj.get('total', 0)}集",
})
return items
def _search_weiguan(self, plat, keyword, page):
body = {
"audience": "", "page": page, "pageSize": 30,
"searchWord": keyword, "subject": "",
}
res = self._post_json(f"{plat['host']}{plat['search']}", body=body)
items = []
for it in res.get("data") or []:
items.append({
"vod_id": f"围观@{it.get('oneId', '')}",
"vod_name": self._s(it.get("title")),
"vod_pic": self._s(it.get("vertPoster")),
"vod_remarks": f"围观短剧 | 集数:{it.get('episodeCount', 0)}",
})
return items
def _search_juwang(self, plat, keyword, page):
url = f"{plat['host']}{plat['search']}/{quote(keyword)}/page/{page}/"
html_text = self._get_text(url)
root = self.html(html_text)
if root is None:
return []
items = []
for li in root.xpath("//section[contains(@class,'container')]//li"):
name_el = li.xpath(".//img/@alt")
href_el = li.xpath(".//a[contains(@class,'image-line')]/@href")
src_el = li.xpath(".//img/@src")
remark_el = li.xpath(".//*[contains(@class,'remarks')]/text()")
name = self._s(name_el[0]) if name_el else ""
href = self._s(href_el[0]) if href_el else ""
pic = self._s(src_el[0]) if src_el else ""
remark = self._s(remark_el[0]).strip() if remark_el else ""
if href:
items.append({
"vod_id": f"剧王@{href}",
"vod_name": name,
"vod_pic": pic,
"vod_remarks": f"剧王短剧 | {remark}",
})
return items
def _search_qixing(self, plat, keyword, page):
url = f"{plat['host']}{plat['search']}"
res = self._post_json(url, body={"text": keyword}, headers=self.xingya_headers)
items = []
for it in (res.get("data") or {}).get("theater", {}).get("search_data") or []:
detail_url = f"{plat['host']}{plat['url2']}?theater_parent_id={it.get('id', '')}"
items.append({
"vod_id": f"七星@{detail_url}",
"vod_name": self._s(it.get("title")),
"vod_pic": self._s(it.get("cover_url")),
"vod_remarks": f"七星短剧 | {it.get('total', 0)}集",
})
return items
def _search_tianquan(self, plat, keyword, page):
url = f"{plat['host']}{plat['search']}={quote(keyword)}&offset={page}"
res = self._get_json(url)
items = []
for it in res.get("data") or []:
items.append({
"vod_id": f"甜圈@{it.get('book_id', '')}",
"vod_name": self._s(it.get("title")),
"vod_pic": self._s(it.get("cover")),
"vod_remarks": f"甜圈短剧 | {self._s(it.get('sub_title'))}",
})
return items
def _search_duanjuwang(self, plat, keyword, page):
host = plat["host"]
url = f"{host}/search.php?q={quote(keyword)}&page={page}"
html = self._get_text(url, headers={
"User-Agent": (
"Mozilla/5.0 (iPhone; CPU iPhone OS 16_0 like Mac OS X) "
"AppleWebKit/605.1.15 (KHTML, like Gecko) Mobile/15E148"
),
"Referer": host + "/",
})
root = self.html(html)
if root is None:
return []
items = []
seen = set()
for card in root.xpath("//li[contains(@class,'col-6')]"):
anchors = card.xpath(".//h3[contains(@class,'f-14')]//a[@href]")
if not anchors:
continue
anchor = anchors[0]
name = self._clean("".join(anchor.xpath(".//text()")))
href = self._s((anchor.xpath("./@href") or [""])[0]).strip()
if not name or not href:
continue
full_url = href if href.startswith("http") else host + href
if full_url in seen:
continue
seen.add(full_url)
pic = self._s((card.xpath(".//img[contains(@class,'lazy')]/@data-original") or [""])[0])
if pic and not pic.startswith(("http://", "https://", "//")):
pic = host + pic
elif pic.startswith("//"):
pic = "https:" + pic
items.append({
"vod_id": f"短剧网@{full_url}",
"vod_name": name,
"vod_pic": pic,
"vod_remarks": "短剧网",
})
return items
# --- play per platform ---
def _play_weiguan(self, play_id):
try:
data = json.loads(play_id) if isinstance(play_id, str) else play_id
except (json.JSONDecodeError, TypeError):
return {"parse": 0, "jx": 0, "url": play_id}
url = data.get("super") or data.get("high") or data.get("normal") or ""
return {"parse": 0, "jx": 0, "url": url}
def _play_juwang(self, play_id):
url = play_id if play_id.startswith("http") else self.platforms["剧王"]["host"] + play_id
html_text = self._get_text(url)
m = re.search(r'"wwm3u8"\s*:\s*"([^"]+)"', html_text)
video_url = ""
if m:
video_url = m.group(1).replace(r"\/", "/").replace(r'\"', '"')
return {
"parse": 0, "jx": 0, "url": video_url,
"header": {
"User-Agent": (
"Mozilla/5.0 (Windows NT 6.1; WOW64) AppleWebKit/537.36 "
"(KHTML, like Gecko) Chrome/50.0.2661.87 Safari/537.36"
),
},
}