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

[patch] 处理PDD登录时get_auth_token方法报出的response返回异常未处理报错问题

Browse Source
This commit is contained in:
haosicheng
2025-10-18 16:12:57 +08:00
parent ceb5d83d75
commit fd6dbfc8da
2 changed files with 226 additions and 29 deletions
Download Patch File Download Diff File Expand all files Collapse all files

249
Utils/Pdd/PddUtils.py
View File

@@ -2307,33 +2307,173 @@ class ChatPdd:
return None
def get_auth_token(self):
url = "https://mms.pinduoduo.com/janus/api/subSystem/getAuthToken"
data = {"subSystemId": 17,
"anti_content": "0asAfx5E-wCElqJNXaKt_UKccG7YNycjZoPYgO1YTmZoApN7QJU5XT_JuYdPZ4uvLPQFgIcyXOKqm8xGrPjy0lxn0gaXr9ac0TynYEJnDwdj-WEJnwK3G4kc3M0TiPgtB11eBeZkB1hkBxUHFOtjfqz-eAkgc2aa4sZuIJwHOptYX14VK1NJSqIYfqmw6jpQTXs574CfWMFxT1RPTIB2MKBjC199pXpkbdtXxnn2mA4C1YdNnTUrNa_Wvn5mYj5XadnDbNT7xNuVeYXrsTsiipUr6xn2AAXKoYmv6j0PL92PtCTMfZzzfTfjutCgvBTzxFw-2LeeRIkFBATU1Npg1ScUFRv-1Ukrs3AklVAbBiEbBJhzcf2cXKfNH50X9QUFCd_Y1FhLW1BBuI-KEUFYK3lZTB3g_BlLDk8ti-JXmbalXzWAb6V2hX0fZE9n2L0GITFInNS"}
response = requests.post(url, cookies=self.cookie, headers=self.headers, json=data).json()
self.auth_token = response["result"]["authToken"]
"""获取拼多多认证token（支持Cookie过期检测）
Returns:
str: "cookie_expired" 表示Cookie已过期，需要通知后端
str: "success" 表示获取成功
str: "error" 表示其他错误
"""
try:
url = "https://mms.pinduoduo.com/janus/api/subSystem/getAuthToken"
data = {"subSystemId": 17,
"anti_content": "0asAfx5E-wCElqJNXaKt_UKccG7YNycjZoPYgO1YTmZoApN7QJU5XT_JuYdPZ4uvLPQFgIcyXOKqm8xGrPjy0lxn0gaXr9ac0TynYEJnDwdj-WEJnwK3G4kc3M0TiPgtB11eBeZkB1hkBxUHFOtjfqz-eAkgc2aa4sZuIJwHOptYX14VK1NJSqIYfqmw6jpQTXs574CfWMFxT1RPTIB2MKBjC199pXpkbdtXxnn2mA4C1YdNnTUrNa_Wvn5mYj5XadnDbNT7xNuVeYXrsTsiipUr6xn2AAXKoYmv6j0PL92PtCTMfZzzfTfjutCgvBTzxFw-2LeeRIkFBATU1Npg1ScUFRv-1Ukrs3AklVAbBiEbBJhzcf2cXKfNH50X9QUFCd_Y1FhLW1BBuI-KEUFYK3lZTB3g_BlLDk8ti-JXmbalXzWAb6V2hX0fZE9n2L0GITFInNS"}
response = requests.post(url, cookies=self.cookie, headers=self.headers, json=data, timeout=10).json()
# 调试日志
self._log(f"[PDD] get_auth_token API响应: {response}", "DEBUG")
# 🔥 检查拼多多API特有的错误格式
if isinstance(response, dict):
# 检查是否有错误码（拼多多的错误响应格式）
if "error_code" in response or "error_msg" in response:
error_code = response.get("error_code", "未知")
error_msg = response.get("error_msg", "未知错误")
self._log(f"⚠️ [PDD] get_auth_token API返回错误 [错误码: {error_code}]: {error_msg}", "WARNING")
# 🎯 特殊处理：Cookie过期（43001)
if error_code == 43001 or "过期" in str(error_msg):
self._log(f"🔄 [PDD] 检测到Cookie已过期，将通知后端重新获取", "INFO")
return "cookie_expired" # 返回特殊标识
else:
self._log(f"❌ [PDD] 拼多多API错误: [{error_code}] {error_msg}", "ERROR")
return "error"
# 检查正常响应的result字段
if "result" not in response:
self._log(f"❌ [PDD] get_auth_token 响应缺少result字段: {response}", "ERROR")
return "error"
result = response["result"]
if not isinstance(result, dict) or "authToken" not in result:
self._log(f"❌ [PDD] get_auth_token result缺少authToken字段: {result}", "ERROR")
return "error"
self.auth_token = result["authToken"]
self._log(f"✅ [PDD] 成功获取auth_token: {self.auth_token[:20] if len(self.auth_token) > 20 else self.auth_token}...", "SUCCESS")
return "success"
except Exception as e:
self._log(f"❌ [PDD] get_auth_token 异常: {e}", "ERROR")
return "error"
def get_mall_id(self):
url = "https://mms.pinduoduo.com/chats/userinfo/realtime?get_response=true"
data = {"get_response": "true"}
response = requests.get(url, cookies=self.cookie, headers=self.headers, params=data).json()
self.mall_id = response["mall_id"]
self.user_id = response["id"]
"""获取拼多多店铺ID和用户ID（支持Cookie过期检测）
Returns:
str: "cookie_expired" 表示Cookie已过期
str: "success" 表示获取成功
str: "error" 表示其他错误
"""
try:
url = "https://mms.pinduoduo.com/chats/userinfo/realtime?get_response=true"
data = {"get_response": "true"}
response = requests.get(url, cookies=self.cookie, headers=self.headers, params=data, timeout=10).json()
# 调试日志
self._log(f"[PDD] get_mall_id API响应: {response}", "DEBUG")
# 检查错误格式
if isinstance(response, dict):
if "error_code" in response or "error_msg" in response:
error_code = response.get("error_code", "未知")
error_msg = response.get("error_msg", "未知错误")
self._log(f"⚠️ [PDD] get_mall_id API返回错误 [错误码: {error_code}]: {error_msg}", "WARNING")
if error_code == 43001 or "过期" in str(error_msg):
self._log(f"🔄 [PDD] 检测到Cookie已过期", "INFO")
return "cookie_expired"
else:
self._log(f"❌ [PDD] API错误: [{error_code}] {error_msg}", "ERROR")
return "error"
# 检查必需字段
if "mall_id" not in response or "id" not in response:
self._log(f"❌ [PDD] get_mall_id 响应缺少必需字段: {response}", "ERROR")
return "error"
self.mall_id = response["mall_id"]
self.user_id = response["id"]
self._log(f"✅ [PDD] 成功获取mall_id: {self.mall_id}, user_id: {self.user_id}", "SUCCESS")
return "success"
except Exception as e:
self._log(f"❌ [PDD] get_mall_id 异常: {e}", "ERROR")
return "error"
def get_assign_cslist(self):
url = "https://mms.pinduoduo.com/latitude/assign/getAssignCsList"
data = {
"wechatCheck": True
}
data = json.dumps(data, separators=(',', ':'))
response = requests.post(url, headers=self.headers, cookies=self.cookie, data=data).json()
cslist = response["result"]["csList"]
keys = cslist.keys()
for key in keys:
username = cslist[key].get("username")
if username == self.csname:
self.csid = key
break
"""获取拼多多客服列表（支持Cookie过期检测）
Returns:
str: "cookie_expired" 表示Cookie已过期
str: "success" 表示获取成功
str: "error" 表示其他错误
"""
try:
url = "https://mms.pinduoduo.com/latitude/assign/getAssignCsList"
data = {
"wechatCheck": True
}
data_str = json.dumps(data, separators=(',', ':'))
response = requests.post(url, headers=self.headers, cookies=self.cookie, data=data_str, timeout=10).json()
# 调试日志
self._log(f"[PDD] get_assign_cslist API响应: {response}", "DEBUG")
# 检查错误格式
if isinstance(response, dict):
if "error_code" in response or "error_msg" in response:
error_code = response.get("error_code", "未知")
error_msg = response.get("error_msg", "未知错误")
self._log(f"⚠️ [PDD] get_assign_cslist API返回错误 [错误码: {error_code}]: {error_msg}", "WARNING")
if error_code == 43001 or "过期" in str(error_msg):
self._log(f"🔄 [PDD] 检测到Cookie已过期", "INFO")
return "cookie_expired"
else:
self._log(f"❌ [PDD] API错误: [{error_code}] {error_msg}", "ERROR")
return "error"
# 检查result和csList字段
if "result" not in response:
self._log(f"❌ [PDD] get_assign_cslist 响应缺少result字段: {response}", "ERROR")
return "error"
result = response["result"]
if not isinstance(result, dict) or "csList" not in result:
self._log(f"❌ [PDD] result缺少csList字段: {result}", "ERROR")
return "error"
cslist = result["csList"]
if not isinstance(cslist, dict):
self._log(f"❌ [PDD] csList格式错误: {type(cslist)}", "ERROR")
return "error"
# 查找匹配的客服ID
found = False
for key, cs_info in cslist.items():
username = cs_info.get("username", "")
if username == self.csname:
self.csid = key
found = True
self._log(f"✅ [PDD] 找到匹配客服: {username} (ID: {key})", "SUCCESS")
break
if not found:
self._log(f"⚠️ [PDD] 未找到匹配的客服名称: {self.csname}", "WARNING")
self._log(f"✅ [PDD] 成功获取客服列表，共 {len(cslist)} 个客服", "SUCCESS")
return "success"
except Exception as e:
self._log(f"❌ [PDD] get_assign_cslist 异常: {e}", "ERROR")
return "error"
def tab_mall(self, uid):
"""执行转接操作"""
@@ -3245,10 +3385,33 @@ class PddListenerForGUI:
self.running = True
self._log("🎉 [PDD] 开始监听平台消息", "SUCCESS")
# 获取认证信息
self.pdd_bot.get_auth_token()
self.pdd_bot.get_mall_id()
self.pdd_bot.get_assign_cslist()
# 🔥 获取认证信息（支持Cookie过期自动恢复）
auth_result = self.pdd_bot.get_auth_token()
if auth_result == "cookie_expired":
self._log("🔄 [PDD] Cookie已过期，通知后端重新获取", "WARNING")
await self._notify_backend_cookie_expired(store_id)
return "cookie_expired" # 返回特殊标识，防止重复发送消息
elif auth_result == "error":
self._log("❌ [PDD] 获取auth_token失败", "ERROR")
return False
mall_result = self.pdd_bot.get_mall_id()
if mall_result == "cookie_expired":
self._log("🔄 [PDD] Cookie已过期，通知后端重新获取", "WARNING")
await self._notify_backend_cookie_expired(store_id)
return "cookie_expired"
elif mall_result == "error":
self._log("❌ [PDD] 获取mall_id失败", "ERROR")
return False
cslist_result = self.pdd_bot.get_assign_cslist()
if cslist_result == "cookie_expired":
self._log("🔄 [PDD] Cookie已过期，通知后端重新获取", "WARNING")
await self._notify_backend_cookie_expired(store_id)
return "cookie_expired"
elif cslist_result == "error":
self._log("❌ [PDD] 获取客服列表失败", "ERROR")
return False
# 启动监听
store = {'id': store_id}
@@ -3265,6 +3428,38 @@ class PddListenerForGUI:
self._log(f"错误详情: {traceback.format_exc()}", "DEBUG")
return False
async def _notify_backend_cookie_expired(self, store_id: str):
"""向后端发送Cookie过期通知
Args:
store_id: 店铺ID
"""
try:
from WebSocket.backend_singleton import get_websocket_manager
ws_manager = get_websocket_manager()
if ws_manager and ws_manager.backend_client:
# 构造通知消息
message = {
"type": "connect_message",
"store_id": store_id,
"status": False,
"content": "cookies失效，需要重新登录"
}
# 发送消息到后端
ws_manager.backend_client.send_message(message)
self._log(f"📤 [PDD] 已向后端发送Cookie过期通知: store_id={store_id}", "INFO")
self._log(f"📋 [PDD] 等待后端重新下发新的Cookie...", "INFO")
else:
self._log("⚠️ [PDD] WebSocket管理器未连接，无法发送通知", "WARNING")
except Exception as e:
self._log(f"❌ [PDD] 发送Cookie过期通知失败: {e}", "ERROR")
import traceback
self._log(f"详细错误: {traceback.format_exc()}", "DEBUG")
def stop_listening(self):
try:
self._log("🛑 开始停止拼多多监听...", "INFO")