import sys import time import json import os import re import uuid import socket import threading import random import requests from datetime import datetime from typing import Optional, Dict from PyQt5.QtWidgets import ( QApplication, QMainWindow, QWidget, QVBoxLayout, QHBoxLayout, QGridLayout, QGroupBox, QLabel, QLineEdit, QSpinBox, QPushButton, QTableWidget, QTableWidgetItem, QAbstractItemView, QTextEdit, QHeaderView, QMessageBox ) from PyQt5.QtCore import pyqtSignal, QObject, Qt from DrissionPage.common import Keys from DrissionPage import ChromiumPage, ChromiumOptions import configure from utils.cloudflare_bypass_for_scraping import CloudflareBypasser from toolkit.vs_cloud_api import VSCloudApi from toolkit.mihomo_tunnel import MihomoTunnel from vs_types import BizLogicError from utils.mouse import HumanMouse from utils.keyboard import HumanKeyboard from utils.fingerprint_utils import FingerprintGenerator from utils.fake_utils import generate_random_account_detail 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: raise Exception(f"读取代理配置文件失败: {e}") # ================= 用于跨线程更新UI的信号类 ================= class GuiSignals(QObject): log_signal = pyqtSignal(str) # 传递日志文本 status_signal = pyqtSignal(int, str) # 传递行号和状态文本 # ================= 核心业务类 ================= class TlsRegistrator: def __init__(self, tls_url, proxy_config: Optional[Dict]=None, capsolver_key: Optional[str]=None, signals: GuiSignals=None): self.proxy_config = proxy_config self.capsolver_key = capsolver_key self.account_detail = None # 动态绑定 self.signals = signals # 绑定UI信号 self.instance_id = uuid.uuid4().hex[:8] self.tls_url = tls_url self.workspace = os.path.abspath(os.path.join("data/temp_browser_data", f"reg_session_{self.instance_id}")) self.page = None self.mouse = None self.keyboard = None self.tunnel = None def _log(self, msg): now = datetime.now().strftime("%Y-%m-%d %H:%M:%S") log_str = f"[{now}][Bot-{self.instance_id}] {msg}" if self.signals: self.signals.log_signal.emit(log_str) else: print(log_str) def _get_free_port(self): with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s: s.bind(('', 0)) return s.getsockname()[1] def save_screenshot(self, name_prefix): try: timestamp = int(time.time()) filename = f"{self.instance_id}_{name_prefix}_{timestamp}.jpg" save_path = os.path.join("data", filename) os.makedirs("data", exist_ok=True) self.page.get_screenshot(path=save_path, full_page=False) self._log(f"Screenshot saved to {save_path}") except Exception as e: self._log(f"Failed to save screenshot: {e}") def init_browser(self): self._log("Initializing browser...") co = ChromiumOptions() port = self._get_free_port() co.set_local_port(port) co.set_user_data_path(self.workspace) 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(message='Mihomo path is null') self.tunnel = MihomoTunnel(mihomo_path, exit_node=exit_node, relay_node=relay_node) local_proxy = self.tunnel.start() co.set_argument(f'--proxy-server={local_proxy}') else: proxy_str = f"{p.get('proto')}://{p.get('ip')}:{p.get('port')}" co.set_argument(f'--proxy-server={proxy_str}') else: self._log("[WARN] No proxy configured!") fingerprint_gen = FingerprintGenerator() specific_fp = fingerprint_gen.generate(self.instance_id) 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={specific_fp.get('seed')}") co.set_argument(f"--fingerprint-platform={specific_fp.get('platform')}") co.set_argument(f"--fingerprint-brand={specific_fp.get('brand')}") self.page = ChromiumPage(co) self.page.get(self.tls_url) time.sleep(5) cf_bypasser = CloudflareBypasser(self.page, log=True) cf_bypasser.bypass(max_retry=8) time.sleep(3) cf_bypasser.handle_waiting_room() self._log("正在初始化拟人化工具...") self.mouse = HumanMouse(self.page, debug=False) self.keyboard = HumanKeyboard(self.page) 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) self._log("浏览器初始化完成。") def solve_captcha(self, page_url: str, task_type: str, site_key: str, use_proxy=False, action: str=None, api_domain: str=None) -> str: if not self.capsolver_key: raise ValueError("Capsolver API key missing") task = {"type": task_type, "websiteURL": page_url, "websiteKey": site_key} if api_domain: task["apiDomain"] = api_domain if use_proxy: proxy = self.proxy_config task["proxyType"] = proxy.get('proto', 'http') task["proxyAddress"] = proxy.get('ip') task["proxyPort"] = int(proxy.get('port')) if proxy.get('username'): task["proxyLogin"] = proxy.get('username') task["proxyPassword"] = proxy.get('password') if action: task["pageAction"] = action payload = {"clientKey": self.capsolver_key, "task": task} res = requests.post("https://api.capsolver.com/createTask", json=payload, timeout=20) if res.status_code != 200 or res.json().get("errorId") != 0: raise Exception(f"Failed to create capsolver task: {res.text}") task_id = res.json().get("taskId") for _ in range(30): r = requests.post("https://api.capsolver.com/getTaskResult", json={"clientKey": self.capsolver_key, "taskId": task_id}, timeout=20) data = r.json() if data.get("status") == "ready": return data["solution"].get("gRecaptchaResponse") or data["solution"].get("token") time.sleep(3) raise Exception("Capsolver task timeout") def register(self): email = self.account_detail.get('email') password = self.account_detail.get('pwd') self._log(f"开始执行自动注册: {email}") btn_selector = '#submit' if not self.page.wait.ele_displayed(btn_selector, timeout=3): register_btn = self.page.ele("tag:a@@href:registration") if register_btn: self.mouse.human_click_ele(register_btn) time.sleep(3) if not self.page.wait.ele_displayed(btn_selector, timeout=10): raise BizLogicError(message=f"Can't find selector={btn_selector}") time.sleep(random.uniform(0.5, 1)) self._log("正在填写邮箱和密码...") self.mouse.human_click_ele(self.page.ele('#email')) self.keyboard.type_text(email, humanize=True) time.sleep(random.uniform(0.2, 0.5)) self.mouse.human_click_ele(self.page.ele('#password')) self.keyboard.type_text(password, humanize=True) time.sleep(random.uniform(0.2, 0.5)) self.mouse.human_click_ele(self.page.ele('#confirm-password')) self.keyboard.type_text(password, humanize=True) time.sleep(random.uniform(0.2, 0.5)) self._log("正在勾选必选条款...") for checkbox_id in ['#terms-and-conditions', '#biometric-data', '#privacy-notice']: self.mouse.human_click_ele(self.page.ele(checkbox_id).next()) time.sleep(random.uniform(0.3, 0.6)) self._log("提交注册...") time.sleep(random.uniform(0.3, 0.6)) self.mouse.human_click_ele(self.page.ele(btn_selector)) self._log("正在等待验证结果 (最多15秒)...") success_dialog = self.page.wait.ele_displayed('tag:h1@text():Check your email inbox', timeout=15) if not success_dialog: self.save_screenshot("failed_submit.png") raise BizLogicError(message='Failed to submit account registration') self._log("✅ 操作成功！已弹出提示：Check your email inbox") return True def activate(self, sent_at=None): email = self.account_detail.get('email') if not sent_at: sent_at = datetime.utcnow().strftime("%Y-%m-%d %H:%M:%S") self._log("获取激活邮件...") content_out = VSCloudApi.Instance().fetch_mail_content( email='hujiarui8@gmail.com', sender='TLSContact', recipient=email, subject_keywords='TLSContact', body_keywords='', sent_date=sent_at, expiry=600 ) match = re.search(r'https://\S+', content_out) activate_link = match.group(0) if match else None if not activate_link: raise Exception("未找到激活链接") self.page.get(activate_link) btn_selector = "#activation-pending-button" if not self.page.wait.ele_displayed(btn_selector, timeout=10): raise BizLogicError(message=f"Wait ele={btn_selector} timeout") self.page.ele(btn_selector).click() time.sleep(3) self._log("✅ 激活成功！") def make_account_useful(self): def fill_date_field(page, selector, date_str): if not date_str: return ele = page.ele(selector) ele.scroll.to_see(center=True) js_detect_format = """ const parts = new Intl.DateTimeFormat().formatToParts(new Date(2023, 11, 31)); let format = []; for (let part of parts) { if (part.type === 'year') format.push('Y'); if (part.type === 'month') format.push('M'); if (part.type === 'day') format.push('D'); } return format; """ date_format = page.run_js(js_detect_format) year, month, day = date_str.split('-') date_dict = {'Y': year, 'M': month.zfill(2), 'D': day.zfill(2)} ele.click() time.sleep(0.1) page.actions.type(Keys.LEFT * 3) time.sleep(0.1) for i, char in enumerate(date_format): val = date_dict[char] page.actions.type(val) time.sleep(0.1) if char == 'Y': if i < 2: page.actions.type(Keys.RIGHT) time.sleep(0.1) # email = self.account_detail.get('email') # password = self.account_detail.get('pwd') # location = self.account_detail.get('location') # self._log("=== 开始执行自动填表流程 ===") # btn_selector = 'tag:button@@text():Login' # if not self.page.wait.ele_displayed(btn_selector, timeout=3): # login_btn = self.page.ele("tag:a@@href:login") # if login_btn: self.mouse.human_click_ele(login_btn) # time.sleep(3) # if not self.page.wait.ele_displayed(btn_selector, timeout=10): # raise BizLogicError(message=f"Can't find selector={btn_selector}") # recpatchav2_token = "" # if self.page.ele('.g-recaptcha') or self.page.ele('xpath://iframe[contains(@src, "recaptcha")]'): # self._log("Solving ReCaptcha...") # recpatchav2_token = self.solve_captcha(self.page.url, "ReCaptchaV2TaskProxyLess", "6LcDpXcfAAAAAM7wOEsF_38DNsL20tTvPTKxpyn0") # input_ele = self.page.ele('tag:label@@text():Email').next() # self.mouse.human_click_ele(input_ele) # time.sleep(random.uniform(0.2, 0.6)) # self.keyboard.type_text(email, humanize=True) # time.sleep(random.uniform(0.5, 1.2)) # input_ele = self.page.ele('tag:label@@text():Password').next() # self.mouse.human_click_ele(input_ele) # time.sleep(random.uniform(0.2, 0.6)) # self.keyboard.type_text(password, humanize=True) # if recpatchav2_token: # inject_recpatchav2_token_js = f""" # var g = document.getElementById('g-recaptcha-response'); # if(g) {{ g.value = "{recpatchav2_token}"; }} # """ # self._log("Inject ReCaptchaV2 Token via JS...") # self.page.run_js(inject_recpatchav2_token_js) # time.sleep(random.uniform(0.5, 1.0)) # self._log("Submitting Login...") # time.sleep(random.uniform(0.3, 0.8)) # login_btn = self.page.ele('tag:button@@text():Login') # self.mouse.human_click_ele(login_btn) # self._log("Waiting for dashboard redirect...") # self.page.wait.url_change('login-actions', exclude=True, timeout=45) # time.sleep(4) # if "login-actions" in self.page.url or "auth" in self.page.url: # raise BizLogicError(message="Login Failed! Invalid credentials or Captcha rejected.") # self._log("Waiting for dashboard...") # self.page.wait.load_start() # time.sleep(5) # self._log("Parsing Dashboard for Travel Group...") # html = self.page.html # js_pattern = r'\\"travelGroups\\":\s*(\[.*?\]),\\"availableCountriesToCreateGroups' # js_match = re.search(js_pattern, html, re.DOTALL) # groups = [] # if js_match: # json_str = js_match.group(1).replace(r'\"', '"') # groups = json.loads(json_str) # travel_group = None # for g in groups: # if location in g.get('vacName', ''): # travel_group = g # break # if not travel_group: # raise BizLogicError(message=f"Travel Group not found for {location}") # formgroup_id = travel_group.get('formGroupId') # self._log(f"Waiting for group button to render: {formgroup_id}") # btn_selector = f'tag:a@@data-testid=btn-select-group' # self._log(f"Select group_id={formgroup_id}...") # self.mouse.human_click_ele(self.page.ele(btn_selector)) # self._log("Waiting for url redirect...") # self.page.wait.url_change('travel-groups', exclude=True, timeout=45) # time.sleep(2) # if "travel-groups" in self.page.url or "auth" in self.page.url: # raise BizLogicError(message="Redirect to service-level Failed!") # btn_selector_add = 'tag:button@@data-testid=btn-add-applicant' # btn_selector_max = 'tag:button@@data-testid=btn-max-number-of-applicants' # error_selector = 'tag:h2@text():Something went wrong' # for attempt in range(2): # try: # btn_selector = btn_selector_add # if not self.page.wait.ele_displayed(btn_selector, timeout=10): # btn_selector = btn_selector_max # if not self.page.ele(btn_selector): # raise BizLogicError(message=f"Can't find selector={btn_selector}") # add_btn = self.page.ele(btn_selector) # add_btn.scroll.to_see(center=True) # time.sleep(random.uniform(0.6, 0.8)) # add_btn.click() # if self.page.wait.ele_displayed(error_selector, timeout=5): # raise BizLogicError("Page shows 'Something went wrong'") # break # except Exception as e: # if attempt == 0: # self.page.refresh() # time.sleep(5) # else: # raise BizLogicError(message=f"Click add applicant failed after retry: {e}") self._log("开始填充表单字段...") visa_type = self.account_detail.get("visa_type") if visa_type: btn = self.page.ele('tag:button@@data-testid=input-visa-type') btn.scroll.to_see(center=True) btn.click() time.sleep(0.5) self.page.ele(f'tag:span@@text():{visa_type}').click(by_js=True) time.sleep(0.5) tavel_purpose = self.account_detail.get("travel_purpose") if tavel_purpose: btn = self.page.ele('tag:button@@data-testid=input-travel-purpose') btn.scroll.to_see(center=True) btn.click() time.sleep(0.5) self.page.ele(f'tag:span@@text():{tavel_purpose}').click(by_js=True) time.sleep(0.5) application_form_id = self.account_detail.get('application_form_id') if application_form_id: ele = self.page.ele('tag:input@@data-testid=f_cai') ele.scroll.to_see(center=True) ele.input(application_form_id) last_name = self.account_detail.get('last_name') if last_name: ele = self.page.ele('tag:input@@data-testid=f_pers_surnames') ele.scroll.to_see(center=True) ele.input(last_name.upper()) first_name = self.account_detail.get('first_name') if first_name: ele = self.page.ele('tag:input@@data-testid=f_pers_givennames') ele.scroll.to_see(center=True) ele.input(first_name.upper()) gender = self.account_detail.get('gender') if gender: try: gender = gender.capitalize() ele = self.page.ele(f'tag:label@@text():{gender}') ele.scroll.to_see(center=True) ele.click() except Exception as e: self._log(e) fill_date_field(self.page, 'tag:input@@data-testid=f_pers_birth_date', self.account_detail.get('birthday')) nationality = self.account_detail.get('nationality') if nationality: nationality = nationality.title() btn = self.page.ele('tag:label@@for=f_pers_nationality').next() btn.scroll.to_see(center=True) btn.click() time.sleep(0.5) self.page.ele(f'tag:li@@role=option@@text():{nationality}').click(by_js=True) time.sleep(0.5) province_residence = self.account_detail.get('province_residence') if province_residence: try: province_residence = province_residence.title() btn = self.page.ele('tag:label@@for=f_pers_province').next() btn.scroll.to_see(center=True) btn.click() time.sleep(0.5) self.page.ele(f'tag:li@@role=option@@text():{province_residence}').click(by_js=True) time.sleep(0.5) except Exception as e: self._log(e) passport_type = self.account_detail.get('passport_type') if passport_type: btn = self.page.ele('tag:label@@for=f_identity_type').next() btn.scroll.to_see(center=True) btn.click() time.sleep(0.5) self.page.ele(f'tag:li@@role=option@@text():{passport_type}').click(by_js=True) time.sleep(0.5) passport_no = self.account_detail.get('passport_no') if passport_no: ele = self.page.ele('tag:input@@data-testid=f_pass_num') ele.scroll.to_see(center=True) ele.input(passport_no.upper()) passport_issue_date = self.account_detail.get('passport_issue_date') if passport_issue_date: try: fill_date_field(self.page, 'tag:input@@data-testid=fi_passport_issue_date', passport_issue_date) time.sleep(0.5) except Exception as e: self._log(e) passport_expiry_date = self.account_detail.get('passport_expiry_date') if passport_expiry_date: try: fill_date_field(self.page, 'tag:input@@data-testid=fi_passport_expiry_date', passport_expiry_date) time.sleep(0.5) except Exception as e: self._log(e) phone_country_code = self.account_detail.get('phone_country_code') phone_number = self.account_detail.get('phone_number') if phone_country_code: div = self.page.ele('tag:label@@for=f_pers_mobile_phone').next() btn = div.ele('tag:button') btn.scroll.to_see(center=True) btn.click() time.sleep(0.5) self.page.ele(f'tag:li@@role=option@@text():+{phone_country_code}').click(by_js=True) time.sleep(0.5) div.ele('tag:input@@type:tel').input(phone_number) fill_date_field(self.page, 'tag:input@@data-testid=fi_trav_origin_departure_date', self.account_detail.get('departure_origin_date')) fill_date_field(self.page, 'tag:input@@data-testid=f_trav_departure_date', self.account_detail.get('arrival_schengen_area_date')) fill_date_field(self.page, 'tag:input@@data-testid=f_trav_arrival_date', self.account_detail.get('departure_schengen_area_date')) submit_btn = self.page.ele('tag:button@@data-testid=btn-submit') submit_btn.scroll.to_see(center=True) time.sleep(1) submit_btn.click() time.sleep(6) confirm_btn = self.page.ele('tag:button@@text():Confirm') if confirm_btn: confirm_btn.scroll.to_see(center=True) confirm_btn.click() time.sleep(6) self._log("✅ 自动填表提交成功！") def upload_account_to_server(self): api_url = 'https://api.text.skin/api/account/add' api_token = 'tok_e946329a60ff45ba807f3f41b0e8b7fc' headers = { 'accept': 'application/json', 'Authorization': f'Bearer {api_token}', 'Content-Type': 'application/json' } payload = { "pool_name": self.account_detail.get("pool_name", "default_pool"), "username": self.account_detail.get("email"), "password": self.account_detail.get("pwd"), "extra_data": self.account_detail } try: self._log(f"Uploading account {self.account_detail['email']} to server...") resp = requests.post(api_url, json=payload, headers=headers, timeout=10) if resp.status_code == 200: self._log(f"✅ [API Upload Success] Server responded: {resp.text}") return True else: self._log(f"❌ [API Upload Failed] Status: {resp.status_code}, Body: {resp.text}") return False except Exception as e: self._log(f"❌ [API Upload Error]: {e}") return False def cleanup(self): self._log("Cleaning up resources...") if self.page: try: self.page.quit() except: pass # ================= PyQt5 GUI 主窗口 ================= class MainWindow(QMainWindow): def __init__(self): super().__init__() self.setWindowTitle("TLS 批量注册工具 v1.0 (PyQt5)") self.resize(1000, 700) self.bot = None self.profiles = [] self.capsolver_key = os.getenv("CAPSOLVER_KEY", "CAP-5441DD341DD3CC2FAEF0BE6FE493EE9A") # 实例化信号类并绑定槽函数 self.signals = GuiSignals() self.signals.log_signal.connect(self.append_log) self.signals.status_signal.connect(self.update_table_status) self.init_ui() def init_ui(self): main_widget = QWidget() main_layout = QVBoxLayout(main_widget) # 1. 基础设置区域 group_settings = QGroupBox("1. 基础设置") layout_settings = QGridLayout() layout_settings.addWidget(QLabel("地区 (如 CN-Beijing):"), 0, 0) self.input_region = QLineEdit("CN-Beijing") layout_settings.addWidget(self.input_region, 0, 1) layout_settings.addWidget(QLabel("代理池名 (从proxies.json读取):"), 0, 2) self.input_pool = QLineEdit("local") layout_settings.addWidget(self.input_pool, 0, 3) layout_settings.addWidget(QLabel("目标 URL:"), 1, 0) self.input_url = QLineEdit("https://visas-fr.tlscontact.com/en-us/country/gb/vac/gbLON2fr") layout_settings.addWidget(self.input_url, 1, 1, 1, 3) group_settings.setLayout(layout_settings) main_layout.addWidget(group_settings) # 2. 生成区域 group_gen = QGroupBox("2. 生成虚拟信息") layout_gen = QHBoxLayout() layout_gen.addWidget(QLabel("生成数量 (1-100):")) self.spin_count = QSpinBox() self.spin_count.setRange(1, 100) self.spin_count.setValue(5) layout_gen.addWidget(self.spin_count) self.btn_generate = QPushButton("生成虚拟信息") self.btn_generate.setStyleSheet("background-color: lightblue;") self.btn_generate.clicked.connect(self.action_generate) layout_gen.addWidget(self.btn_generate) layout_gen.addStretch() group_gen.setLayout(layout_gen) main_layout.addWidget(group_gen) # 3. 表格展示区域 self.table = QTableWidget() self.table.setColumnCount(5) self.table.setHorizontalHeaderLabels(["序号", "邮箱", "密码", "姓名", "当前状态"]) self.table.horizontalHeader().setSectionResizeMode(QHeaderView.Stretch) self.table.setSelectionBehavior(QAbstractItemView.SelectRows) self.table.setEditTriggers(QAbstractItemView.NoEditTriggers) main_layout.addWidget(self.table) # 4. 浏览器操作区域 group_actions = QGroupBox("3. 浏览器操作 (需先在表格中点击选中一行)") layout_actions = QHBoxLayout() self.btn_start = QPushButton("[全局] 启动浏览器") self.btn_start.clicked.connect(self.action_start_browser) layout_actions.addWidget(self.btn_start) self.btn_register = QPushButton("1. 自动注册并激活") self.btn_register.setStyleSheet("background-color: #d0f0c0;") self.btn_register.clicked.connect(self.action_register) layout_actions.addWidget(self.btn_register) self.btn_fill = QPushButton("2. 自动填表") self.btn_fill.setStyleSheet("background-color: #ffd700;") self.btn_fill.clicked.connect(self.action_fill) layout_actions.addWidget(self.btn_fill) self.btn_upload = QPushButton("3. 上报并标记已用") self.btn_upload.setStyleSheet("background-color: #ffb6c1;") self.btn_upload.clicked.connect(self.action_upload) layout_actions.addWidget(self.btn_upload) self.btn_close = QPushButton("[全局] 关闭浏览器") self.btn_close.setStyleSheet("color: red;") self.btn_close.clicked.connect(self.action_close_browser) layout_actions.addWidget(self.btn_close) group_actions.setLayout(layout_actions) main_layout.addWidget(group_actions) # 5. 日志区域 group_log = QGroupBox("运行日志") layout_log = QVBoxLayout() self.text_log = QTextEdit() self.text_log.setReadOnly(True) layout_log.addWidget(self.text_log) group_log.setLayout(layout_log) main_layout.addWidget(group_log) self.setCentralWidget(main_widget) # ---------------- 槽函数：UI 更新 ---------------- def append_log(self, text): self.text_log.append(text) self.text_log.moveCursor(self.text_log.textCursor().End) def update_table_status(self, row, status): item = QTableWidgetItem(status) item.setTextAlignment(Qt.AlignCenter) self.table.setItem(row, 4, item) # ---------------- 业务逻辑触发 ---------------- def run_in_thread(self, target_func, *args): t = threading.Thread(target=target_func, args=args, daemon=True) t.start() def action_generate(self): count = self.spin_count.value() region = self.input_region.text().strip() pool = self.input_pool.text().strip() self.append_log(f"开始生成 {count} 个 {region} 地区的虚拟信息...") self.profiles.clear() self.table.setRowCount(0) for i in range(count): try: acc = generate_random_account_detail(region) acc['pool_name'] = pool self.profiles.append(acc) self.table.insertRow(i) self.table.setItem(i, 0, QTableWidgetItem(str(i+1))) self.table.setItem(i, 1, QTableWidgetItem(acc.get('email', ''))) self.table.setItem(i, 2, QTableWidgetItem(acc.get('pwd', ''))) name = f"{acc.get('first_name','')} {acc.get('last_name','')}" self.table.setItem(i, 3, QTableWidgetItem(name)) self.update_table_status(i, "就绪") except Exception as e: self.append_log(f"生成时出错: {e}") self.append_log("✅ 虚拟信息生成完毕！") def action_start_browser(self): pool = self.input_pool.text().strip() url = self.input_url.text().strip() def _task(): try: self.signals.log_signal.emit(f"正在读取代理池 {pool} ...") proxies = load_proxies(pool) proxy = random.choice(proxies) self.signals.log_signal.emit(f"已选取代理: {proxy.get('ip')}") self.bot = TlsRegistrator( tls_url=url, proxy_config=proxy, capsolver_key=self.capsolver_key, signals=self.signals ) self.bot.init_browser() except Exception as e: self.signals.log_signal.emit(f"❌ 浏览器启动失败: {e}") self.run_in_thread(_task) def _get_selected_row(self): row = self.table.currentRow() if row < 0: QMessageBox.warning(self, "警告", "请先在表格中点击选中需要操作的虚拟账号！") return None return row def action_register(self): row = self._get_selected_row() if row is None: return if not self.bot: QMessageBox.warning(self, "警告", "请先启动浏览器！") return profile = self.profiles[row] def _task(): try: self.bot.account_detail = profile self.signals.status_signal.emit(row, "注册中...") sent_at = datetime.utcnow().strftime("%Y-%m-%d %H:%M:%S") self.bot.register() self.signals.status_signal.emit(row, "激活中...") self.bot.activate(sent_at=sent_at) self.signals.status_signal.emit(row, "注册激活完成") except Exception as e: self.signals.log_signal.emit(f"❌ 注册过程异常: {e}") self.signals.status_signal.emit(row, "注册失败") self.run_in_thread(_task) def action_fill(self): row = self._get_selected_row() if row is None: return if not self.bot: QMessageBox.warning(self, "警告", "请先启动浏览器！") return profile = self.profiles[row] def _task(): try: self.bot.account_detail = profile self.signals.status_signal.emit(row, "填表中...") self.bot.make_account_useful() self.signals.status_signal.emit(row, "填表完成") except Exception as e: self.signals.log_signal.emit(f"❌ 填表过程异常: {e}") self.signals.status_signal.emit(row, "填表失败") self.run_in_thread(_task) def action_upload(self): row = self._get_selected_row() if row is None: return if not self.bot: return profile = self.profiles[row] def _task(): try: self.bot.account_detail = profile self.signals.status_signal.emit(row, "上报中...") success = self.bot.upload_account_to_server() if success: self.signals.status_signal.emit(row, "已使用(上报成功)") else: self.signals.status_signal.emit(row, "上报失败") except Exception as e: self.signals.log_signal.emit(f"❌ 上报异常: {e}") self.signals.status_signal.emit(row, "上报异常") self.run_in_thread(_task) def action_close_browser(self): if self.bot: self.bot.cleanup() self.bot = None self.append_log("✅ 浏览器已关闭。") def closeEvent(self, event): """主窗口关闭时，确保清理资源""" if self.bot: self.bot.cleanup() event.accept() if __name__ == "__main__": app = QApplication(sys.argv) window = MainWindow() window.show() sys.exit(app.exec_())