import time import json import random import re import os import uuid import shutil import socket from datetime import datetime from typing import List, Dict, Optional, Any, Callable from urllib.parse import urljoin, urlparse, urlencode, parse_qs from concurrent.futures import ThreadPoolExecutor from DrissionPage import ChromiumPage, ChromiumOptions import configure from vs_plg import IVSPlg from vs_types import VSPlgConfig, AppointmentType, VSQueryResult, VSBookResult, AvailabilityStatus, TimeSlot, DateAvailability, NotFoundError, PermissionDeniedError, RateLimiteddError, SessionExpiredOrInvalidError, BizLogicError from utils.cloudflare_bypass_for_scraping import CloudflareBypasser from toolkit.mihomo_tunnel import MihomoTunnel from utils.mouse import HumanMouse from utils.keyboard import HumanKeyboard from utils.fingerprint_utils import FingerprintGenerator class BrowserResponse: """模拟 requests.Response""" def __init__(self, result_dict): result_dict = result_dict or {} self.status_code = result_dict.get('status', 0) self.text = result_dict.get('body', '') self.headers = result_dict.get('headers', {}) self.url = result_dict.get('url', '') self._json = None def json(self): if self._json is None: if not self.text: return {} try: self._json = json.loads(self.text) except: self._json = {} return self._json class TlsPlugin(IVSPlg): """ TLSContact 签证预约插件 (DrissionPage 版) """ def __init__(self, group_id: str): self.group_id = group_id self.config: Optional[VSPlgConfig] = None self.free_config: Dict[str, Any] = {} self.is_healthy = True self.logger = None self.mouse = None self.keyboard = None self.page: Optional[ChromiumPage] = None self.travel_group: Optional[Dict] = None self.instance_id = uuid.uuid4().hex[:8] self.root_workspace = os.path.abspath(os.path.join("data/temp_browser_data", f"{self.group_id}.{self.instance_id}")) self.user_data_path = os.path.join(self.root_workspace, "user_data") if not os.path.exists(self.root_workspace): os.makedirs(self.root_workspace) self.tunnel = None self.session_create_time: float = 0 def get_group_id(self) -> str: return self.group_id def set_log(self, logger: Callable[[str], None]): self.logger = logger def _log(self, message): if self.logger: self.logger(f'[TlsPlugin] [{self.group_id}] {message}') else: print(f'[TlsPlugin] [{self.group_id}] {message}') def set_config(self, config: VSPlgConfig): self.config = config self.free_config = config.free_config or {} def keep_alive(self): try: self.page.refresh() self.page.wait.load_start(timeout=2) self.page.wait.doc_loaded() time.sleep(random.uniform(1, 3)) self._check_page_is_session_expired_or_invalid('Book your appointment', html = self.page.html) self.simulate_random_human_clicks() except SessionExpiredOrInvalidError as e: self.is_healthy = False except Exception as e: self._log(f"Unexpected error in keep_alive: {e}") def simulate_random_human_clicks(self, min_x=300, max_x=800, min_y=400, max_y=600, min_clicks=1, max_clicks=2): """ 在指定区域内模拟人类随机移动鼠标并点击数次。 :param min_x: X坐标最小范围 :param max_x: X坐标最大范围 :param min_y: Y坐标最小范围 :param max_y: Y坐标最大范围 :param min_clicks: 随便点击的最少次数 :param max_clicks: 随便点击的最多次数 """ click_count = random.randint(min_clicks, max_clicks) self._log(f"Starting random human simulation: will click {click_count} times in the area.") for i in range(click_count): rand_x = random.randint(min_x, max_x) rand_y = random.randint(min_y, max_y) self._log(f"[{i+1}/{click_count}] Moving mouse to ({rand_x}, {rand_y}) and clicking") self.mouse.click(rand_x, rand_y, humanize=True) if i < click_count - 1: sleep_time = random.uniform(0.5, 1.8) self._log(f"Resting for {sleep_time:.2f} seconds before next click...") time.sleep(sleep_time) self._log("Random human clicks simulation completed.") def health_check(self) -> bool: if not self.is_healthy: return False if self.page is None: return False try: if not self.page.run_js("return 1;"): return False except: return False if self.config.session_max_life > 0: current_time = time.time() elapsed_time = current_time - self.session_create_time if elapsed_time > self.config.session_max_life: self._log(f"Session expired.") return False return True def _save_screenshot(self, name_prefix): try: timestamp = int(time.time()) filename = f"{self.instance_id}_{name_prefix}_{timestamp}.jpg" save_path = os.path.join("data", filename) os.makedirs("data", exist_ok=True) self.page.get_screenshot(path=save_path, full_page=False) self._log(f"Screenshot saved to {save_path}") except Exception as e: self._log(f"Failed to save screenshot: {e}") def create_session(self): """ 全浏览器会话创建：过盾 -> JS注入登录 -> 状态机自动路由导航 -> 到达目标页 """ self._log(f"Initializing Session (ID: {self.instance_id})...") captcha_future = None captcha_executor = ThreadPoolExecutor(max_workers=1) login_captcha_cfg = self.free_config.get("login_captcha", {}) if login_captcha_cfg.get('solve_advance'): login_page = login_captcha_cfg.get("page_url") site_key = login_captcha_cfg.get("site_key") task_type = login_captcha_cfg.get("task") self._log(f"🚀 Early starting background Captcha solve for sitekey={site_key}") rc_params = { "type": task_type, "page": login_page, "siteKey": site_key, "apiToken": self.free_config.get("capsolver_key", "") } captcha_future = captcha_executor.submit(self._solve_recaptcha, rc_params) co = ChromiumOptions() def get_free_port(): with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s: s.bind(('', 0)) return s.getsockname()[1] debug_port = get_free_port() self._log(f"Assigned Debug Port: {debug_port}") co.set_local_port(debug_port) co.set_user_data_path(self.user_data_path) chrome_path = configure.CHROME_PATH if not chrome_path: chrome_path = os.getenv("CHROME_BIN") if chrome_path and os.path.exists(chrome_path): co.set_paths(browser_path=chrome_path) if self.config.proxy and self.config.proxy.ip: p = self.config.proxy self._log(f'Current proxy id={p.id}') if p.username and p.password: self._log(f"Starting Proxy Tunnel for {p.ip}...") exit_node = { "name": "ExitNode", "type": p.proto, "server": p.ip, "port": p.port, "username": p.username, "password": p.password } relay_node = None if configure.MIHOMO_RELAY_NODES: relay_node = random.choice(configure.MIHOMO_RELAY_NODES) mihomo_path = configure.MIHOMO_BIN_PATH if not mihomo_path: mihomo_path = os.getenv("MIHOMO_BIN") if not mihomo_path: raise BizLogicError(message='Mihomo path is null, You need set mihomo bin path in configure or os env') self.tunnel = MihomoTunnel(mihomo_path, exit_node=exit_node, relay_node=relay_node) local_proxy = self.tunnel.start() self._log(f"Tunnel started at {local_proxy}") co.set_argument(f'--proxy-server={local_proxy}') else: proxy_str = f"{p.proto}://{p.ip}:{p.port}" co.set_argument(f'--proxy-server={proxy_str}') else: self._log("[WARN] No proxy configured!") specific_fp = FingerprintGenerator().generate(self.config.account.username) fp_seed = specific_fp.get("seed") fp_platform = specific_fp.get("platform") fp_brand = specific_fp.get("brand") self._log(f'browser fingerprint seed={fp_seed}') co.headless(False) co.set_argument('--no-sandbox') co.set_argument('--disable-dev-shm-usage') co.set_argument('--window-size=1920,1080') co.set_argument('--disable-blink-features=AutomationControlled') # co.set_argument('--ignore-gpu-blocklist') # co.set_argument('--enable-webgl') # co.set_argument('--use-gl=angle') # co.set_argument('--use-angle=swiftshader') co.set_argument(f"--fingerprint={fp_seed}") co.set_argument(f"--fingerprint-platform={fp_platform}") co.set_argument(f"--fingerprint-brand={fp_brand}") try: self.page = ChromiumPage(co) # --- 预检指纹信息 --- if self.config.debug: self.page.get('https://example.com') js_script = """ function getFingerprint() { let webglVendor = 'Unknown'; let webglRenderer = 'Unknown'; try { let canvas = document.createElement('canvas'); let gl = canvas.getContext('webgl') || canvas.getContext('experimental-webgl'); if (gl) { let debugInfo = gl.getExtension('WEBGL_debug_renderer_info'); if (debugInfo) { webglVendor = gl.getParameter(debugInfo.UNMASKED_VENDOR_WEBGL); webglRenderer = gl.getParameter(debugInfo.UNMASKED_RENDERER_WEBGL); } } } catch(e) {} return { "User-Agent": navigator.userAgent, "Platform": navigator.userAgentData ? navigator.userAgentData.platform : navigator.platform, "Brands": navigator.userAgentData ? navigator.userAgentData.brands.map(b => b.brand).join(', ') : 'Not Supported', "CPU Cores": navigator.hardwareConcurrency, "Language": navigator.language, "Timezone": Intl.DateTimeFormat().resolvedOptions().timeZone, "WebGL Vendor": webglVendor, "WebGL Renderer": webglRenderer }; } return getFingerprint(); """ fp_data = self.page.run_js(js_script) self._log("================ 预检浏览器指纹数据 ================") self._log(json.dumps(fp_data, indent=4, ensure_ascii=False)) self._log("====================================================") # --- 初始化访问与过盾 --- tls_url = self.free_config.get('tls_url', '') self._log(f"Navigating: {tls_url}") self.page.get(tls_url) time.sleep(5) cf_bypasser = CloudflareBypasser(self.page, log=self.config.debug) if not cf_bypasser.bypass(max_retry=6): raise BizLogicError("Cloudflare bypass timeout") time.sleep(3) cf_bypasser.handle_waiting_room() # --- 初始化人类行为模拟工具 --- self._log("Init humanize tools...") self.mouse = HumanMouse(self.page, debug=self.config.debug) self.keyboard = HumanKeyboard(self.page) viewport_width = self.page.rect.viewport_size[0] viewport_height = self.page.rect.viewport_size[1] init_x = random.randint(10, viewport_width - 10) init_y = random.randint(10, viewport_height - 10) self.mouse.move(init_x, init_y) max_steps = 10 session_created = False has_submitted_login = False for step in range(max_steps): self.page.wait.load_start() current_url = self.page.url self._log(f"--- [Router Step {step+1}] Current URL: {current_url} ---") cloudflare_blocked_indicators = [ "Sorry, you have been blocked" in self.page.html, "You are being rate limited" in self.page.html, "Cloudflare Ray ID" in self.page.html ] if any(cloudflare_blocked_indicators): raise BizLogicError(message="Blocked by Cloudflare WAF. Need to change IP or browser fingerprint.") # 状态 1：到达终极目标页面 (成功退出条件) if "appointment-booking" in current_url or self.page.ele('tag:button@text():Book your appointment', timeout=1): btn_selector = 'tag:button@text():Book your appointment' if self.page.wait.ele_displayed(btn_selector, timeout=10): self.session_create_time = time.time() self._log("✅ Login & Navigation Success! Reached appointment-booking.") session_created = True break # 状态 2：遇到没有申请人的拦截页 (致命错误退出条件) no_applicant_indicators = [ "Add a new applicant" in self.page.html, "You have not yet added an applicant" in self.page.html, "applicants-information" in current_url ] if any(no_applicant_indicators): raise BizLogicError(message="No applicant added. Cannot proceed to booking.") if current_url == tls_url: # 状态 3：首页/登录入口页 -> 需要点击进入登录 if self.page.ele("tag:a@@href:login", timeout=1) and not self.page.ele('tag:label@@text():Email', timeout=1): self._log("State: Login Portal. Clicking login link...") login_link = self.page.ele("tag:a@@href:login") self.mouse.human_click_ele(login_link) time.sleep(3) continue if self.page.ele("tag:svg@@data-testid=user-button", timeout=1): self._log("State: Already login, logout now...") user_btn = self.page.ele("tag:svg@@data-testid=user-button") self.mouse.human_click_ele(user_btn) time.sleep(1.5) logout_btn = self.page.ele("#logout") self.mouse.human_click_ele(logout_btn) time.sleep(1.5) self.page.get(tls_url) time.sleep(3) continue # 状态 4：真正的登录表单页 if self.page.ele('tag:label@@text():Email', timeout=1) and not has_submitted_login: self._log("State: Login Form. Processing credentials and Captcha...") recaptchav2_token = "" if not captcha_future and (self.page.ele('.g-recaptcha') or self.page.ele('xpath://iframe[contains(@src, "recaptcha")]')): rec_iframe = self.page.ele('xpath://iframe[contains(@src, "recaptcha")]') rec_iframe_src = rec_iframe.attr('src') rec_parsed = urlparse(rec_iframe_src) rec_params = parse_qs(rec_parsed.query) rec_sitekey = rec_params.get("k", [None])[0] rec_size = rec_params.get("size", [None])[0] if 'normal' == rec_size: self._log(f"Found dynamic sitekey={rec_sitekey}. Starting async Captcha solver...") rc_params = { "type": "ReCaptchaV2TaskProxyLess", "page": current_url, "siteKey": rec_sitekey, "apiToken": self.free_config.get("capsolver_key", "") } captcha_future = captcha_executor.submit(self._solve_recaptcha, rc_params) username = self.config.account.username password = self.config.account.password input_ele = self.page.ele('tag:label@@text():Email').next() self.mouse.human_click_ele(input_ele) time.sleep(random.uniform(0.2, 0.6)) self.keyboard.type_text(username, humanize=True) time.sleep(random.uniform(0.5, 1.2)) input_ele = self.page.ele('tag:label@@text():Password').next() self.mouse.human_click_ele(input_ele) time.sleep(random.uniform(0.2, 0.6)) self.keyboard.type_text(password, humanize=True) # 注入 Token if captcha_future: self._log("Waiting for background Captcha result...") try: # 设一个合理的超时，防止死锁 recaptchav2_token = captcha_future.result(timeout=120) self._log("Background Captcha solved successfully!") except Exception as e: raise BizLogicError(f"Captcha solving failed or timed out: {e}") # 注入 Token if recaptchav2_token: inject_js = f"var g = document.getElementById('g-recaptcha-response'); if(g) {{ g.value = '{recaptchav2_token}'; }}" self.page.run_js(inject_js) time.sleep(random.uniform(0.5, 1.0)) self._log("Submitting Login...") login_btn = self.page.ele('tag:button@@text():Login') self.mouse.human_click_ele(login_btn) has_submitted_login = True time.sleep(3) continue # 状态 5：Travel Groups 页面 if "travel-groups" in current_url: self._log("State: Travel Groups. Selecting targeted group...") groups = self._parse_travel_groups(self.page.html) location = self.free_config.get('location') self.travel_group = next((g for g in groups if location in g['location']), None) if not self.travel_group or not self.travel_group.get("submitted"): self._save_screenshot("group_not_found") raise NotFoundError(f"Group not found for {location}") formgroup_id = self.travel_group.get('group_number') btn_selector = f'tag:button@@name=formGroupId@@value={formgroup_id}' if self.page.wait.eles_loaded(btn_selector, timeout=10): buttons = self.page.eles(btn_selector) select_btn = next((btn for btn in reversed(buttons) if btn.rect.size[0] > 0 and btn.rect.size[1] > 0), None) if select_btn: time.sleep(random.uniform(0.5, 1.2)) self.mouse.human_click_ele(select_btn) time.sleep(3) continue else: self._log("[WARN] Select button found but not visible.") else: self._log(f"[WARN] Wait timeout for group button {formgroup_id}") # 状态 6：中间过渡页，需点击 "Book Appointment" 继续往下走 if self.page.ele('#book-appointment-btn', timeout=1): self._log("State: Intermediate Dashboard. Clicking Book Appointment button...") self.mouse.human_click_ele(self.page.ele('#book-appointment-btn')) time.sleep(3) continue # 状态 7：登录失败校验 或 未知加载状态 if "login-actions" in current_url and has_submitted_login: self._log("Waiting on login-actions... (Might be authenticating or invalid credentials)") time.sleep(2) if self.page.ele('text:Invalid username or password', timeout=1): # 假设网页上有错误提示 raise BizLogicError(message="Login Failed! Invalid credentials or Captcha rejected.") continue self._log("State: Transitioning or Unknown. Waiting 2 seconds...") time.sleep(2) if not session_created: raise BizLogicError(f"Failed to reach appointment-booking after {max_steps} navigation steps. Stuck at: {self.page.url}") except Exception as e: self._log(f"Session Create Error: {e}") if self.config.debug: self._save_screenshot("create_session_except") self.cleanup() raise e def query(self, apt_type: AppointmentType) -> VSQueryResult: res = VSQueryResult() res.success = False slots = [] self._log(f"Executing silent JS fetch...") resp = self._perform_request("GET", self.page.url, retry_count=1) self._check_page_is_session_expired_or_invalid('Book your appointment', resp.text) slots = self._parse_appointment_slots(resp.text) if slots: res.success = True earliest_date = slots[0]["date"] earliest_dt = datetime.strptime(earliest_date, "%Y-%m-%d") res.availability_status = AvailabilityStatus.Available res.earliest_date = earliest_dt date_map: dict[datetime, list[TimeSlot]] = {} for s in slots: date_str = s["date"] dt = datetime.strptime(date_str, "%Y-%m-%d") date_map.setdefault(dt, []).append( TimeSlot(time=s["time"], label=str(s.get("label", ""))) ) res.availability = [DateAvailability(date=d, times=slots) for d, slots in date_map.items()] self._log(f"Slot Found! -> {slots}") else: self._log("No slots available.") res.success = False res.availability_status = AvailabilityStatus.NoneAvailable # TODO(TEST): 临时测试预约提交 if configure.TLS_TEST_BOOK_AFTER_QUERY: test_date = "2026-06-10" test_time = "09:00" test_label = "" test_dt = datetime.strptime(test_date, "%Y-%m-%d") query_res = VSQueryResult() query_res.success = True query_res.availability_status = AvailabilityStatus.Available query_res.earliest_date = test_dt query_res.availability = [ DateAvailability( date=test_dt, times=[TimeSlot(time=test_time, label=test_label)] ) ] self._log(f"[TEST] using fixed June slot: {test_date} {test_time} {test_label}") test_userinput = { "support_pta": False, "expected_end_date": "2100-01-01", "expected_start_date": "2000-01-01" } try: self.book(query_res, test_userinput) except Exception as e: self._log(f"[TEST] book() after query failed: {e}") self.is_healthy = False return res def book_bak(self, slot_info: VSQueryResult, user_inputs: Dict = None) -> VSBookResult: res = VSBookResult() res.success = False exp_start = user_inputs.get('expected_start_date', '') exp_end = user_inputs.get('expected_end_date', '') support_pta = user_inputs.get('support_pta', True) target_labels = [''] if support_pta: target_labels.append('pta') available_dates_str =[ da.date.strftime("%Y-%m-%d") for da in slot_info.availability if da.date ] valid_dates_list = self._filter_dates(available_dates_str, exp_start, exp_end) if not valid_dates_list: raise NotFoundError(message="No dates match user constraints") all_possible_slots =[] for da in slot_info.availability: if not da.date: continue date_str = da.date.strftime("%Y-%m-%d") if date_str in valid_dates_list: for t in da.times: if t.label in target_labels: all_possible_slots.append({ "date": date_str, "time_obj": t, "label": t.label }) if not all_possible_slots: raise NotFoundError(message="No suitable slot found (after label filtering)") selected_slot = random.choice(all_possible_slots) selected_date = selected_slot["date"] selected_time = selected_slot["time_obj"] selected_label = selected_slot["label"] self._log(f"Found {len(all_possible_slots)} valid slots. selected slot: {selected_date} {selected_time.time} {selected_label}") self.page.listen.start('/workflow/appointment-booking', method='POST') page_url = self.page.url location = self.travel_group.get('location') mapper = { "London": "gbLON2fr", "Dublin": "ieDUB2fr" } location_id = mapper.get(location) form_group_id =self.travel_group.get('group_number') api_token = self.free_config.get("capsolver_key", "") rc_params = { "type": "ReCaptchaV3Task", "page": page_url, "action": "book", "siteKey": "6LcTpXcfAAAAAM3VojNhyV-F1z92ADJIvcSZ39Y9", "apiToken": api_token, "proxy": True } g_token = self._solve_recaptcha(rc_params) ACTION_ID = "6033ac5e6e4ac04f59a4b74a9c5dd312876dd46bd9" js_script = f""" (async function() {{ const url = "{page_url}"; const groupId = "{form_group_id}"; const locationId = "{location_id}"; const selectedDate = "{selected_date}"; const selectedTime = "{selected_time.time}"; const selectedLabel = "{selected_label}"; const captchaToken = "{g_token}"; const actionId = "{ACTION_ID}"; let castleToken = ''; try {{ const castleModule = window.__webpack_require__(93773); if (castleModule && castleModule.createRequestToken) {{ castleToken = await castleModule.createRequestToken(); console.log('Castle token obtained:', castleToken); }} }} catch(e) {{ console.error('Failed to get Castle token:', e); }} let routerStateTree = ''; try {{ const routerModule = window.__webpack_require__(11807); const routerState = routerModule.getCurrentAppRouterState(); if (routerState && routerState.tree) {{ const prepareModule = window.__webpack_require__(16378); routerStateTree = prepareModule.prepareFlightRouterStateForRequest(routerState.tree); console.log('Router state tree obtained'); }} }} catch(e) {{ console.error('Failed to get router state:', e); }} const formData = new FormData(); formData.append('1_formGroupId', groupId); formData.append('1_lang', 'en-us'); formData.append('1_process', 'APPOINTMENT'); formData.append('1_location', locationId); formData.append('1_date', selectedDate); formData.append('1_time', selectedTime); formData.append('1_appointmentLabel', selectedLabel); formData.append('1_castleRequestToken', castleToken); formData.append('1_captchaToken', captchaToken); formData.append('0', '[{{"status":"IDLE"}},"$K1"]'); const headers = {{ 'Accept': 'text/x-component', 'Next-Action': actionId, 'Next-Router-State-Tree': routerStateTree, }}; const response = await fetch(url, {{ method: 'POST', headers: headers, body: formData, credentials: 'include' }}); const text = await response.text(); const responseHeaders = {{}}; response.headers.forEach((value, key) => {{ responseHeaders[key] = value; }}); const result = {{ status: response.status, body: text, headers: responseHeaders, url: response.url, redirected: response.redirected, ok: response.ok }}; return result; }})(); """ self._log("Submitting booking request via JS Fetch...") self.page.run_js(js_script) packet = self.page.listen.wait(timeout=15) if not packet: raise BizLogicError(message='Listening data failed') self.page.listen.stop() self._log(f"URL: {packet.url}") self._log(f"POST Body: {packet.request.postData}") self._log(f"POST Stat: {packet.response.status}") self._log(f"POST head: {packet.response.headers}") self._log(f"POST Resp: {packet.response.raw_body}") redirect_location = packet.response.headers.get('location', '') or '' appointment_confirmation_indicators = [ "order-summary" in redirect_location, "partner-services" in redirect_location, "appointment-confirmation" in redirect_location, ] if any(appointment_confirmation_indicators): self._log(f"Booking Success!") res.success = True res.book_date = selected_date res.book_time = selected_time.time return res return res def book(self, slot_info: VSQueryResult, user_inputs: Dict = None) -> VSBookResult: res = VSBookResult() res.success = False exp_start = user_inputs.get('expected_start_date', '') exp_end = user_inputs.get('expected_end_date', '') support_pta = user_inputs.get('support_pta', True) target_labels = [''] if support_pta: target_labels.append('pta') available_dates_str =[ da.date.strftime("%Y-%m-%d") for da in slot_info.availability if da.date ] valid_dates_list = self._filter_dates(available_dates_str, exp_start, exp_end) if not valid_dates_list: raise NotFoundError(message="No dates match user constraints") all_possible_slots =[] for da in slot_info.availability: if not da.date: continue date_str = da.date.strftime("%Y-%m-%d") if date_str in valid_dates_list: for t in da.times: if t.label in target_labels: all_possible_slots.append({ "date": date_str, "time_obj": t, "label": t.label }) if not all_possible_slots: raise NotFoundError(message="No suitable slot found (after label filtering)") selected_slot = random.choice(all_possible_slots) selected_date = selected_slot["date"] selected_time = selected_slot["time_obj"] selected_label = selected_slot["label"] self._log(f"Found {len(all_possible_slots)} valid slots. selected slot: {selected_date} {selected_time.time} {selected_label}") self.page.listen.start('/workflow/appointment-booking', method='POST') js_update_form = f""" try {{ const buttons = Array.from(document.querySelectorAll('button[type="submit"]')); const submitBtn = buttons.find(btn => {{ return btn.textContent.trim().toLowerCase().includes('book your appointment'); }}); if (!submitBtn) return 'Submit button not found'; const form = submitBtn.closest('form'); if (!form) return 'Correct form not found'; function setReactValue(input, value) {{ if (!input) return; input.value = value; }} setReactValue(form.querySelector('input[name="date"]'), '{selected_date}'); setReactValue(form.querySelector('input[name="time"]'), '{selected_time.time}'); setReactValue(form.querySelector('input[name="appointmentLabel"]'), '{selected_label}'); submitBtn.removeAttribute('disabled'); submitBtn.classList.remove('opacity-50', 'cursor-not-allowed'); return 'form_updated'; }} catch (e) {{ return e.toString(); }} """ update_res = self.page.run_js(js_update_form) self._log(f"Form update triggered: {update_res}") if update_res != 'form_updated': raise BizLogicError(message=f"Failed to update form: {update_res}") submit_btn = self.page.ele('tag:button@@type=submit@@text():Book your appointment') if not submit_btn: raise BizLogicError(message="Submit button not found for mouse click") self._log("Moving mouse to submit button and clicking") self.mouse.human_click_ele(submit_btn) packet = self.page.listen.wait(timeout=10) if not packet: raise BizLogicError(message='Listening data failed') self.page.listen.stop() self._log(f"URL: {packet.url}") self._log(f"POST Body: {packet.request.postData}") self._log(f"POST Resp: {packet.response.body}") self._log("Waiting for Next.js to process the form submission...") for _ in range(10): try: current_page_url = self.page.url current_page_html = self.page.html appointment_confirmation_indicators = [ "order-summary" in current_page_url, "partner-services" in current_page_url, "appointment-confirmation" in current_page_url, "Change my appointment" in current_page_html, "Book a new appointment" in current_page_html, ] if any(appointment_confirmation_indicators): self._log(f"✅ BOOKING SUCCESS! Redirected to: {current_page_url}") res.success = True res.label = selected_label res.book_date = selected_date res.book_time = selected_time.time self._save_screenshot("book_slot_success") break toast_selector = 'tag:div@role=alert' toast_ele = self.page.ele(toast_selector, timeout=0.5) if toast_ele: error_msg = toast_ele.text self._log(f"❌ BOOKING FAILED! Detected popup: {error_msg}") break time.sleep(0.5) except Exception: pass return res def _perform_request(self, method, url, headers=None, data=None, json_data=None, params=None, retry_count=0): """ 在浏览器上下文中注入 JS 执行 Fetch """ if not self.page: raise BizLogicError("Browser not initialized") if params: from urllib.parse import urlencode if '?' in url: url += '&' + urlencode(params) else: url += '?' + urlencode(params) fetch_options = { "method": method.upper(), "headers": headers or {}, "credentials": "include" } # Body 处理 if json_data: fetch_options['body'] = json.dumps(json_data) fetch_options['headers']['Content-Type'] = 'application/json' elif data: if isinstance(data, dict): from urllib.parse import urlencode fetch_options['body'] = urlencode(data) fetch_options['headers']['Content-Type'] = 'application/x-www-form-urlencoded' else: fetch_options['body'] = data js_script = f""" const url = "{url}"; const options = {json.dumps(fetch_options)}; return fetch(url, options) .then(async response => {{ const text = await response.text(); const headers = {{}}; response.headers.forEach((value, key) => headers[key] = value); return {{ status: response.status, body: text, headers: headers, url: response.url }}; }}) .catch(error => {{ return {{ status: 0, body: error.toString(), headers: {{}}, url: url }}; }}); """ res_dict = self.page.run_js(js_script, timeout=30) resp = BrowserResponse(res_dict) if resp.status_code == 200: return resp elif resp.status_code == 401: self.is_healthy = False raise SessionExpiredOrInvalidError() elif resp.status_code == 403: if retry_count < 2: self._log(f"HTTP 403 Detected. Cloudflare session expired? Attempting refresh (Try {retry_count+1}/2)...") if self._refresh_firewall_session(): self._log("Firewall session refreshed. Retrying request...") return self._perform_request(method, url, headers, data, json_data, params, retry_count+1) else: self._log("Failed to refresh firewall session.") raise PermissionDeniedError(f"HTTP 403: {resp.text[:100]}") elif resp.status_code == 429: self.is_healthy = False raise RateLimiteddError() else: if resp.status_code == 0: raise BizLogicError(f"Network Error: {resp.text}") raise BizLogicError(message=f"HTTP Error {resp.status_code}: {resp.text[:100]}") def _refresh_firewall_session(self) -> bool: """ 主动刷新页面以触发 Cloudflare 挑战并尝试通过 """ try: self._log("Refreshing page to trigger Cloudflare...") self.page.refresh() cf = CloudflareBypasser(self.page, log=self.config.debug) success = cf.bypass(max_retry=6) if success: title = self.page.title.lower() if "access denied" in title: return False time.sleep(2) return True return False except Exception as e: self._log(f"Error during firewall refresh: {e}") return False def _solve_recaptcha(self, params) -> str: """调用 VSCloudApi 解决 ReCaptcha""" key = params.get("apiToken") if not key: raise NotFoundError("Api-token required") submit_url = "https://api.capsolver.com/createTask" task = { "type": params.get("type"), "websiteURL": params.get("page"), "websiteKey": params.get("siteKey"), } if params.get("action"): task["pageAction"] = params.get("action") if params.get("proxy"): p = self.config.proxy task["proxyType"] = p.proto task["proxyAddress"] = p.ip task["proxyPort"] = p.port if p.username: task["proxyLogin"] = p.username task["proxyPassword"] = p.password payload = {"clientKey": key, "task": task} import requests as req r = req.post(submit_url, json=payload, timeout=20) if r.status_code != 200: raise BizLogicError(message="Failed to submit capsolver task") task_id = r.json().get("taskId") for _ in range(20): r = req.post("https://api.capsolver.com/getTaskResult", json={"clientKey": key, "taskId": task_id}, timeout=20) if r.status_code == 200: d = r.json() if d.get("status") == "ready": return d["solution"]["gRecaptchaResponse"] time.sleep(3) raise BizLogicError(message="Capsolver task timeout") def _parse_travel_groups(self, html_content) -> List[Dict]: groups = [] js_pattern = r'\\"travelGroups\\":\s*(\[.*?\]),\\"availableCountriesToCreateGroups' js_match = re.search(js_pattern, html_content, re.DOTALL) if js_match: json_str = js_match.group(1).replace(r'\"', '"') data = json.loads(json_str) for g in data: groups.append({ 'group_name': g.get('groupName'), 'group_number': g.get('formGroupId'), 'location': g.get('vacName'), 'submitted': g.get('submitted') }) else: self._log('Parsed travel group page, but not found travelGroups') return groups def _parse_appointment_slots(self, html_content) -> List[Dict]: slots = [] pattern = r'"availableAppointments\\":\s*(\[.*\]),\\"showFlexiAppointment' match = re.search(pattern, html_content, re.DOTALL) if match: json_str = match.group(1).replace(r'\"', '"') data = json.loads(json_str) for day in data: d_str = day.get('day') for s in day.get('slots', []): labels = s.get('labels', []) lbl = None if 'pta' in labels: lbl = 'pta' elif 'ptaw' in labels: lbl = 'ptaw' elif '' in labels: lbl = '' if lbl is not None: slots.append({ 'date': d_str, 'time': s.get('time'), 'label': lbl }) return slots def _check_page_is_session_expired_or_invalid(self, keyword, html: str) -> bool: if not html: self.is_healthy = False raise SessionExpiredOrInvalidError() html_lower = html.lower() if keyword.lower() not in html_lower: session_expire_or_invalid_indicators = [ 'redirected automatically' in html_lower, 'login' in html_lower and 'password' in html_lower, 'session expired' in html_lower ] if any(session_expire_or_invalid_indicators): self.is_healthy = False raise SessionExpiredOrInvalidError() def _filter_dates(self, dates: List[str], start_str: str, end_str: str) -> List[str]: if not start_str or not end_str: return dates valid_dates = [] s_date = datetime.strptime(start_str[:10], "%Y-%m-%d") e_date = datetime.strptime(end_str[:10], "%Y-%m-%d") for date_str in dates: curr_date = datetime.strptime(date_str, "%Y-%m-%d") if s_date <= curr_date <= e_date: valid_dates.append(date_str) random.shuffle(valid_dates) return valid_dates # --- 资源清理核心方法 --- def cleanup(self): """ 销毁浏览器并彻底删除临时文件 """ if self.page: try: self.page.quit(force=True) except Exception: pass self.page = None if os.path.exists(self.root_workspace): for _ in range(3): try: time.sleep(0.2) shutil.rmtree(self.root_workspace, ignore_errors=True) break except Exception as e: self._log(f"Cleanup retry: {e}") time.sleep(0.5) if os.path.exists(self.root_workspace): self._log(f"[WARN] Failed to fully remove workspace: {self.root_workspace}") if self.tunnel: try: self.tunnel.stop() except: pass self.tunnel = None def __del__(self): """ 析构函数：当对象被垃圾回收时自动调用 """ self.cleanup()