mirror of
https://github.com/KenanZhu/AutoLibrary.git
synced 2026-06-17 23:13:03 +08:00
chore(LibChecker): add new LibOperator: LibChecker.py
We introduce a new LibOperator which is used to check if a given user date is available for reserve or check-in. This is ready for the upcoming feature: 'Auto Check-In'.
This commit is contained in:
+207
@@ -0,0 +1,207 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
"""
|
||||
Copyright (c) 2025 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.
|
||||
"""
|
||||
import re
|
||||
import time
|
||||
import queue
|
||||
|
||||
from datetime import datetime, timedelta
|
||||
from selenium.webdriver.common.by import By
|
||||
from selenium.webdriver.support.ui import WebDriverWait
|
||||
from selenium.webdriver.support import expected_conditions as EC
|
||||
|
||||
from LibOperator import LibOperator
|
||||
|
||||
|
||||
class LibChecker(LibOperator):
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
input_queue: queue.Queue,
|
||||
output_queue: queue.Queue,
|
||||
driver
|
||||
):
|
||||
|
||||
super().__init__(input_queue, output_queue)
|
||||
|
||||
self.__driver = driver
|
||||
|
||||
|
||||
def _waitResponseLoad(
|
||||
self
|
||||
) -> bool:
|
||||
|
||||
pass
|
||||
|
||||
|
||||
def __navigateToReserveRecordPage(
|
||||
self
|
||||
) -> bool:
|
||||
|
||||
try:
|
||||
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
|
||||
return True
|
||||
|
||||
|
||||
def __getReserveRecord(
|
||||
self,
|
||||
wanted_date: str,
|
||||
wanted_status: str
|
||||
) -> dict:
|
||||
|
||||
if wanted_date is None:
|
||||
self._showTrace("日期未指定, 无法检查当前预约状态")
|
||||
return None
|
||||
self._showTrace(f"正在检查用户在日期 {wanted_date} 是否有预约状态为 {wanted_status} 的预约记录......")
|
||||
date_obj = datetime.strptime(wanted_date, "%Y-%m-%d").date()
|
||||
|
||||
checked_count = 0
|
||||
max_check_times = 3 # we only check (3*4=)12 reservations
|
||||
|
||||
if not self.__navigateToReserveRecordPage():
|
||||
return None
|
||||
for _ in range(max_check_times):
|
||||
try:
|
||||
# check if there's any reservation on the date
|
||||
reservations = self.__driver.find_elements(
|
||||
By.CSS_SELECTOR, ".myReserveList dl"
|
||||
)
|
||||
except:
|
||||
self._showTrace("加载预约记录失败 !")
|
||||
return None
|
||||
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"
|
||||
)
|
||||
info_elements = reservation.find_elements(
|
||||
By.CSS_SELECTOR, "a"
|
||||
)
|
||||
except:
|
||||
self._showTrace(f"解析第 {i + 1} 条预约记录时发生未知错误 !")
|
||||
continue
|
||||
is_wanted = any(wanted_status in status.text for status in info_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:
|
||||
self._showTrace(f"无法解析第 {i + 1} 条预约记录的日期 ! 该记录的时间为 {time_str}")
|
||||
continue
|
||||
# reservation is later than the given date, check the next one
|
||||
if datetime.strptime(date_str, "%Y-%m-%d").date() > date_obj:
|
||||
continue
|
||||
# reservation is earlier than the given date, can reserve
|
||||
if datetime.strptime(date_str, "%Y-%m-%d").date() < date_obj:
|
||||
return None
|
||||
# query the wanted status
|
||||
if is_wanted:
|
||||
self._showTrace(f"寻找到第 {i + 1} 条预约记录, 状态为 {wanted_status}")
|
||||
time_match = re.search(r"(\d{1,2}:\d{2}) -- (\d{1,2}:\d{2})", time_str)
|
||||
if time_match is None:
|
||||
self._showTrace(f"无法解析第 {i + 1} 条预约记录的时间 ! 该记录的时间为 {time_str}")
|
||||
continue
|
||||
return {
|
||||
"index": i,
|
||||
"date": date_str,
|
||||
"time_str": time_match.group(0),
|
||||
"status": wanted_status
|
||||
}
|
||||
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:
|
||||
self._showTrace("该用户无法加载更多预约记录")
|
||||
break
|
||||
except:
|
||||
self._showTrace("加载更多预约记录失败 !")
|
||||
break
|
||||
return None
|
||||
|
||||
|
||||
def canReserve(
|
||||
self,
|
||||
date: str
|
||||
) -> bool:
|
||||
|
||||
# no reserved or using record in the given date
|
||||
# then can reserve
|
||||
if self.__getReserveRecord(date, "已预约") is None:
|
||||
if self.__getReserveRecord(date, "使用中") is None:
|
||||
self._showTrace(f"用户在日期 {date} 可以预约")
|
||||
return True
|
||||
self._showTrace(f"用户在日期 {date} 有使用中的预约, 无法预约")
|
||||
self._showTrace(f"用户在日期 {date} 已存在有效预约, 无法预约")
|
||||
return False
|
||||
|
||||
|
||||
def canCheckin(
|
||||
self,
|
||||
date: str
|
||||
) -> bool:
|
||||
|
||||
# have a reserved record in the given date
|
||||
record = self.__getReserveRecord(date, "已预约")
|
||||
if record is not None:
|
||||
time_match = re.search(r"(\d{1,2}:\d{2})", record["time_str"])
|
||||
if time_match:
|
||||
begin_time = time_match.group(0)
|
||||
begin_time = datetime.strptime(f"{date} {begin_time}", "%Y-%m-%d %H:%M")
|
||||
time_diff = datetime.now() - begin_time
|
||||
time_diff_seconds = time_diff.total_seconds()
|
||||
# before 30 minutes, cant checkin
|
||||
if time_diff_seconds < -30*60:
|
||||
self._showTrace(
|
||||
f"用户在日期 {date} 的预约开始时间为 {begin_time}, "
|
||||
f"距离当前时间还有 {abs(time_diff_seconds)/60:.2f} 分钟, 无法签到"
|
||||
)
|
||||
return False
|
||||
# before in 30 minutes, can checkin
|
||||
elif -30*60 <= time_diff_seconds < 0:
|
||||
self._showTrace(
|
||||
f"用户在日期 {date} 的预约开始时间为 {begin_time}, "
|
||||
f"距离当前时间还有 {abs(time_diff_seconds)/60:.2f} 分钟, 可以签到"
|
||||
)
|
||||
return True
|
||||
# past less than 30 minutes, can checkin
|
||||
elif 0 <= time_diff_seconds < 30*60:
|
||||
self._showTrace(
|
||||
f"用户在日期 {date} 的预约开始时间为 {begin_time}, "
|
||||
f"当前时间已经 {abs(time_diff_seconds)/60:.2f} 分钟, 可以签到"
|
||||
)
|
||||
return True
|
||||
else:
|
||||
self._showTrace(f"用户在日期 {date} 的预约时间格式错误, 无法签到")
|
||||
self._showTrace(f"用户在日期 {date} 有没有有效预约记录, 无法签到")
|
||||
return False
|
||||
Reference in New Issue
Block a user