import time import json import random import socket import uuid import shutil import re import os import base64 from concurrent.futures import ThreadPoolExecutor from datetime import datetime from typing import List, Dict, Optional, Any, Callable from urllib.parse import urlencode, urlparse 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 toolkit.mihomo_tunnel import MihomoTunnel from toolkit.vs_cloud_api import VSCloudApi from utils.mouse import HumanMouse from utils.keyboard import HumanKeyboard from utils.fingerprint_utils import FingerprintGenerator class BrowserResponse: 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 # ========================================== # 2. ItaPlugin 核心逻辑 # ========================================== class ItaPlugin(IVSPlg): 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.page: Optional[ChromiumPage] = None # Prenotami 特有配置 self._service_id = 0 self.ita_url = 'https://prenotami.esteri.it' 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]) -> None: self.logger = logger def _log(self, message): if self.logger: self.logger(f'[ItaPlugin] [{self.group_id}] {message}') else: print(f'[ItaPlugin] [{self.group_id}] {message}') def set_config(self, config: VSPlgConfig): self.config = config self.free_config = config.free_config or {} # Service ID (e.g., 1321 for Ireland, 5059 for Guangzhou) self._service_id = self.free_config.get('service_id', 0) def keep_alive(self): pass def health_check(self) -> bool: if not self.is_healthy or not self.page: return False try: if not self.page.run_js("return 1;"): return False except: return False if self.config.session_max_life > 0: if time.time() - self.session_create_time > self.config.session_max_life: self._log("Session expired.") return False return True def create_session(self): """ 全浏览器会话创建：过盾 -> JS注入登录 -> 状态机自动路由导航 -> 到达目标页 """ self._log(f"Initializing Session (ID: {self.instance_id})...") 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(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("====================================================") self.page = ChromiumPage(co) ita_url = self.ita_url self._log(f"Navigating to {ita_url}") self.page.get(ita_url) self._log("Init humanize tools...") self.mouse = HumanMouse(self.page, debug=True) self.keyboard = HumanKeyboard(self.page) self._log("Random mouse start position...") 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) switch_en_btn = self.page.ele('tag:a@@href=/Language/ChangeLanguage?lang=2') self.mouse.human_click_ele(switch_en_btn) self.page.wait.load_start() time.sleep(5) max_steps = 15 stuck_counter = 0 last_url = "" session_created = False has_submitted_login = False username = self.config.account.username password = self.config.account.password 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 if stuck_counter >= 3: self._log("[WARN] Page stucked, try to refresh...") self.page.refresh() self._random_sleep(5, 8) stuck_counter = 0 continue server_error_indicators = ["502 Bad Gateway", "503 Service Temporarily Unavailable", "error 1020"] 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})...") self.page.refresh() self._random_sleep(10, 15) continue # Cloudflare 拦截 cloudflare_blocked_indicators = [ "Sorry, you have been blocked", "You are being rate limited", "Cloudflare Ray ID" ] if any(indicator in current_html_content for indicator in cloudflare_blocked_indicators): raise BizLogicError(message="Blocked by Cloudflare WAF. Need to change IP or browser fingerprint.") # 遇到五秒盾先绕盾 (保留你原有的逻辑) if "just a moment" in current_title or "cloudflare" in current_title: # 假设你有 cf_bypasser 实例 # if not cf_bypasser.bypass(max_retry=3): continue self._log("[State] CF 5-second shield detected. Waiting...") time.sleep(5) continue # 状态 0: 成功到达目标预约页面 (Target Page) if "/Services" in current_url and self.page.ele('.app-menu', timeout=1): self._log("🎉 Successfully reached the Booking Services page! Session created successfully!") self.session_create_time = time.time() session_created = True break # 状态 1: 登录前的初始首页 elif self.page.ele('#pingid-button', timeout=1): self._log("[State] Initial Landing Page detected. Clicking login button...") login_btn = self.page.ele('#pingid-button') self.mouse.human_click_ele(login_btn) self._log("Redirecting to PingID...") self._random_sleep(3, 5) # 替代 time.sleep(5) continue # 状态 2: 真正的表单登录页 (PingID) elif self.page.ele('@name=callback_1', timeout=1): self._log("[State] PingID Login Form detected. Submitting credentials...") print("正在输入账号...") user_input = self.page.ele('@name=callback_1') self.mouse.human_click_ele(user_input) user_input.input(username, clear=True) time.sleep(1) # 模拟人手停顿 print("正在输入密码...") pwd_input = self.page.ele('@type=password') self.mouse.human_click_ele(pwd_input) pwd_input.input(password, clear=True) time.sleep(1) print("正在点击登录按钮...") submit_btn = self.page.ele('tag:button@type=submit') self.mouse.human_click_ele(submit_btn) has_submitted_login = True self._log("Login form submitted. Waiting for dashboard to load...") self._random_sleep(5, 7) continue # 状态 3: 登录成功后的主控制台 (导航栏页面) elif self.page.ele('.app-menu', timeout=1): self._log("[State] Dashboard Menu detected.") switch_en_btn = self.page.ele('tag:a@@href=/Language/ChangeLanguage?lang=2', timeout=0.5) if switch_en_btn: self._log("Found 'English' language switch button. Clicking to change language...") self.mouse.human_click_ele(switch_en_btn) self._random_sleep(4, 6) continue # 动作 B: 如果没有英文切换按钮（说明已经是英文或不存在），则查找并点击 Book 按钮 book_btn = self.page.ele('@href=/Services', timeout=0.5) if book_btn: self._log("Found 'Book' menu element. Clicking to proceed to reservation...") self.mouse.human_click_ele(book_btn) self._random_sleep(4, 6) continue self._log("[WARN] On Dashboard, but neither 'English' nor 'Book' button was found. Waiting...") time.sleep(2) continue else: self._log("[State] In unknown or transitional state. Waiting for next polling cycle...") 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}") # ------------------------------------------------------------- # 2. Query Availability # ------------------------------------------------------------- def query(self, apt_type: AppointmentType) -> VSQueryResult: res = VSQueryResult() res.success = False res.availability_status = AvailabilityStatus.NoneAvailable # 假设要预约的服务和到访原因 TARGET_SERVICE = "National and Schengen Visas" REASON_FOR_VISIT = "Tourism" # 对应 value 42 的选项文本 # ========================================== # 步骤 1：在服务列表中寻找并点击目标服务的 Book # ========================================== print("等待服务列表加载...") self.page.wait.eles_loaded('@aria-controls=dataTableServices', timeout=10) print(f"正在查找 [{TARGET_SERVICE}] 的 Book 按钮...") # 使用 XPath：找包含 TARGET_SERVICE 文本的行(tr)，再找该行里包含 Book 的超链接(a) target_book_btn_xpath = f'xpath://tr[contains(., "{TARGET_SERVICE}")]//a[contains(text(), "Book")]' book_btn = self.page.ele(target_book_btn_xpath) if not book_btn: print(f"未找到 {TARGET_SERVICE} 的 Book 按钮，可能当前无号，程序退出。") # 这里可以根据你的逻辑 return 或者 raise Exception else: self.mouse.human_click_ele(book_btn) self.page.wait.load_start() time.sleep(3) # ========================================== # 步骤 2：填写预约表单 # ========================================== print("等待预约表单加载...") self.page.wait.eles_loaded('#bookingForm', timeout=10) print("1. 选择预约类型...") typeofbooking_ddl = self.page.ele('#typeofbookingddl') # DrissionPage select 方法直接按文本选中，会自动触发网页的 JS 联动 typeofbooking_ddl.select('Individual booking') time.sleep(1) print("2. 选择到访原因...") reason_ddl = self.page.ele('#ddls_0') reason_ddl.select(REASON_FOR_VISIT) time.sleep(1) print("3. 填写备注...") notes_input = self.page.ele('#BookingNotes') notes_input.clear() notes_input.input('N/A') # 选填，填入 N/A 或者留空 time.sleep(1) # ========================================== # 步骤 3：处理 OTP 验证码 # ========================================== print("4. 点击发送 OTP 验证码...") otp_send_btn = self.page.ele('#otp-send') self.mouse.human_click_ele(otp_send_btn) # 这是一个 AJAX 请求，会转圈圈。我们需要等待成功提示出现 print("等待验证码发送成功的提示...") self.page.wait.ele_displayed('#IdOtpSent', timeout=15) # 等待绿字 "New code sent!" 显示 print("验证码已发送！") print("5. 填写默认验证码 123456 ...") otp_input = self.page.ele('#otp-input') self.mouse.human_click_ele(otp_input) otp_input.input('123456') time.sleep(1) # ========================================== # 步骤 4：勾选隐私政策并提交 # ========================================== print("6. 勾选隐私政策...") privacy_checkbox = self.page.ele('#PrivacyCheck') # 判断一下如果没勾上才去点，防止重复点击取消了 if not privacy_checkbox.states.is_checked: self.mouse.human_click_ele(privacy_checkbox) time.sleep(1) print("7. 点击 Forward 提交表单...") forward_btn = self.page.ele('#btnAvanti') self.mouse.human_click_ele(forward_btn) # 提交后页面会跳转，等待加载开始 self.page.wait.load_start() print("表单已提交！当前页面：", self.page.url) time.sleep(5) valid_dates = [] if valid_dates: res.success = True res.availability_status = AvailabilityStatus.Available earliest_date = valid_dates[0] earliest_dt = datetime.strptime(earliest_date, "%Y-%m-%d") res.earliest_date = earliest_dt for day in valid_days: # 查询具体 Slot slot_url = f"{self._host}/BookingCalendar/RetrieveTimeSlots" slot_payload = { "selectedDay": day, # YYYY-MM-DD "idService": str(self._service_id) } resp_slot = self._perform_request("POST", slot_url, json_data=slot_payload) time_slots = self._parse_time_slots(resp_slot.text) ts_list = [] if time_slots: # 转换结构 for ts in time_slots: # ts: {'id': 123, 'start': '10:00', 'end': '10:30', 'remain': 1} ts_list.append(TimeSlot( time=f"{ts['start']} - {ts['end']}", label=str(ts['id']) # 将 ID 存入 label 以便 book 使用 )) res.availability.append(DateAvailability(date=datetime.strptime(day, "%d-%m-%Y"), times=ts_list)) return res # ------------------------------------------------------------- # 3. Book # ------------------------------------------------------------- def book(self, slot_info: VSQueryResult, user_inputs: Dict = None) -> VSBookResult: res = VSBookResult() res.success = False if not slot_info.availability: raise NotFoundError("No slots to book") target_dt = slot_info.availability[0].date target_date = target_dt.strftime("%Y-%m-%d") # 取第一个时间段 target_slot = slot_info.availability[0].times[0] slot_id = target_slot.label # 我们在 query 里把 ID 存在了 label slot_text = target_slot.time # "10:00 - 10:30" # 1. 获取 OTP (GenerateOTP) self._log("Requesting OTP...") otp_url = f"{self._host}/BookingCalendar/GenerateOTP?ServiceID={self._service_id}" self._perform_request("POST", otp_url) # 2. 等待并读取邮件 self._log("Waiting for email code...") time.sleep(10) # 稍微等一下发信 email_account = self.config.account.email # 使用 CloudAPI 读取 (假设已配置) otp_code = VSCloudApi.Instance().get_email_verify_code(email_account) if not otp_code: raise BizLogicError("Failed to retrieve OTP code") self._log(f"Got OTP: {otp_code}") # 3. 提交详细信息 (Fill User Info) # 这是最复杂的一步，涉及文件上传 (Multipart) self._log("Submitting User Details & Files...") # 准备文件 (转 Base64 传给 JS) passport_pdf_path = user_inputs.get('passport_pdf_path') irp_pdf_path = user_inputs.get('irp_pdf_path') def file_to_b64(path): if not path or not os.path.exists(path): return "" with open(path, "rb") as f: return base64.b64encode(f.read()).decode('utf-8') ppt_b64 = file_to_b64(passport_pdf_path) irp_b64 = file_to_b64(irp_pdf_path) # 构造 JS FormData 提交脚本 # 注意：这里需要根据 Service ID (Dublin/Canton) 动态调整字段 ID # 下面以 Dublin (1321) 的字段为例，如果是 Canton 需要修改 _Id 和 _TipoDatoAddizionale # 为了通用性，这里演示 Dublin 的结构，请根据实际 Service ID 调整 mapping # 假设是 Dublin (根据提供的源码分析) boundary = '----WebKitFormBoundaryRandomString' submit_url = f"{self._host}/Services/Booking/{self._service_id}" # 注入 JS 执行 js_submit = f""" const url = "{submit_url}"; const fd = new FormData(); // 基础字段 fd.append('ServizioDescrizione', 'D Visa Application'); fd.append('MessaggioRassicuranteWaitingList', 'True'); fd.append('isWaitingListEnabled', 'False'); fd.append('IDServizioConsolare', '35'); fd.append('IDServizioErogato', '{self._service_id}'); fd.append('IdTipoPrenotazione', '1'); // Single fd.append('NumMaxAccompagnatori', '3'); fd.append('NumAccompagnatoriSelected', '0'); // 动态字段 (Dublin 示例) // [0] Other citizenship -> User Input fd.append('DatiAddizionaliPrenotante[0]._Descrizione', 'Other citizenship/s'); fd.append('DatiAddizionaliPrenotante[0]._testo', '{user_inputs.get("citizen", "China")}'); fd.append('DatiAddizionaliPrenotante[0]._Obbligatorio', 'False'); fd.append('DatiAddizionaliPrenotante[0]._Id', '61738'); fd.append('DatiAddizionaliPrenotante[0]._TipoDatoAddizionale.IDTipoDatoAddizionale', '26'); fd.append('DatiAddizionaliPrenotante[0]._TipoDatoAddizionale.IDTipoControllo', '2'); // [1] Full address -> User Input fd.append('DatiAddizionaliPrenotante[1]._Descrizione', 'Full residence address'); fd.append('DatiAddizionaliPrenotante[1]._testo', '{user_inputs.get("address", "")}'); fd.append('DatiAddizionaliPrenotante[1]._Obbligatorio', 'True'); fd.append('DatiAddizionaliPrenotante[1]._Id', '61739'); fd.append('DatiAddizionaliPrenotante[1]._TipoDatoAddizionale.IDTipoDatoAddizionale', '25'); fd.append('DatiAddizionaliPrenotante[1]._TipoDatoAddizionale.IDTipoControllo', '2'); // [2] Passport Num fd.append('DatiAddizionaliPrenotante[2]._Descrizione', 'Passport number'); fd.append('DatiAddizionaliPrenotante[2]._testo', '{user_inputs.get("passport", "")}'); fd.append('DatiAddizionaliPrenotante[2]._Obbligatorio', 'True'); fd.append('DatiAddizionaliPrenotante[2]._Id', '61740'); fd.append('DatiAddizionaliPrenotante[2]._TipoDatoAddizionale.IDTipoDatoAddizionale', '2'); fd.append('DatiAddizionaliPrenotante[2]._TipoDatoAddizionale.IDTipoControllo', '2'); // [3] Reason (Select) fd.append('DatiAddizionaliPrenotante[3]._Descrizione', 'Reason for visit'); fd.append('DatiAddizionaliPrenotante[3]._Obbligatorio', 'True'); fd.append('DatiAddizionaliPrenotante[3]._Id', '61741'); fd.append('DatiAddizionaliPrenotante[3]._TipoDatoAddizionale.IDTipoDatoAddizionale', '34'); fd.append('DatiAddizionaliPrenotante[3]._TipoDatoAddizionale.IDTipoControllo', '3'); fd.append('DatiAddizionaliPrenotante[3]._idSelezionato', '42'); // 42 = Tourism? Need verify // OTP fd.append('otp-input', '{otp_code}'); fd.append('PrivacyCheck', 'true'); // 文件处理 (Base64 -> Blob -> FormData) // 注意：这里假设页面上有文件上传的对应 ID，或者我们直接硬编码 FormData // 原始抓包并未显示文件字段名，通常是 File_0, File_1 // 我们需要将 base64 转 blob async function addFile(b64, name, filename) {{ if(!b64) return; const res = await fetch(`data:application/pdf;base64,${{b64}}`); const blob = await res.blob(); fd.append(name, blob, filename); }} // 并行处理文件 await Promise.all([ addFile('{ppt_b64}', 'File_0', 'passport.pdf'), // 假设 File_0 是护照 addFile('{irp_b64}', 'File_1', 'irp.pdf') // 假设 File_1 是 IRP ]); // 发送 POST return fetch(url, {{ method: 'POST', body: fd }}).then(async r => {{ return {{ status: r.status, url: r.url, text: await r.text() }}; }}).catch(e => {{ return {{ status: 0, text: e.toString() }}; }}); """ result_dict = self.page.run_js(js_submit) resp = BrowserResponse(result_dict) if resp.status_code == 302 or "BookingCalendar" in resp.url: self._log("User Info Submitted Successfully.") else: self._log(f"User Info Submit Failed: {resp.text[:100]}") # 如果 OTP 错误，页面会返回特定错误信息 if "Codice errato" in resp.text: raise BizLogicError("Invalid OTP Code") return res # Fail # 4. 最终确认预约 (InsertNewBooking) self._log("Finalizing Booking...") final_url = f"{self._host}/BookingCalendar/InsertNewBooking" final_payload = { "idCalendarioGiornaliero": slot_id, "selectedDay": target_date, "selectedHour": slot_text # "10:00 - 10:30(2)" } # 这里用 Form-UrlEncoded resp_final = self._perform_request("POST", final_url, data=final_payload) if resp_final.status_code == 200: self._log("Booking Confirmed!") res.success = True res.book_date = target_date res.book_time = slot_text else: self._log(f"Final Booking Failed: {resp_final.status_code}") return res def _perform_request(self, method, url, headers=None, data=None, json_data=None): """JS Fetch Wrapper""" if not self.page: raise BizLogicError("Browser not init") fetch_opts = { "method": method.upper(), "headers": headers or {}, "credentials": "include" } if json_data: fetch_opts['body'] = json.dumps(json_data) fetch_opts['headers']['Content-Type'] = 'application/json; charset=UTF-8' elif data: if isinstance(data, dict): from urllib.parse import urlencode fetch_opts['body'] = urlencode(data) fetch_opts['headers']['Content-Type'] = 'application/x-www-form-urlencoded; charset=UTF-8' else: fetch_opts['body'] = data js = f""" return fetch("{url}", {json.dumps(fetch_opts)}) .then(async r => {{ const h = {{}}; r.headers.forEach((v, k) => h[k] = v); return {{ status: r.status, body: await r.text(), headers: h, url: r.url }}; }}).catch(e => {{ return {{ status: 0, body: e.toString() }}; }}); """ return BrowserResponse(self.page.run_js(js, timeout=60)) # 文件上传可能较慢，给60s def _parse_valid_days(self, text): # 提取 DateLibere (YYYY-MM-DD) # 格式: {"DateLibere":"22/10/2024 00:00:00","SlotLiberi":1,"SlotRimanenti":1} # 原始正则: r'{"DateLibere":"(.*?)","SlotLiberi":\d+,"SlotRimanenti":(-?\d+)}' days = [] try: matches = re.findall(r'{"DateLibere":"(.*?)".*?"SlotRimanenti":(-?\d+)}', text) for d_str, rem in matches: if int(rem) != -1: # 22/10/2024 -> 2024-10-22 dt = datetime.strptime(d_str[:10], "%d/%m/%Y") days.append(dt.strftime("%Y-%m-%d")) except: pass return days def _parse_time_slots(self, text): # 提取 IDCalendarioServizioGiornaliero, StartTime, EndTime, Remain slots = [] try: # 原始逻辑比较复杂，这里简化正则 # 查找 SlotRimanenti > 0 的记录 # 关键是 IDCalendarioServizioGiornaliero raw_list = json.loads(text) # Prenotami 返回的是一个 JSON 列表字符串 for item in raw_list: remain = item.get('SlotRimanenti', -1) if remain > 0: start = item['OrarioInizioFascia'] end = item['OrarioFineFascia'] s_time = f"{start['Hours']:02d}:{start['Minutes']:02d}" e_time = f"{end['Hours']:02d}:{end['Minutes']:02d}" slots.append({ 'id': item['IDCalendarioServizioGiornaliero'], 'start': s_time, 'end': e_time, 'remain': remain }) except: pass return slots # --- 资源清理核心方法 --- def cleanup(self): """ 销毁浏览器并彻底删除临时文件 """ # 1. 关闭浏览器 if self.page: try: self.page.quit() # 这会关闭 Chrome 进程 except Exception: pass # 忽略已关闭的错误 self.page = None # 2. 删除文件 # 注意：Chrome 关闭后可能需要几百毫秒释放文件锁，稍微等待 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: # 如果删除失败(通常是Windows文件占用)，重试 self._log(f"Cleanup retry: {e}") time.sleep(0.5) # 如果依然存在，打印警告（虽然 ignore_errors=True 会掩盖报错，但可以 check exists） if os.path.exists(self.root_workspace): self._log(f"[WARN] Failed to fully remove workspace: {self.root_workspace}") # 3. [新增] 关闭代理隧道 if self.tunnel: try: self.tunnel.stop() except: pass self.tunnel = None def __del__(self): """ 析构函数：当对象被垃圾回收时自动调用 """ self.cleanup()