feat: Implement login business logic, update base UI, and add new API configuration and assets
This commit is contained in:
parent
6f1e63539f
commit
8464338ab3
91
framework/business/login.py
Normal file
91
framework/business/login.py
Normal file
@ -0,0 +1,91 @@
|
||||
from framework.core.base_ui import BaseUI
|
||||
from framework.config.settings import Config
|
||||
from framework.core.logger import get_logger
|
||||
import time
|
||||
|
||||
logger = get_logger("LoginPage")
|
||||
|
||||
class LoginPage(BaseUI):
|
||||
# --- 元素定位器 ---
|
||||
# 根据截图:登录按钮 Class 是 loginBtn
|
||||
USERNAME_INPUT = "input[placeholder*='账号'], #account"
|
||||
PASSWORD_INPUT = "input[placeholder*='密码'], #password"
|
||||
LOGIN_BUTTON = "button.loginBtn"
|
||||
|
||||
# 登录成功后的特征元素(比如侧边栏或侧边标题)
|
||||
SUCCESS_INDICATOR = ".ant-layout-sider, .user-name"
|
||||
|
||||
def __init__(self, headless=False):
|
||||
super().__init__(headless=headless)
|
||||
|
||||
def login(self, username, password):
|
||||
"""执行登录流程"""
|
||||
# 1. 访问登录页
|
||||
login_url = f"{Config.BASE_URL}/login"
|
||||
self.navigate(login_url)
|
||||
|
||||
# 2. 等待输入框可见
|
||||
logger.info("等待登录页面加载...")
|
||||
self.wait_for_selector(self.USERNAME_INPUT)
|
||||
|
||||
# 3. 输入账号密码
|
||||
self.fill(self.USERNAME_INPUT, username)
|
||||
self.fill(self.PASSWORD_INPUT, password)
|
||||
|
||||
# 4. 点击登录
|
||||
logger.info("正在点击登录按钮...")
|
||||
time.sleep(1) # 增加稳定性
|
||||
self.click(self.LOGIN_BUTTON)
|
||||
|
||||
# 5. 增强型事件监听 (实时监控所有 URL 变化,防止错过极速重定向)
|
||||
logger.info("正在实时监控重定向链路...")
|
||||
success_flag = {"done": False}
|
||||
|
||||
def _check_url(frame):
|
||||
url = frame.url if hasattr(frame, "url") else str(frame)
|
||||
if "/setting/profile" in url or "bearer=" in url:
|
||||
logger.info(f"⌛ 侦测到重定向轨迹: {url[:80]}...")
|
||||
success_flag["done"] = True
|
||||
|
||||
self.page.on("framenavigated", _check_url)
|
||||
|
||||
# 给重定向留出充足时间
|
||||
start_time = time.time()
|
||||
while time.time() - start_time < 20:
|
||||
if success_flag["done"]:
|
||||
# 如果侦测到了成功轨迹,最后再确认一下最终落点
|
||||
time.sleep(3) # 缓冲渲染
|
||||
if "/login" not in self.page.url or "bearer" in self.page.url:
|
||||
logger.info(f"✅ 登录验证通过!最终 URL: {self.page.url}")
|
||||
return True
|
||||
else:
|
||||
# 补救措施:既然拿到了 token 轨迹,尝试强制冲入
|
||||
logger.info("尝试补救强制跳转...")
|
||||
self.page.goto(f"{Config.BASE_URL}/setting/profile", wait_until="domcontentloaded")
|
||||
time.sleep(2)
|
||||
if "/login" not in self.page.url:
|
||||
return True
|
||||
time.sleep(0.5)
|
||||
|
||||
# 最终判定失败
|
||||
self.page.screenshot(path="login_debug.png")
|
||||
logger.error(f"无法确认登录成功状态。最终 URL: {self.page.url}")
|
||||
return False
|
||||
|
||||
if __name__ == "__main__":
|
||||
# 本地测试脚本
|
||||
ui = LoginPage(headless=False)
|
||||
try:
|
||||
ui.start()
|
||||
# 提示用户手动输入账号密码进行本地测试
|
||||
user = input("请输入测试账号: ")
|
||||
pwd = input("请输入测试密码: ")
|
||||
|
||||
success = ui.login(user, pwd)
|
||||
if success:
|
||||
print("✅ 登录成功")
|
||||
else:
|
||||
print("❌ 登录失败或未观察到跳转")
|
||||
time.sleep(100)
|
||||
finally:
|
||||
ui.stop()
|
||||
@ -1,22 +1,67 @@
|
||||
"""
|
||||
UI 自动化基类占位 (推荐使用 Playwright)
|
||||
"""
|
||||
from playwright.sync_api import sync_playwright
|
||||
from framework.core.logger import get_logger
|
||||
import time
|
||||
|
||||
logger = get_logger("BaseUI")
|
||||
|
||||
class BaseUI:
|
||||
def __init__(self, browser_type="chromium"):
|
||||
self.browser_type = browser_type
|
||||
# TODO: 集成 playwright/selenium 初始化
|
||||
pass
|
||||
def __init__(self, headless=False):
|
||||
self.playwright = None
|
||||
self.browser = None
|
||||
self.context = None
|
||||
self.page = None
|
||||
self.headless = headless
|
||||
|
||||
def start(self):
|
||||
"""启动浏览器并伪装"""
|
||||
self.playwright = sync_playwright().start()
|
||||
# 增加伪装配置
|
||||
self.browser = self.playwright.chromium.launch(
|
||||
headless=self.headless,
|
||||
args=["--disable-blink-features=AutomationControlled"]
|
||||
)
|
||||
# 设置真实的 User-Agent
|
||||
user_agent = "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/122.0.0.0 Safari/537.36"
|
||||
self.context = self.browser.new_context(user_agent=user_agent)
|
||||
self.page = self.context.new_page()
|
||||
|
||||
# 实时打印浏览器控制台日志,方便排查白屏
|
||||
self.page.on("console", lambda msg: logger.info(f"[BROWSER LOG] {msg.text}"))
|
||||
self.page.on("pageerror", lambda exc: logger.error(f"[BROWSER ERROR] {exc}"))
|
||||
|
||||
logger.info("Browser started with spoofing and logging")
|
||||
return self.page
|
||||
|
||||
def stop(self):
|
||||
"""关闭浏览器"""
|
||||
if self.browser:
|
||||
self.browser.close()
|
||||
if self.playwright:
|
||||
self.playwright.stop()
|
||||
logger.info("Browser stopped")
|
||||
|
||||
def navigate(self, url):
|
||||
logger.info(f"Navigate to {url}")
|
||||
# self.page.goto(url)
|
||||
pass
|
||||
# 使用更稳健的加载策略:DOM 加载完即继续,后续靠 wait_for_selector
|
||||
self.page.goto(url, wait_until="domcontentloaded", timeout=30000)
|
||||
# 给 2 秒缓冲时间让脚本执行
|
||||
time.sleep(2)
|
||||
|
||||
def click(self, selector):
|
||||
logger.info(f"Clicking {selector}")
|
||||
# self.page.click(selector)
|
||||
pass
|
||||
def click(self, selector, timeout=5000):
|
||||
try:
|
||||
logger.info(f"Clicking: {selector}")
|
||||
self.page.click(selector, timeout=timeout)
|
||||
except Exception as e:
|
||||
self.page.screenshot(path="click_failed.png")
|
||||
logger.error(f"Click failed on {selector}, saved screenshot to click_failed.png")
|
||||
raise e
|
||||
|
||||
def fill(self, selector, value):
|
||||
logger.info(f"Filling {selector} with value")
|
||||
self.page.fill(selector, value)
|
||||
|
||||
def wait_for_selector(self, selector, timeout=10000):
|
||||
self.page.wait_for_selector(selector, timeout=timeout)
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user