"""Tab 4: account management UI.""" from __future__ import annotations from ..widgets import * from ..workers import AccountLoginCheckWorker as _RealAccountLoginCheckWorker class AccountDialog(QDialog): """Dialog for adding or editing one account.""" def __init__(self, parent=None, account=None, default_port=9222, config=None): super().__init__(parent) self._account = account self._config = config self.setWindowTitle("编辑账号" if account else "新增账号") self.account_name_edit = QLineEdit() self.alias_edit = QLineEdit() self.region_host_edit = QLineEdit(accounts.DEFAULT_REGION_HOST) self.debug_port_spin = QSpinBox() self.debug_port_spin.setRange(1, 65535) self.debug_port_spin.setValue(int(default_port)) self.password_edit = QLineEdit() self.password_edit.setEchoMode(QLineEdit.Password) self.note_edit = QPlainTextEdit() self.note_edit.setMaximumHeight(76) self.slug_edit = QLineEdit() self.slug_edit.setReadOnly(True) self.user_data_dir_edit = QLineEdit() self.user_data_dir_edit.setReadOnly(True) if account is not None: self.account_name_edit.setText(account.account_name) self.alias_edit.setText(account.alias) self.region_host_edit.setText(account.region_host) self.debug_port_spin.setValue(int(account.debug_port)) self.password_edit.setText(account.password or "") self.note_edit.setPlainText(account.note or "") self.slug_edit.setText(account.slug) self.user_data_dir_edit.setText(account.user_data_dir) form = QFormLayout() form.addRow("账号名", self.account_name_edit) form.addRow("别名", self.alias_edit) form.addRow("地区", self.region_host_edit) form.addRow("调试端口", self.debug_port_spin) form.addRow("密码", self.password_edit) form.addRow("备注", self.note_edit) form.addRow("slug", self.slug_edit) form.addRow("数据目录", self.user_data_dir_edit) buttons = QDialogButtonBox(QDialogButtonBox.Ok | QDialogButtonBox.Cancel) buttons.accepted.connect(self.accept) buttons.rejected.connect(self.reject) layout = QVBoxLayout(self) layout.addLayout(form) layout.addWidget(buttons) self.alias_edit.textChanged.connect(self._update_path_preview) self._update_path_preview() def _update_path_preview(self): alias = self.alias_edit.text().strip() if not alias: self.slug_edit.clear() self.user_data_dir_edit.clear() return try: slug = account_config.make_slug(alias) self.slug_edit.setText(slug) if self._account is not None and alias == self._account.alias: self.user_data_dir_edit.setText(self._account.user_data_dir) else: self.user_data_dir_edit.setText( accounts.preview_user_data_dir(alias, config=self._config) ) except Exception: self.slug_edit.clear() self.user_data_dir_edit.clear() def values(self): return { "account_name": self.account_name_edit.text().strip(), "alias": self.alias_edit.text().strip(), "region_host": self.region_host_edit.text().strip(), "debug_port": self.debug_port_spin.value(), "password": self.password_edit.text(), "note": self.note_edit.toPlainText().strip(), } def _new_account_dialog(*args, **kwargs): dialog_class = _package_attr("AccountDialog", AccountDialog) return dialog_class(*args, **kwargs) def AccountLoginCheckWorker(*args, **kwargs): return _call_package_attr("AccountLoginCheckWorker", _RealAccountLoginCheckWorker, *args, **kwargs) class AccountsTab(QWidget): COLUMNS = ["账号名", "别名", "地区", "端口", "登录状态", "备注"] def __init__(self, parent=None, db_path=None, config=None, status_callback=None): super().__init__(parent) self.config = appconfig.load_config() if config is None else config self.db_path = _database_path(db_path, self.config) self.status_callback = status_callback self.account_rows = [] self.login_statuses = {} self.threads = [] self.table = QTableWidget(0, len(self.COLUMNS)) self.table.setHorizontalHeaderLabels(self.COLUMNS) self.table.setSelectionBehavior(QAbstractItemView.SelectRows) self.table.setSelectionMode(QAbstractItemView.SingleSelection) self.table.setEditTriggers(QAbstractItemView.NoEditTriggers) self.table.horizontalHeader().setSectionResizeMode(QHeaderView.Stretch) self.table.verticalHeader().setVisible(False) self.add_button = QPushButton("新增") self.edit_button = QPushButton("编辑") self.delete_button = QPushButton("删除") self.delete_button.setObjectName("deleteAccountButton") self.delete_button.setStyleSheet(_danger_outline_button_style("deleteAccountButton")) self.launch_button = QPushButton("启动登录") self.check_button = QPushButton("检测登录") self.shortcut_button = QPushButton("快捷方式") toolbar = QHBoxLayout() for button in ( self.add_button, self.edit_button, self.delete_button, self.launch_button, self.check_button, self.shortcut_button, ): toolbar.addWidget(button) toolbar.addStretch(1) self.empty_label = QLabel("") layout = QVBoxLayout(self) layout.setContentsMargins(18, 18, 18, 18) layout.addLayout(toolbar) layout.addWidget(self.table, 1) layout.addWidget(self.empty_label) self.add_button.clicked.connect(self.add_account) self.edit_button.clicked.connect(self.edit_account) self.delete_button.clicked.connect(self.delete_account) self.launch_button.clicked.connect(self.launch_login) self.check_button.clicked.connect(self.check_login) self.shortcut_button.clicked.connect(self.create_shortcut) self.table.itemSelectionChanged.connect(self._update_button_state) self.table.doubleClicked.connect(self.edit_account) self.refresh_accounts() def _set_status(self, message, level=None): _emit_status(self.status_callback, message, level=level) def _selected_account(self): selected = self.table.selectionModel().selectedRows() if not selected: return None row = selected[0].row() if row < 0 or row >= len(self.account_rows): return None return self.account_rows[row] def _update_button_state(self): has_selection = self._selected_account() is not None for button in ( self.edit_button, self.delete_button, self.launch_button, self.check_button, self.shortcut_button, ): button.setEnabled(has_selection) def refresh_accounts(self): try: self.account_rows = accounts.list_accounts( path=self.db_path, config=self.config, ) except Exception as exc: self.account_rows = [] self._set_status(f"账号读取失败：{exc}") self.table.setRowCount(len(self.account_rows)) for row, account in enumerate(self.account_rows): status = self.login_statuses.get(account.alias, "未知") values = [ account.account_name, account.alias, account.region_host, str(account.debug_port), _login_status_display(status), account.note or "", ] for column, value in enumerate(values): item = QTableWidgetItem(value) if column == 4: item.setForeground(_login_status_color(status)) self.table.setItem(row, column, item) self.empty_label.setText("" if self.account_rows else "暂无账号") self._update_button_state() def _show_error(self, message): QMessageBox.warning(self, "账号管理", str(message)) self._set_status(str(message)) def add_account(self, checked=False): try: default_port = accounts.next_debug_port( path=self.db_path, config=self.config, ) except Exception: default_port = appconfig.default_debug_port(self.config) dialog = _new_account_dialog(self, default_port=default_port, config=self.config) if dialog.exec() != QDialog.Accepted: return values = dialog.values() if self._should_warn_plaintext_password(values): self._show_plaintext_password_warning() try: accounts.create_account( path=self.db_path, config=self.config, **values, ) except Exception as exc: self._show_error(exc) return self.refresh_accounts() self._set_status("账号已新增") def edit_account(self, checked=False): account = self._selected_account() if account is None: return dialog = _new_account_dialog( self, account=account, default_port=account.debug_port, config=self.config, ) if dialog.exec() != QDialog.Accepted: return values = dialog.values() if self._should_warn_plaintext_password(values, account): self._show_plaintext_password_warning() try: updated = accounts.update_account( account.alias, path=self.db_path, config=self.config, **values, ) except Exception as exc: self._show_error(exc) return if updated.alias != account.alias: self.login_statuses.pop(account.alias, None) self.refresh_accounts() self._set_status("账号已保存") def delete_account(self, checked=False): account = self._selected_account() if account is None: return answer = QMessageBox.question( self, "删除账号", f"确认删除账号「{account.alias}」？", QMessageBox.Yes | QMessageBox.No, QMessageBox.No, ) if answer != QMessageBox.Yes: return try: accounts.delete_account(account.alias, path=self.db_path, config=self.config) except Exception as exc: self._show_error(exc) return self.login_statuses.pop(account.alias, None) self.refresh_accounts() self._set_status("账号已删除") def launch_login(self, checked=False): account = self._selected_account() if account is None: return run_id = _safe_create_run_log( "chrome_launch", db_path=self.db_path, total=1, options={"alias": account.alias, "debug_port": account.debug_port}, ) started = time.monotonic() _safe_add_run_log_event( run_id, f"step=launch_chrome result=start detail=账号 {account.alias} debug_port={account.debug_port}", db_path=self.db_path, account=account, ) try: launch_result = accounts.launch_for_login(account, config=self.config) except Exception as exc: elapsed_ms = _elapsed_ms(started) safe_error = diagnostics.redact_log_text(str(exc) or exc.__class__.__name__) _safe_add_run_log_event( run_id, f"step=launch_chrome result=failed detail={safe_error} elapsed_ms={elapsed_ms}", db_path=self.db_path, account=account, level="error", ) _safe_write_diagnostic_log( "Chrome启动失败", level="ERROR", step="launch_chrome", account=account, elapsed_ms=elapsed_ms, payload={"alias": account.alias, "debug_port": account.debug_port, "error": safe_error}, exc=exc, log_dir=diagnostics.DEFAULT_LOG_DIR, ) _safe_finish_run_log( run_id, db_path=self.db_path, status="failed", done=0, failed_count=1, summary_json={"ok": False, "alias": account.alias, "error": safe_error}, ) self._show_error(safe_error) return elapsed_ms = _elapsed_ms(started) action = (launch_result or {}).get("action") or "launched" pid = (launch_result or {}).get("pid") target_id = (launch_result or {}).get("target_id") or "" url = (launch_result or {}).get("url") or "" event_result = "reused" if action == "reused" else "launched" _safe_add_run_log_event( run_id, ( f"step=launch_chrome result={event_result} detail=账号 {account.alias} " f"pid={pid or ''} target_id={target_id} url={url} elapsed_ms={elapsed_ms}" ), db_path=self.db_path, account=account, ) _safe_finish_run_log( run_id, db_path=self.db_path, status="done", done=1, success_count=1, failed_count=0, summary_json={ "ok": True, "alias": account.alias, "debug_port": account.debug_port, "action": action, "pid": pid, "target_id": target_id, "url": url, }, ) self.login_statuses[account.alias] = "已启动" self.refresh_accounts() if action == "reused": self._set_status("该账号 Chrome 已打开，已复用现有窗口") else: self._set_status("Chrome 已启动，请人工登录") def check_login(self, checked=False): account = self._selected_account() if account is None: return self.login_statuses[account.alias] = "检测中" self.refresh_accounts() worker = AccountLoginCheckWorker( account, db_path=self.db_path, config=self.config, diagnostic_log_dir=diagnostics.DEFAULT_LOG_DIR, ) worker.finished.connect(self._on_login_check_finished) worker.failed.connect( lambda _task_id, error, alias=account.alias: self._on_login_check_failed( alias, error, ) ) thread = run_worker(worker, start=False) thread.finished.connect(lambda: self._forget_thread(thread)) self.threads.append(thread) thread.start() def _forget_thread(self, thread): if thread in self.threads: self.threads.remove(thread) def _on_login_check_finished(self, payload): if payload.get("ok") is False and not payload.get("alias"): return alias = payload.get("alias") status = payload.get("status") or {} if alias: self.login_statuses[alias] = accounts.login_status_text(status) self.refresh_accounts() self._set_status("登录状态已刷新") def _on_login_check_failed(self, alias, error): self.login_statuses[alias] = "检测失败" self.refresh_accounts() self._set_status(f"登录检测失败：{error}") def create_shortcut(self, checked=False): account = self._selected_account() if account is None: return try: shortcut_path = accounts.create_shortcut(account, config=self.config) except Exception as exc: self._show_error(exc) return self._set_status(f"快捷方式已生成：{shortcut_path}") QMessageBox.information(self, "账号管理", f"快捷方式已生成：\n{shortcut_path}") def _should_warn_plaintext_password(self, values, account=None): new_password = str((values or {}).get("password") or "") current_password = str(getattr(account, "password", None) or "") return bool(new_password) and new_password != current_password def _show_plaintext_password_warning(self): QMessageBox.warning( self, PLAINTEXT_SECRET_TITLE, PLAINTEXT_PASSWORD_WARNING, )