coordinate/plugins/tls_plugin.py

import time
import json
import random
import re
import os
from datetime import datetime
from typing import List, Dict, Optional, Any
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 vs_log_macros import VSC_INFO, VSC_ERROR, VSC_DEBUG, VSC_WARN
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.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_config(self, config: VSPlgConfig):
        self.config = config
        try:
            self.free_config = json.loads(config.free_config) if config.free_config else {}
        except:
            self.free_config = {}

    def health_check(self) -> bool:
        return self.is_healthy

    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}_{self.group_id}_{timestamp}.html"
        with open(filename, "w", encoding="utf-8") as f:
            f.write(content)
        VSC_INFO("tls_plg", "[%s] HTML saved to: %s", self.group_id, 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 create_session(self):
        """
        创建会话：处理 Cloudflare -> 登录 -> 获取 Travel Group
        """        
        # 1. 初始化 Session
        curlopt = {
            const.CurlOpt.MAXAGE_CONN: 1800,
            const.CurlOpt.MAXLIFETIME_CONN: 1800,
            const.CurlOpt.VERBOSE: False,
        }
        
        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)
        self._save_debug_html(resp.text, '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)
        self._save_debug_html(resp.text, 'Travel_Groups_Page')
        # 6. 解析 Travel Groups
        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 group found for city {target_city}")
        VSC_INFO("tls_plg", "[%s] Session created. Group: %s", self.group_id, 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"))
        

        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,
        }

        resp = self._perform_request("GET", url, headers=headers, params=params)
        
        self._save_debug_html(resp.text, 'Query_Slot_Page')
        self._check_session_expired_page(resp.text)

        # 检测关键词
        if not "availableAppointments" in resp.text:
            raise NotFoundError(message='Query result not found availableAppointments')


        # 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.availability_status = AvailabilityStatus.NoneAvailable

        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
        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)
        self._save_debug_html(resp.text, 'Book_Appointment_Page')
        if resp.status_code == 303: 
            res.success = True
            res.book_date = target_date
            res.book_time = target_time
            return res
        else:
            VSC_WARN('tls_plg', 'Expected Status is 303, but got {resp.status_code}')
            res.success = False            
        return res
    
    def _perform_request(self, method, url, headers=None, data=None, json_data=None, params=None):
        """
        统一 HTTP 请求封装，严格复刻 C++ 逻辑：
        1. 发送 OPTIONS 请求
        2. 发送实际请求
        """
        print(f'[perform request] {method} {url}')
        resp = self.session.request(method, url, headers=headers, data=data, json=json_data, params=params, timeout=30)
        VSC_INFO('tls_plg', resp.text)
        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 盾
        """
        VSC_INFO("tls_plg", f"[{self.group_id}] 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
        VSC_INFO("tls_plg", "[%s] Cloudflare 5s challenge solved.", self.group_id)

    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:
            VSC_WARN('tls_plg', 'Parsed travel group page, but not found travelGroups')
        return groups

    def _parse_appointment_slots(self, html: str) -> List[Dict]:
        slots = []
        pattern = r'availableAppointments\\?":\s*(\[.*?\])(?:,\\?"|\},)'
        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
                            })
        else:
            VSC_WARN("tls_plg", 'Parsed appointment slots page, but not found availableAppointments')
        return slots

    def _check_session_expired_page(self, html: str) -> bool:
        if not html:
            self.is_healthy = False
            raise SessionExpiredOrInvalidError()
        if 'availableAppointments' 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()