cmshoppe/app/gui/tabs/accounts.py

"""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,
        )