coordinate/vfs_registration_bot.py

import os
import re
import uuid
import random
import socket
import json
import time
import string
import shutil
import logging
import base64
import requests
import argparse
from datetime import datetime, timezone
from urllib.parse import urlparse, urlencode

from bs4 import BeautifulSoup
from cryptography.hazmat.primitives import serialization, hashes
from cryptography.hazmat.primitives.asymmetric import padding
from cryptography.hazmat.backends import default_backend

# 假设这些是你本地的依赖
import configure
from DrissionPage import ChromiumPage, ChromiumOptions
from vs_types import RateLimiteddError, BizLogicError
from toolkit.mihomo_tunnel import MihomoTunnel
from utils.cloudflare_bypass_for_scraping import CloudflareBypasser

# ==========================================
# 日志与常量配置
# ==========================================
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s [%(levelname)s] %(message)s',
    datefmt='%H:%M:%S'
)
logger = logging.getLogger("VFSRegistrar")

VFS_PUBLIC_KEY_PEM = """-----BEGIN PUBLIC KEY-----
MIIBIjANBgkqhkiG9w0BAQEFAAOCAQ8AMIIBCgKCAQEAuupFgB+lYIOtSxrRoHzc
LmCZKJ6+oSbgqgOPzFMM0TasOeLw0NXEn1XfIzXdx75+tegNKwyIZumoh0yhubKs
t59GV321kN0iquYRHrdh3ygfDDHlS9rROQeBqRga0ncSADtbLMrBPqXJjPCoV76y
t92towriKoH75BhiazY0mghm4LjmAWrV0u/GNpV3tk9bxbtHEXGaFmxCJqjg+7x6
1e5wXLfvpj9w1QsiSWOSJxLOyICz/9ByxXycQQFdNmjnnnwco9Gt/Mi33NYH71j0
5oXIjklFC4lvJqaqSY5lS7Vwb9oCt9zX9J0Yz4z4e/3V+0jgRnWOFGofyks4FKe2
GQIDAQAB
-----END PUBLIC KEY-----"""


# ==========================================
# 辅助工具类
# ==========================================
class VFSHelper:
    @staticmethod
    def generate_mobile_number(country_code=353, e164_format=False):
        if country_code == 353:  # Ireland
            prefix = random.choice(['83', '85', '86', '87', '89'])
            number = f"{prefix}{''.join([str(random.randint(0, 9)) for _ in range(7)])}"
            return f"+353{number}" if e164_format else number
        elif country_code == 44:  # UK
            prefix_second = random.choice(['1', '2', '3', '4', '5', '7', '8', '9'])
            number = f"7{prefix_second}{''.join([str(random.randint(0, 9)) for _ in range(8)])}"
            return f"+44{number}" if e164_format else number
        elif country_code == 86:  # China
            prefixes = ["130", "131", "132", "133", "135", "136", "138", "139", "150", "158", "159", "186"]
            prefix = random.choice(prefixes)
            number = f"{prefix}{''.join([str(random.randint(0, 9)) for _ in range(8)])}"
            return f"+86{number}" if e164_format else number
        return "".join([str(random.randint(0, 9)) for _ in range(10)])

    @staticmethod
    def generate_password(length=12):
        chars = string.ascii_letters + string.digits + "@#$%"
        while True:
            pwd = ''.join(random.choices(chars, k=length))
            if (any(c.islower() for c in pwd) and 
                any(c.isupper() for c in pwd) and 
                any(c.isdigit() for c in pwd) and 
                any(c in "@#$%" for c in pwd)):
                return pwd

    @staticmethod
    def encrypt_password(password: str) -> str:
        public_key = serialization.load_pem_public_key(
            VFS_PUBLIC_KEY_PEM.encode(), backend=default_backend()
        )
        ciphertext = public_key.encrypt(
            password.encode(),
            padding.OAEP(
                mgf=padding.MGF1(algorithm=hashes.SHA256()),
                algorithm=hashes.SHA256(),
                label=None
            )
        )
        return base64.b64encode(ciphertext).decode()

    @staticmethod
    def get_client_source() -> str:
        timestamp = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%S")
        payload = f"GA;{timestamp}Z"
        return VFSHelper.encrypt_password(payload)


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:
            try:
                self._json = json.loads(self.text) if self.text else {}
            except json.JSONDecodeError:
                self._json = {}
        return self._json


# ==========================================
# API 交互与代理工具
# ==========================================
def load_proxies(pool_name):
    """从 config/proxies.json 读取对应的代理池"""
    config_path = os.path.join(os.path.dirname(__file__), 'config', 'proxies.json')
    try:
        with open(config_path, 'r', encoding='utf-8') as f:
            data = json.load(f)
            proxies = data.get(pool_name, [])
            if not proxies:
                raise ValueError(f"代理池 '{pool_name}' 为空或不存在！")
            return proxies
    except Exception as e:
        logger.error(f"读取代理配置文件失败: {e}")
        exit(1)

def upload_account_to_server(account, api_url, api_token):
    """将注册成功的账号上报到中心服务器"""
    headers = {
        'accept': 'application/json',
        'Authorization': f'Bearer {api_token}',
        'Content-Type': 'application/json'
    }

    payload = {
        "pool_name": account.get("pool_name"),
        "username": account.get("username"),
        "password": account.get("password"),
        "extra_data": {
            "country_code": account.get("country_code"),
            "mission_code": account.get("mission_code"),
            "phone_country_code": account.get("phone_country_code"),
            "phone_number": account.get("phone_number"),
            "created_at": datetime.now().strftime("%Y-%m-%d %H:%M:%S")
        }
    }

    try:
        logger.info(f"Uploading account {account['username']} to server...")
        resp = requests.post(api_url, json=payload, headers=headers, timeout=10)
        if resp.status_code == 200:
            logger.info(f"✅ [API Upload Success] Server responded: {resp.text}")
            return True
        else:
            logger.error(f"❌ [API Upload Failed] Status: {resp.status_code}, Body: {resp.text}")
            return False
    except Exception as e:
        logger.error(f"❌ [API Upload Error]: {e}")
        return False


# ==========================================
# 核心自动化类
# ==========================================
class VFSRegistrationBot:
    def __init__(self, vfs_url, proxy, email_api_token, master_email):
        self.vfs_url = vfs_url
        self.proxy_config = proxy
        self.email_api_token = email_api_token
        self.master_email = master_email
        
        self.page = None
        self.tunnel = None
        self.instance_id = uuid.uuid4().hex[:8]
        self.workspace = os.path.abspath(os.path.join("data/temp_browser_data", f"reg_session_{self.instance_id}"))
        self.active_proxy_url = None  # 供 request 请求复用

    def _log(self, msg):
        logger.info(f"[TLS-Reg-{self.instance_id}] {msg}")

    def _init_browser(self):
        """初始化浏览器配置"""
        co = ChromiumOptions()
        
        # 查找可用端口
        with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
            s.bind(('', 0))
            port = s.getsockname()[1]
            
        co.set_local_port(port)
        
        chrome_path = configure.CHROME_PATH or os.getenv("CHROME_BIN")
        if chrome_path and os.path.exists(chrome_path):
            co.set_paths(browser_path=chrome_path)

        if self.proxy_config and self.proxy_config.get("ip"):
            p = self.proxy_config
            if p.get('username') and p.get('password'):
                self._log(f"Starting Proxy Tunnel for {p.get('ip')}...")
                exit_node = {
                    "name": "ExitNode",
                    "type": p.get('proto'),
                    "server": p.get('ip'),
                    "port": p.get('port'),
                    "username": p.get('username'),
                    "password": p.get('password')
                }
                relay_node = random.choice(configure.MIHOMO_RELAY_NODES) if configure.MIHOMO_RELAY_NODES else None
                mihomo_path = configure.MIHOMO_BIN_PATH or os.getenv("MIHOMO_BIN")
                if not mihomo_path:
                    raise BizLogicError('Mihomo path is null. Set mihomo bin path in configure or os env')
                
                self.tunnel = MihomoTunnel(mihomo_path, exit_node=exit_node, relay_node=relay_node)
                self.active_proxy_url = self.tunnel.start()
                self._log(f"Tunnel started at {self.active_proxy_url}")
                co.set_argument(f'--proxy-server={self.active_proxy_url}')
            else:
                self.active_proxy_url = f"{p.get('proto')}://{p.get('ip')}:{p.get('port')}"
                co.set_argument(f'--proxy-server={self.active_proxy_url}')
        else:
            self._log("[WARN] No proxy configured!")

        co.headless(False)
        co.set_argument('--no-sandbox')
        co.set_argument('--disable-gpu')
        co.set_argument('--disable-dev-shm-usage')
        co.set_argument('--window-size=1920,1080')
        co.set_argument('--disable-blink-features=AutomationControlled')
        
        self.page = ChromiumPage(co)
        self.page.set.timeouts(15)

    def _perform_js_fetch(self, method, url, headers=None, data=None, json_data=None, retry_count=0):
        """注入JS执行Fetch请求，绕过部分指纹检测"""
        if not self.page:
            raise BizLogicError("Browser not initialized")
        if retry_count > 3:
            raise BizLogicError("Max retries exceeded for request")

        fetch_options = {
            "method": method.upper(),
            "headers": headers or {},
            "credentials": "include"
        }
        
        if json_data:
            fetch_options['body'] = json.dumps(json_data)
            fetch_options['headers']['Content-Type'] = 'application/json'
        elif data:
            fetch_options['body'] = urlencode(data) if isinstance(data, dict) else str(data)
            fetch_options['headers']['Content-Type'] = 'application/x-www-form-urlencoded'

        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(err => ({{ status: 0, body: err.toString() }}));
        """
        
        try:
            res_dict = self.page.run_js(js_script, timeout=60)
            resp = BrowserResponse(res_dict)
            
            if resp.status_code == 200:
                return resp
            
            if resp.status_code == 403 and ("cloudflare" in resp.text.lower() or "Just a moment" in resp.text):
                logger.warning(f"Cloudflare 403 detected. Retrying ({retry_count+1})...")
                new_token = self._refresh_turnstile()
                if new_token and json_data and "captcha_api_key" in json_data:
                    json_data["captcha_api_key"] = new_token
                return self._perform_js_fetch(method, url, headers, data, json_data, retry_count + 1)
            
            if resp.status_code == 429:
                raise RateLimiteddError(f"Rate Limit: {resp.text[:100]}")
                
            return resp
        except Exception as e:
            logger.error(f"JS Execution Error: {e}")
            raise BizLogicError(f"Fetch failed: {e}")

    def _handle_cookie_banner(self):
        try:
            self.page.run_js("""
            var btn = document.getElementById('onetrust-accept-btn-handler');
            if(btn) { btn.click(); return true; }
            var banner = document.getElementById('onetrust-banner-sdk');
            if(banner) { banner.remove(); return true; }
            """)
        except:
            pass

    def _refresh_turnstile(self):
        self._log("Attempting to refresh Turnstile token...")
        try:
            self.page.run_js('try{window.turnstile.reset()}catch(e){}')
            cf_bypasser = CloudflareBypasser(self.page, log=True)
            
            for i in range(30):
                time.sleep(1)
                try:
                    ele = self.page.ele('@name=cf-turnstile-response')
                    if ele and ele.value:
                        self._log("Turnstile token obtained.")
                        return ele.value
                except:
                    pass
                if i > 5:
                    try: cf_bypasser.click_verification_button(is_dfs=False)
                    except: pass
        except Exception as e:
            logger.error(f"Turnstile refresh failed: {e}")
        return None

    def _wait_for_activation_link(self, username, max_wait_sec=60):
        self._log(f"Waiting for email to {username}...")
        start_time = time.time()

        url = "https://api.text.skin/api/email-authorizations/fetch"
        headers = {
            "Authorization": f"Bearer {self.email_api_token}",
            "Content-Type": "application/json",
            "Accept": "application/json, text/plain, */*"
        }

        while time.time() - start_time < max_wait_sec:
            try:
                utc_now_str = datetime.now(timezone.utc).strftime('%Y-%m-%d %H:%M:%S')
                params = {
                    "email": self.master_email,
                    "sender": 'donotreply at vfsglobal.com',
                    "recipient": username,
                    "subjectKeywords": 'Welcome',
                    "bodyKeywords": 'ActivateAccount',
                    "sentDate": utc_now_str,
                    "expiry": str(60)
                }
        
                # 已经去掉 proxies=proxies 参数
                resp = requests.post(url, headers=headers, params=params, json={}, timeout=60)
                
                if resp.status_code != 200:
                    logger.warning(f"Email API returned status {resp.status_code}.")
                    time.sleep(15)
                    continue

                try:
                    result = resp.json()
                except json.JSONDecodeError:
                    time.sleep(15)
                    continue

                if result.get('code') != 0:
                    logger.debug(f"API Message: {result.get('message')}")
                else:
                    content = result.get('data', {}).get('body', "")
                    if content:
                        soup = BeautifulSoup(content, "html.parser")
                        link = soup.find("a", string="ActivateAccount")
                        if link:
                            return link["href"].replace(" ", "").replace("\n", "").replace("\r", "").strip()
            except Exception as e:
                logger.warning(f"Error fetching email: {e}")
            
            time.sleep(15)
            self._log("Checking email again...")
            
        return None

    def register(self, account):
        try:
            self._init_browser()
            self._log(f"Opening {self.vfs_url}")
            
            self.page.set.timeouts(page_load=30, script=30)
            try:
                self.page.get(self.vfs_url, retry=0, timeout=30)
            except Exception:
                logger.warning("Page load timed out. Stopping loading manually.")
                self.page.stop_loading()
            
            cf_token = None
            cf_bypasser = CloudflareBypasser(self.page, log=True)
            
            for _ in range(40):
                time.sleep(1)
                if "page-not-found" in self.page.url:
                    logger.error("❌ [BLOCKED] Redirected to 'Page Not Found'. Aborting.")
                    return False
                if "403" in self.page.title and "Just a moment" not in self.page.title:
                    logger.error("❌ [BLOCKED] 403 Forbidden detected. Aborting.")
                    return False
                
                self._handle_cookie_banner()
                
                try:
                    ele = self.page.ele('@name=cf-turnstile-response')
                    if ele and ele.value:
                        cf_token = ele.value
                        break
                except: pass
                
                try: cf_bypasser.click_verification_button(is_dfs=False)
                except: pass
            
            if not cf_token:
                raise BizLogicError("Failed to obtain initial Cloudflare token")

            post_data = {
                'emailid': account['username'],
                'password': VFSHelper.encrypt_password(account['password']),
                'confirmPassword': VFSHelper.encrypt_password(account['password']),
                'processPerDataAgreed': True,
                'intTransPerDataAgreed': True,
                'termAndConditionAgreed': True,
                'missioncode': account['mission_code'],
                'countrycode': account['country_code'],
                'languageCode': 'en',
                'dialcode': str(account['phone_country_code']),
                'contact': account['phone_number'],
                'captcha_version': 'cloudflare-v1',
                'captcha_api_key': cf_token,
                'cultureCode': 'en-US',
                'IsSpecialUser': False,
            }
            
            headers = {
                'content-type': 'application/json;charset=utf-8',
                'accept': 'application/json, text/plain, */*',
                'route': f"{account['country_code']}/en/{account['mission_code']}",
                'clientsource': VFSHelper.get_client_source(),
            }

            self._log(f"Submitting registration for {account['username']}")
            
            country_code = account.get('country_code', '').lower()
            if country_code == 'chn':
                api_domain = 'https://lift-apicn.vfsglobal.com'
            else:
                api_domain = 'https://lift-api.vfsglobal.com'
                
            register_endpoint = f"{api_domain}/user/registration"
            self._log(f"Using API Endpoint: {register_endpoint}")

            resp = self._perform_js_fetch(
                'POST', 
                register_endpoint,
                headers=headers, 
                json_data=post_data
            )
            
            self._log(f'register resp={resp.text}')
            resp_data = resp.json()
            
            if resp_data.get("code") == "200":
                self._log("Registration API success. Waiting for email...")
                activate_link = self._wait_for_activation_link(account['username'])
                
                if activate_link:
                    self._log(f"Activating account: {activate_link}")
                    activate_tab = self.page.new_tab(activate_link)
                    try:
                        success_ele = activate_tab.ele('Activation Successful', timeout=30)
                        if success_ele:
                            logger.info(f"✅ Account {account['username']} activated successfully.")
                            return True
                        else:
                            logger.error("Activation verification failed. Success text not found.")
                            return False
                    finally:
                        activate_tab.close()
                else:
                    logger.error("Timeout waiting for activation email.")
            else:
                logger.error(f"Registration failed: {resp_data}")

        except Exception as e:
            logger.error(f"Registration process exception: {e}", exc_info=True)
        return False
    
    def cleanup(self):
        self._log("Cleaning up resources...")
        if self.page:
            try: self.page.quit()
            except: pass
        if self.tunnel:
            try: self.tunnel.stop()
            except: pass
        if os.path.exists(self.workspace):
            time.sleep(1)
            shutil.rmtree(self.workspace, ignore_errors=True)


# ==========================================
# 主流程 & 命令行参数解析
# ==========================================
def parse_arguments():
    parser = argparse.ArgumentParser(description="VFS Global 自动注册机器人")
    
    # 核心参数
    parser.add_argument('--target', type=int, default=20, help="需要注册的目标账号数量")
    parser.add_argument('--vfs-url', type=str, required=True, help="VFS 注册页面的完整 URL, 如: https://visa.vfsglobal.com/chn/en/aut/register")
    parser.add_argument('--proxy-pool', type=str, default='proxy-cheap', help="使用的代理池名称 (在 proxies.json 中定义)")
    
    # 账号生成相关参数
    parser.add_argument('--account-prefix', type=str, default="cn_at", help="生成的邮箱前缀")
    parser.add_argument('--account-pool', type=str, default="cn.at.sentinel", help="上报到服务端的账号池名称")
    parser.add_argument('--phone-country', type=int, default=86, help="生成的手机号国家代码")
    parser.add_argument('--email-domain', type=str, default="text.skin", help="邮箱域名")
    
    # API 与 上报配置
    parser.add_argument('--upload-url', type=str, default="https://api.text.skin/api/account/add", help="账号上报服务器 URL")
    parser.add_argument('--upload-token', type=str, default="tok_e946329a60ff45ba807f3f41b0e8b7fc", help="账号上报服务器 API Token")
    parser.add_argument('--email-token', type=str, default="tok_e946329a60ff45ba807f3f41b0e8b7fc", help="获取邮件的 API Token")
    parser.add_argument('--master-email', type=str, default="hujiarui8@gmail.com", help="接收验证邮件的主邮箱")

    return parser.parse_args()


def generate_account_details(args):
    """根据命令行参数和 URL 生成账号数据字典"""
    # 从 URL 中提取 from_country 和 to_country
    match = re.search(r'vfsglobal\.com/([^/]+)/[^/]+/([^/]+)/register', args.vfs_url)
    if not match:
        raise ValueError(f"无法从 URL 解析国家代码: {args.vfs_url}")
        
    from_country = match.group(1)
    to_country = match.group(2)
    
    rand_suffix = ''.join(random.choices(string.ascii_lowercase + string.digits, k=6))
    username = f"{args.account_prefix}_{rand_suffix}@{args.email_domain}"
    phone = VFSHelper.generate_mobile_number(args.phone_country)
    
    return {
        'pool_name': args.account_pool,
        'country_code': from_country,
        'mission_code': to_country,
        'phone_country_code': args.phone_country,
        'phone_number': phone,
        'username': username,
        'password': VFSHelper.generate_password(),
    }


def main():
    args = parse_arguments()
    proxies = load_proxies(args.proxy_pool)
    logger.info(f"[*] 成功加载代理数量: {len(proxies)} 个")
    
    success_accounts = []
    logger.info(">>> Starting Registration Bot <<<")
    
    while len(success_accounts) < args.target:
        proxy_config = random.choice(proxies)
        account_detail = generate_account_details(args)
        
        logger.info(f"Processing Account: {account_detail['username']}")
        bot = None
        try:
            bot = VFSRegistrationBot(
                vfs_url=args.vfs_url,
                proxy=proxy_config,
                email_api_token=args.email_token,
                master_email=args.master_email
            )
            is_success = bot.register(account_detail)
            
            if is_success:
                success_accounts.append(account_detail)
                logger.info(f"Progress: {len(success_accounts)}/{args.target}")
                
                upload_account_to_server(account_detail, args.upload_url, args.upload_token)
                
                # 实时保存结果，防止中途退出丢失
                with open("registered_accounts.json", "w", encoding='utf-8') as f:
                    json.dump(success_accounts, f, indent=4, ensure_ascii=False)
            else:
                logger.warning("Registration failed. Retrying with new account details...")

            time.sleep(5)  # 避免请求过于频繁
            
        except Exception as e:
            logger.error(f"[ERROR] 注册发生致命错误 | 代理 IP: {proxy_config.get('ip')} | 异常信息: {e}")
            
        finally:
            if bot:
                bot.cleanup()

    logger.info(f">>> All tasks completed. Registered {len(success_accounts)} accounts. <<<")

if __name__ == "__main__":
    main()