yml/live-hub-py
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

登录 ok 了

Browse Source
This commit is contained in:
yml2213
2026-06-21 22:48:14 +08:00
parent 828f9889ed
commit 4dca691890
8 changed files with 197 additions and 133 deletions
Download Patch File Download Diff File Expand all files Collapse all files
geetest/network.py
+84 -43
View File
@@ -2,9 +2,40 @@ import time
import requests
import json
import re
from typing import Tuple
from typing import Mapping, Optional, Tuple
REQUEST_TIMEOUT = (3.05, 12)
PASSPORT_REFERER = "https://passport.douyu.com/"
def _get(
url: str,
*,
params: Optional[dict] = None,
headers: Optional[dict] = None,
proxies: Optional[Mapping[str, str]] = None,
) -> requests.Response:
"""发送极验 GET 请求，确保使用同一个代理出口。"""
return requests.get(
url,
params=params,
headers=headers,
proxies=dict(proxies or {}),
timeout=REQUEST_TIMEOUT,
)
def _parse_jsonp_response(response: requests.Response, source: str) -> dict:
"""解析极验 JSONP 响应。"""
response.raise_for_status()
match = re.search(r'\((.*)\)$', response.text)
if not match:
raise ValueError(f"{source} 无法解析 JSONP 响应")
data = json.loads(match.group(1))
if data.get('status') == 'error':
raise ValueError(f"{source} 失败: {data.get('user_error', data.get('error', '未知错误'))}")
return data
def _parse_json_response(response: requests.Response, source: str) -> dict:
@@ -83,18 +114,18 @@ def get_challenge_gt() -> Tuple[str, str]:
def get_js_address(gt:str) -> dict:
def get_js_address(gt: str, proxies: Optional[Mapping[str, str]] = None) -> dict:
headers = {
'accept': '*/*',
'accept-language': 'zh-CN,zh;q=0.9',
'referer': 'https://demos.geetest.com/',
'sec-ch-ua': '"Chromium";v="140", "Not=A?Brand";v="24", "Google Chrome";v="140"',
'referer': PASSPORT_REFERER,
'sec-ch-ua': '"Google Chrome";v="149", "Chromium";v="149", "Not)A;Brand";v="24"',
'sec-ch-ua-mobile': '?0',
'sec-ch-ua-platform': '"Windows"',
'sec-ch-ua-platform': '"macOS"',
'sec-fetch-dest': 'script',
'sec-fetch-mode': 'no-cors',
'sec-fetch-site': 'same-site',
'user-agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/140.0.0.0 Safari/537.36',
'sec-fetch-site': 'cross-site',
'user-agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/149.0.0.0 Safari/537.36',
}
params = {
@@ -102,62 +133,72 @@ def get_js_address(gt:str) -> dict:
'callback': 'geetest_' + str(int(round(time.time() * 1000))),
}
response = requests.get(
'https://apiv6.geetest.com/gettype.php',
response = _get(
'https://api.geetest.com/gettype.php',
params=params,
headers=headers,
timeout=REQUEST_TIMEOUT,
proxies=proxies,
)
response.raise_for_status()
match = re.search(r'\((.*)\)$', response.text)
if match:
json_str = match.group(1)
return json.loads(json_str)
else:
raise ValueError("无法解析 JSONP 响应")
return _parse_jsonp_response(response, "极验 gettype")
def get_c_s(gt:str, challenge:str, w:str) -> Tuple[list[int],str]:
def get_c_s(
gt: str,
challenge: str,
w: str,
proxies: Optional[Mapping[str, str]] = None,
) -> Tuple[list[int], str]:
headers = {
'accept': '*/*',
'accept-language': 'zh-CN,zh;q=0.9',
'referer': 'https://demos.geetest.com/',
'sec-ch-ua': '"Chromium";v="140", "Not=A?Brand";v="24", "Google Chrome";v="140"',
'referer': PASSPORT_REFERER,
'sec-ch-ua': '"Google Chrome";v="149", "Chromium";v="149", "Not)A;Brand";v="24"',
'sec-ch-ua-mobile': '?0',
'sec-ch-ua-platform': '"Windows"',
'sec-ch-ua-platform': '"macOS"',
'sec-fetch-dest': 'script',
'sec-fetch-mode': 'no-cors',
'sec-fetch-site': 'same-site',
'user-agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/140.0.0.0 Safari/537.36',
'sec-fetch-site': 'cross-site',
'user-agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/149.0.0.0 Safari/537.36',
}
response = requests.get(
response = _get(
'https://api.geetest.com/get.php?gt=' + gt + '&challenge=' + challenge + '&lang=zh-cn&pt=0&client_type=web&w=' + w + '&callback=geetest_' + str(
int(round(time.time() * 1000))),
headers=headers,
timeout=REQUEST_TIMEOUT,
proxies=proxies,
)
response.raise_for_status()
data = _parse_jsonp_response(response, "极验 get.php")
return data['data']['c'], data['data']['s']
# 检查是否是错误响应
if response.text.startswith('{'):
# 直接JSON响应
data = json.loads(response.text)
if 'status' in data and data['status'] == 'error':
raise ValueError(f"极验请求失败: {data.get('user_error', data.get('error', '未知错误'))}")
match = re.search(r'\((.*)\)$', response.text)
if match:
json_str = match.group(1)
data = json.loads(json_str)
def req_fullpage_validate(
gt: str,
challenge: str,
w: str,
proxies: Optional[Mapping[str, str]] = None,
) -> dict:
"""HAR 中的 fullpage 最终校验，成功后直接返回 validate。"""
headers = {
'Accept': '*/*',
'Accept-Language': 'zh-CN,zh;q=0.9',
'Connection': 'keep-alive',
'Referer': PASSPORT_REFERER,
'Sec-Fetch-Dest': 'script',
'Sec-Fetch-Mode': 'no-cors',
'Sec-Fetch-Site': 'cross-site',
'User-Agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/149.0.0.0 Safari/537.36',
'sec-ch-ua': '"Google Chrome";v="149", "Chromium";v="149", "Not)A;Brand";v="24"',
'sec-ch-ua-mobile': '?0',
'sec-ch-ua-platform': '"macOS"',
}
# 检查是否是错误响应
if 'status' in data and data['status'] == 'error':
raise ValueError(f"极验请求失败: {data.get('user_error', data.get('error', '未知错误'))}")
return data['data']['c'], data['data']['s']
else:
raise ValueError("无法解析 JSONP 响应")
response = _get(
'https://api.geetest.com/ajax.php?gt=' + gt + '&challenge=' + challenge + '&lang=zh-cn&pt=0&client_type=web&w=' + w + '&callback=geetest_' + str(
int(round(time.time() * 1000))),
headers=headers,
proxies=proxies,
)
return _parse_jsonp_response(response, "极验 fullpage ajax.php")
def req_slide(gt:str, challenge:str, w2:str) -> None:
headers = {