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

彻底优化项目结构,使用 web

Browse Source
This commit is contained in:
yml2213
2026-06-22 13:39:03 +08:00
parent 347edb8103
commit 63c1ffa046
46 changed files with 127 additions and 3077 deletions
Download Patch File Download Diff File Expand all files Collapse all files
core/__init__.py
+5
View File
@@ -0,0 +1,5 @@
"""核心业务逻辑"""
from .models import Account, ProxyConfig
__all__ = ["Account", "ProxyConfig"]
core/douyu/__init__.py
+8
View File
@@ -0,0 +1,8 @@
"""斗鱼登录模块"""
from .login import DouyuLogin
from .email_verifier import EmailVerifier
from .proxy import ProxyManager
from .whitelist import WhitelistManager
__all__ = ["DouyuLogin", "EmailVerifier", "ProxyManager", "WhitelistManager"]
core/douyu/crypto.py
+68
View File
@@ -0,0 +1,68 @@
"""斗鱼专用加密模块 - 用户名密码加密"""
import base64
import hashlib
from Crypto.PublicKey import RSA
from Crypto.Cipher import PKCS1_v1_5
from Crypto.Cipher import AES
from Crypto.Cipher import ARC4
# 斗鱼RSA公钥（从JS中提取）
DOUYU_RSA_PUBLIC_KEY = """-----BEGIN PUBLIC KEY-----
MIGfMA0GCSqGSIb3DQEBAQUAA4GNADCBiQKBgQDHfGXRkF+RiLA71KAHOFBaWGBy
J7M6D3MDAsFHo2JMBDm2Kfj6V3GFMI7B2JQ3qGl0jCk6ILT1jQ+IFhLvLR3cXPaC
HT5xYa0hzJpMNO3bLSuJhzY5jQNqRMWfbcV4FLB2JBaFfWcY7RWQ2pCE6jjDnMHM
o2kz+dJoGnZM0b99VwIDAQAB
-----END PUBLIC KEY-----"""
# 斗鱼AES密钥（16位）
DOUYU_AES_KEY = "1234567890abcdef"
# 斗鱼登录页 cryptoData 使用的 RC4 密钥
DOUYU_RC4_KEY = "7TkbRSEWvVWebXbr"
def md5(text: str) -> str:
"""MD5加密"""
return hashlib.md5(text.encode('utf-8')).hexdigest()
def rsa_encrypt(text: str, public_key: str = DOUYU_RSA_PUBLIC_KEY) -> str:
"""RSA加密"""
key = RSA.import_key(public_key)
cipher = PKCS1_v1_5.new(key)
encrypted = cipher.encrypt(text.encode('utf-8'))
return base64.b64encode(encrypted).decode('utf-8')
def aes_encrypt(text: str, key: str = DOUYU_AES_KEY) -> str:
"""AES加密"""
key_bytes = key.encode('utf-8')
text_bytes = text.encode('utf-8')
# 填充到16的倍数
padding_len = 16 - (len(text_bytes) % 16)
text_bytes += bytes([padding_len] * padding_len)
cipher = AES.new(key_bytes, AES.MODE_ECB)
encrypted = cipher.encrypt(text_bytes)
return base64.b64encode(encrypted).decode('utf-8')
def encrypt_username(username: str) -> str:
"""加密用户名（斗鱼使用RSA加密）"""
return rsa_encrypt(username)
def encrypt_password(password: str) -> str:
"""加密密码（斗鱼使用MD5）"""
return md5(password)
def encrypt_nickname_or_phone(text: str) -> str:
"""加密昵称或手机号"""
key = DOUYU_RC4_KEY.encode('utf-8')
cipher = ARC4.new(key)
encrypted = cipher.encrypt(text.encode('utf-8'))
return base64.b64encode(encrypted).decode('utf-8')
core/douyu/email_verifier.py
+349
View File
@@ -0,0 +1,349 @@
"""邮箱验证模块 - IMAP获取验证码"""
import imaplib
import email
import html
import re
import time
from datetime import datetime, timedelta
from email.header import decode_header
from email.utils import parsedate_to_datetime
from typing import Optional
from loguru import logger
class EmailVerifier:
"""邮箱验证器"""
def __init__(
self,
imap_server: str,
imap_port: int,
username: str,
password: str,
timeout: float = 12,
mailbox: str = "INBOX",
lookback_minutes: int = 10,
max_messages: int = 20,
):
self.imap_server = imap_server
self.imap_port = imap_port
self.username = username
self.password = password
self.timeout = timeout
self.mailbox = mailbox
self.lookback_minutes = lookback_minutes
self.max_messages = max_messages
self._connection: Optional[imaplib.IMAP4_SSL] = None
def connect(self) -> None:
"""连接IMAP服务器"""
try:
logger.info(f"连接IMAP服务器: {self.imap_server}:{self.imap_port}")
self._validate_login_text()
self._connection = imaplib.IMAP4_SSL(
self.imap_server,
self.imap_port,
timeout=self.timeout,
)
self._connection.login(self.username, self.password)
logger.info("IMAP连接成功")
except Exception as e:
logger.error(f"IMAP连接失败: {e}")
raise
def _validate_login_text(self) -> None:
"""提前检查IMAP登录字段，给出比ascii编码异常更明确的提示。"""
for label, value in (("邮箱账号", self.username), ("邮箱密码/授权码", self.password)):
try:
value.encode("ascii")
except UnicodeEncodeError as exc:
raise ValueError(
f"{label}包含中文或其他非ASCII字符，IMAP无法登录；"
"请检查导入格式是否为：用户名|密码|邮箱|邮箱密码"
) from exc
def disconnect(self) -> None:
"""断开IMAP连接"""
if self._connection:
try:
self._connection.logout()
except:
pass
self._connection = None
def get_verification_code(
self,
max_wait: int = 60,
interval: int = 2,
after_timestamp: Optional[float] = None,
allow_old_seconds: int = 15,
) -> str:
"""
获取斗鱼验证码
Args:
max_wait: 最大等待时间（秒）
interval: 轮询间隔（秒）
after_timestamp: 发起发送验证码请求的时间戳，用于过滤旧邮件
allow_old_seconds: 邮件服务器时间允许向前偏移的秒数
Returns:
6位验证码
"""
logger.info(f"等待斗鱼验证码邮件，最大等待 {max_wait} 秒...")
self.connect()
try:
deadline = time.monotonic() + max_wait
while time.monotonic() < deadline:
code = self._fetch_latest_code(
after_timestamp=after_timestamp,
allow_old_seconds=allow_old_seconds,
)
if code:
logger.success(f"获取到验证码: {code}")
return code
logger.debug("未找到验证码，等待中...")
time.sleep(interval)
raise TimeoutError("等待验证码超时")
finally:
self.disconnect()
def _fetch_latest_code(
self,
after_timestamp: Optional[float] = None,
allow_old_seconds: int = 15,
) -> Optional[str]:
"""从IMAP获取最新验证码"""
try:
if not self._connection:
raise RuntimeError("IMAP未连接")
status, _ = self._connection.select(self.mailbox, readonly=True)
if status != 'OK':
logger.warning(f"选择邮箱目录失败: {self.mailbox}")
return None
since = self._build_since_date(after_timestamp, allow_old_seconds)
# IMAP命令只能稳定发送ASCII条件，中文主题/发件人改到本地解析过滤。
status, messages = self._connection.search(None, 'SINCE', since)
if status != 'OK' or not messages[0]:
return None
message_ids = messages[0].split()
recent_ids = list(reversed(message_ids[-self.max_messages:]))
logger.debug(f"扫描最近 {len(recent_ids)} 封邮件，SINCE {since}")
for message_id in recent_ids:
msg = self._fetch_message(message_id)
if not msg:
continue
if not self._is_recent_enough(msg, after_timestamp, allow_old_seconds):
continue
subject = self._decode_subject(msg.get('Subject', ''))
from_addr = msg.get('From', '')
body = self._get_email_body(msg)
if not self._is_douyu_email(subject, from_addr, body):
continue
code = self._extract_verification_code(body)
if code:
logger.info(f"从邮件中提取到验证码，主题: {subject}")
return code
return None
except imaplib.IMAP4.abort as e:
logger.warning(f"IMAP连接中断，准备下轮重连: {e}")
self.disconnect()
self.connect()
return None
except Exception as e:
logger.error(f"获取邮件失败: {e}")
return None
def _build_since_date(
self,
after_timestamp: Optional[float],
allow_old_seconds: int,
) -> str:
"""构造IMAP SINCE日期，月份固定用英文缩写。"""
if after_timestamp:
since_dt = datetime.fromtimestamp(
max(0, after_timestamp - allow_old_seconds)
)
else:
since_dt = datetime.now() - timedelta(minutes=self.lookback_minutes)
return self._format_imap_date(since_dt)
def _format_imap_date(self, value: datetime) -> str:
"""格式化IMAP日期，避免系统locale影响月份名称。"""
months = [
"Jan", "Feb", "Mar", "Apr", "May", "Jun",
"Jul", "Aug", "Sep", "Oct", "Nov", "Dec",
]
return f"{value.day:02d}-{months[value.month - 1]}-{value.year}"
def _fetch_message(self, message_id: bytes) -> Optional[email.message.Message]:
"""获取邮件完整内容，使用PEEK避免标记已读。"""
status, msg_data = self._connection.fetch(message_id, '(BODY.PEEK[])')
if status != 'OK':
return None
raw_message = self._join_fetch_payload(msg_data)
if not raw_message:
return None
return email.message_from_bytes(raw_message)
def _join_fetch_payload(self, msg_data) -> bytes:
"""合并IMAP fetch返回中的邮件字节内容。"""
chunks = []
for item in msg_data:
if isinstance(item, tuple) and len(item) >= 2 and isinstance(item[1], bytes):
chunks.append(item[1])
elif isinstance(item, bytes) and item.startswith(b'From '):
chunks.append(item)
return b''.join(chunks)
def _is_recent_enough(
self,
msg: email.message.Message,
after_timestamp: Optional[float],
allow_old_seconds: int,
) -> bool:
"""检查邮件时间是否晚于本次发送验证码请求。"""
if not after_timestamp:
return True
message_time = self._parse_message_time(msg)
if not message_time:
logger.debug("邮件缺少Date头，保守纳入候选")
return True
if message_time.timestamp() < after_timestamp - allow_old_seconds:
subject = self._decode_subject(msg.get('Subject', ''))
logger.debug(f"跳过旧邮件: {subject}")
return False
return True
def _parse_message_time(self, msg: email.message.Message) -> Optional[datetime]:
"""解析邮件Date头。"""
date_header = msg.get('Date')
if not date_header:
return None
try:
return parsedate_to_datetime(date_header)
except (TypeError, ValueError):
return None
def _is_douyu_email(self, subject: str, from_addr: str, body: str = "") -> bool:
"""判断是否是斗鱼的邮件"""
douyu_keywords = ['斗鱼', 'douyu', 'douyutv']
verify_keywords = ['验证码', '安全验证', '登录验证', '动态码', '校验码']
combined = f"{subject}\n{from_addr}\n{body[:500]}".lower()
return (
any(keyword in combined for keyword in douyu_keywords)
or any(keyword in subject for keyword in verify_keywords)
)
def _decode_subject(self, subject: str) -> str:
"""解码邮件主题"""
if not subject:
return ""
decoded_parts = decode_header(subject)
result = []
for part, charset in decoded_parts:
if isinstance(part, bytes):
result.append(part.decode(charset or 'utf-8', errors='ignore'))
else:
result.append(part)
return ' '.join(result)
def _get_email_body(self, msg: email.message.Message) -> str:
"""获取邮件正文"""
parts = []
if msg.is_multipart():
for part in msg.walk():
content_type = part.get_content_type()
if part.get_content_disposition() == 'attachment':
continue
if content_type == 'text/plain' or content_type == 'text/html':
decoded = self._decode_part_payload(part)
if decoded:
parts.append(decoded)
else:
decoded = self._decode_part_payload(msg)
if decoded:
parts.append(decoded)
return "\n".join(parts)
def _decode_part_payload(self, part: email.message.Message) -> str:
"""解码邮件片段内容。"""
try:
payload = part.get_payload(decode=True)
if payload is None:
raw_payload = part.get_payload()
return raw_payload if isinstance(raw_payload, str) else ""
charset = part.get_content_charset() or 'utf-8'
return payload.decode(charset, errors='ignore')
except Exception:
return ""
def _extract_verification_code(self, text: str) -> Optional[str]:
"""从文本中提取验证码"""
# 清理HTML标签和实体，方便匹配中文邮件模板。
text = html.unescape(text)
text = re.sub(r'<[^>]+>', ' ', text)
text = re.sub(r'\s+', ' ', text)
# 查找6位数字验证码
# 注意：中文冒号和英文冒号需要分别处理
patterns = [
r'(?:验证码|校验码|动态码|安全码)\s*(?:是|为|:|：)?\s*(\d{6})',
r'(?:您的|你的|本次)?(?:验证码|校验码|动态码|安全码)\s*(?:是|为)?\s*(\d{6})',
r'(\d{6})\s*(?:是|为)?(?:您的|你的|本次)?(?:验证码|校验码|动态码|安全码)',
r'(?:verification\s+code|security\s+code|code)\s*(?:is|:|：)?\s*(\d{6})',
r'\b(\d{6})\b', # 候选邮件已经过滤为斗鱼验证码邮件，最后再兜底匹配6位数字。
]
for pattern in patterns:
match = re.search(pattern, text, re.IGNORECASE)
if match:
code = match.group(1)
return code
return None
def get_email_config_for_account(email_address: str) -> dict:
"""根据邮箱地址自动配置IMAP服务器"""
configs = {
'bdhg.xyz': {'server': 'mail.bdhg.xyz', 'port': 993},
'qq.com': {'server': 'imap.qq.com', 'port': 993},
'163.com': {'server': 'imap.163.com', 'port': 993},
'126.com': {'server': 'imap.126.com', 'port': 993},
'gmail.com': {'server': 'imap.gmail.com', 'port': 993},
'outlook.com': {'server': 'outlook.office365.com', 'port': 993},
'hotmail.com': {'server': 'outlook.office365.com', 'port': 993},
}
domain = email_address.split('@')[-1].lower()
return configs.get(domain, {'server': f'imap.{domain}', 'port': 993})
core/douyu/login.py
+551
View File
@@ -0,0 +1,551 @@
"""斗鱼登录核心模块"""
import re
import json
import time
import requests
from typing import Mapping, Optional, Tuple
from urllib.parse import urlsplit, urlunsplit
from loguru import logger
from core.models import Account
from .crypto import encrypt_password, encrypt_nickname_or_phone
from .email_verifier import EmailVerifier
from .proxy import ProxyManager, get_proxy_manager
from core.geetest import run_solver
from core.geetest.v3_slide.solver import (
_generate_seed, get_w1, get_w2,
)
from core.geetest.common.network import (
get_js_address,
get_c_s,
req_fullpage_validate,
)
class LoginResult:
"""登录结果"""
def __init__(self, success: bool, cookie: str = "", message: str = "", code: str = ""):
self.success = success
self.cookie = cookie
self.message = message
self.code = code # 需要验证时的code
class DouyuLogin:
"""斗鱼登录器"""
# 斗鱼API地址
LOGIN_API = "https://passport.douyu.com/wgapi/member/passport/login"
SEND_EMAIL_API = "https://passport.douyu.com/wgapi/member/passport/remotelogin/sendemail"
VERIFY_API = "https://passport.douyu.com/wgapi/member/passport/remotelogin/verify"
LOGIN_CALLBACK_API = "https://www.douyu.com/api/passport/login"
WEBLOGIN_API = "https://msg.douyu.com/webLogin"
LOGIN_REFERER = (
"https://passport.douyu.com/index/login?"
"passport_reg_callback=PASSPORT_REG_SUCCESS_CALLBACK&"
"passport_login_callback=PASSPORT_LOGIN_SUCCESS_CALLBACK&"
"passport_close_callback=PASSPORT_CLOSE_CALLBACK&"
"passport_dp_callback=PASSPORT_DP_CALLBACK&"
"type=login&client_id=1&"
"state=https%3A%2F%2Fwww.douyu.com%2Fdirectory"
)
REQUEST_TIMEOUT = (10, 30)
def __init__(
self,
account: Account,
proxy: Optional[str | Mapping[str, str]] = None,
proxy_api_url: Optional[str] = None,
timeout: tuple[float, float] = REQUEST_TIMEOUT,
max_geetest_retries: int = 5,
):
self.account = account
self.proxy = proxy
self.timeout = timeout
self.max_geetest_retries = max_geetest_retries
self.session = requests.Session()
# 初始化代理管理器
self.proxy_manager = get_proxy_manager(proxy_api_url) if proxy_api_url else None
self._setup_session()
def _setup_session(self) -> None:
"""配置Session"""
# 只使用配置文件里显式传入的代理，避免系统环境变量悄悄影响请求。
self.session.trust_env = False
self.session.headers.update({
'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',
'Accept': 'application/json, text/javascript, */*; q=0.01',
'Accept-Language': 'zh-CN,zh;q=0.9',
'Content-Type': 'application/x-www-form-urlencoded; charset=UTF-8',
'Referer': self.LOGIN_REFERER,
'Origin': 'https://passport.douyu.com',
'X-Requested-With': 'XMLHttpRequest',
})
# 设置代理
self._apply_proxy()
def _apply_proxy(self, proxy: str = None) -> None:
"""应用代理到Session"""
if proxy:
# 使用指定的代理
self.session.proxies = {
'http': proxy,
'https': proxy,
}
elif self.proxy:
# 使用配置的代理
if isinstance(self.proxy, str):
self.session.proxies = {
'http': self.proxy,
'https': self.proxy,
}
else:
self.session.proxies = {
scheme: url
for scheme, url in self.proxy.items()
if url
}
else:
# 从代理管理器获取代理
if self.proxy_manager:
new_proxy = self.proxy_manager.get_proxy()
if new_proxy:
self.session.proxies = {
'http': new_proxy,
'https': new_proxy,
}
def _refresh_proxy(self) -> Optional[str]:
"""刷新代理IP"""
if not self.proxy_manager:
return None
new_proxy = self.proxy_manager.get_proxy()
if new_proxy:
self._apply_proxy(new_proxy)
logger.info(f"已切换代理: {new_proxy}")
return new_proxy
def _safe_url(self, url: str) -> str:
"""隐藏查询参数，避免日志泄露登录回调 code。"""
parsed = urlsplit(url)
return urlunsplit((parsed.scheme, parsed.netloc, parsed.path, "", ""))
def _request(self, method: str, url: str, max_retries: int = 3, **kwargs) -> requests.Response:
"""
统一发送请求，附带分段超时和更明确的错误信息。
代理连接失败时自动重试获取新的代理IP。
"""
timeout = kwargs.pop('timeout', self.timeout)
safe_url = self._safe_url(url)
for attempt in range(max_retries):
started = time.monotonic()
try:
response = self.session.request(method, url, timeout=timeout, **kwargs)
elapsed = time.monotonic() - started
logger.debug(
f"{method.upper()} {safe_url} -> {response.status_code} "
f"({elapsed:.2f}s)"
)
return response
except requests.Timeout as exc:
elapsed = time.monotonic() - started
raise TimeoutError(
f"{method.upper()} {safe_url} 超时，耗时 {elapsed:.1f}s，"
f"timeout={timeout}"
) from exc
except requests.ProxyError as exc:
elapsed = time.monotonic() - started
logger.warning(f"代理连接失败，尝试 {attempt + 1}/{max_retries}: {exc}")
if attempt < max_retries - 1:
# 刷新代理
self._refresh_proxy()
time.sleep(1)
continue
raise ConnectionError(
f"{method.upper()} {safe_url} 代理连接失败，已重试 {max_retries} 次"
) from exc
except requests.RequestException as exc:
elapsed = time.monotonic() - started
raise ConnectionError(
f"{method.upper()} {safe_url} 请求失败，耗时 {elapsed:.1f}s: {exc}"
) from exc
def _request_json(self, method: str, url: str, source: str, **kwargs) -> dict:
"""请求 JSON 接口，并在响应异常时输出可定位的信息。"""
response = self._request(method, url, **kwargs)
response.raise_for_status()
body = response.text.strip()
if not body:
raise ValueError(f"{source} 返回为空，无法解析 JSON")
try:
return response.json()
except json.JSONDecodeError as exc:
preview = body[:200].replace("\n", "\\n")
raise ValueError(f"{source} 返回的不是有效 JSON: {preview}") from exc
def login(self) -> LoginResult:
"""
完整登录流程
Returns:
LoginResult: 登录结果，包含cookie
"""
logger.info(f"开始登录账号: {self.account.username}")
try:
# 1️⃣ 第一次登录（获取极验参数）
logger.info("步骤1: 第一次登录，获取极验参数...")
gt, challenge, code_token, initial_cookies = self._first_login()
# 2️⃣ 极验 fullpage 验证
logger.info("步骤2: 极验 fullpage 验证...")
validate, seccode = self._solve_geetest(gt, challenge)
# 3️⃣ 第二次登录（带极验）
logger.info("步骤3: 第二次登录（带极验验证）...")
remote_code = self._second_login(gt, challenge, validate, seccode, code_token)
# 4️⃣ 发送邮箱验证
logger.info("步骤4: 发送邮箱验证...")
email_sent_at = time.time()
self._send_email_verify(remote_code)
# 5️⃣ IMAP获取验证码
logger.info("步骤5: 获取邮箱验证码...")
verify_code = self._get_email_code(after_timestamp=email_sent_at)
# 6️⃣ 提交验证码
logger.info("步骤6: 提交验证码...")
login_url = self._submit_verify_code(remote_code, verify_code)
# 7️⃣ 完成登录获取Cookie
logger.info("步骤7: 完成登录，获取Cookie...")
cookie = self._complete_login(login_url)
logger.success(f"登录成功! Cookie长度: {len(cookie)}")
return LoginResult(success=True, cookie=cookie, message="登录成功")
except Exception as e:
logger.error(f"登录失败: {e}")
return LoginResult(success=False, message=str(e))
def _first_login(self) -> Tuple[str, str, str, dict]:
"""
第一次登录，获取极验参数
Returns:
(gt, challenge, code_token, cookies)
"""
# 加密用户名和密码
encrypted_username = encrypt_nickname_or_phone(self.account.username)
encrypted_password = encrypt_password(self.account.password)
data = {
'type': '1',
'nicknameOrPhoneEncrypt': encrypted_username,
'password': encrypted_password,
'biz_type': '1',
'room_id': '0',
'redirect_url': self.LOGIN_REFERER,
't': str(int(time.time() * 1000)),
'client_id': '1',
'did': '',
'lang': '',
'isMultiAccount': '0',
}
payload = self._request_json(
'post',
self.LOGIN_API,
'第一次登录接口',
data=data,
)
logger.debug(f"第一次登录响应: {json.dumps(payload, ensure_ascii=False)[:200]}")
if payload.get('error') != 81:
error_msg = payload.get('msg', '未知错误')
raise ValueError(f"第一次登录失败: {error_msg}")
# 提取极验参数
geetest_data = payload.get('data', {}).get('geetest', {})
code_data = geetest_data.get('code_data', {})
code_token = geetest_data.get('code_token', '')
gt = code_data.get('gt', '')
challenge = code_data.get('challenge', '')
if not gt or not challenge:
raise ValueError("获取极验参数失败")
logger.info(f"获取极验参数成功: gt={gt[:10]}..., challenge={challenge[:10]}...")
return gt, challenge, code_token, self.session.cookies.get_dict()
def _solve_geetest(self, gt: str, challenge: str) -> Tuple[str, str]:
"""
解决极验 fullpage 验证（带重试机制）
Args:
gt: 极验gt参数
challenge: 极验challenge参数（第一次登录返回的）
Returns:
(validate, seccode)
"""
logger.info("开始极验 fullpage 验证...")
for attempt in range(self.max_geetest_retries):
try:
logger.info(f"极验验证尝试 {attempt + 1}/{self.max_geetest_retries}")
# 按斗鱼登录页 HAR：fullpage 智能检测流程，不进入图片滑块。
str_16 = _generate_seed()
proxies = dict(self.session.proxies)
# 获取JS地址
get_js_address(gt, proxies=proxies)
# 获取第一个w值
w1 = get_w1(gt, challenge, str_16)
# 获取c和s
c, s = get_c_s(gt, challenge, w1, proxies=proxies)
# 获取第二个w值
w2 = get_w2(gt, challenge, c, s, str_16)
# HAR 中 ajax.php 直接返回 validate。
result = req_fullpage_validate(gt, challenge, w2, proxies=proxies)
# 从 fullpage ajax.php 响应中提取 validate
if isinstance(result, dict):
data = result.get('data', {})
validate = data.get('validate', '') or result.get('validate', '')
success = data.get('result') == 'success' or result.get('success') == 1
message = data.get('result') or result.get('message', '')
if success and validate:
seccode = f"{validate}|jordan"
logger.success(f"极验 fullpage 验证成功! validate={validate[:20]}...")
return validate, seccode
else:
logger.warning(f"极验验证失败: {message}，重试中...")
# 刷新代理
self._refresh_proxy()
time.sleep(1)
continue
else:
validate = str(result)
if validate:
seccode = f"{validate}|jordan"
logger.success(f"极验 fullpage 验证成功! validate={validate[:20]}...")
return validate, seccode
except Exception as e:
logger.warning(f"极验验证异常: {e}，重试中...")
# 刷新代理
self._refresh_proxy()
# 增加等待时间，避免请求过于频繁
time.sleep(2)
continue
raise ValueError(f"极验验证失败，已重试 {self.max_geetest_retries} 次")
def _second_login(self, gt: str, challenge: str, validate: str,
seccode: str, code_token: str) -> str:
"""
第二次登录（带极验验证）
Args:
gt: 极验gt参数
challenge: 第一次登录返回的challenge
validate: 极验验证返回的validate
seccode: 极验验证返回的seccode
code_token: 第一次登录返回的code_token
Returns:
remote_code: 用于邮箱验证的code
"""
encrypted_username = encrypt_nickname_or_phone(self.account.username)
encrypted_password = encrypt_password(self.account.password)
# 参考HAR文件中的完整参数
# 注意：geetest_challenge应该使用第一次登录返回的challenge
data = {
'type': '1',
'nicknameOrPhoneEncrypt': encrypted_username,
'password': encrypted_password,
'room_id': '0',
'code_type': '1',
'code_token': code_token,
'gt_version': 'v3',
'geetest_challenge': challenge,
'geetest_validate': validate,
'geetest_seccode': seccode,
'code_data[geetest_challenge]': challenge,
'code_data[geetest_validate]': validate,
'code_data[geetest_seccode]': seccode,
'code_data[gt_version]': 'v3',
'code_data[code]': '',
'redirect_url': self.LOGIN_REFERER,
't': str(int(time.time() * 1000)),
'client_id': '1',
'did': '',
'lang': '',
'isMultiAccount': '0',
'biz_type': '1',
}
logger.debug(f"第二次登录参数: challenge={challenge[:20]}..., validate={validate[:20]}...")
payload = self._request_json(
'post',
self.LOGIN_API,
'第二次登录接口',
data=data,
)
logger.debug(f"第二次登录响应: {json.dumps(payload, ensure_ascii=False)[:200]}")
if payload.get('error') != 130014:
error_msg = payload.get('msg', '未知错误')
raise ValueError(f"第二次登录失败: {error_msg}")
# 提取remote_code
remote_code = payload.get('data', {}).get('remoteLogin', {}).get('code', '')
if not remote_code:
raise ValueError("获取remote_code失败")
logger.info(f"获取remote_code成功: {remote_code[:20]}...")
return remote_code
def _send_email_verify(self, remote_code: str) -> None:
"""发送邮箱验证"""
data = {
'code': remote_code,
'client_id': '1',
}
payload = self._request_json(
'post',
self.SEND_EMAIL_API,
'发送验证邮件接口',
data=data,
)
if payload.get('error') != 0:
raise ValueError(f"发送验证邮件失败: {payload.get('msg')}")
logger.info("验证邮件已发送")
def _get_email_code(self, after_timestamp: Optional[float] = None) -> str:
"""获取邮箱验证码"""
verifier = EmailVerifier(
imap_server=self.account.email_imap_server,
imap_port=self.account.email_imap_port,
username=self.account.email,
password=self.account.email_password,
)
return verifier.get_verification_code(
max_wait=60,
after_timestamp=after_timestamp,
)
def _submit_verify_code(self, remote_code: str, verify_code: str) -> str:
"""
提交验证码
Returns:
login_url: 登录回调URL
"""
data = {
'verify_type': '2',
'captcha': verify_code,
'isMultiAccount': '0',
'code': remote_code,
'client_id': '1',
'redirect_url': '//www.douyu.com/api/passport/login',
}
payload = self._request_json(
'post',
self.VERIFY_API,
'提交验证码接口',
data=data,
)
if payload.get('error') != 0:
raise ValueError(f"提交验证码失败: {payload.get('msg')}")
login_url = payload.get('data', {}).get('url', '')
if not login_url:
raise ValueError("获取登录URL失败")
# 补全URL
if login_url.startswith('//'):
login_url = 'https:' + login_url
logger.info(f"获取登录URL成功: {login_url[:50]}...")
return login_url
def _complete_login(self, login_url: str) -> str:
"""
完成登录，获取Cookie
Returns:
cookie: 完整的Cookie字符串
"""
# 访问登录URL
response = self._request('get', login_url)
response.raise_for_status()
# 尝试访问webLogin获取用户信息
try:
code_match = re.search(r'code=([^&]+)', login_url)
if code_match:
code = code_match.group(1)
weblogin_url = f"{self.WEBLOGIN_API}?code={code}"
response2 = self._request('get', weblogin_url)
if response2.status_code == 200:
logger.info("WebLogin成功")
except Exception as e:
logger.warning(f"WebLogin请求失败（不影响登录）: {e}")
# 收集所有Cookie
cookies = self.session.cookies.get_dict()
# 格式化Cookie字符串
cookie_str = '; '.join([f"{k}={v}" for k, v in cookies.items()])
return cookie_str
def save_cookie(self, cookie: str, filepath: str) -> None:
"""保存Cookie到文件"""
Path(filepath).parent.mkdir(parents=True, exist_ok=True)
with open(filepath, 'w', encoding='utf-8') as f:
json.dump({
'username': self.account.username,
'cookie': cookie,
'timestamp': int(time.time()),
}, f, ensure_ascii=False, indent=2)
logger.info(f"Cookie已保存到: {filepath}")
core/douyu/proxy.py
+115
View File
@@ -0,0 +1,115 @@
"""代理管理模块"""
import re
import requests
from typing import Optional, List
from loguru import logger
class ProxyManager:
"""代理管理器"""
def __init__(self, api_url: str = None):
self.api_url = api_url or "http://api.xiequ.cn/VAD/GetIp.aspx?act=get&uid=106015&vkey=97111DB5379E38E3BC2FF09A1B00A0C7&num=1&time=30&plat=1&re=0&type=0&so=1&ow=1&spl=1&addr=&db=1"
self.current_proxy: Optional[str] = None
def get_proxy(self) -> Optional[str]:
"""
从代理API获取代理IP
Returns:
代理URL，格式: http://ip:port
"""
try:
logger.info("获取代理IP...")
response = requests.get(self.api_url, timeout=10)
response.raise_for_status()
text = response.text.strip()
logger.debug(f"代理API响应: {text}")
# 解析IP:Port格式
match = re.search(r'(\d+\.\d+\.\d+\.\d+):(\d+)', text)
if match:
ip = match.group(1)
port = match.group(2)
proxy = f"http://{ip}:{port}"
self.current_proxy = proxy
logger.info(f"获取到代理: {proxy}")
return proxy
else:
logger.warning(f"无法解析代理地址: {text}")
return None
except Exception as e:
logger.error(f"获取代理失败: {e}")
return None
def get_proxies_dict(self, proxy: str = None) -> dict:
"""
获取requests使用的proxies字典
Args:
proxy: 代理URL，如果不提供则使用当前代理
Returns:
proxies字典
"""
proxy = proxy or self.current_proxy
if proxy:
return {
'http': proxy,
'https': proxy,
}
return {}
def verify_proxy(self, proxy: str = None) -> bool:
"""
验证代理是否可用
Args:
proxy: 代理URL
Returns:
是否可用
"""
proxy = proxy or self.current_proxy
if not proxy:
return False
try:
response = requests.get(
'https://httpbin.org/ip',
proxies={'http': proxy, 'https': proxy},
timeout=10
)
if response.status_code == 200:
data = response.json()
logger.info(f"代理验证成功，当前IP: {data.get('origin')}")
return True
return False
except Exception as e:
logger.error(f"代理验证失败: {e}")
return False
# 全局代理管理器实例
_proxy_manager: Optional[ProxyManager] = None
def get_proxy_manager(api_url: str = None) -> ProxyManager:
"""获取全局代理管理器实例"""
global _proxy_manager
if _proxy_manager is None:
_proxy_manager = ProxyManager(api_url)
return _proxy_manager
def get_proxy() -> Optional[str]:
"""获取代理URL的便捷函数"""
return get_proxy_manager().get_proxy()
def get_proxies_dict() -> dict:
"""获取proxies字典的便捷函数"""
return get_proxy_manager().get_proxies_dict()
core/douyu/whitelist.py
+268
View File
@@ -0,0 +1,268 @@
"""代理IP白名单管理模块"""
import json
import re
from typing import Optional
from urllib.parse import urlencode
import requests
from loguru import logger
class WhitelistManager:
"""协固代理IP白名单管理器"""
MEMO_PREFIX = "douyu_auto"
BASE_URL = "http://op.xiequ.cn/IpWhiteList.aspx"
def __init__(self, uid: str, ukey: str):
self.uid = uid
self.ukey = ukey
self._memo = self.MEMO_PREFIX
@property
def memo(self) -> str:
"""当前机器的固定备注"""
return self._memo
def _build_url(self, **params) -> str:
"""构建请求URL"""
base_params = {
"uid": self.uid,
"ukey": self.ukey,
}
base_params.update(params)
query = urlencode(base_params)
return f"{self.BASE_URL}?{query}"
def get_whitelist_json(self) -> list[dict]:
"""
获取白名单列表（JSON格式）
Returns:
[{"IP": "x.x.x.x", "MEMO": "备注"}, ...]
"""
try:
url = self._build_url(act="getjson")
response = requests.get(url, timeout=10)
response.raise_for_status()
text = response.text.strip()
if not text or text == "[]":
return []
data = json.loads(text)
if isinstance(data, list):
return data
# 处理 {"data": [...]} 格式
if isinstance(data, dict):
items = data.get("data", [])
if isinstance(items, list):
return items
return []
except Exception as e:
logger.error(f"获取白名单失败: {e}")
return []
def add_ip(self, ip: str) -> bool:
"""
添加IP到白名单
Args:
ip: 要添加的IP地址
"""
try:
url = self._build_url(act="add", ip=ip, meno=self._memo)
response = requests.get(url, timeout=10)
response.raise_for_status()
text = response.text.strip()
logger.debug(f"添加白名单响应: {text}")
# 成功通常返回 "ok" 或类似信息
if "ok" in text.lower() or "success" in text.lower() or "添加成功" in text:
logger.info(f"白名单添加成功: {ip} (备注: {self._memo})")
return True
# 检查是否已存在
if "已存在" in text or "exist" in text.lower():
logger.info(f"白名单已存在: {ip}")
return True
logger.warning(f"白名单添加结果: {text}")
return False
except Exception as e:
logger.error(f"添加白名单失败: {e}")
return False
def delete_ip(self, ip: str) -> bool:
"""
删除指定IP
Args:
ip: 要删除的IP地址
"""
try:
url = self._build_url(act="del", ip=ip)
response = requests.get(url, timeout=10)
response.raise_for_status()
text = response.text.strip()
logger.debug(f"删除白名单响应: {text}")
if "ok" in text.lower() or "success" in text.lower() or "删除成功" in text:
logger.info(f"白名单删除成功: {ip}")
return True
logger.warning(f"白名单删除结果: {text}")
return False
except Exception as e:
logger.error(f"删除白名单失败: {e}")
return False
def get_memo_ip(self) -> Optional[str]:
"""
获取当前备注对应的IP
Returns:
IP地址，如果不存在返回None
"""
records = self.get_whitelist_json()
for record in records:
if record.get("MEMO") == self._memo:
return record.get("IP")
return None
def get_memo_records(self) -> list[dict]:
"""
获取当前备注的所有记录
Returns:
匹配备注的记录列表
"""
records = self.get_whitelist_json()
return [r for r in records if r.get("MEMO") == self._memo]
def sync_ip(self, current_ip: str) -> tuple[bool, str]:
"""
同步白名单IP
检查当前备注是否有记录：
- 如果IP相同，无需操作
- 如果IP不同，删除旧的并添加新的
- 如果IP已存在但备注不同（如手动添加无备注），删除后重新添加
- 如果无记录，添加新的
Args:
current_ip: 当前出口IP
Returns:
(是否成功, 消息)
"""
try:
existing_ip = self.get_memo_ip()
# IP相同，无需更新
if existing_ip == current_ip:
msg = f"白名单IP已是最新的: {current_ip}"
logger.info(msg)
return True, msg
# 有旧记录，先删除
if existing_ip:
logger.info(f"白名单IP变化: {existing_ip} -> {current_ip}")
self.delete_ip(existing_ip)
# IP已存在但备注不同（或备注为空），先删除后重新添加
records = self.get_whitelist_json()
if any(r.get('IP') == current_ip for r in records):
logger.info(f"白名单IP {current_ip} 已存在但备注不同，删除后重新添加")
self.delete_ip(current_ip)
# 添加新IP
if self.add_ip(current_ip):
if existing_ip:
msg = f"白名单IP已更新: {existing_ip} -> {current_ip}"
else:
msg = f"白名单IP已添加: {current_ip}"
logger.info(msg)
return True, msg
return False, "白名单添加失败"
except Exception as e:
msg = f"白名单同步失败: {e}"
logger.error(msg)
return False, msg
def test_connection(self) -> tuple[bool, str]:
"""
测试白名单API连接
Returns:
(是否成功, 消息)
"""
try:
records = self.get_whitelist_json()
count = len(records)
my_records = [r for r in records if r.get("MEMO", "").startswith(self.MEMO_PREFIX)]
msg = f"连接成功，白名单共 {count} 条记录，其中本机相关 {len(my_records)} 条"
logger.info(msg)
return True, msg
except Exception as e:
msg = f"连接失败: {e}"
logger.error(msg)
return False, msg
def get_exit_ip_via_proxy(proxy: str) -> Optional[str]:
"""
通过代理获取出口IP
Args:
proxy: 代理URL，格式 http://ip:port
Returns:
出口IP地址
"""
targets = [
"https://qifu-api.baidubce.com/ip/local/geo/v1/district",
"https://myip.ipip.net",
"https://4.ipw.cn",
]
proxies = {"http": proxy, "https": proxy}
for url in targets:
try:
response = requests.get(
url,
proxies=proxies,
timeout=6,
headers={"User-Agent": "Mozilla/5.0"},
)
response.raise_for_status()
# 尝试解析IP
content_type = response.headers.get("content-type", "").lower()
if "json" in content_type:
data = response.json()
ip = data.get("ip") or data.get("origin")
if ip:
return str(ip).split(",")[0].strip()
# 从文本中提取IP
text = response.text.strip()
match = re.search(r"(\d{1,3}(?:\.\d{1,3}){3})", text)
if match:
return match.group(1)
except Exception:
continue
return None
core/geetest/__init__.py
+5
View File
@@ -0,0 +1,5 @@
"""Geetest 验证码求解器"""
from .v3_slide.solver import run_solver
__all__ = ["run_solver"]
core/geetest/common/__init__.py
+1
View File
@@ -0,0 +1 @@
"""Geetest 公共工具模块"""
core/geetest/common/crypto.py
+246
View File
@@ -0,0 +1,246 @@
import random
import hashlib
from typing import Any
from Crypto.PublicKey import RSA
from Crypto.Cipher import PKCS1_v1_5, AES
# 随机产生4个字符组成的字符串
def four_random_chart() -> str:
return hex(int(65536 * (1 + random.random())))[2:][1:]
# PKCS#1 v1.5 填充 + RSA 加密
def parse_jsbn_bigint(n_obj:dict[Any, int]) -> int:
DB = 28
DV = 1 << DB
t = n_obj['t']
result = 0
for i in range(t):
result += n_obj[i] * (DV ** i)
return result
def encrypt_data(plaintext:str) -> str:
cipher = PKCS1_v1_5.new(public_key)
encrypted = cipher.encrypt(plaintext.encode('utf-8'))
# 转换为十六进制，确保偶数长度
hex_result = encrypted.hex()
if len(hex_result) % 2 == 1:
hex_result = '0' + hex_result
return hex_result
def RSA_jiami_r(str_16:str) -> str:
global public_key
# 你的数据
n_data = {
0: 134982529, 1: 254232810, 2: 164556709, 3: 234907349,
4: 134685994, 5: 35463984, 6: 258277946, 7: 12518857,
8: 44638621, 9: 93783641, 10: 212253739, 11: 62792472,
12: 186688352, 13: 109500232, 14: 182488077, 15: 261196188,
16: 26354094, 17: 103248217, 18: 106891695, 19: 165771045,
20: 41530993, 21: 263704736, 22: 111785174, 23: 12753611,
24: 232116673, 25: 155524985, 26: 218291229, 27: 122452343,
28: 248250238, 29: 118739550, 30: 251169095, 31: 129059733,
32: 149835464, 33: 5498868, 34: 71719731, 35: 154456417,
36: 49635,
't': 37, 's': 0
}
e = 65537
n = parse_jsbn_bigint(n_data)
# 构造公钥
public_key = RSA.construct((n, e))
encrypted = encrypt_data(str_16)
return encrypted
# AES加密
# 加密模式: AES-CBC
# 密钥长度: 128位
# IV: 固定为 "0000000000000000"
def parse_string_to_wordarray(text:str) -> list[int]:
"""将字符串转换为 WordArray 格式"""
length = len(text)
words = []
for i in range(length):
# 计算在 words 数组中的索引
word_index = i >> 2 # 相当于 i // 4
# 确保 words 数组足够长
while len(words) <= word_index:
words.append(0)
# 获取字符的 ASCII 码
char_code = ord(text[i]) & 0xFF
# 计算位移量
shift = 24 - (i % 4) * 8
# 将字符添加到对应的 word 中
words[word_index] |= char_code << shift
return words
def AES_O(plaintext:str, str_16:str) -> list[int]:
# 密钥
key_words = parse_string_to_wordarray(str_16)
key = b''.join(w.to_bytes(4, 'big') for w in key_words)
# IV
iv = b'0000' * 4 # "0000000000000000"
# 填充(PKCS7)
pad_len = 16 - len(plaintext) % 16
plaintext_padded = plaintext.encode() + bytes([pad_len] * pad_len)
# 加密
cipher = AES.new(key, AES.MODE_CBC, iv)
ciphertext = cipher.encrypt(plaintext_padded)
# 结果是字节数组
return list(ciphertext)
# 自定义base64编码
def geetest_base64_encode(data:list[int]) -> dict[str, Any]:
"""极验自定义Base64编码"""
# 配置
charset = 'ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789()'
pad_char = '.'
# 位掩码 (这些是打乱的)
masks = [7274496, 9483264, 19220, 235]
mask_bits = 24 # 总位数
def extract_bits(value, mask):
"""根据掩码提取位"""
result = 0
# 从高位到低位遍历
for i in range(mask_bits - 1, -1, -1):
# 如果掩码的第i位是1
if (mask >> i) & 1:
# 提取value的第i位,添加到结果
result = (result << 1) | ((value >> i) & 1)
return result
encoded = ""
padding = ""
length = len(data)
# 每3字节一组
i = 0
while i < length:
if i + 2 < length:
# 完整3字节
block = (data[i] << 16) | (data[i + 1] << 8) | data[i + 2]
# 使用4个掩码提取
encoded += charset[extract_bits(block, masks[0])]
encoded += charset[extract_bits(block, masks[1])]
encoded += charset[extract_bits(block, masks[2])]
encoded += charset[extract_bits(block, masks[3])]
i += 3
else:
# 处理剩余
remainder = length - i
if remainder == 2:
block = (data[i] << 16) | (data[i + 1] << 8)
encoded += charset[extract_bits(block, masks[0])]
encoded += charset[extract_bits(block, masks[1])]
encoded += charset[extract_bits(block, masks[2])]
padding = pad_char
elif remainder == 1:
block = data[i] << 16
encoded += charset[extract_bits(block, masks[0])]
encoded += charset[extract_bits(block, masks[1])]
padding = pad_char + pad_char
break
return {
"res": encoded,
"end": padding
}
def encrypt_string(e:str, t:list[int], n:str) -> str:
"""
JS加密函数的Python实现
参数:
e: 原始字符串
t: 加密参数数组
n: 十六进制字符串
"""
if not t or not n:
return e
o = 0 # 偏移量
i = e # 结果字符串
s = t[0] # 12
a = t[2] # 98
_ = t[4] # 43
# 每次读取2个字符（十六进制）
while o < len(n):
r = n[o:o + 2] # 取2个字符
if len(r) < 2:
break
o += 2
# 解析十六进制
c = int(r, 16)
# 转换为字符
l = chr(c)
# 计算插入位置: (s * c^2 + a * c + _) % len(e)
u = (s * c * c + a * c + _) % len(e)
# 在位置u插入字符
i = i[:u] + l + i[u:]
return i
def simple_md5(message:str) -> str:
"""
简化版MD5实现，结构更清晰
"""
# 使用内置hashlib验证结果
def verify_result(msg):
return hashlib.md5(msg.encode()).hexdigest()
# 轮移位常量
shifts = [
7, 12, 17, 22, 7, 12, 17, 22, 7, 12, 17, 22, 7, 12, 17, 22,
5, 9, 14, 20, 5, 9, 14, 20, 5, 9, 14, 20, 5, 9, 14, 20,
4, 11, 16, 23, 4, 11, 16, 23, 4, 11, 16, 23, 4, 11, 16, 23,
6, 10, 15, 21, 6, 10, 15, 21, 6, 10, 15, 21, 6, 10, 15, 21
]
# K常数（与JavaScript版本中的常数对应）
K = [
0xd76aa478, 0xe8c7b756, 0x242070db, 0xc1bdceee,
0xf57c0faf, 0x4787c62a, 0xa8304613, 0xfd469501,
0x698098d8, 0x8b44f7af, 0xffff5bb1, 0x895cd7be,
0x6b901122, 0xfd987193, 0xa679438e, 0x49b40821,
0xf61e2562, 0xc040b340, 0x265e5a51, 0xe9b6c7aa,
0xd62f105d, 0x02441453, 0xd8a1e681, 0xe7d3fbc8,
0x21e1cde6, 0xc33707d6, 0xf4d50d87, 0x455a14ed,
0xa9e3e905, 0xfcefa3f8, 0x676f02d9, 0x8d2a4c8a,
0xfffa3942, 0x8771f681, 0x6d9d6122, 0xfde5380c,
0xa4beea44, 0x4bdecfa9, 0xf6bb4b60, 0xbebfbc70,
0x289b7ec6, 0xeaa127fa, 0xd4ef3085, 0x04881d05,
0xd9d4d039, 0xe6db99e5, 0x1fa27cf8, 0xc4ac5665,
0xf4292244, 0x432aff97, 0xab9423a7, 0xfc93a039,
0x655b59c3, 0x8f0ccc92, 0xffeff47d, 0x85845dd1,
0x6fa87e4f, 0xfe2ce6e0, 0xa3014314, 0x4e0811a1,
0xf7537e82, 0xbd3af235, 0x2ad7d2bb, 0xeb86d391
]
# 实际实现...
return verify_result(message)
core/geetest/common/imaging.py
+89
View File
@@ -0,0 +1,89 @@
import requests
import cv2
import numpy as np
from PIL import Image
REQUEST_TIMEOUT = (3.05, 12)
# 将 Image 转换为 Mat，通过 flag 可以控制颜色
def pilImgToCv2(img: Image.Image, flag=cv2.COLOR_RGB2BGR):
return cv2.cvtColor(np.asarray(img), flag)
# 识别图片缺口返回滑块距离
def shibie(img: Image.Image, slice: Image.Image):
# 通过 pilImgToCv2 将图片置灰
# 背景图和滑块图都需要做相同处理
grayImg = pilImgToCv2(img, cv2.COLOR_BGR2GRAY)
# showImg(grayImg) # 可以通过它来看处理后的图片效果
graySlice = pilImgToCv2(slice, cv2.COLOR_BGR2GRAY)
# 做边缘检测进一步降低干扰，阈值可以自行调整
grayImg = cv2.Canny(grayImg, 255, 255)
# showImg(grayImg) # 可以通过它来看处理后的图片效果
graySlice = cv2.Canny(graySlice, 255, 255)
# 通过模板匹配两张图片，找出缺口的位置
result = cv2.matchTemplate(grayImg, graySlice, cv2.TM_CCOEFF_NORMED)
maxLoc = cv2.minMaxLoc(result)[3]
# 匹配出来的滑动距离
distance = maxLoc[0]
# 下面的逻辑是在图片画出一个矩形框来标记匹配到的位置，可以直观的看到匹配结果，去掉也可以的
sliceHeight, sliceWidth = graySlice.shape[:2]
# 左上角
x, y = maxLoc
# 右下角
x2, y2 = x + sliceWidth, y + sliceHeight
resultBg = pilImgToCv2(img, cv2.COLOR_RGB2BGR)
cv2.rectangle(resultBg, (x, y), (x2, y2), (0, 0, 255), 2)
# showImg(resultBg) # 可以通过它来看处理后的图片效果
return distance
# 还原图片
def restore_geetest_image(input_path:str, output_path:str) -> None:
"""
还原极验打乱的验证码图像
"""
Ut = [
39, 38, 48, 49, 41, 40, 46, 47, 35, 34, 50, 51, 33, 32, 28, 29,
27, 26, 36, 37, 31, 30, 44, 45, 43, 42, 12, 13, 23, 22, 14, 15,
21, 20, 8, 9, 25, 24, 6, 7, 3, 2, 0, 1, 11, 10, 4, 5, 19, 18, 16, 17
]
# 打开混淆图像
img = Image.open(input_path)
new_img = Image.new("RGB", (260,160))
r = 160
for _ in range(len(Ut)):
a = r / 2
c = Ut[_] % 26 * 12 + 1
u = 80 if Ut[_] > 25 else 0
l = img.crop(box=(c, u, c + 10, u + 80))
new_img.paste(l, box=(_ % 26 * 10, 80 if _ > 25 else 0))
new_img.save(output_path)
print(f"图像已还原并保存到: {output_path}")
# 下载图片
def download_picture(bg:str, fullbg:str, slice:str) -> int:
for i in range(3):
if i == 0:
url = "https://static.geetest.com/"+bg
response = requests.get(url, timeout=REQUEST_TIMEOUT)
response.raise_for_status()
with open("bg.jpg", "wb") as f:
f.write(response.content)
restore_geetest_image("bg.jpg", "bg.jpg")
else:
if i == 1:
url = "https://static.geetest.com/" + fullbg
response = requests.get(url, timeout=REQUEST_TIMEOUT)
response.raise_for_status()
with open("fullbg.jpg", "wb") as f:
f.write(response.content)
restore_geetest_image("fullbg.jpg", "fullbg.jpg")
else:
url = "https://static.geetest.com/" + slice
response = requests.get(url, timeout=REQUEST_TIMEOUT)
response.raise_for_status()
with open("slice.jpg", "wb") as f:
f.write(response.content)
return shibie(Image.open('fullbg.jpg'), Image.open('slice.jpg'))
core/geetest/common/network.py
+338
View File
@@ -0,0 +1,338 @@
import time
import requests
import json
import re
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:
response.raise_for_status()
body = response.text.strip()
if not body:
raise ValueError(f"{source} 返回为空，无法解析 JSON")
try:
return response.json()
except json.JSONDecodeError as exc:
preview = body[:200].replace("\n", "\\n")
raise ValueError(f"{source} 返回的不是有效 JSON: {preview}") from exc
def get_challenge_gt_bak() -> Tuple[str, str]:
headers = {
'accept': 'application/json, text/javascript, */*; q=0.01',
'accept-language': 'zh-CN,zh;q=0.9',
'priority': 'u=1, i',
'referer': 'https://demos.geetest.com/slide-float.html',
'sec-ch-ua': '"Chromium";v="140", "Not=A?Brand";v="24", "Google Chrome";v="140"',
'sec-ch-ua-mobile': '?0',
'sec-ch-ua-platform': '"Windows"',
'sec-fetch-dest': 'empty',
'sec-fetch-mode': 'cors',
'sec-fetch-site': 'same-origin',
'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',
'x-requested-with': 'XMLHttpRequest',
}
params = {
't': str(int(round(time.time() * 1000))),
}
response = requests.get(
'https://demos.geetest.com/gt/register-slide',
params=params,
headers=headers,
timeout=10,
)
data = _parse_json_response(response, "极验 demo register-slide")
return data["gt"], data["challenge"]
def get_challenge_gt() -> Tuple[str, str]:
headers = {
'User-Agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/146.0.0.0 Safari/537.36',
'Accept': 'application/json, text/javascript, */*; q=0.01',
'Content-Type': 'application/x-www-form-urlencoded',
'referer': 'https://passport.douyu.com/member/login?state=https%3A%2F%2Fwww.douyu.com%2Fmember%2FcpSecurity%2Fcheck_geetest_status',
}
data = {
'type': '1',
'nicknameOrPhoneEncrypt': '03ILaBwtmmCm0A==',
'password': '57219dddec71c31b7647683fa5306103',
'biz_type': '1',
}
response = requests.post(
'https://passport.douyu.com/wgapi/member/passport/login',
headers=headers,
data=data,
timeout=10,
)
payload = _parse_json_response(response, "斗鱼登录接口")
try:
code_data = payload["data"]["geetest"]["code_data"]
return code_data["gt"], code_data["challenge"]
except KeyError as exc:
preview = json.dumps(payload, ensure_ascii=False)[:300]
raise ValueError(f"斗鱼登录接口返回中缺少极验参数: {preview}") from exc
def get_js_address(gt: str, proxies: Optional[Mapping[str, str]] = None) -> dict:
headers = {
'accept': '*/*',
'accept-language': 'zh-CN,zh;q=0.9',
'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': '"macOS"',
'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',
}
params = {
'gt': gt,
'callback': 'geetest_' + str(int(round(time.time() * 1000))),
}
response = _get(
'https://api.geetest.com/gettype.php',
params=params,
headers=headers,
proxies=proxies,
)
return _parse_jsonp_response(response, "极验 gettype")
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': 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': '"macOS"',
'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',
}
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,
proxies=proxies,
)
data = _parse_jsonp_response(response, "极验 get.php")
return data['data']['c'], data['data']['s']
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"',
}
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 = {
'Accept': '*/*',
'Accept-Language': 'zh-CN,zh;q=0.9',
'Connection': 'keep-alive',
'Referer': 'https://demos.geetest.com/',
'Sec-Fetch-Dest': 'script',
'Sec-Fetch-Mode': 'no-cors',
'Sec-Fetch-Site': 'cross-site',
'Sec-Fetch-Storage-Access': 'active',
'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-ch-ua': '"Chromium";v="140", "Not=A?Brand";v="24", "Google Chrome";v="140"',
'sec-ch-ua-mobile': '?0',
'sec-ch-ua-platform': '"Windows"',
}
response = requests.get(
'https://api.geevisit.com/ajax.php?gt=' + gt + '&challenge=' + challenge + '&lang=zh-cn&pt=0&client_type=web&w=' + w2 + '&callback=geetest_' + str(
int(round(time.time() * 1000))),
headers=headers,
timeout=REQUEST_TIMEOUT,
)
response.raise_for_status()
def get_picture(gt:str, challenge:str) -> tuple[str, str, list[int], str, str, str]:
headers = {
'Accept': '*/*',
'Accept-Language': 'zh-CN,zh;q=0.9',
'Connection': 'keep-alive',
'Referer': 'https://demos.geetest.com/',
'Sec-Fetch-Dest': 'script',
'Sec-Fetch-Mode': 'no-cors',
'Sec-Fetch-Site': 'cross-site',
'Sec-Fetch-Storage-Access': 'active',
'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-ch-ua': '"Chromium";v="140", "Not=A?Brand";v="24", "Google Chrome";v="140"',
'sec-ch-ua-mobile': '?0',
'sec-ch-ua-platform': '"Windows"',
}
params = {
'is_next': 'true',
'type': 'slide3',
'gt': gt,
'challenge': challenge,
'lang': 'zh-cn',
'https': 'true',
'protocol': 'https://',
'offline': 'false',
'product': 'embed',
'api_server': 'api.geevisit.com',
'isPC': 'true',
'autoReset': 'true',
'width': '100%',
'callback': 'geetest_'+str(int(round(time.time() * 1000))),
}
response = requests.get(
'https://api.geevisit.com/get.php',
params=params,
headers=headers,
timeout=REQUEST_TIMEOUT,
)
response.raise_for_status()
print(response.text)
match = re.search(r'\((.*)\)$', response.text)
if match:
json_str = match.group(1)
data = json.loads(json_str)
# 检查验证码类型
if 'data' in data and isinstance(data['data'], dict):
# 新版极验格式
inner_data = data['data']
return (
inner_data.get('bg', ''),
inner_data.get('fullbg', ''),
inner_data.get('c', []),
inner_data.get('s', ''),
inner_data.get('slice', ''),
inner_data.get('challenge', '')
)
else:
# 旧版极验格式
return (
data.get('bg', ''),
data.get('fullbg', ''),
data.get('c', []),
data.get('s', ''),
data.get('slice', ''),
data.get('challenge', '')
)
else:
raise ValueError("无法解析 JSONP 响应")
def req_end(gt:str, challenge:str, w:str) -> dict:
headers = {
'Accept': '*/*',
'Accept-Language': 'zh-CN,zh;q=0.9',
'Connection': 'keep-alive',
'Referer': 'https://demos.geetest.com/',
'Sec-Fetch-Dest': 'script',
'Sec-Fetch-Mode': 'no-cors',
'Sec-Fetch-Site': 'cross-site',
'Sec-Fetch-Storage-Access': 'active',
'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-ch-ua': '"Chromium";v="140", "Not=A?Brand";v="24", "Google Chrome";v="140"',
'sec-ch-ua-mobile': '?0',
'sec-ch-ua-platform': '"Windows"',
}
response = requests.get(
'https://api.geevisit.com/ajax.php?gt='+gt+'&challenge='+challenge+'&lang=zh-cn&%24_BCm=0&client_type=web&w='+w+'&callback=geetest_'+str(int(round(time.time() * 1000))),
headers=headers,
timeout=REQUEST_TIMEOUT,
)
response.raise_for_status()
print(response.text)
match = re.search(r'\((.*)\)$', response.text)
if match:
json_str = match.group(1)
data = json.loads(json_str)
print(f"极验验证响应: {data}")
# 检查验证是否成功
if data.get('success') == 1:
return data
else:
# 如果验证失败，返回错误信息
return {
'success': 0,
'message': data.get('message', '验证失败'),
'validate': ''
}
else:
raise ValueError("无法解析 JSONP 响应")
core/geetest/common/performance.py
+113
View File
@@ -0,0 +1,113 @@
import time
import random
from typing import Optional
def generate_fake_performance_timing(base_time: Optional[int] = None) -> dict[str, int]:
"""
生成伪造的浏览器性能时间戳数据
Args:
base_time: 基准时间戳（毫秒），默认使用当前时间
Returns:
dict: 包含所有性能时间戳的字典
"""
if base_time is None:
base_time = int(time.time() * 1000)
# 定义合理的时间间隔范围（毫秒）
intervals = {
'fetch': random.randint(1, 2),
'domain_lookup_start': random.randint(3, 5),
'domain_lookup': random.randint(5, 15),
'connect': random.randint(50, 150),
'ssl_offset': random.randint(30, 50),
'request': random.randint(1, 5),
'response': random.randint(20, 100),
'response_end': random.randint(1, 3),
'unload_start': random.randint(1, 3),
'unload': random.randint(1, 5),
'dom_loading': random.randint(1, 3),
'dom_interactive': random.randint(50, 200),
'dom_content_loaded': random.randint(1, 3),
'load_event': random.randint(0, 5)
}
timing = {}
# 按照时间顺序构建
timing['navigationStart'] = base_time
timing['fetchStart'] = timing['navigationStart'] + intervals['fetch']
timing['domainLookupStart'] = timing['fetchStart'] + intervals['domain_lookup_start']
timing['domainLookupEnd'] = timing['domainLookupStart'] + intervals['domain_lookup']
timing['connectStart'] = timing['domainLookupEnd']
timing['secureConnectionStart'] = timing['connectStart'] + intervals['ssl_offset']
timing['connectEnd'] = timing['connectStart'] + intervals['connect']
timing['requestStart'] = timing['connectEnd'] + intervals['request']
timing['responseStart'] = timing['requestStart'] + intervals['response']
timing['responseEnd'] = timing['responseStart'] + intervals['response_end']
timing['unloadEventStart'] = timing['responseEnd'] + intervals['unload_start']
timing['unloadEventEnd'] = timing['unloadEventStart'] + intervals['unload']
timing['domLoading'] = timing['unloadEventEnd'] + intervals['dom_loading']
timing['domInteractive'] = timing['domLoading'] + intervals['dom_interactive']
timing['domContentLoadedEventStart'] = timing['domInteractive']
timing['domContentLoadedEventEnd'] = timing['domInteractive'] + intervals['dom_content_loaded']
timing['domComplete'] = timing['domContentLoadedEventEnd']
timing['loadEventStart'] = timing['domComplete']
timing['loadEventEnd'] = timing['loadEventStart'] + intervals['load_event']
# 无重定向的情况
timing['redirectStart'] = 0
timing['redirectEnd'] = 0
return timing
def __ease_out_expo(sep):
'''
轨迹相关操作
'''
if sep == 1:
return 1
else:
return 1 - pow(2, -10 * sep)
def get_slide_track(distance):
"""
根据滑动距离生成滑动轨迹
:param distance: 需要滑动的距离
:return: 滑动轨迹<type 'list'>: [[x,y,t], ...]
x: 已滑动的横向距离
y: 已滑动的纵向距离, 除起点外, 均为0
t: 滑动过程消耗的时间, 单位: 毫秒
"""
if not isinstance(distance, int) or distance < 0:
raise ValueError(f"distance类型必须是大于等于0的整数: distance: {distance}, type: {type(distance)}")
# 初始化轨迹列表
slide_track = [
[random.randint(-50, -10), random.randint(-50, -10), 0],
[0, 0, 0],
]
# 共记录count次滑块位置信息
count = 10 + int(distance / 2)
# 初始化滑动时间
t = random.randint(50, 100)
# 记录上一次滑动的距离
_x = 0
_y = 0
for i in range(count):
# 已滑动的横向距离
x = round(__ease_out_expo(i / count) * distance)
# y = round(__ease_out_expo(i / count) * 14)
# 滑动过程消耗的时间
t += random.randint(10, 50)
if x == _x:
continue
slide_track.append([x, _y, t])
_x = x
slide_track.append(slide_track[-1])
return slide_track, slide_track[-1][2]
core/geetest/common/trajectory.py
+628
View File
@@ -0,0 +1,628 @@
import random
import math
from typing import Any, Optional
# 生成类人的鼠标轨迹
def generate_realistic_trajectory(start_x:int, start_y:int, end_x:int, end_y:int, start_time:int) -> list[Any]:
"""生成类人的鼠标轨迹"""
trajectory = []
# 计算距离和步数
distance = ((end_x - start_x) ** 2 + (end_y - start_y) ** 2) ** 0.5
steps = int(distance / 2) + random.randint(5, 15) # 根据距离动态调整步数
current_time = start_time
current_x, current_y = start_x, start_y
for i in range(steps):
# 使用贝塞尔曲线模拟自然移动
progress = i / steps
# 添加缓动函数（开始快，中间慢，结束快）
if progress < 0.3:
eased = progress / 0.3 * 0.2
elif progress < 0.7:
eased = 0.2 + (progress - 0.3) / 0.4 * 0.5
else:
eased = 0.7 + (progress - 0.7) / 0.3 * 0.3
# 计算目标位置（添加随机抖动）
target_x = start_x + (end_x - start_x) * eased
target_y = start_y + (end_y - start_y) * eased
# 添加微小的随机偏移（模拟手抖）
jitter_x = random.uniform(-0.5, 0.5)
jitter_y = random.uniform(-0.5, 0.5)
current_x = int(target_x + jitter_x)
current_y = int(target_y + jitter_y)
# 随机时间间隔（3-25ms，符合人类反应）
time_delta = random.choices(
[3, 4, 5, 6, 7, 8, 10, 12, 15, 20, 24],
weights=[5, 8, 10, 12, 10, 8, 5, 3, 2, 1, 1]
)[0]
current_time += time_delta
trajectory.append([
"move",
current_x,
current_y,
current_time,
"pointermove"
])
# 偶尔在同一位置停留（模拟视觉确认）
if random.random() < 0.15:
trajectory.append([
"move",
current_x,
current_y,
current_time + random.randint(5, 15),
"pointermove"
])
current_time += random.randint(5, 15)
# 到达目标后的悬停
hover_time = random.randint(50, 150)
for _ in range(random.randint(2, 5)):
current_time += random.randint(8, 25)
trajectory.append([
"move",
end_x + random.randint(-1, 1),
end_y + random.randint(-1, 1),
current_time,
"pointermove"
])
current_time += hover_time
# 点击事件
trajectory.append([
"down",
end_x,
end_y,
current_time,
"pointerdown"
])
trajectory.append([
"focus",
current_time + 1
])
click_duration = random.randint(80, 130)
trajectory.append([
"up",
end_x,
end_y,
current_time + click_duration,
"pointerup"
])
return trajectory
# 处理原始轨迹数组
def process_mouse_trajectory(events:list[Any], max_records: Optional[int] = None) -> dict[str,Any]:
"""
处理鼠标/触摸轨迹数据，将绝对坐标转换为相对坐标和时间差
参数:
events: 原始事件数据列表
max_records: 最大保留记录数（None表示保留全部）
返回:
处理后的事件列表
"""
if not events or len(events) == 0:
return []
# 初始化变量
prev_x = 0 # 上一个X坐标
prev_y = 0 # 上一个Y坐标
prev_time = 0 # 上一个时间戳
result = [] # 结果数组
first_event = None # 第一个事件
last_event = None # 最后一个事件
# 移动类事件（包含坐标信息）
MOVE_EVENTS = ["move", "mousemove", "touchmove", "pointermove"]
# 点击类事件（仅时间信息）
CLICK_EVENTS = ["down", "up", "click", "mousedown", "mouseup",
"touchstart", "touchend", "pointerdown", "pointerup"]
# 特殊事件（仅时间信息）
TIME_ONLY_EVENTS = ["focus", "blur", "keydown", "keyup"]
# 如果设置了最大记录数，只处理最后N条
start_index = 0
if max_records and len(events) > max_records:
start_index = len(events) - max_records
# 遍历事件
for i in range(start_index, len(events)):
event = events[i]
event_type = event[0]
# 处理移动类事件（包含X, Y坐标）
if event_type in MOVE_EVENTS:
x = event[1]
y = event[2]
timestamp = event[3]
# 记录第一个和最后一个事件
if first_event is None:
first_event = event
last_event = event
# 计算相对坐标差值
delta_x = x - prev_x
delta_y = y - prev_y
# 计算时间差
if prev_time == 0:
time_diff = 0 # 第一个事件时间差为0
else:
time_diff = timestamp - prev_time
# 添加到结果数组
result.append([
event_type,
[delta_x, delta_y],
time_diff
])
# 更新上一次的值
prev_x = x
prev_y = y
prev_time = timestamp
# 处理点击类事件（包含坐标但只记录时间差）
elif event_type in CLICK_EVENTS:
timestamp = event[3] if len(event) > 3 else event[1]
# 计算时间差
if prev_time == 0:
time_diff = 0
else:
time_diff = timestamp - prev_time
# 添加到结果数组（坐标差为[0,0]）
result.append([
event_type,
[0, 0],
time_diff
])
prev_time = timestamp
# 处理仅时间类事件（如focus）
elif event_type in TIME_ONLY_EVENTS:
timestamp = event[1]
# 计算时间差
if prev_time == 0:
time_diff = 0
else:
time_diff = timestamp - prev_time
# 添加到结果数组（仅包含时间差）
result.append([
event_type,
time_diff
])
prev_time = timestamp
return {
"data": result,
"first_event": first_event,
"last_event": last_event,
"total_events": len(result)
}
def compress_trajectory(e:list[Any]) -> str:
"""
压缩轨迹数据的完整实现
Args:
e: 轨迹数据列表
Returns:
压缩后的 Base64 编码字符串
"""
# 事件类型映射
p = {
"move": 0,
"down": 1,
"up": 2,
"scroll": 3,
"focus": 4,
"blur": 5,
"unload": 6,
"unknown": 7
}
def h(e, t):
"""
填充二进制字符串
Args:
e: 数值
t: 目标长度
Returns:
填充后的二进制字符串
"""
n = bin(e)[2:] # 转为二进制并去掉 '0b' 前缀
r = ""
o = len(n) + 1
while o <= t:
r += "0"
o += 1
return r + n
def f(e):
"""
压缩事件类型数组
Args:
e: 事件类型列表
Returns:
压缩后的二进制字符串
"""
t = []
n = len(e)
r = 0
# 游程编码（Run-Length Encoding）
while r < n:
o = e[r]
i = 0
while True:
if 16 <= i:
break
s = r + i + 1
if n <= s:
break
if e[s] != o:
break
i += 1
r = r + 1 + i
a = p[o]
if i != 0:
t.append(8 | a) # 设置重复标志位
t.append(i - 1)
else:
t.append(a)
# 编码长度信息
_ = h(32768 | n, 16)
c = ""
for l in range(len(t)):
c += h(t[l], 4)
return _ + c
def c(e, t):
"""
对数组每个元素应用函数
Args:
e: 输入数组
t: 转换函数
Returns:
转换后的数组
"""
n = []
for r in range(len(e)):
n.append(t(e[r]))
return n
def d(e, t):
"""
压缩数值数组（游程编码 + 变长编码）
Args:
e: 数值数组
t: 是否为坐标数据（需要过滤符号位）
Returns:
压缩后的二进制字符串
"""
# 第一步：限制数值范围到 [-32767, 32767]
def limit_value(val):
limit = 32767
return max(-limit, min(limit, val))
e = c(e, limit_value)
# 第二步：游程编码（Run-Length Encoding）
n = len(e)
r = 0
o = []
while r < n:
i = 1
s = e[r]
a = abs(s)
# 统计连续相同的值
while r + i < n and e[r + i] == s and a < 127 and i < 127:
i += 1
if i > 1:
# 重复值编码格式：
# 位15: 符号标志 (1=负数49152, 0=正数32768)
# 位14-7: 重复次数 (i)
# 位6-0: 绝对值 (a)
o.append((49152 if s < 0 else 32768) | (i << 7) | a)
else:
o.append(s)
r += i
e = o
# 第三步：变长编码
r = [] # 存储每个数字的十六进制位数
o = [] # 存储实际数值
for val in e:
# 计算需要多少个十六进制位（每位4 bit）
if val == 0:
bits = 1
else:
# 方法1：使用对数（与原JS一致）
bits = math.ceil(math.log(abs(val) + 1) / math.log(16))
# 方法2：直接计算（更快）
# bits = len(format(abs(val), 'x'))
bits = max(1, bits)
r.append(h(bits - 1, 2)) # 2位十六进制存储位数信息
o.append(h(abs(val), 4 * bits)) # 实际值
i = "".join(r) # 元数据
s = "".join(o) # 数据
# 第四步：符号位编码（仅用于坐标数据）
if t:
# 过滤掉：0值 和 已编码符号的压缩值（位15=1）
filtered = [x for x in e if x != 0 and (x >> 15) != 1]
n = "".join(["1" if x < 0 else "0" for x in filtered])
else:
n = ""
# 最终格式：[头部16位][元数据][数据][符号位]
# 头部：最高位置1 + 数组长度
return h(32768 | len(e), 16) + i + s + n
# 主函数：数据分离
t = [] # 事件类型
n = [] # 时间差
r = [] # X坐标
o = [] # Y坐标
for i in range(len(e)):
a = e[i]
length = len(a)
t.append(a[0])
n.append(a[1] if length == 2 else a[2])
if length == 3:
r.append(a[1][0])
o.append(a[1][1])
# 压缩各部分
c_str = f(t) + d(n, False) + d(r, True) + d(o, True)
# 填充到6的倍数
l = len(c_str)
if l % 6 != 0:
c_str += h(0, 6 - l % 6)
# Base64编码
def u(e):
"""Base64编码"""
t = ""
n = len(e) // 6
base64_chars = "()*,-./0123456789:?@ABCDEFGHIJKLMNOPQRSTUVWXYZ_abcdefghijklmnopqrstuvwxyz~"
for r in range(n):
# 每次取6位二进制
binary_str = e[6 * r: 6 * (r + 1)]
index = int(binary_str, 2)
t += base64_chars[index]
return t
return u(c_str)
class TrajectoryEncoder:
def __init__(self) -> None:
self.CHARSET = "()*,-./0123456789:?@ABCDEFGHIJKLMNOPQRSTUVWXYZ_abcdefghijklmnopqr"
self.BASE = len(self.CHARSET) # 64
self.DIRECTION_CHARS = "stuvwxyz~"
self.DIRECTION_PATTERNS = [
[1, 0], # s
[2, 0], # t
[1, -1], # u
[1, 1], # v
[0, 1], # w
[0, -1], # x
[3, 0], # y
[2, -1], # z
[2, 1] # ~
]
def encode_number(self, num):
"""编码单个数值为64进制"""
abs_num = abs(num)
high_index = abs_num // self.BASE
low_index = abs_num % self.BASE
result = ""
# 负数标记
if num < 0:
result += "!"
# 高位（当值>=64时）
if high_index > 0 and high_index < self.BASE:
result += "$"
result += self.CHARSET[high_index]
# 低位
result += self.CHARSET[low_index]
return result
def compress_trajectory(self, points):
"""压缩轨迹：计算相邻点差值"""
compressed = []
time_accumulator = 0
for i in range(len(points) - 1):
dx = points[i + 1][0] - points[i][0] # 不要用abs
dy = points[i + 1][1] - points[i][1] # 不要用abs
dt = abs(points[i + 1][2] - points[i][2]) # 时间可以用abs
# 跳过完全相同的点
if dx == 0 and dy == 0 and dt == 0:
continue
# 位置不变只累积时间
if dx == 0 and dy == 0:
time_accumulator += dt
else:
compressed.append([dx, dy, dt + time_accumulator])
time_accumulator = 0
# 处理剩余时间
if time_accumulator != 0:
compressed.append([0, 0, time_accumulator])
return compressed
def get_direction_code(self, dx, dy):
"""识别是否匹配方向模式"""
for i, pattern in enumerate(self.DIRECTION_PATTERNS):
if dx == pattern[0] and dy == pattern[1]:
return self.DIRECTION_CHARS[i]
return None
def encode(self, trajectory):
"""
编码轨迹
trajectory: [[x, y, timestamp], ...]
返回: 编码后的字符串
"""
compressed = self.compress_trajectory(trajectory)
x_encoded = []
y_encoded = []
t_encoded = []
for dx, dy, dt in compressed:
direction_code = self.get_direction_code(dx, dy)
if direction_code:
# 匹配到方向模式，只记录y
y_encoded.append(direction_code)
else:
# 不匹配，完整编码x和y
x_encoded.append(self.encode_number(dx))
y_encoded.append(self.encode_number(dy))
# 时间总是编码
t_encoded.append(self.encode_number(dt))
# 拼接：x坐标 !! y坐标 !! 时间戳
return "".join(x_encoded) + "!!" + "".join(y_encoded) + "!!" + "".join(t_encoded)
def encrypt_string(self,e, t, n):
"""
JS加密函数的Python实现
参数:
e: 原始字符串
t: 加密参数数组
n: 十六进制字符串
"""
if not t or not n:
return e
o = 0 # 偏移量
i = e # 结果字符串
s = t[0] # 12
a = t[2] # 98
_ = t[4] # 43
# 每次读取2个字符（十六进制）
while o < len(n):
r = n[o:o + 2] # 取2个字符
if len(r) < 2:
break
o += 2
# 解析十六进制
c = int(r, 16)
# 转换为字符
l = chr(c)
# 计算插入位置: (s * c^2 + a * c + _) % len(e)
u = (s * c * c + a * c + _) % len(e)
# 在位置u插入字符
i = i[:u] + l + i[u:]
return i
def H(t:int, e:str) -> str:
# 解析后缀
n = e[-2:]
r = []
for char in n:
o = ord(char)
r.append(o - 87 if o > 57 else o - 48)
n = 36 * r[0] + r[1]
# 计算目标值
a = round(t) + n
# 构建字符池
_ = [[], [], [], [], []]
c = {}
u = 0
for char in e[:-2]:
if char not in c:
c[char] = 1
_[u].append(char)
u = (u + 1) % 5
# 生成结果
f = a
d = 4
p = ""
g = [1, 2, 5, 10, 50]
while f > 0:
if f >= g[d]:
h = int(random.random() * len(_[d]))
p += _[d][h]
f -= g[d]
else:
_.pop(d)
g.pop(d)
d -= 1
return p
core/geetest/v3_slide/__init__.py
+5
View File
@@ -0,0 +1,5 @@
"""Geetest v3 滑块验证码求解器"""
from .solver import run_solver
__all__ = ["run_solver"]
core/geetest/v3_slide/solver.py
+159
View File
@@ -0,0 +1,159 @@
import time
import random
import json
from core.geetest.common.trajectory import generate_realistic_trajectory, process_mouse_trajectory, compress_trajectory, TrajectoryEncoder, \
H
from core.geetest.common.crypto import four_random_chart, RSA_jiami_r, AES_O, geetest_base64_encode, encrypt_string, simple_md5
from core.geetest.common.imaging import download_picture
from core.geetest.common.network import get_challenge_gt, get_js_address, get_c_s, req_slide, get_picture, req_end
from core.geetest.common.performance import generate_fake_performance_timing, get_slide_track
def _generate_seed() -> str:
return four_random_chart() + four_random_chart() + four_random_chart() + four_random_chart()
def get_w1(gt:str, challenge:str, str_16:str) -> str:
r = RSA_jiami_r(str_16)
plaintext = '{"gt":"' + gt + '","challenge":"' + challenge + '","offline":false,"new_captcha":true,"product":"float","width":"300px","https":true,"api_server":"apiv6.geetest.com","protocol":"https://","type":"fullpage","static_servers":["static.geetest.com/","static.geevisit.com/"],"voice":"/static/js/voice.1.2.6.js","click":"/static/js/click.3.1.2.js","beeline":"/static/js/beeline.1.0.1.js","fullpage":"/static/js/fullpage.9.2.0-guwyxh.js","slide":"/static/js/slide.7.9.3.js","geetest":"/static/js/geetest.6.0.9.js","aspect_radio":{"slide":103,"click":128,"voice":128,"beeline":50},"cc":16,"ww":true,"i":"-1!!-1!!-1!!-1!!-1!!-1!!-1!!-1!!-1!!-1!!-1!!-1!!-1!!-1!!-1!!-1!!-1!!-1!!-1!!-1!!-1!!-1!!-1!!-1!!-1!!-1!!-1!!-1!!-1!!-1!!-1!!-1!!-1!!-1!!-1!!-1!!-1!!-1!!-1!!-1!!-1!!-1!!-1!!-1!!-1!!-1!!-1!!-1!!-1!!-1!!-1!!-1!!-1!!-1!!-1!!-1!!-1!!-1!!-1!!-1!!-1!!-1!!-1!!-1!!-1!!-1!!-1!!-1!!-1!!-1!!-1!!-1!!-1!!-1"}'
o = AES_O(plaintext, str_16)
i = geetest_base64_encode(o)
return i['res'] + i['end'] + r
def get_w2(gt:str, challenge:str, c:list[int], s:str,str_16:str) -> str:
# 伪造浏览器性能数据
fake_timing = generate_fake_performance_timing()
# 映射
web_load_time = {
"a": fake_timing["navigationStart"],
"b": fake_timing["unloadEventStart"],
"c": fake_timing["unloadEventEnd"],
"d": fake_timing["redirectStart"],
"e": fake_timing["redirectEnd"],
"f": fake_timing["fetchStart"],
"g": fake_timing["domainLookupStart"],
"h": fake_timing["domainLookupEnd"],
"i": fake_timing["connectStart"],
"j": fake_timing["connectEnd"],
"k": fake_timing["secureConnectionStart"],
"l": fake_timing["requestStart"],
"m": fake_timing["responseStart"],
"n": fake_timing["responseEnd"],
"o": fake_timing["domLoading"],
"p": fake_timing["domInteractive"],
"q": fake_timing["domContentLoadedEventStart"],
"r": fake_timing["domContentLoadedEventEnd"],
"s": fake_timing["domComplete"],
"t": fake_timing["loadEventStart"],
"u": fake_timing["loadEventEnd"]
}
first_time = int(round(time.time() * 1000)) # 伪造脚本开始运行时间
guiji_yuanshu_shuzu = generate_realistic_trajectory(
start_x=random.randint(400, 600), # 起始位置随机
start_y=random.randint(400, 500),
end_x=853,
end_y=288,
start_time=first_time
)
trajectory = process_mouse_trajectory(guiji_yuanshu_shuzu)["data"]
compressed = compress_trajectory(trajectory)
tt = encrypt_string(compressed, c, s)
passtime = str(int(round(time.time() * 1000)) - first_time)
rp = simple_md5(gt + challenge + passtime)
plaintext = '{"lang":"zh-cn","type":"fullpage","tt":"'+tt+'","light":"DIV_0","s":"c7c3e21112fe4f741921cb3e4ff9f7cb","h":"321f9af1e098233dbd03f250fd2b5e21","hh":"39bd9cad9e425c3a8f51610fd506e3b3","hi":"09eb21b3ae9542a9bc1e8b63b3d9a467","vip_order":-1,"ct":-1,"ep":{"v":"9.2.0-guwyxh","te":false,"$_BBn":true,"ven":"Google Inc. (AMD)","ren":"ANGLE (AMD, AMD Radeon RX 6750 GRE 12GB (0x000073DF) Direct3D11 vs_5_0 ps_5_0, D3D11)","fp":'+json.dumps(guiji_yuanshu_shuzu[0], separators=(',', ':'))+',"lp":'+json.dumps(guiji_yuanshu_shuzu[-1], separators=(',', ':'))+',"em":{"ph":0,"cp":0,"ek":"11","wd":1,"nt":0,"si":0,"sc":0},"tm":'+json.dumps(web_load_time, separators=(',', ':'))+',"dnf":"dnf","by":0},"passtime":'+passtime+',"rp":"'+rp+'","captcha_token":"112439067","tsfq":"xovrayel"}'
result = geetest_base64_encode(AES_O(plaintext, str_16))
return result['res']+result['end']+result['end']
def get_w3(str_16:str, challenge:str, hkjl:int, c:list[int], s:str, gt:str) -> str:
encoder = TrajectoryEncoder()
u = RSA_jiami_r(str_16)
# 伪造浏览器性能数据
fake_timing = generate_fake_performance_timing()
# 映射
web_load_time = {
"a": fake_timing["navigationStart"],
"b": fake_timing["unloadEventStart"],
"c": fake_timing["unloadEventEnd"],
"d": fake_timing["redirectStart"],
"e": fake_timing["redirectEnd"],
"f": fake_timing["fetchStart"],
"g": fake_timing["domainLookupStart"],
"h": fake_timing["domainLookupEnd"],
"i": fake_timing["connectStart"],
"j": fake_timing["connectEnd"],
"k": fake_timing["secureConnectionStart"],
"l": fake_timing["requestStart"],
"m": fake_timing["responseStart"],
"n": fake_timing["responseEnd"],
"o": fake_timing["domLoading"],
"p": fake_timing["domInteractive"],
"q": fake_timing["domContentLoadedEventStart"],
"r": fake_timing["domContentLoadedEventEnd"],
"s": fake_timing["domComplete"],
"t": fake_timing["loadEventStart"],
"u": fake_timing["loadEventEnd"]
}
trajectory = get_slide_track(hkjl)[0]
print(trajectory)
# trajectory = generator.generate(target_x)
userresponse = H(trajectory[-1][0], challenge)
result1 = encoder.encode(trajectory)
aa = encoder.encrypt_string(result1, c, s)
passtime = str(trajectory[-1][2])
rp = simple_md5(gt + challenge[:32] + passtime)
plaintext = '{"lang":"zh-cn","userresponse":"'+userresponse+'","passtime":'+passtime+',"imgload":50,"aa":"'+aa+'","ep":{"v":"7.9.3","$_BIT":false,"me":true,"tm":'+json.dumps(web_load_time, separators=(',', ':'))+',"td":-1},"h9s9":"1816378497","rp":"'+rp+'"}'
h = geetest_base64_encode(AES_O(plaintext, str_16))['res']
return h+u
def run_solver() -> None:
# 16位字符串
str_16 = _generate_seed()
# 访问第一个链接获取gt challeneg
gt, challenge = get_challenge_gt()
# 访问第二个链接获取js文件地址（后续需要用到）
get_js_address(gt)
# 获取第一个w值
w1 = get_w1(gt, challenge, str_16)
# 访问第三个链接获取数组c与字符串s
c, s = get_c_s(gt, challenge, w1)
# 获取第二个w值
w2 = get_w2(gt, challenge, c, s, str_16)
# 访问第四个链接确认进入滑动
req_slide(gt, challenge, w2)
# 访问第五个链接获取最后一个w值所需要的加密参数
bg, fullbg, c, s, slice, challenge = get_picture(gt, challenge)
# 识别缺口位置得到滑块距离
hkjl = download_picture(bg, fullbg, slice)
# 获取第三个w值
w3 = get_w3(str_16,challenge,hkjl,c,s,gt)
# 最后的验证
message = req_end(gt, challenge, w3)
core/models.py
+27
View File
@@ -0,0 +1,27 @@
"""核心数据模型"""
from dataclasses import dataclass
@dataclass
class Account:
"""账号配置"""
username: str
password: str
email: str
email_password: str
email_imap_server: str
email_imap_port: int = 993
@dataclass
class ProxyConfig:
"""代理配置"""
enabled: bool = False
api_url: str = ""
http: str = ""
https: str = ""
# 白名单配置
whitelist_enabled: bool = False
whitelist_uid: str = ""
whitelist_ukey: str = ""