import os import time import random from datetime import datetime from typing import List, Dict, Callable from vs_types import BookerStandaloneConfig, QueryWaitMode, VSPlgConfig, VSQueryResult, VSBookResult from vs_plg_factory import VSPlgFactory from utils.safe_redis_cli import SafeRedisClient from toolkit.vs_cloud_api import VSCloudApi from toolkit.thread_pool import ThreadPool class BookerStandalone: def __init__(self, config: BookerStandaloneConfig, redis_conf: Dict, logger: Callable[[str], None] = None): self.m_logger = logger self.m_factory = VSPlgFactory() self.m_cfg = config self.redis_client = SafeRedisClient(redis_conf, self.m_logger) self.m_instance = None self.m_running = False self.m_next_query_time = 0.0 self.m_last_login_time = 0.0 self.m_current_dates = {} def _log(self, message): if self.m_logger: self.m_logger(f'[BOOKER] {message}') else: print(f'[BOOKER] {message}') def _get_wait_interval(self) -> float: """根据配置计算下一次查询的等待时间""" if self.m_cfg.query_wait.mode == QueryWaitMode.Loop: return 1.0 elif self.m_cfg.query_wait.mode == QueryWaitMode.Fixed: return float(self.m_cfg.query_wait.fixed_wait) elif self.m_cfg.query_wait.mode == QueryWaitMode.Random: return random.uniform(self.m_cfg.query_wait.random_min, self.m_cfg.query_wait.random_max) return 30.0 def _is_within_active_hours(self) -> bool: """ 判断当前是否在允许创建实例的时间段内 """ start_str = self.m_cfg.active_time_start end_str = self.m_cfg.active_time_end current_bj_time = datetime.utcnow().time() start_time = datetime.strptime(start_str, "%H:%M").time() end_time = datetime.strptime(end_str, "%H:%M").time() return start_time <= current_bj_time <= end_time def update_config(self, new_cfg: BookerStandaloneConfig): """ 动态更新配置 """ pass def _notify_date_changes(self, query_result: VSQueryResult): current_earliest_date = query_result.earliest_date apt_type = query_result.apt_type if current_earliest_date: last_date = self.m_current_dates.get(apt_type.routing_key) if last_date != current_earliest_date: self.m_current_dates[apt_type.routing_key] = current_earliest_date def _push_to_wx(): try: push_content = ( f"📢【档期变化通知】\n" f"最早日期: {query_result.earliest_date}\n" f"目标国家: {apt_type.country}\n" f"递交城市: {apt_type.city}\n" f"签证类型: {apt_type.visa_type}\n" f"Routing: {apt_type.routing_key}" ) VSCloudApi.Instance().push_weixin_text(push_content) except Exception as e: self._log(f"Failed to notify to cloud: {e}") ThreadPool.getInstance().enqueue(_push_to_wx) def _notify_book_result(self, book_result: VSBookResult): if book_result.success: def _update_cloud_success(): try: push_content = ( f"🎉 【预定成功通知】\n" f"━━━━━━━━━━━━━━━\n" f"预约账号: {book_result.account}\n" f"预约日期: {book_result.book_date}\n" f"预约时间: {book_result.book_time}\n" f"预约编号: {book_result.urn}\n" f"支付链接: {book_result.payment_link if book_result.payment_link else '无需支付/暂无'}\n" f"━━━━━━━━━━━━━━━\n" ) VSCloudApi.Instance().push_weixin_text(push_content) except Exception as e: self._log(f"Failed to update success state to cloud: {e}") ThreadPool.getInstance().enqueue(_update_cloud_success) def _countdown_wait(self, seconds: float, wait_for='') -> bool: """倒计时等待，动态刷新同一行，返回 False 表示被 stop 中断""" end = time.time() + seconds last_remaining = None while self.m_running and time.time() < end: remaining = int(end - time.time()) if remaining != last_remaining: print(f"[BOOKER] Next {wait_for} in {remaining} seconds... (stop to exit)", end='\r', flush=True) last_remaining = remaining time.sleep(1) print() return self.m_running def _init_instance(self): """初始化单实例并进行登录创建会话""" if self.m_instance is not None: self.m_instance.cleanup() now = time.time() if now < self.m_last_login_time + self.m_cfg.login_interval: wait = self.m_last_login_time + self.m_cfg.login_interval - now self._countdown_wait(wait, wait_for='init instance') plugin_name = self.m_cfg.plugin_config.plugin_name class_name = "".join(part.title() for part in plugin_name.split('_')) plugin_path = os.path.join(self.m_cfg.plugin_config.lib_path, self.m_cfg.plugin_config.plugin_bin) self.m_factory.register_plugin(plugin_name, plugin_path, class_name) plg_cfg = VSPlgConfig() plg_cfg.debug = self.m_cfg.debug plg_cfg.account = self.m_cfg.account proxy = VSCloudApi.Instance().get_next_proxy(self.m_cfg.proxy_pool, proxy_cd=600) plg_cfg.proxy = type(plg_cfg.proxy)(**proxy) plg_cfg.free_config = self.m_cfg.free_config plg_cfg.session_max_life = self.m_cfg.session_max_life self.m_instance = self.m_factory.create("single_task", plugin_name) self.m_instance.set_log(self.m_logger) self.m_instance.set_config(plg_cfg) self.m_instance.create_session() self.m_last_login_time = time.time() self.m_next_query_time = time.time() + self._get_wait_interval() self._log("Session created successfully.") def start(self): """ 单线程无限循环：心跳保活 -> 时间判定 -> 查询 -> 预定 """ self.m_running = True self._log("Auto Booker Started.") while self.m_running: if not self._is_within_active_hours(): continue try: self._init_instance() break except Exception as e: self._log(f"Failed to create session: {e}. Retrying in 10s...") time.sleep(10) while self.m_running: try: now = time.time() if not self._is_within_active_hours(): continue if not self.m_instance.health_check(): self._log("Health check failed. Session dead. Recreating session...") self._init_instance() continue if now < self.m_next_query_time: wait = self.m_next_query_time - now self._countdown_wait(wait, wait_for="query slot") apt_types = self.m_cfg.appointment_types weights = [float(t.weight) for t in apt_types] apt_type = random.choices(apt_types, weights=weights, k=1)[0] self._log(f"Querying slots for {apt_type.routing_key}...") query_result = self.m_instance.query(apt_type) query_result.apt_type = apt_type self.m_next_query_time = time.time() + self._get_wait_interval() self._notify_date_changes(query_result) if query_result.success: self._log("🔥 SLOT FOUND! Initiating AUTO-BOOKING...") # 5. 执行预定 book_result = self.m_instance.book(query_result, self.m_cfg.user_preferences) self._notify_book_result(book_result) if book_result.success: self._log(f"🎉 BOOKING SUCCESSFUL for {apt_type.routing_key}!") break except Exception as e: self._log(f"Loop Exception: {e}") def stop(self): """外部中断时调用""" self.m_running = False