haosicheng/shuidrop_gui
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

实现下发需要验证码结构给后端

Browse Source
This commit is contained in:
jjz
2025-09-16 11:17:30 +08:00
parent faef5d2316
commit 256e6c21a5
3 changed files with 141 additions and 94 deletions
Download Patch File Download Diff File Expand all files Collapse all files

219
Utils/Pdd/PddUtils.py
View File

@@ -15,10 +15,12 @@ import json
import random
import time
import traceback
import execjs
import websockets
import requests
import asyncio
import execjs
# execjs 已移除 - 不再需要本地生成anti_content
from datetime import datetime
from typing import Dict, Any, Optional, Callable
@@ -997,70 +999,7 @@ class ImgDistance:
)
class AutiContent:
"""anti_content生成类"""
def __init__(self):
self.ctx = None
try:
# 使用新的auti_content.js文件
current_dir = "static/js"
auti_js_path = os.path.join(current_dir, "auti_content.js")
if os.path.exists(auti_js_path):
# 暂时禁用JS文件加载，直接使用fallback避免编码问题
logger.info("检测到auti_content.js文件，但为避免编码问题使用fallback实现")
self.ctx = None
else:
logger.warning("auti_content.js文件不存在，使用fallback实现")
# fallback JS代码
jscode = """
function auti_content() {
// 生成类似格式的anti_content
var chars = 'ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789-_';
var result = '';
for (var i = 0; i < 40; i++) {
result += chars.charAt(Math.floor(Math.random() * chars.length));
}
return result;
}
function gzipCompress(data) {
return data; // 简单返回原数据
}
"""
self.ctx = execjs.compile(jscode)
except Exception as e:
logger.error(f"初始化AutiContent失败: {e}")
# 最终fallback
self.ctx = None
def get_auti_content(self):
"""获取anti_content"""
try:
if self.ctx:
# 尝试调用JS函数，如果遇到编码问题则降级处理
result = self.ctx.call("auti_content")
# 检查结果是否包含非ASCII字符
try:
result.encode('ascii')
logger.info(f"成功生成auti_content: {result[:20]}...")
return result
except UnicodeEncodeError:
logger.warning("JS生成的auti_content包含特殊字符，使用fallback")
raise Exception("包含特殊字符")
except Exception as e:
logger.error(f"调用auti_content函数失败: {e}")
# fallback处理
import uuid
import random
import string
# 生成类似格式的anti_content字符串
chars = string.ascii_letters + string.digits + '-_'
result = ''.join(random.choice(chars) for _ in range(40))
logger.warning(f"使用fallback生成auti_content: {result}")
return result
# AutiContent类已移除 - 后端会提供所有必要的anti_content
def gzip_compress(self, data):
"""压缩数据"""
@@ -1117,7 +1056,7 @@ class EncryptTool:
return t
class PddLogin(AutiContent):
class PddLogin:
"""拼多多登录核心类"""
def __init__(self, log_callback=None):
@@ -1624,11 +1563,11 @@ class PddLogin(AutiContent):
return captcha_collect
# 获取滑块加密参数key和iv
def vc_pre_ck_b(self):
def vc_pre_ck_b(self, backend_anti_content):
self.headers.pop("anti-content", None)
url = "https://apiv2.pinduoduo.net/api/phantom/vc_pre_ck_b"
payload = {
"anti_content": self.get_auti_content(),
"anti_content": backend_anti_content,
"verify_auth_token": self.verify_auth_token,
"sdk_type": 3,
"client_time": int(round(time.time() * 1000))
@@ -1638,10 +1577,10 @@ class PddLogin(AutiContent):
return response.json().get("salt")
# 获取滑块图片
def obtain_captcha(self):
def obtain_captcha(self, backend_anti_content):
url = "https://apiv2.pinduoduo.net/api/phantom/obtain_captcha"
payload = {
"anti_content": self.get_auti_content(),
"anti_content": backend_anti_content,
"verify_auth_token": self.verify_auth_token,
"captcha_collect": ""
}
@@ -1650,13 +1589,13 @@ class PddLogin(AutiContent):
return response.json().get("pictures")
# 滑块验证
def verify(self, captcha_collect, verify_code):
def verify(self, captcha_collect, verify_code, backend_anti_content):
url = "https://apiv2.pinduoduo.net/api/phantom/user_verify"
payload = {
"verify_code": verify_code,
"captcha_collect": captcha_collect,
"verify_auth_token": self.verify_auth_token,
"anti_content": self.get_auti_content()
"anti_content": backend_anti_content
}
response = requests.post(url, headers=self.headers, json=payload, cookies=self.cookies)
self.cookies.update(response.cookies.get_dict())
@@ -1665,15 +1604,19 @@ class PddLogin(AutiContent):
return result
# 发送验证码通知给后端，并获取验证码
def request_verification_code(self, username, store_id):
def request_verification_code(self, username, store_id, backend_anti_content):
"""向后端请求获取手机验证码"""
self._log(f"开始请求验证码，用户名: {username}, 店铺ID: {store_id}", "INFO")
self.headers["anti-content"] = self.get_auti_content()
# 使用后端下发的anti_content
anti_content = backend_anti_content
self._log(f"使用后端下发的anti_content", "DEBUG")
self.headers["anti-content"] = anti_content
url = "https://mms.pinduoduo.com/janus/api/user/getLoginVerificationCode"
payload = {
"username": username,
"crawlerInfo": self.get_auti_content()
"crawlerInfo": anti_content
}
response = requests.post(url, headers=self.headers, json=payload, cookies=self.cookies)
self._log(f"发送验证码请求结果: {response.text}")
@@ -1711,10 +1654,83 @@ class PddLogin(AutiContent):
import traceback
self._log(f"错误详情: {traceback.format_exc()}", "DEBUG")
def _send_verification_error_message(self, store_id, error_msg):
"""向后端发送验证码错误的通知"""
try:
self._log(f"开始发送验证码错误通知，店铺ID: {store_id}, 错误: {error_msg}", "INFO")
from WebSocket.backend_singleton import get_backend_client
backend = get_backend_client()
if backend:
message = {
"type": "connect_message",
"store_id": store_id,
"status": False,
"content": error_msg
}
self._log(f"准备发送验证码错误消息: {message}", "DEBUG")
backend.send_message(message)
self._log("✅ 成功向后端发送验证码错误通知", "SUCCESS")
else:
self._log("❌ 后端客户端为空，无法发送验证码错误通知", "ERROR")
except Exception as e:
self._log(f"❌ 发送验证码错误通知失败: {e}", "ERROR")
import traceback
self._log(f"错误详情: {traceback.format_exc()}", "DEBUG")
def _send_login_success_message(self, store_id):
"""向后端发送登录成功的通知"""
try:
self._log(f"开始发送登录成功通知，店铺ID: {store_id}", "INFO")
from WebSocket.backend_singleton import get_backend_client
backend = get_backend_client()
if backend:
message = {
"type": "connect_message",
"store_id": store_id,
"status": True # 登录成功
}
self._log(f"准备发送登录成功消息: {message}", "DEBUG")
backend.send_message(message)
self._log("✅ 成功向后端发送登录成功通知", "SUCCESS")
else:
self._log("❌ 获取后端客户端失败", "ERROR")
except Exception as e:
self._log(f"❌ 发送登录成功通知失败: {e}", "ERROR")
import traceback
self._log(f"错误详情: {traceback.format_exc()}", "DEBUG")
def _send_login_failure_message(self, store_id, error_msg):
"""向后端发送登录失败的通知"""
try:
self._log(f"开始发送登录失败通知，店铺ID: {store_id}, 错误: {error_msg}", "INFO")
from WebSocket.backend_singleton import get_backend_client
backend = get_backend_client()
if backend:
message = {
"type": "connect_message",
"store_id": store_id,
"status": False, # 登录失败
"content": error_msg # 官方返回的失败原因
}
self._log(f"准备发送登录失败消息: {message}", "DEBUG")
backend.send_message(message)
self._log("✅ 成功向后端发送登录失败通知", "SUCCESS")
else:
self._log("❌ 获取后端客户端失败", "ERROR")
except Exception as e:
self._log(f"❌ 发送登录失败通知失败: {e}", "ERROR")
import traceback
self._log(f"错误详情: {traceback.format_exc()}", "DEBUG")
def login_with_params(self, login_params, store_id=None, error_count=0, success_count=0):
"""使用后端下发的登录参数进行登录"""
self._log("🚀 [PddLogin] 开始使用参数登录", "INFO")
self._log(f"📋 [PddLogin] 登录参数: username={login_params.get('username', 'N/A')}, 包含code={bool(login_params.get('code'))}", "DEBUG")
# 检查验证码字段（兼容 code 和 verification_code）
verification_code = login_params.get("verification_code") or login_params.get("code", "")
self._log(f"📋 [PddLogin] 登录参数: username={login_params.get('username', 'N/A')}, 包含验证码={bool(verification_code)}", "DEBUG")
self.headers["anti-content"] = login_params.get("anti_content", "")
@@ -1724,8 +1740,8 @@ class PddLogin(AutiContent):
"username": login_params.get("username", ""),
"password": login_params.get("password", ""), # 后端已加密
"passwordEncrypt": True,
"verificationCode": "",
"mobileVerifyCode": login_params.get("code", ""), # 检查是否包含验证码
"verificationCode": verification_code,
"mobileVerifyCode": "", # 使用兼容的验证码字段
"sign": "",
"touchevent": {
"mobileInputEditStartTime": ts - random.randint(10000, 20000),
@@ -1779,6 +1795,11 @@ class PddLogin(AutiContent):
"crawlerInfo": login_params.get("anti_content", "")
}
# 添加详细的请求日志
self._log(f"🔍 [Debug] 登录请求URL: {self.login_api}", "DEBUG")
self._log(f"🔍 [Debug] 登录请求payload: {payload}", "DEBUG")
self._log(f"🔍 [Debug] 登录请求headers: {dict(self.headers)}", "DEBUG")
response = requests.post(self.login_api, headers=self.headers, json=payload, cookies=self.cookies)
self.cookies.update(response.cookies.get_dict())
self._log(f"登录响应状态码: {response.status_code}")
@@ -1823,29 +1844,43 @@ class PddLogin(AutiContent):
return False
elif "需要手机验证" in response.text:
# 检查是否已经包含验证码
if login_params.get("code"):
# 已有验证码但仍需要验证，可能验证码错误或过期
self._log("验证码可能错误或过期", "WARNING")
return False
# 检查是否已经包含验证码（兼容 code 和 verification_code）
if verification_code:
# 已有验证码但仍需要验证，验证码可能错误、过期或者其他问题
self._log(f"带验证码登录失败，验证码: {verification_code}", "WARNING")
# 检查响应中的具体错误信息
response_data = response.json()
error_msg = response_data.get('errorMsg', '验证码验证失败')
self._log(f"服务器返回错误: {error_msg}", "WARNING")
# 不要重新发送验证码请求，直接报告验证失败
self._send_verification_error_message(store_id, error_msg) # 直接使用官方错误信息
return "verification_code_error" # 返回特殊状态，避免重复发送消息
else:
# 第一次需要验证码，调用发送验证码方法
self._log("检测到需要手机验证码，正在调用发送验证码方法", "INFO")
username = login_params.get("username")
self._log(f"为用户 {username} 发送验证码", "INFO")
self.request_verification_code(username, store_id)
backend_anti_content = login_params.get("anti_content")
self._log(f"为用户 {username} 发送验证码，使用后端anti_content", "INFO")
# 传递后端下发的anti_content
self.request_verification_code(username, store_id, backend_anti_content)
return "need_verification_code"
else:
# 登录成功或其他情况
response_data = response.json()
if response_data.get("success"):
self._log("登录成功", "SUCCESS")
self._log("🎉 登录成功！", "SUCCESS")
# 发送登录成功通知给后端
self._send_login_success_message(store_id)
return self.cookies
else:
self._log(f"登录失败: {response_data.get('errorMsg', '未知错误')}", "ERROR")
return False
# 登录失败，发送失败通知给后端
error_msg = response_data.get('errorMsg', '未知错误')
self._log(f"登录失败: {error_msg}", "ERROR")
self._send_login_failure_message(store_id, error_msg)
return "login_failure" # 返回特殊状态，避免重复发送消息
# ===== 登录相关类集成结束 =====
@@ -2810,6 +2845,12 @@ class PddListenerForGUI:
if login_result == "need_verification_code":
self._log("⚠️ [PDD] 需要手机验证码，已通知后端，等待重新下发包含验证码的登录参数", "WARNING")
return "need_verification_code" # 返回特殊标识，避免被覆盖
elif login_result == "verification_code_error":
self._log("⚠️ [PDD] 验证码错误，已通知后端", "WARNING")
return "verification_code_error" # 返回特殊标识，避免重复发送消息
elif login_result == "login_failure":
self._log("⚠️ [PDD] 登录失败，已发送失败通知给后端", "WARNING")
return "login_failure" # 返回特殊标识，避免重复发送消息
elif not login_result:
self._log("❌ [PDD] 登录失败", "ERROR")
return False