From 8464338ab3ce45cc9f2e965f7c2b7a043868e2d2 Mon Sep 17 00:00:00 2001 From: "hangyu.tao" Date: Tue, 17 Mar 2026 20:10:32 +0800 Subject: [PATCH] feat: Implement login business logic, update base UI, and add new API configuration and assets --- framework/business/login.py | 91 +++++++++++++++++++++++++++++++++++++ framework/core/base_ui.py | 65 ++++++++++++++++++++++---- 2 files changed, 146 insertions(+), 10 deletions(-) create mode 100644 framework/business/login.py diff --git a/framework/business/login.py b/framework/business/login.py new file mode 100644 index 0000000..cf7817b --- /dev/null +++ b/framework/business/login.py @@ -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() diff --git a/framework/core/base_ui.py b/framework/core/base_ui.py index 5367b27..f636ec8 100644 --- a/framework/core/base_ui.py +++ b/framework/core/base_ui.py @@ -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)