feat: add zxzj player parsing

This commit is contained in:
Harold
2026-04-20 11:00:59 +08:00
parent 92a8d61e6d
commit a530a8c99b
2 changed files with 108 additions and 0 deletions
+43
View File
@@ -164,6 +164,49 @@ class TestZXZJSpider(unittest.TestCase):
)
self.assertEqual(groups, [])
def test_decrypt_url_round_trip_with_fixture(self):
self.assertEqual(
self.spider._decrypt_url("d6f636e256c607d6168756e2f67464544434241456469667f2f2a33707474786"),
"https://video.example.com",
)
def test_extract_iframe_url_reads_jx_target(self):
html = '<script>var player_data = {"url":"https://jx.zxzj.example/player?id=1"};</script>'
self.assertEqual(
self.spider._extract_iframe_url(html),
"https://jx.zxzj.example/player?id=1",
)
@patch.object(Spider, "_request_html")
def test_player_content_returns_decoded_direct_url_for_zxzj(self, mock_request_html):
mock_request_html.side_effect = [
'<script>var player_data = {"url":"https://jx.zxzj.example/player?id=1"};</script>',
'<script>var result_v2 = {"data":"d6f636e256c607d6168756e2f67464544434241456469667f2f2a33707474786"};</script>',
]
result = self.spider.playerContent("zxzj", "vodplay/999-1-1.html", {})
self.assertEqual(result["parse"], 0)
self.assertEqual(result["jx"], 0)
self.assertEqual(result["url"], "https://video.example.com")
self.assertEqual(result["header"]["Referer"], "https://jx.zxzj.example/player?id=1")
@patch.object(Spider, "_request_html")
def test_player_content_falls_back_to_play_page_when_decrypt_fails(self, mock_request_html):
mock_request_html.side_effect = [
'<script>var player_data = {"url":"https://jx.zxzj.example/player?id=1"};</script>',
'<script>var result_v2 = {"data":"broken"};</script>',
]
result = self.spider.playerContent("zxzj", "vodplay/999-1-1.html", {})
self.assertEqual(result["parse"], 1)
self.assertEqual(result["jx"], 1)
self.assertEqual(result["url"], "https://www.zxzjhd.com/vodplay/999-1-1.html")
def test_player_content_returns_direct_netdisk_link(self):
result = self.spider.playerContent("quark", "https://pan.quark.cn/s/demo", {})
self.assertEqual(
result,
{"parse": 0, "jx": 0, "playUrl": "", "url": "https://pan.quark.cn/s/demo", "header": {}},
)
if __name__ == "__main__":
unittest.main()
+65
View File
@@ -379,3 +379,68 @@ class Spider(BaseSpider):
"vod_play_url": "$$$".join(play_url),
}
return {"list": [vod]}
def _decrypt_url(self, encrypted_data):
raw = str(encrypted_data or "").strip()
if not raw or len(raw) % 2 != 0:
return ""
try:
reversed_hex = raw[::-1]
decoded = []
for index in range(0, len(reversed_hex), 2):
decoded.append(chr(int(reversed_hex[index : index + 2], 16)))
text = "".join(decoded)
split_len = max((len(text) - 7) // 2, 0)
candidate = text[:split_len] + text[split_len + 7 :]
return candidate if candidate.startswith("http") else ""
except Exception:
return ""
def _extract_iframe_url(self, html):
matched = re.search(r'"url"\s*:\s*"(https:[^"]*?jx\.zxzj[^"]*?)"', str(html or ""))
if matched:
return matched.group(1).replace("\\/", "/")
matched = re.search(r"player_[a-z0-9_]+\s*=\s*(\{[\s\S]*?\})\s*;?", str(html or ""), re.I)
if not matched:
return ""
try:
payload = json.loads(matched.group(1))
except Exception:
return ""
value = str(payload.get("url") or "").replace("\\/", "/")
return value if "jx.zxzj" in value else ""
def _extract_result_v2_data(self, html):
matched = re.search(r"result_v2\s*=\s*(\{[\s\S]*?\})\s*;", str(html or ""))
if not matched:
return ""
try:
payload = json.loads(matched.group(1))
except Exception:
return ""
return str(payload.get("data") or payload.get("url") or "").strip()
def playerContent(self, flag, id, vipFlags):
raw_flag = str(flag or "").lower()
raw_id = str(id or "").strip()
if raw_flag in ("baidu", "quark", "uc", "aliyun", "xunlei"):
return {"parse": 0, "jx": 0, "playUrl": "", "url": raw_id, "header": {}}
play_url = raw_id if raw_id.startswith("http") else self._build_url("/" + raw_id.lstrip("/"))
play_html = self._request_html(play_url, referer=self.headers["Referer"])
iframe_url = self._extract_iframe_url(play_html)
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)
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}
return {
"parse": 0,
"jx": 0,
"playUrl": "",
"url": final_url,
"header": {"Referer": iframe_url, "User-Agent": self.headers["User-Agent"]},
}