import time import json import os import re import uuid import socket import shutil import requests from datetime import datetime, timedelta from DrissionPage import ChromiumPage, ChromiumOptions from utils.cloudflare_bypass_for_scraping import CloudflareBypasser from toolkit.vs_cloud_api import VSCloudApi class TlsRegistrator: def __init__(self, proxy_config: dict, capsolver_key: str): self.proxy_config = proxy_config self.capsolver_key = capsolver_key # 隔离的用户数据目录 self.instance_id = uuid.uuid4().hex[:8] self.workspace = os.path.abspath(os.path.join("data", f"reg_session_{self.instance_id}")) self.page = None def _log(self, msg): print(f"[TLS-Reg-{self.instance_id}] {msg}") def _get_free_port(self): """获取可用端口，防止 DrissionPage 解析日志报错""" with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s: s.bind(('', 0)) return s.getsockname()[1] def init_browser(self): """初始化独立、配置好代理的浏览器环境""" self._log("Initializing browser...") co = ChromiumOptions() # 1. 端口与路径隔离 port = self._get_free_port() co.set_local_port(port) co.set_user_data_path(self.workspace) # 2. 代理配置 (支持账号密码) if self.proxy_config and self.proxy_config.get("ip"): p = self.proxy_config if p.get("username") and p.get("password"): proxy_str = f"{p['username']}:{p['password']}@{p['ip']}:{p['port']}" else: proxy_str = f"{p.get('scheme', 'http')}://{p['ip']}:{p['port']}" self._log(f"Setting proxy: {p['ip']}:{p['port']}") co.set_proxy(proxy_str) # 3. 反爬及稳定性配置 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) def solve_captcha(self, page_url: str, site_key: str) -> str: """调用 Capsolver 解决验证码 (无依赖版本)""" if not self.capsolver_key: raise ValueError("Capsolver API key is missing!") self._log("Submitting captcha task to Capsolver...") task = { "type": "ReCaptchaV3TaskProxyLess", "websiteURL": page_url, "websiteKey": site_key, "pageAction": "register" } payload = { "clientKey": self.capsolver_key, "task": task } print(f'createTask payload={payload}') create_res = requests.post("https://api.capsolver.com/createTask", json=payload, timeout=20) if create_res.status_code != 200 or create_res.json().get("errorId") != 0: raise Exception(f"Failed to create capsolver task: {create_res.text}") task_id = create_res.json().get("taskId") self._log(f"Captcha Task created: {task_id}. Waiting for solution...") for _ in range(30): res = requests.post( "https://api.capsolver.com/getTaskResult", json={"clientKey": self.capsolver_key, "taskId": task_id}, timeout=20 ) data = res.json() self._log(f"data={data}") if data.get("status") == "ready": self._log("Captcha solved successfully!") return data.get("solution", {}).get("gRecaptchaResponse") or data.get("solution", {}).get("token") time.sleep(3) raise Exception("Capsolver task timeout") def register(self, email, password, issuer_id="cnCNG2fr"): """执行自动注册""" try: self.init_browser() # 1. 访问注册页面并过盾 reg_url = f"https://visas-fr.tlscontact.com/en-us/registration?issuerId={issuer_id}" self._log(f"Navigating to {reg_url}") self.page.get(reg_url) cf_bypasser = CloudflareBypasser(self.page, log=True) cf_bypasser.bypass() time.sleep(3) cf_bypasser.handle_waiting_room() site_key = "6LcTpXcfAAAAAM3VojNhyV-F1z92ADJIvcSZ39Y9" captcha_token = self.solve_captcha(reg_url, site_key) self._log("Constructing Next.js fetch request...") action_id = "608511a5a6e58da30dc13b591255b358caa2193367" router_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%22(auth)%22%2C%7B%22children%22%3A%5B%22registration%22%2C%7B%22children%22%3A%5B%22__PAGE__%22%2C%7B%7D%2Cnull%2Cnull%5D%7D%2Cnull%2Cnull%5D%7D%2Cnull%2Cnull%5D%7D%2Cnull%2Cnull%2Ctrue%5D%7D%2Cnull%2Cnull%5D' payload =[ { "email": email, "password": password, "locale": "en", "issuerId": issuer_id, "consentList": ["SURVEY"] }, captcha_token ] body_str = json.dumps(payload).replace("'", "\\'") js_script = f""" const url = "{reg_url}"; const headers = {{ 'next-action': '{action_id}', 'next-router-state-tree': '{router_state}', 'accept': 'text/x-component', 'content-type': 'text/plain;charset=UTF-8' }}; const bodyData = '{body_str}'; return fetch(url, {{ method: 'POST', headers: headers, body: bodyData }}) .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 => {{ return {{ status: 0, body: err.toString(), headers: {{}} }}; }}); """ self._log("Submitting registration...") res_dict = self.page.run_js(js_script) # 4. 解析结果 (处理 Next.js Server Action 响应) status = res_dict.get('status') body_text = res_dict.get('body', '') resp_headers = {str(k).lower(): v for k, v in res_dict.get('headers', {}).items()} action_redirect = resp_headers.get('x-action-redirect', '') # 【核心修改点】：适配 Next.js 响应中的 "status":"OK" is_success = ( status == 303 or (status == 200 and ("login" in action_redirect or '"status":"OK"' in body_text)) ) if is_success: self._log("✅ Registration SUCCESS!") self._log(f"Response snippet: {body_text[:100]}...") return True else: self._log(f"❌ Registration FAILED! Status: {status}") self._log(f"Response Body: {body_text[:500]}") return False except Exception as e: self._log(f"Error during registration: {e}") return False def activate(self, email, sent_at=None): email_box = 'visafly666@gmail.com' sender = 'TLSContact' recipient = email subject_keywords = 'Activate' body_keywords = '' content_out = VSCloudApi.Instance().fetch_mail_content( email=email_box, sender=sender, recipient=recipient, subject_keywords=subject_keywords, body_keywords=body_keywords, sent_date=sent_at, expiry=600 ) print(f'activate email content={content_out}') match = re.search(r'https://\S+', content_out) activate_link = match.group(0) if match else None tab = self.page.new_tab(activate_link) btn_selector = "Activate" if not tab.wait.ele_displayed(btn_selector, timeout=10): return False tab.ele(btn_selector).click() def cleanup(self): """清理浏览器进程和缓存文件夹""" self._log("Cleaning up resources...") if self.page: try: self.page.quit() except: pass if os.path.exists(self.workspace): time.sleep(1) # 等待文件锁释放 shutil.rmtree(self.workspace, ignore_errors=True) if __name__ == "__main__": # ================= 配置区域 ================= PROXY_CONFIG = { "scheme": "http", "ip": "127.0.0.1", "port": "7890", "username": "", "password": "" } CAPSOLVER_KEY = "CAP-5441DD341DD3CC2FAEF0BE6FE493EE9A" TARGET_EMAIL = "lisi39@gmail-app.com" TARGET_PWD = "Visafly@111" TARGET_ISSUER = "cnCNG2fr" # ============================================ bot = TlsRegistrator(proxy_config=PROXY_CONFIG, capsolver_key=CAPSOLVER_KEY) now_utc = datetime.utcnow() formatted_utc_time = (now_utc - timedelta(minutes=4)).strftime("%Y-%m-%d %H:%M:%S") # success = bot.register(email=TARGET_EMAIL, password=TARGET_PWD, issuer_id=TARGET_ISSUER) # if not success: # print("\n--> 流程完成：账号注册失败，请检查日志。") bot.activate(email=TARGET_EMAIL, sent_at=formatted_utc_time)