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

fix: improve juquanquan player requests

Browse Source
This commit is contained in:
Harold
2026-04-19 19:19:20 +08:00
parent 5940fcc32e
commit cccf5bbfd4
2 changed files with 95 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files
py/tests/test_剧圈圈.py
+41 -1
View File
@@ -1,6 +1,7 @@
import unittest import unittest
from importlib.machinery import SourceFileLoader from importlib.machinery import SourceFileLoader
from pathlib import Path from pathlib import Path
from requests.exceptions import ConnectionError
from unittest.mock import patch from unittest.mock import patch
@@ -171,6 +172,34 @@ class TestJuQuanQuanSpider(unittest.TestCase):
self.assertEqual(mock_fetch.call_count, 1) self.assertEqual(mock_fetch.call_count, 1)
self.assertEqual(mock_post.call_count, 1) self.assertEqual(mock_post.call_count, 1)
@patch.object(Spider, "_curl_request")
@patch.object(Spider, "fetch")
def test_request_with_headers_falls_back_to_curl_on_get_error(self, mock_fetch, mock_curl_request):
mock_fetch.side_effect = ConnectionError("dns failed")
mock_curl_request.return_value = {"body": "<html>ok</html>", "headers": {"set-cookie": ["a=b; Path=/"]}, "status_code": 200}
result = self.spider._request_with_headers("https://www.jqqzx.cc/play/62215-5-1.html")
self.assertEqual(result["body"], "<html>ok</html>")
self.assertEqual(result["status_code"], 200)
self.assertEqual(mock_curl_request.call_count, 1)
@patch.object(Spider, "_curl_request")
@patch.object(Spider, "post")
def test_request_with_headers_falls_back_to_curl_on_post_error(self, mock_post, mock_curl_request):
mock_post.side_effect = ConnectionError("dns failed")
mock_curl_request.return_value = {"body": '{"code":200}', "headers": {}, "status_code": 200}
result = self.spider._request_with_headers(
"https://www.jqqzx.cc/jx/api.php",
headers={"X-Requested-With": "XMLHttpRequest"},
data="vid=demo",
)
self.assertEqual(result["body"], '{"code":200}')
self.assertEqual(result["status_code"], 200)
self.assertEqual(mock_curl_request.call_count, 1)
@patch.object(Spider, "_request_with_headers") @patch.object(Spider, "_request_with_headers")
def test_player_content_returns_direct_media_url(self, mock_request_with_headers): def test_player_content_returns_direct_media_url(self, mock_request_with_headers):
mock_request_with_headers.return_value = { mock_request_with_headers.return_value = {
@@ -201,10 +230,21 @@ class TestJuQuanQuanSpider(unittest.TestCase):
"status_code": 200, "status_code": 200,
}, },
] ]
self.spider._decode_url = lambda value: "https://video.example/fallback.m3u8" def decode_stub(value):
if value == "error://apiRes_dummy":
return "https://video.example/fallback.m3u8"
return ""
self.spider._decode_url = decode_stub
result = self.spider.playerContent("线路A", "play/123-1-1", {}) result = self.spider.playerContent("线路A", "play/123-1-1", {})
self.assertEqual(result["parse"], 0) self.assertEqual(result["parse"], 0)
self.assertEqual(result["url"], "https://video.example/fallback.m3u8") self.assertEqual(result["url"], "https://video.example/fallback.m3u8")
api_call = mock_request_with_headers.call_args_list[2]
self.assertEqual(api_call.args[0], "https://www.jqqzx.cc/jx/api.php")
self.assertEqual(api_call.kwargs["data"], "vid=https%3A//middle.example/embed%3Fid%3D1")
self.assertEqual(api_call.kwargs["headers"]["Content-Type"], "application/x-www-form-urlencoded; charset=UTF-8")
self.assertEqual(api_call.kwargs["headers"]["Accept"], "*/*")
self.assertEqual(api_call.kwargs["headers"]["X-Requested-With"], "XMLHttpRequest")
@patch.object(Spider, "_request_with_headers") @patch.object(Spider, "_request_with_headers")
def test_player_content_falls_back_to_play_page_when_player_data_missing(self, mock_request_with_headers): def test_player_content_falls_back_to_play_page_when_player_data_missing(self, mock_request_with_headers):
py/剧圈圈.py
+54 -9
View File
@@ -3,6 +3,7 @@ import base64
import hashlib import hashlib
import json import json
import re import re
import subprocess
import sys import sys
from urllib.parse import quote, unquote, urljoin from urllib.parse import quote, unquote, urljoin
@@ -252,15 +253,57 @@ class Spider(BaseSpider):
request_headers = dict(self.headers) request_headers = dict(self.headers)
if headers: if headers:
request_headers.update(headers) request_headers.update(headers)
if data is None: try:
response = self.fetch(target, headers=request_headers, timeout=10) if data is None:
else: response = self.fetch(target, headers=request_headers, timeout=10)
response = self.post(target, data=data, headers=request_headers, timeout=10) else:
return { response = self.post(target, data=data, headers=request_headers, timeout=10)
"body": response.text or "", return {
"headers": getattr(response, "headers", {}) or {}, "body": response.text or "",
"status_code": response.status_code, "headers": getattr(response, "headers", {}) or {},
} "status_code": response.status_code,
}
except Exception:
return self._curl_request(target, headers=request_headers, data=data)
def _curl_request(self, url, headers=None, data=None):
command = ["curl", "-L", "--silent", "--show-error", "-D", "-", url]
for key, value in (headers or {}).items():
command.extend(["-H", f"{key}: {value}"])
if data is not None:
command.extend(["-X", "POST", "--data", data])
completed = subprocess.run(command, capture_output=True, text=True, check=True, timeout=20)
raw = completed.stdout or ""
marker = "\r\n\r\n" if "\r\n\r\n" in raw else "\n\n"
chunks = [chunk for chunk in raw.split(marker) if chunk.strip()]
header_block = ""
body = ""
for index, chunk in enumerate(chunks):
if chunk.lstrip().startswith("HTTP/"):
header_block = chunk
body = marker.join(chunks[index + 1:]) if index + 1 < len(chunks) else ""
if not header_block and chunks:
header_block = chunks[0]
body = marker.join(chunks[1:]) if len(chunks) > 1 else ""
header_lines = header_block.splitlines()
status_line = header_lines[0] if header_lines else ""
matched = re.search(r"HTTP/\S+\s+(\d+)", status_line)
status_code = int(matched.group(1)) if matched else 200
parsed_headers = {}
for line in header_lines[1:]:
if ":" not in line:
continue
key, value = line.split(":", 1)
low_key = key.strip().lower()
clean_value = value.strip()
existing = parsed_headers.get(low_key)
if existing is None:
parsed_headers[low_key] = clean_value
elif isinstance(existing, list):
existing.append(clean_value)
else:
parsed_headers[low_key] = [existing, clean_value]
return {"body": body, "headers": parsed_headers, "status_code": status_code}
def _get_set_cookies(self, headers): def _get_set_cookies(self, headers):
raw = headers.get("set-cookie") or headers.get("Set-Cookie") or [] raw = headers.get("set-cookie") or headers.get("Set-Cookie") or []
@@ -354,6 +397,8 @@ class Spider(BaseSpider):
api_res = self._request_with_headers( api_res = self._request_with_headers(
self._build_url("/jx/api.php"), self._build_url("/jx/api.php"),
headers={ headers={
"Content-Type": "application/x-www-form-urlencoded; charset=UTF-8",
"Accept": "*/*",
"Referer": player_page, "Referer": player_page,
"Origin": self.host, "Origin": self.host,
"X-Requested-With": "XMLHttpRequest", "X-Requested-With": "XMLHttpRequest",