feat: Web管理系统 + Docker支持
- 多账号管理（异步登录、状态轮询） - 购物车预售商品同步（倒计时/定时开售） - 定时抢购（自动刷新、SKU选择、重试机制） - 账号隔离调度（同账号顺序、跨账号并行） - Web面板（任务分组、实时倒计时、批量操作） - Dockerfile + docker-compose
This commit is contained in:
151
test_login.py
Normal file
151
test_login.py
Normal file
@@ -0,0 +1,151 @@
|
||||
"""
|
||||
微店登录流程测试脚本
|
||||
用法: python test_login.py
|
||||
"""
|
||||
import asyncio
|
||||
import json
|
||||
from playwright.async_api import async_playwright
|
||||
from utils.stealth import stealth_async
|
||||
|
||||
LOGIN_URL = "https://sso.weidian.com/login/index.php"
|
||||
TEST_PHONE = "15556986013"
|
||||
TEST_PASSWORD = "Precious171259"
|
||||
|
||||
|
||||
async def test_login():
|
||||
login_result = {"success": False, "msg": "未收到响应", "cookies": []}
|
||||
|
||||
p = await async_playwright().start()
|
||||
browser = await p.chromium.launch(headless=False, args=["--disable-gpu"])
|
||||
device = p.devices["iPhone 13"]
|
||||
context = await browser.new_context(**device)
|
||||
page = await context.new_page()
|
||||
await stealth_async(page)
|
||||
|
||||
# 监听登录 API 响应
|
||||
async def on_response(response):
|
||||
url = response.url
|
||||
if "user/login" in url or "user/bindcheck" in url:
|
||||
try:
|
||||
# 打印请求参数
|
||||
req = response.request
|
||||
print(f"\n[网络] {req.method} {url}")
|
||||
print(f"[网络] 请求头: {dict(req.headers)}")
|
||||
post_data = req.post_data
|
||||
print(f"[网络] 请求体: {post_data}")
|
||||
# 打印响应
|
||||
data = await response.json()
|
||||
status = data.get("status", {})
|
||||
print(f"[网络] 响应 status_code={status.get('status_code')}")
|
||||
if status.get("status_code") == 0 and "user/login" in url:
|
||||
login_result["success"] = True
|
||||
login_result["msg"] = "登录成功"
|
||||
login_result["cookies"] = data.get("result", {}).get("cookie", [])
|
||||
elif "user/login" in url:
|
||||
login_result["msg"] = status.get("status_reason", "未知错误")
|
||||
except Exception as e:
|
||||
print(f"[网络] 解析失败: {e}")
|
||||
|
||||
page.on("response", on_response)
|
||||
|
||||
# Step 1: 打开登录页
|
||||
print("[1] 打开登录页...")
|
||||
await page.goto(LOGIN_URL, wait_until="networkidle", timeout=15000)
|
||||
await asyncio.sleep(1)
|
||||
await page.screenshot(path="debug_step1.png")
|
||||
print(f" URL: {page.url}")
|
||||
|
||||
# Step 2: 点击"登录"进入表单
|
||||
print("[2] 点击 #login_init_by_login...")
|
||||
await page.locator("#login_init_by_login").click(timeout=5000)
|
||||
await asyncio.sleep(1.5)
|
||||
await page.screenshot(path="debug_step2.png")
|
||||
|
||||
# Step 3: 点击"账号密码登录" tab
|
||||
print("[3] 点击'账号密码登录' tab...")
|
||||
try:
|
||||
await page.locator("h4.login_content_h4 span", has_text="账号密码登录").click(timeout=3000)
|
||||
await asyncio.sleep(0.5)
|
||||
except Exception:
|
||||
print(" 已在密码登录 tab")
|
||||
|
||||
await page.screenshot(path="debug_step3.png")
|
||||
|
||||
# Step 4: 填写手机号和密码
|
||||
print(f"[4] 填写手机号: {TEST_PHONE}")
|
||||
phone_input = page.locator('input[placeholder*="手机号"]').first
|
||||
await phone_input.click()
|
||||
await phone_input.fill("")
|
||||
await page.keyboard.type(TEST_PHONE, delay=50)
|
||||
|
||||
print("[5] 填写密码...")
|
||||
pwd_input = page.locator('input[placeholder*="登录密码"], input[type="password"]').first
|
||||
await pwd_input.click()
|
||||
await pwd_input.fill("")
|
||||
await page.keyboard.type(TEST_PASSWORD, delay=50)
|
||||
|
||||
await asyncio.sleep(0.5)
|
||||
await page.screenshot(path="debug_step4.png")
|
||||
print(" 截图已保存: debug_step4.png")
|
||||
|
||||
# Step 5: 点击登录按钮
|
||||
print("[6] 点击 #login_pwd_submit...")
|
||||
await page.locator("#login_pwd_submit").click(timeout=5000)
|
||||
|
||||
# 等待 API 响应
|
||||
print(" 等待 API 响应...")
|
||||
await asyncio.sleep(5)
|
||||
await page.screenshot(path="debug_step5.png")
|
||||
print(f" 当前URL: {page.url}")
|
||||
|
||||
# 检查结果
|
||||
print(f"\n[结果] success={login_result['success']}, msg={login_result['msg']}")
|
||||
print(f"[结果] 拿到 {len(login_result['cookies'])} 个 cookie")
|
||||
|
||||
if login_result["success"] and login_result["cookies"]:
|
||||
# 把 API 返回的 cookie 写入浏览器
|
||||
for c in login_result["cookies"]:
|
||||
await context.add_cookies([{
|
||||
"name": c.get("name", ""),
|
||||
"value": c.get("value", ""),
|
||||
"domain": c.get("domain", ".weidian.com"),
|
||||
"path": c.get("path", "/"),
|
||||
"httpOnly": c.get("httpOnly", False),
|
||||
"secure": c.get("secure", False),
|
||||
"sameSite": "Lax",
|
||||
}])
|
||||
|
||||
# 保存 storage_state
|
||||
await context.storage_state(path="test_auth_state.json")
|
||||
print("[结果] cookie 已保存到 test_auth_state.json")
|
||||
|
||||
# 验证: 访问个人页
|
||||
print("\n[验证] 访问个人页...")
|
||||
await page.goto(
|
||||
"https://h5.weidian.com/decoration/uni-mine/",
|
||||
wait_until="networkidle", timeout=15000
|
||||
)
|
||||
await asyncio.sleep(2)
|
||||
await page.screenshot(path="debug_verify.png")
|
||||
verify_url = page.url
|
||||
print(f"[验证] URL: {verify_url}")
|
||||
if "login" in verify_url.lower():
|
||||
print("[验证] 失败 - 被重定向到登录页")
|
||||
else:
|
||||
print("[验证] 成功 - cookie 有效!")
|
||||
else:
|
||||
print("[结果] 登录未成功")
|
||||
body = await page.locator("body").text_content()
|
||||
print(f"[页面] {body[:300]}")
|
||||
|
||||
print("\n按回车关闭浏览器...")
|
||||
try:
|
||||
await asyncio.get_event_loop().run_in_executor(None, input)
|
||||
except Exception:
|
||||
pass
|
||||
await browser.close()
|
||||
await p.stop()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
asyncio.run(test_login())
|
||||
Reference in New Issue
Block a user