mirror of
https://github.com/KenanZhu/AutoLibrary.git
synced 2026-06-17 23:13:03 +08:00
refactor(pages): 将 LoginPage 日志回调从方法参数改为构造器注入
消除 login() 方法签名中的 tracer/log_level 参数，通过构造器可选注入 tracer 统一日志模式，避免 Page Object 对外暴露 MsgBase 内部细节。 Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
This commit is contained in:
@@ -164,7 +164,7 @@ class AutoLibPages(MsgBase):
|
||||
self._showTrace("未配置图书馆参数 !", self.TraceLevel.ERROR)
|
||||
return False
|
||||
url: str = lib_config.get("host_url") + lib_config.get("login_url")
|
||||
self.__login_page = LoginPage(self.__driver)
|
||||
self.__login_page = LoginPage(self.__driver, tracer=self._showTrace)
|
||||
self.__driver.set_page_load_timeout(5)
|
||||
try:
|
||||
self.__driver.get(url)
|
||||
@@ -244,8 +244,6 @@ class AutoLibPages(MsgBase):
|
||||
password,
|
||||
captcha_solver=self.__captcha_handler.solveCaptcha,
|
||||
auto_captcha=auto_captcha,
|
||||
tracer=self._showTrace,
|
||||
log_level=self.TraceLevel,
|
||||
max_attempts=login_config.get("max_attempt", 3),
|
||||
):
|
||||
return 1
|
||||
|
||||
+20
-11
@@ -7,7 +7,7 @@ This software is provided "as is", without any warranty of any kind.
|
||||
You may use, modify, and distribute this file under the terms of the MIT License.
|
||||
See the LICENSE file for details.
|
||||
"""
|
||||
from typing import Callable
|
||||
from typing import Callable, Optional
|
||||
|
||||
from selenium.common.exceptions import (
|
||||
ElementNotInteractableException,
|
||||
@@ -37,9 +37,21 @@ class LoginPage:
|
||||
def __init__(
|
||||
self,
|
||||
driver: WebDriver,
|
||||
tracer: Optional[Callable[..., None]] = None,
|
||||
) -> None:
|
||||
|
||||
self._driver: WebDriver = driver
|
||||
self._tracer: Optional[Callable[..., None]] = tracer
|
||||
|
||||
def _trace(
|
||||
self,
|
||||
msg: str,
|
||||
level: int = 20,
|
||||
no_log: bool = False,
|
||||
) -> None:
|
||||
|
||||
if self._tracer:
|
||||
self._tracer(msg, level, no_log)
|
||||
|
||||
def navigate(
|
||||
self,
|
||||
@@ -177,16 +189,13 @@ class LoginPage:
|
||||
password: str,
|
||||
captcha_solver: Callable[["LoginPage", bool], str],
|
||||
auto_captcha: bool,
|
||||
tracer: Callable[..., None],
|
||||
log_level: type,
|
||||
max_attempts: int = 5,
|
||||
) -> bool:
|
||||
|
||||
ERR = log_level.ERROR
|
||||
for attempt in range(max_attempts):
|
||||
tracer(
|
||||
self._trace(
|
||||
f"用户 {username} 第 {attempt + 1} 次尝试登录......",
|
||||
20, no_log=True,
|
||||
no_log=True,
|
||||
)
|
||||
if not self.fillCredentials(username, password):
|
||||
continue
|
||||
@@ -195,16 +204,16 @@ class LoginPage:
|
||||
continue
|
||||
if not self.fillCaptcha(captcha_text):
|
||||
continue
|
||||
tracer("尝试登录...", 20, no_log=True)
|
||||
self._trace("尝试登录...", no_log=True)
|
||||
if not self.clickLogin():
|
||||
continue
|
||||
if self.waitLoginSuccess():
|
||||
tracer(f"用户 {username} 第 {attempt + 1} 次登录成功 !")
|
||||
self._trace(f"用户 {username} 第 {attempt + 1} 次登录成功 !")
|
||||
return True
|
||||
else:
|
||||
err_msg = (
|
||||
self._trace(
|
||||
"登录页面加载失败 ! : "
|
||||
"用户账号或者密码错误/验证码错误, 具体以页面提示为准"
|
||||
"用户账号或者密码错误/验证码错误, 具体以页面提示为准",
|
||||
level=40,
|
||||
)
|
||||
tracer(err_msg, ERR)
|
||||
return False
|
||||
|
||||
@@ -1,162 +0,0 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
"""
|
||||
Copyright (c) 2026 KenanZhu.
|
||||
All rights reserved.
|
||||
|
||||
This software is provided "as is", without any warranty of any kind.
|
||||
You may use, modify, and distribute this file under the terms of the MIT License.
|
||||
See the LICENSE file for details.
|
||||
|
||||
AutoLibrary 真实运行测试脚本。
|
||||
在 venv 中运行:
|
||||
py -3 test_pages_refactor.py [--mode MODE]
|
||||
|
||||
MODE 可选值 (默认 1):
|
||||
1 = 只预约
|
||||
2 = 只签到
|
||||
3 = 预约 + 签到
|
||||
4 = 只续约
|
||||
7 = 全部 (预约 + 签到 + 续约)
|
||||
"""
|
||||
import os
|
||||
import sys
|
||||
import argparse
|
||||
|
||||
SRC = os.path.dirname(os.path.abspath(__file__))
|
||||
if SRC not in sys.path:
|
||||
sys.path.insert(0, SRC)
|
||||
|
||||
|
||||
def getAppConfigDir() -> str:
|
||||
appData = os.environ.get("APPDATA", "")
|
||||
if not appData:
|
||||
appData = os.path.join(os.path.expanduser("~"), "AppData", "Roaming")
|
||||
return os.path.join(appData, "AutoLibrary", "configs")
|
||||
|
||||
|
||||
def main():
|
||||
parser = argparse.ArgumentParser(description="AutoLibrary 真实运行测试")
|
||||
parser.add_argument(
|
||||
"--mode", type=int, default=1,
|
||||
help="运行模式 bitmask: 1=预约 2=签到 4=续约 (默认 1)"
|
||||
)
|
||||
parser.add_argument(
|
||||
"--group", type=int, default=0,
|
||||
help="只运行第 N 个启用的任务组 (0=全部, 默认 0)"
|
||||
)
|
||||
parser.add_argument(
|
||||
"--headless", action="store_true",
|
||||
help="使用 headless 模式运行浏览器"
|
||||
)
|
||||
args = parser.parse_args()
|
||||
|
||||
# ---- 1. 初始化 ConfigManager ----
|
||||
from managers.config.ConfigManager import instance as configInstance
|
||||
from managers.config.ConfigUtils import ConfigUtils
|
||||
from utils.JSONReader import JSONReader
|
||||
|
||||
configDir = getAppConfigDir()
|
||||
if not os.path.isdir(configDir):
|
||||
print(f"[FAIL] 配置目录不存在: {configDir}")
|
||||
print("请先启动一次 AutoLibrary GUI 以生成配置文件。")
|
||||
return 1
|
||||
|
||||
try:
|
||||
configInstance(configDir)
|
||||
except ValueError:
|
||||
pass
|
||||
|
||||
configPaths = ConfigUtils.getAutomationConfigPaths()
|
||||
runPath = configPaths.get("run")
|
||||
userPath = configPaths.get("user")
|
||||
|
||||
if not runPath or not os.path.isfile(runPath):
|
||||
print(f"[FAIL] run.json 不存在: {runPath}")
|
||||
return 1
|
||||
if not userPath or not os.path.isfile(userPath):
|
||||
print(f"[FAIL] user.json 不存在: {userPath}")
|
||||
return 1
|
||||
|
||||
print(f"[INFO] run : {runPath}")
|
||||
print(f"[INFO] user : {userPath}")
|
||||
|
||||
# ---- 2. 加载配置 ----
|
||||
runConfig = JSONReader(runPath).data()
|
||||
userConfig = JSONReader(userPath).data()
|
||||
|
||||
if args.mode is not None:
|
||||
runConfig["mode"]["run_mode"] = args.mode
|
||||
if args.headless:
|
||||
runConfig["web_driver"]["headless"] = True
|
||||
|
||||
groups = userConfig.get("groups", [])
|
||||
if not groups:
|
||||
print("[FAIL] user.json 中没有任务组")
|
||||
return 1
|
||||
|
||||
print(f"[INFO] 运行模式: {runConfig['mode']['run_mode']}")
|
||||
if args.headless:
|
||||
print("[INFO] Headless 模式已启用")
|
||||
|
||||
# ---- 3. 创建 AutoLib 并运行 ----
|
||||
from pages.AutoLibPages import AutoLibPages
|
||||
import queue
|
||||
import threading
|
||||
|
||||
for gi, group in enumerate(groups):
|
||||
if args.group > 0 and gi + 1 != args.group:
|
||||
continue
|
||||
if not group.get("enabled", True):
|
||||
print(f"[SKIP] 任务组 {gi + 1} '{group.get('name', '未命名')}' 已禁用")
|
||||
continue
|
||||
|
||||
users = group.get("users", [])
|
||||
enabledUsers = [u for u in users if u.get("enabled", True)]
|
||||
if not enabledUsers:
|
||||
print(f"[SKIP] 任务组 {gi + 1} 没有启用的用户")
|
||||
continue
|
||||
|
||||
print(f"\n{'=' * 60}")
|
||||
print(f"任务组 {gi + 1}/{len(groups)}: '{group.get('name', '未命名')}'")
|
||||
print(f"启用的用户: {len(enabledUsers)}/{len(users)}")
|
||||
print(f"{'=' * 60}")
|
||||
|
||||
outputQueue = queue.Queue()
|
||||
stopConsumer = threading.Event()
|
||||
traceLines = []
|
||||
|
||||
def consumeTrace():
|
||||
while not stopConsumer.is_set():
|
||||
try:
|
||||
msg = outputQueue.get(timeout=0.3)
|
||||
traceLines.append(msg)
|
||||
print(msg)
|
||||
except queue.Empty:
|
||||
continue
|
||||
|
||||
consumer = threading.Thread(target=consumeTrace, daemon=True)
|
||||
consumer.start()
|
||||
|
||||
try:
|
||||
autoLib = AutoLibPages(
|
||||
input_queue=queue.Queue(),
|
||||
output_queue=outputQueue,
|
||||
run_config=runConfig,
|
||||
)
|
||||
autoLib.run({"users": enabledUsers})
|
||||
autoLib.close()
|
||||
except Exception as e:
|
||||
print(f"[FAIL] 运行异常: {e}")
|
||||
import traceback
|
||||
traceback.print_exc()
|
||||
return 1
|
||||
finally:
|
||||
stopConsumer.set()
|
||||
consumer.join(timeout=2)
|
||||
|
||||
print("\n[OK] 测试完成")
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
Reference in New Issue
Block a user