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, MOUSE_PROFILES 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 set_log(self, logger: Callable[[str], None]): self.logger = logger def _log(self, message): if self.logger: self.logger(f'[TlsPlugin] [{self.group_id}] [{self.instance_id}] {message}') else: print(f'[TlsPlugin] [{self.group_id}] [{self.instance_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_mouse_move() 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_mouse_move(self, min_x=100, max_x=800, min_y=100, max_y=800, min_points=1, max_points=2): """ 在指定区域内模拟人类随机移动鼠标并点击数次。 :param min_x: X坐标最小范围 :param max_x: X坐标最大范围 :param min_y: Y坐标最小范围 :param max_y: Y坐标最大范围 :param min_point: 随便移动的最少次数 :param max_point: 随便移动的最多次数 """ move_cnt = random.randint(min_points, max_points) self._log(f"Starting random human simulation: will move {move_cnt} times in the area.") for i in range(move_cnt): rand_x = random.randint(min_x, max_x) rand_y = random.randint(min_y, max_y) self._log(f"[{i+1}/{move_cnt}] Moving mouse to ({rand_x}, {rand_y})") self.mouse.move(rand_x, rand_y, humanize=True) self._log("Random human move 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}") self._log(f"Account id={self.config.account.id}, proxy id={self.config.proxy.id}") 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 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}") 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) if 'Attention Required! | Cloudflare' in self.page.title and 'Sorry, you have been blocked' in self.page.html: self._log(f'Block by cloudflare, try refresh...') self.page.refresh() self.page.wait.load_start(timeout=2) self.page.wait.doc_loaded() 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...") profile_name = random.choice(list(MOUSE_PROFILES.keys())) self._log(f"[HumanMouse] current mouse profiles: {profile_name}") self.mouse = HumanMouse(self.page, timing=MOUSE_PROFILES[profile_name], 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 stuck_counter = 0 last_url = "" session_created = False has_submitted_login = False other_langs = ['fr-fr', 'ar-ar', 'cb-ph', 'id-id', 'km-kh', 'mk-mk', 'ru-ru', 'sq-al', 'th-th', 'tl-ph', 'uz-uz', 'vi-vn', 'zh-cn', 'hy-am', 'zh-hk'] for step in range(max_steps): self.page.wait.doc_loaded() time.sleep(1) current_url = self.page.url current_title = self.page.title.lower() current_html_content = self.page.html self._log(f"--- [Router Step {step+1}] Current URL: {current_url} ---") if current_url == last_url: stuck_counter += 1 else: last_url = current_url stuck_counter = 0 # 在某个页面卡住3轮以上,直接重试 if stuck_counter >= 3: self._log("[WARN] Page stucked, try to refresh...") self.page.refresh() self.page.wait.load_start(timeout=5) stuck_counter = 0 continue server_error_indicators = [ "502 Bad Gateway", "503 Service Temporarily Unavailable", ] # 网络出现故障,直接重试 if any(err in current_html_content for err in server_error_indicators): self._log(f"[WARN] Server network error, try to refresh (Step: {step})...") time.sleep(2) self.page.refresh() self.page.wait.load_start(timeout=5) continue # 特征1: 必须是

标签，且包含指定文案 (绝对不会匹配到