import time import json import random import re import os from datetime import datetime from typing import List, Dict, Optional, Any, Callable from urllib.parse import urljoin, urlparse from curl_cffi import requests, const from bs4 import BeautifulSoup from vs_plg import IVSPlg from vs_types import VSPlgConfig, VSQueryResult, VSBookResult, AvailabilityStatus, NotFoundError, PermissionDeniedError, RateLimiteddError, SessionExpiredOrInvalidError, BizLogicError from toolkit.vs_cloud_api import VSCloudApi class TlsPlugin(IVSPlg): """ TLSContact 签证预约插件 适配法国签证 (FR) 流程 """ 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.session: Optional[requests.Session] = None self.travel_group: Optional[Dict] = None self.user_agent = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36" def get_group_id(self) -> str: return self.group_id def set_log(self, logger: Callable[[str], None]) -> None: self.logger = logger def set_config(self, config: VSPlgConfig): self.config = config self.free_config = config.free_config or {} def health_check(self) -> bool: return self.is_healthy def create_session(self): """ 创建会话：处理 Cloudflare -> 登录 -> 获取 Travel Group """ # 1. 初始化 Session curlopt = { const.CurlOpt.MAXAGE_CONN: 1800, const.CurlOpt.MAXLIFETIME_CONN: 1800, const.CurlOpt.VERBOSE: self.config.debug, } self.session = requests.Session( proxy=self._get_proxy_url(), impersonate="chrome124", curl_options=curlopt, use_thread_local_curl=False, http_version=const.CurlHttpVersion.V2TLS ) embassy = self.free_config.get('center', {}) if not embassy: raise NotFoundError(message="center not found in free config") # 2. 解决 Cloudflare 5s 盾 self._solve_cloudflare5S_challenge() # 3. 获取登录页面参数 (OIDC) login_page = "https://visas-fr.tlscontact.com/en-us/login" params = { "issuerId": embassy["code"], "country": embassy["country"], "vac": embassy["code"], "redirect": f"/en-us/country/{embassy['country']}/vac/{embassy['code']}" } headers = { 'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8', 'Accept-Language': 'zh-CN,zh;q=0.9,en;q=0.8', 'Referer': f'https://visas-fr.tlscontact.com/en-us/country/{embassy["country"]}/vac/{embassy["code"]}', 'User-Agent': self.user_agent, } resp = self._perform_request("GET", login_page, headers=headers, params=params) if self.config.debug: self._save_debug_html(resp.text, prefix='Tls_Login_Page') # 解析 Keycloak 登录地址 soup = BeautifulSoup(resp.text, 'html.parser') form = soup.find('form') if not form: raise NotFoundError(message="Login form not found") action = form.get('action') authenticate_url = action if action.startswith('http') else urljoin(resp.url, action) # 4. 解决 ReCaptcha V2 (登录验证码) api_token = self.free_config.get("capsolver_key", "") if not api_token: raise NotFoundError(message="Missing 'capsolver_key' in free_config, captcha might fail.") rc_params = { "type": "ReCaptchaV2TaskProxyLess", "page": authenticate_url, "siteKey": "6LcDpXcfAAAAAM7wOEsF_38DNsL20tTvPTKxpyn0", "apiToken": api_token, "proxy": self._get_proxy_url() } g_token = self._solve_recaptcha(rc_params) # 5. 提交登录 payload = { 'username': self.config.account.username, 'password': self.config.account.password, 'g-recaptcha-response': g_token } headers['Content-Type'] = 'application/x-www-form-urlencoded' resp = self._perform_request("POST", authenticate_url, headers=headers, data=payload) if self.config.debug: self._save_debug_html(resp.text, prefix='Tls_Travel_Groups_Page') # 6. 解析 Travel Groups self._check_page_is_session_expired_or_invalid("My travel group", resp.text) groups = self._parse_travel_groups(resp.text) # 选择匹配城市的 Group target_city = embassy['city'].lower() for g in groups: if g['location'].lower() == target_city: self.travel_group = g break if not self.travel_group: raise NotFoundError(message=f"No matched group found for city {target_city}") self._log(f"Session created. Group: {self.travel_group['group_number']}") def query(self) -> VSQueryResult: res = VSQueryResult() res.success = False embassy = self.free_config.get('center', {}) group_num = self.travel_group['group_number'] interest_month = self.free_config.get("interest_month", time.strftime("%m-%Y")) max_retries = self.free_config.get("max_retries", 2) url = f'https://visas-fr.tlscontact.com/en-us/{group_num}/workflow/appointment-booking' params = { 'location': embassy["code"], 'month': interest_month, } headers = { 'accept': '*/*', 'accept-language': 'zh-CN,zh;q=0.9,en;q=0.8', 'referer': f'{url}?location={embassy["code"]}', 'user-agent': self.user_agent, } for attempt in range(1, max_retries + 1): try: resp = self._perform_request("GET", url, headers=headers, params=params) if self.config.debug: self._save_debug_html(resp.text, prefix='Tls_Query_Slot_Page') break # ✅ 请求成功，跳出重试循环 except PermissionDeniedError: self._log(f"Query Appointment-booking blocked (403), attempt {attempt}/{max_retries}") # 最后一次就不再绕盾了 if attempt >= max_retries: raise PermissionDeniedError() self._solve_cloudflare5S_challenge() self._log("Cloudflare bypass success, retrying...") continue self._check_page_is_session_expired_or_invalid('Book your appointment', resp.text) # 3. 解析 Slots all_slots = self._parse_appointment_slots(resp.text) target_labels = self.free_config.get("target_labels", ["", "pta"]) available = [s for s in all_slots if s.get("label") in target_labels] res.city = self.free_config.get('city', '') res.country = self.free_config.get('country', '') res.visa_type = self.free_config.get('visa_type', '') res.routing_key = self.free_config.get('routing_key', '') if available: res.success = True res.availability_status = AvailabilityStatus.Available res.earliest_date = available[0]['date'] date_map = {} for s in available: d = s['date'] date_map.setdefault(d, []) ts = VSQueryResult.DateAvailability.TimeSlot() ts.time = s['time'] ts.label = f"{s['type']}" date_map[d].append(ts) for d, slots in date_map.items(): da = VSQueryResult.DateAvailability() da.date = d da.times = slots res.availability.append(da) else: res.success = False res.availability_status = AvailabilityStatus.NoneAvailable return res def book(self, slot_info: VSQueryResult, user_input: Dict = None) -> VSBookResult: res = VSBookResult() res.success = False target_date = slot_info.availability[0].date target_time = slot_info.availability[0].times[0].time target_label = "" embassy = self.free_config.get('center', {}) group_num = self.travel_group['group_number'] interest_month = self.free_config.get("interest_month", time.strftime("%m-%Y")) # 1. 解决 ReCaptcha V3 page_url = f'https://visas-fr.tlscontact.com/en-us/{group_num}/workflow/appointment-booking?location={embassy["code"]}&month={interest_month}' api_token = self.free_config.get("capsolver_key", "") rc_params = { "type": "ReCaptchaV3Task", "page": page_url, "action": "book", "siteKey": "6LcTpXcfAAAAAM3VojNhyV-F1z92ADJIvcSZ39Y9", "apiToken": api_token, "proxy": self._get_proxy_url() } g_token = self._solve_recaptcha(rc_params) # 2. 构造请求 url = f'https://visas-fr.tlscontact.com/en-us/{group_num}/workflow/appointment-booking' next_action = '601f284bf7ee33b6578ad0fad426fae18c232707f2' next_state = '%5B%22%22%2C%7B%22children%22%3A%5B%5B%22lang%22%2C%22en-us%22%2C%22d%22%5D%2C%7B%22children%22%3A%5B%5B%22groupId%22%2C%22$GROUPID$%22%2C%22d%22%5D%2C%7B%22children%22%3A%5B%22workflow%22%2C%7B%22children%22%3A%5B%22appointment-booking%22%2C%7B%22children%22%3A%5B%22__PAGE__%22%2C%7B%7D%2Cnull%2Cnull%5D%7D%2Cnull%2Cnull%2Ctrue%5D%7D%2Cnull%2Cnull%5D%7D%2Cnull%2Cnull%5D%7D%2Cnull%2Cnull%2Ctrue%5D%7D%2Cnull%2Cnull%5D' headers = { 'Next-Action': next_action, 'Referer': page_url, 'Next-Router-State-Tree': next_state.replace("$GROUPID$", group_num), 'Accept': 'text/x-component', 'User-Agent': self.user_agent, } params = { 'location': embassy["code"], 'month': interest_month, } boundary = "----WebKitFormBoundary" + "".join( random.choices("abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789", k=16) ) headers["Content-Type"] = f"multipart/form-data; boundary={boundary}" form_fields = { '1_formGroupId': str(group_num), '1_lang': 'en-us', '1_process': 'APPOINTMENT', '1_location': embassy["code"], '1_date': target_date, '1_time': target_time, '1_appointmentLabel': target_label, '1_captcha_token': g_token, '0': '[{"status":"IDLE"},"$K1"]' } body_parts = [] for name, value in form_fields.items(): body_parts.append(f"--{boundary}\r\n") body_parts.append(f'Content-Disposition: form-data; name="{name}"\r\n') body_parts.append("\r\n") body_parts.append(f"{value}\r\n") body_parts.append(f"--{boundary}--\r\n") body = "".join(body_parts).encode("utf-8") resp = self.session.post(url, params=params, headers=headers, data=body, allow_redirects=False) if self.config.debug: self._save_debug_html(resp.text, prefix='Tls_Book_Appointment_Page') if resp.status_code == 303: res.success = True res.book_date = target_date res.book_time = target_time return res else: self._log(f'Expected Status is 303, but got {resp.status_code}') res.success = False return res def _log(self, message): if self.logger: self.logger(f'[TlsPlugin] [{self.group_id}] {message}') def _save_debug_html(self, content: str, prefix: str = "debug"): save_dir = "debug_pages" if not os.path.exists(save_dir): os.makedirs(save_dir) timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") filename = f"{save_dir}/{prefix}_{timestamp}.html" with open(filename, "w", encoding="utf-8") as f: f.write(content) self._log(f"HTML saved to: {filename}") def _get_proxy_url(self): # 构造代理 proxy_url = "" if self.config.proxy.ip: s = self.config.proxy if s.username: proxy_url = f"{s.scheme}://{s.username}:{s.password}@{s.ip}:{s.port}" else: proxy_url = f"{s.scheme}://{s.ip}:{s.port}" return proxy_url def _perform_request(self, method, url, headers=None, data=None, json_data=None, params=None): """ 统一 HTTP 请求封装，严格复刻 C++ 逻辑： 1. 发送 OPTIONS 请求 2. 发送实际请求 """ resp = self.session.request(method, url, headers=headers, data=data, json=json_data, params=params, timeout=30) if self.config.debug: self._log(f'[perform request] Response={resp.text}\nMethod={method}, Url={url}, Data={data}, JsonData={json_data}, Params={params}') if resp.status_code == 200: return resp elif resp.status_code == 401: self.is_healthy = False raise SessionExpiredOrInvalidError() elif resp.status_code == 403: raise PermissionDeniedError() elif resp.status_code == 429: self.is_healthy = False raise RateLimiteddError() else: raise BizLogicError(message=f"HTTP Error {resp.status_code}: {resp.text[:100]}") def _solve_cloudflare5S_challenge(self): """ 解决 Cloudflare 5s 盾 """ self._log(f"Solving Cloudflare 5s...") embassy = self.free_config.get('center', {}) website_url = f'https://visas-fr.tlscontact.com/en-us/country/{embassy["country"]}' # 1. 格式化代理字符串, 这里的接口要求格式通常是: host:port:user:pass (根据你的脚本示例) p = self.config.proxy if p.username: proxy_str = f"{p.ip}:{p.port}:{p.username}:{p.password}" else: proxy_str = f"{p.ip}:{p.port}" # 2. 提交任务 task = VSCloudApi.Instance().submit_anticloudflare_task(proxy_str, website_url) # 3. 等待结果 task_id = str(task['id']) result = VSCloudApi.Instance().get_anticloudflare_result(task_id) parsed = json.loads(result.get('result', '{}')) cookies_list = parsed.get('cookies', []) for cookie in cookies_list: if cookie['name'] in ['__cf_bm', 'cf_clearance']: self.session.cookies.set( cookie['name'], cookie['value'], domain=cookie['domain'], path='/' ) ua = parsed.get('userAgent') if ua: self.user_agent = ua self.session.headers['User-Agent'] = ua self._log("Cloudflare 5s challenge solved.") def _solve_recaptcha(self, params) -> str: """ 调用 Capsolver """ key = params.get("apiToken") if not key: raise NotFoundError(message="Api-token is required for recaptcha solver") 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 = urlparse(params.get("proxy")) task["proxyType"] = p.scheme task["proxyAddress"] = p.hostname task["proxyPort"] = p.port if p.username: task["proxyLogin"] = p.username task["proxyPassword"] = p.password payload = {"clientKey": key, "task": task} r = requests.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 = requests.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: str) -> List[Dict]: groups = [] js_pattern = r'\\"travelGroups\\":\s*(\[.*?\]),\\"availableCountriesToCreateGroups' js_match = re.search(js_pattern, html, 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') }) else: self._log('Parsed travel group page, but not found travelGroups') return groups def _parse_appointment_slots(self, html: str) -> List[Dict]: slots = [] pattern = r'"availableAppointments\\":\s*(\[.*\]),\\"showFlexiAppointment' match = re.search(pattern, html, 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 = "" stype = "" cost = "" if 'pta' in labels: lbl = 'pta' stype = "Prime" elif 'ptaw' in labels: lbl = 'ptaw' stype = "Prime Weekend" elif '' in labels: lbl = '' stype = "Standard" if lbl or not labels: slots.append({ 'date': d_str, 'time': s.get('time'), 'label': lbl, 'type': stype, 'cost': cost }) return slots else: self._log('Parsed appointment slot page, but not found availableAppointments') return slots def _check_page_is_session_expired_or_invalid(self, keyword, html: str) -> bool: if not html: self.is_healthy = False raise SessionExpiredOrInvalidError() if keyword not in html: if 'redirected automatically' in html.lower(): self.is_healthy = False raise SessionExpiredOrInvalidError() if 'login' in html.lower() and 'password' in html.lower(): self.is_healthy = False raise SessionExpiredOrInvalidError() if 'session expired!' in html.lower() and 'for security reasons, your session has expired. please log in again to continue.' in html.lower() and 'you will be redirected automatically in 10 seconds.' in html.lower(): self.is_healthy = False raise SessionExpiredOrInvalidError() if 'temporarily blocked!' in html.lower() and 'Your session has been temporarily suspended due to the high number of your access to this page.' in html.lower() and 'You can try to access your account again in 2 hours.' in html.lower(): self.is_healthy = False raise SessionExpiredOrInvalidError()