1
1
mirror of https://github.com/KenanZhu/AutoLibrary.git synced 2026-06-18 07:23:03 +08:00

Compare commits

...

8 Commits

6 changed files with 73 additions and 127 deletions
+16 -5
View File
@@ -17,6 +17,7 @@ from selenium.webdriver.support import expected_conditions as EC
from selenium.webdriver.edge.service import Service
from MsgBase import MsgBase
from LibChecker import LibChecker
from LibLogin import LibLogin
from LibLogout import LibLogout
from LibReserve import LibReserve
@@ -95,12 +96,21 @@ class AutoLib(MsgBase):
except Exception as e:
self._showTrace(f"浏览器驱动初始化失败: {e}")
return False
# init library operators
self._showTrace(f"浏览器驱动已初始化, 类型: {self.__driver_type}, 路径: {self.__driver_path}")
return True
def __initLibOperators(
self
):
if not self.__driver:
self._showTrace(f"浏览器驱动未初始化, 请先初始化浏览器驱动 !")
return
self.__lib_checker = LibChecker(self._input_queue, self._output_queue, self.__driver)
self.__lib_login = LibLogin(self._input_queue, self._output_queue, self.__driver)
self.__lib_logout = LibLogout(self._input_queue, self._output_queue, self.__driver)
self.__lib_reserve = LibReserve(self._input_queue, self._output_queue, self.__driver)
self._showTrace(f"浏览器驱动已初始化, 类型: {self.__driver_type}, 路径: {self.__driver_path}")
return True
def __waitResponseLoad(
@@ -169,7 +179,7 @@ class AutoLib(MsgBase):
}
# reserve
if run_mode["auto_reserve"]:
if self.__lib_reserve.canReserve(reserve_info.get("date")):
if self.__lib_checker.canReserve(reserve_info.get("date")):
if self.__lib_reserve.reserve(reserve_info):
self._showTrace(f"用户 {username} 预约成功 !")
result = 0
@@ -202,6 +212,7 @@ class AutoLib(MsgBase):
else:
if not self.__initDriverUrl():
return
self.__initLibOperators()
user_counter = {"current": 0, "success": 0, "failed": 0, "passed": 0}
users = self.__users_config_reader.get("users")
@@ -243,5 +254,5 @@ class AutoLib(MsgBase):
self._showTrace(f"浏览器驱动已关闭")
return True
else:
self._showTrace(f"浏览器驱动未初始化无需关闭")
self._showTrace(f"浏览器驱动未初始化, 无需关闭")
return False
+40 -11
View File
@@ -134,6 +134,41 @@ class LibLogin(LibOperator):
return False
def __solveCaptcha(
self,
auto_captcha: bool = True,
) -> str:
max_attempts = 5
for _ in range(max_attempts):
if auto_captcha:
captcha_text = self.__autoRecognizeCaptcha()
else:
self._showTrace(f"用户未配置自动识别验证码, 请手动输入验证码 !")
captcha_text = self.__manualRecognizeCaptcha()
if captcha_text:
return captcha_text
self._showTrace(f"验证码识别失败 {max_attempts} 次, 请检查验证码是否正确 !")
return ""
def __fillCaptchaElement(
self,
captcha_text: str,
) -> bool:
try:
captcha_element = self.__driver.find_element(By.NAME, "answer")
captcha_element.clear()
captcha_element.send_keys(captcha_text)
return True
except Exception as e:
self._showTrace(f"验证码填写失败 ! : {e}")
self.__refreshCaptcha()
return False
def login(
self,
username: str,
@@ -153,17 +188,11 @@ class LibLogin(LibOperator):
password,
):
continue
while True:
if auto_captcha:
captcha_text = self.__autoRecognizeCaptcha()
else:
self._showTrace(f"用户未配置自动识别验证码, 请手动输入验证码 !")
captcha_text = self.__manualRecognizeCaptcha()
if captcha_text:
break
captcha_element = self.__driver.find_element(By.NAME, "answer")
captcha_element.clear()
captcha_element.send_keys(captcha_text)
captcha_text = self.__solveCaptcha(auto_captcha)
if not captcha_text:
continue
if not self.__fillCaptchaElement(captcha_text):
continue
self._showTrace("尝试登录...")
try:
self.__driver.find_element(
-6
View File
@@ -44,19 +44,13 @@ class LibLogout(LibOperator):
if self.__driver is None:
self._showTrace("未提供有效 WebDriver 实例 !")
return False
try:
self.__driver.find_element(
By.XPATH, "//a[@href='/logout']"
).click()
self._showTrace(f"用户 {username} 注销成功 !")
return True
except Exception as e:
self._showTrace(f"用户 {username} 注销失败 ! : {e}")
return False
+8 -97
View File
@@ -83,9 +83,9 @@ class LibReserve(LibOperator):
raise
if "预定好了" in title or "预约成功" in title or "操作成功" in title:
if len(contents) >= 6:
date_val = contents[1].split(" ")[1].strip() if " " in contents[1] else contents[1].strip()
time_val = contents[2].split(" ")[1].strip() if " " in contents[2] else contents[2].strip()
seat_val = contents[3].split(" ")[1].strip() if " " in contents[3] else contents[3].strip()
date_val = contents[1].split(" : ")[1].strip() if " : " in contents[1] else contents[1].strip()
time_val = contents[2].split(" : ")[1].strip() if " : " in contents[2] else contents[2].strip()
seat_val = contents[3].split(" : ")[1].strip() if " : " in contents[3] else contents[3].strip()
checkin_val = contents[5].strip()
self._showTrace(f"\n"\
f" 预约成功 !\n"\
@@ -137,10 +137,12 @@ class LibReserve(LibOperator):
raise ValueError(f"房间 '{reserve_info['room']}' 不存在")
if reserve_info.get("seat_id") is None:
raise ValueError("未指定座位")
if reserve_info["seat_id"] == "":
raise ValueError("未指定座位号")
except ValueError as e:
self._showTrace(
f"预约信息错误 ! : {e}, "\
f"由于缺少必要的预约信息 无法开始预约流程, 请检查预约信息是否完整"
f"由于缺少必要的预约信息, 无法开始预约流程, 请检查预约信息是否完整"
)
return False
@@ -224,7 +226,7 @@ class LibReserve(LibOperator):
reserve_info["begin_time"]["time"], reserve_info["end_time"]["time"] = end_time_str, begin_time_str
reserve_info["begin_time"]["prefer_early"], reserve_info["end_time"]["prefer_early"] = \
reserve_info["end_time"]["prefer_early"], reserve_info["begin_time"]["prefer_early"]
self._showTrace("预约开始时间晚于预约结束时间自动调换开始时间和结束时间")
self._showTrace("预约开始时间晚于预约结束时间, 自动调换开始时间和结束时间")
# update the begin_mins and end_mins after swap
begin_time_str, end_time_str = end_time_str, begin_time_str
@@ -246,7 +248,7 @@ class LibReserve(LibOperator):
reserve_info["end_time"]["time"] = self.__minsToTime(new_end_mins)
self._showTrace("预约持续时间超过8小时, 自动设置为 8 小时")
self._showTrace(
f"预约信息检查完成准备预约 "
f"预约信息检查完成, 准备预约 "
f"{reserve_info['date']} "
f"{reserve_info['begin_time']["time"]} - "
f"{reserve_info['end_time']["time"]} "
@@ -489,97 +491,6 @@ class LibReserve(LibOperator):
return True
def canReserve(
self,
date: str
) -> bool:
if date is None:
self._showTrace("日期未指定, 无法检查预约状态")
return True
else:
self._showTrace(f"正在检查用户在日期 {date} 是否可预约......")
date_obj = datetime.strptime(date, "%Y-%m-%d").date()
try:
# we need to navigate to the history page to check if we can reserve
WebDriverWait(self.__driver, 5).until(
EC.element_to_be_clickable((By.XPATH, "//a[@href='/history?type=SEAT']"))
).click()
WebDriverWait(self.__driver, 2).until(
EC.presence_of_element_located((By.CLASS_NAME, "myReserveList"))
)
except:
self._showTrace("加载预约记录页面失败 !")
return False
checked_count = 0
max_attemots = 3 # we only check (3*4=)12 reservations
for _ in range(max_attemots):
try:
# check if there's any reservation on the date
reservations = self.__driver.find_elements(
By.CSS_SELECTOR, ".myReserveList dl"
)
except:
self._showTrace("加载预约记录失败 !")
return False
for i in range(checked_count, len(reservations) - 1): # the last one is load button
reservation = reservations[i]
try:
time_element = reservation.find_element(
By.CSS_SELECTOR, "dt"
)
status_elements = reservation.find_elements(
By.CSS_SELECTOR, "a"
)
is_reserved = any("已预约" in status.text for status in status_elements)
# process time element to get the date string
time_str = time_element.text.strip()
today = datetime.now().date()
if "明天" in time_str:
target_date = today + timedelta(days=1)
date_str = target_date.strftime("%Y-%m-%d")
elif "今天" in time_str:
target_date = today
date_str = target_date.strftime("%Y-%m-%d")
elif "昨天" in time_str:
target_date = today - timedelta(days=1)
date_str = target_date.strftime("%Y-%m-%d")
else:
date_match = re.search(r'(\d{4}-\d{1,2}-\d{1,2})', time_str)
if date_match:
date_str = date_match.group(1)
else:
continue
# reservation is earlier than the given date, can reserve
if datetime.strptime(date_str, "%Y-%m-%d").date() < date_obj:
self._showTrace(f"用户在 {date} 可预约")
return True
# reservation is later than the given date, check the next one
elif datetime.strptime(date_str, "%Y-%m-%d").date() > date_obj:
continue
# compare with the given date
if date_str == date and is_reserved:
self._showTrace(f"用户在 {date} 已存在有效预约, 无法预约")
return False
except:
self._showTrace(f"解析第 {i + 1} 条预约记录时发生未知错误 !")
continue
checked_count = len(reservations) - 1
# load new reservations if still not sure
try:
more_btn = self.__driver.find_element(By.ID, "moreBtn")
if more_btn.is_displayed() and more_btn.is_enabled():
self.__driver.execute_script("arguments[0].scrollIntoView(true);", more_btn)
self.__driver.execute_script("arguments[0].click();", more_btn)
else:
break
except:
break
self._showTrace(f"用户在 {date} 可预约")
return True
def reserve(
self,
reserve_info: dict
+2 -8
View File
@@ -70,7 +70,7 @@ class AutoLibWorker(QThread):
os.path.exists(path) for path in self.__config_paths.values()
):
self.showTraceSignal.emit(
"配置文件路径不存在请检查配置文件路径是否正确。"
"配置文件路径不存在, 请检查配置文件路径是否正确。"
)
return False
return True
@@ -102,7 +102,7 @@ class AutoLibWorker(QThread):
self.showTraceSignal.emit("AutoLibrary 运行结束")
except Exception as e:
self.showTraceSignal.emit(
f"AutoLibrary 运行时发生异常{e}"
f"AutoLibrary 运行时发生异常 : {e}"
)
finally:
self.finishedSignal.emit()
@@ -127,10 +127,6 @@ class ALMainWindow(QMainWindow, Ui_ALMainWindow):
self.setupUi(self)
self.__input_queue = queue.Queue()
self.__output_queue = queue.Queue()
self.__auto_lib = AutoLib(
self.__input_queue,
self.__output_queue,
)
self.__config_paths = {
"system":
f"{QDir.toNativeSeparators(QFileInfo(sys.executable).absoluteDir().absoluteFilePath("system.json"))}",
@@ -172,8 +168,6 @@ class ALMainWindow(QMainWindow, Ui_ALMainWindow):
if self.__timer and self.__timer.isActive():
self.__timer.stop()
if self.__auto_lib:
self.__auto_lib.close()
if self.__alConfigWidget:
self.__alConfigWidget.close()
super().closeEvent(event)
+7
View File
@@ -0,0 +1,7 @@
Copyright 2025 KenanZhu
Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the “Software”), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions:
The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED “AS IS”, WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.