From 30db2fbf9a8cff3fc80522219e8eba03bdbfa777 Mon Sep 17 00:00:00 2001 From: KenanZhu <3471685733@qq.com> Date: Fri, 7 Nov 2025 22:48:46 +0800 Subject: [PATCH] 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'. --- LibChecker.py | 207 ++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 207 insertions(+) create mode 100644 LibChecker.py diff --git a/LibChecker.py b/LibChecker.py new file mode 100644 index 0000000..8349530 --- /dev/null +++ b/LibChecker.py @@ -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