261 lines
8.4 KiB
Python
261 lines
8.4 KiB
Python
"""
|
||
小红书验证码登录流程测试脚本
|
||
测试完整的验证码发送和登录流程
|
||
"""
|
||
import asyncio
|
||
import sys
|
||
from xhs_login import XHSLoginService
|
||
|
||
|
||
async def test_send_verification_code(phone: str, proxy_index: int = 0):
|
||
"""
|
||
测试发送验证码流程
|
||
|
||
Args:
|
||
phone: 手机号
|
||
proxy_index: 代理索引 (0 或 1)
|
||
"""
|
||
print(f"\n{'='*60}")
|
||
print(f"📱 测试发送验证码流程")
|
||
print(f"{'='*60}")
|
||
|
||
# 从代理配置获取代理信息
|
||
from damai_proxy_config import get_proxy_config
|
||
proxy_config = get_proxy_config(proxy_index)
|
||
proxy_server = proxy_config['server'].replace('http://', '')
|
||
proxy_url = f"http://{proxy_config['username']}:{proxy_config['password']}@{proxy_server}"
|
||
|
||
print(f"✅ 使用代理: 代理{proxy_index + 1}")
|
||
print(f" 代理服务器: {proxy_config['server']}")
|
||
print(f" 手机号: {phone}")
|
||
|
||
# 创建登录服务
|
||
login_service = XHSLoginService()
|
||
|
||
try:
|
||
# 初始化浏览器（使用代理）
|
||
user_agent = 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36'
|
||
await login_service.init_browser(proxy=proxy_url, user_agent=user_agent)
|
||
print("✅ 浏览器初始化成功（已启用代理）")
|
||
|
||
# 发送验证码
|
||
print(f"\n📤 正在发送验证码到 {phone}...")
|
||
result = await login_service.send_verification_code(phone)
|
||
|
||
if result.get('success'):
|
||
print(f"✅ 验证码发送成功！")
|
||
print(f" 消息: {result.get('message')}")
|
||
return login_service # 返回服务实例供后续登录使用
|
||
else:
|
||
print(f"❌ 验证码发送失败: {result.get('error')}")
|
||
return None
|
||
|
||
except Exception as e:
|
||
print(f"❌ 发送验证码过程异常: {str(e)}")
|
||
import traceback
|
||
traceback.print_exc()
|
||
return None
|
||
|
||
|
||
async def test_login_with_code(login_service: XHSLoginService, phone: str, code: str):
|
||
"""
|
||
测试使用验证码登录
|
||
|
||
Args:
|
||
login_service: XHSLoginService实例
|
||
phone: 手机号
|
||
code: 验证码
|
||
"""
|
||
print(f"\n{'='*60}")
|
||
print(f"🔑 测试使用验证码登录")
|
||
print(f"{'='*60}")
|
||
|
||
print(f" 手机号: {phone}")
|
||
print(f" 验证码: {code}")
|
||
|
||
try:
|
||
# 执行登录
|
||
result = await login_service.login(phone, code)
|
||
|
||
if result.get('success'):
|
||
print("✅ 登录成功！")
|
||
|
||
# 显示获取到的Cookies信息
|
||
cookies = result.get('cookies', {})
|
||
print(f" 获取到 {len(cookies)} 个Cookie")
|
||
|
||
# 保存完整Cookies到文件
|
||
cookies_full = result.get('cookies_full', [])
|
||
if cookies_full:
|
||
import json
|
||
with open('cookies.json', 'w', encoding='utf-8') as f:
|
||
json.dump(cookies_full, f, ensure_ascii=False, indent=2)
|
||
print(" ✅ 已保存完整Cookies到 cookies.json")
|
||
|
||
# 显示用户信息
|
||
user_info = result.get('user_info', {})
|
||
if user_info:
|
||
print(f" 用户信息: {list(user_info.keys())}")
|
||
|
||
return result
|
||
else:
|
||
print(f"❌ 登录失败: {result.get('error')}")
|
||
return result
|
||
|
||
except Exception as e:
|
||
print(f"❌ 登录过程异常: {str(e)}")
|
||
import traceback
|
||
traceback.print_exc()
|
||
return {"success": False, "error": str(e)}
|
||
|
||
|
||
async def test_complete_login_flow(phone: str, code: str = None, proxy_index: int = 0):
|
||
"""
|
||
测试完整的登录流程
|
||
|
||
Args:
|
||
phone: 手机号
|
||
code: 验证码（如果为None，则只测试发送验证码）
|
||
proxy_index: 代理索引
|
||
"""
|
||
print("="*60)
|
||
print("🔄 测试完整登录流程")
|
||
print("="*60)
|
||
|
||
# 步骤1: 发送验证码
|
||
print("\n📋 步骤1: 发送验证码")
|
||
login_service = await test_send_verification_code(phone, proxy_index)
|
||
|
||
if not login_service:
|
||
print("❌ 发送验证码失败，终止流程")
|
||
return
|
||
|
||
# 如果提供了验证码，则执行登录
|
||
if code:
|
||
print("\n📋 步骤2: 使用验证码登录")
|
||
result = await test_login_with_code(login_service, phone, code)
|
||
|
||
if result.get('success'):
|
||
print("\n🎉 完整登录流程成功！")
|
||
else:
|
||
print(f"\n❌ 完整登录流程失败: {result.get('error')}")
|
||
else:
|
||
print("\n⚠️ 提供了验证码参数才可完成登录步骤")
|
||
print(" 请在手机上查看验证码，然后调用登录方法")
|
||
|
||
# 清理资源
|
||
await login_service.close_browser()
|
||
|
||
|
||
async def test_multiple_proxies_login(phone: str, proxy_indices: list = [0, 1]):
|
||
"""
|
||
测试使用多个代理进行登录
|
||
|
||
Args:
|
||
phone: 手机号
|
||
proxy_indices: 代理索引列表
|
||
"""
|
||
print("="*60)
|
||
print("🔄 测试多代理登录")
|
||
print("="*60)
|
||
|
||
for i, proxy_idx in enumerate(proxy_indices):
|
||
print(f"\n🧪 测试代理 {proxy_idx + 1} (第 {i+1} 次尝试)")
|
||
|
||
# 由于验证码只能发送一次，这里只测试发送验证码
|
||
login_service = await test_send_verification_code(phone, proxy_idx)
|
||
|
||
if login_service:
|
||
print(f" ✅ 代理 {proxy_idx + 1} 发送验证码成功")
|
||
await login_service.close_browser()
|
||
else:
|
||
print(f" ❌ 代理 {proxy_idx + 1} 发送验证码失败")
|
||
|
||
# 在测试之间添加延迟
|
||
if i < len(proxy_indices) - 1:
|
||
print(" ⏳ 等待3秒后测试下一个代理...")
|
||
await asyncio.sleep(3)
|
||
|
||
|
||
def show_usage_examples():
|
||
"""显示使用示例"""
|
||
print("="*60)
|
||
print("💡 使用示例")
|
||
print("="*60)
|
||
|
||
print("\n1️⃣ 仅发送验证码:")
|
||
print(" # 发送验证码到手机号，使用代理1")
|
||
print(" await test_send_verification_code('13800138000', proxy_index=0)")
|
||
|
||
print("\n2️⃣ 完整登录流程:")
|
||
print(" # 完整流程：发送验证码 + 登录")
|
||
print(" await test_complete_login_flow('13800138000', '123456', proxy_index=0)")
|
||
|
||
print("\n3️⃣ 多代理测试:")
|
||
print(" # 测试多个代理")
|
||
print(" await test_multiple_proxies_login('13800138000', [0, 1])")
|
||
|
||
|
||
async def main():
|
||
"""主函数"""
|
||
show_usage_examples()
|
||
|
||
print(f"\n{'='*60}")
|
||
print("🎯 选择测试模式")
|
||
print("="*60)
|
||
|
||
print("\n1. 发送验证码测试")
|
||
print("2. 完整登录流程测试")
|
||
print("3. 多代理测试")
|
||
|
||
try:
|
||
choice = input("\n请选择测试模式 (1-3, 默认为1): ").strip()
|
||
|
||
if choice not in ['1', '2', '3']:
|
||
choice = '1'
|
||
|
||
phone = input("请输入手机号: ").strip()
|
||
|
||
if not phone:
|
||
print("❌ 手机号不能为空")
|
||
return
|
||
|
||
if choice == '1':
|
||
proxy_choice = input("请选择代理 (0 或 1, 默认为0): ").strip()
|
||
if proxy_choice not in ['0', '1']:
|
||
proxy_choice = '0'
|
||
proxy_idx = int(proxy_choice)
|
||
|
||
await test_send_verification_code(phone, proxy_idx)
|
||
|
||
elif choice == '2':
|
||
code = input("请输入验证码 (留空则只测试发送): ").strip()
|
||
proxy_choice = input("请选择代理 (0 或 1, 默认为0): ").strip()
|
||
if proxy_choice not in ['0', '1']:
|
||
proxy_choice = '0'
|
||
proxy_idx = int(proxy_choice)
|
||
|
||
await test_complete_login_flow(phone, code if code else None, proxy_idx)
|
||
|
||
elif choice == '3':
|
||
await test_multiple_proxies_login(phone)
|
||
|
||
print(f"\n{'='*60}")
|
||
print("✅ 测试完成！")
|
||
print("="*60)
|
||
|
||
except KeyboardInterrupt:
|
||
print("\n\n⚠️ 测试被用户中断")
|
||
except Exception as e:
|
||
print(f"\n❌ 测试过程中出现错误: {str(e)}")
|
||
import traceback
|
||
traceback.print_exc()
|
||
|
||
|
||
if __name__ == "__main__":
|
||
# Windows环境下设置事件循环策略
|
||
if sys.platform == 'win32':
|
||
asyncio.set_event_loop_policy(asyncio.WindowsProactorEventLoopPolicy())
|
||
|
||
# 运行测试
|
||
asyncio.run(main()) |