"""PySide6 GUI entry point.""" from __future__ import annotations import os import sys try: from PySide6.QtWidgets import ( QAbstractItemView, QApplication, QDialog, QDialogButtonBox, QFormLayout, QHBoxLayout, QHeaderView, QLabel, QLineEdit, QMainWindow, QMessageBox, QPlainTextEdit, QPushButton, QSpinBox, QTableWidget, QTableWidgetItem, QTabWidget, QVBoxLayout, QWidget, ) QT_IMPORT_ERROR = None except ModuleNotFoundError as exc: QApplication = None QMainWindow = object QTabWidget = None QVBoxLayout = None QWidget = object QT_IMPORT_ERROR = exc TAB_TITLES = [ "① 导入采集", "② AI生成", "③ 更新shopee", "④ 账号管理", "⑤ 设置", ] TAB_STYLE = """ QTabWidget::pane { border-top: 1px solid #c9d1d9; } QTabBar::tab { min-width: 128px; min-height: 34px; padding: 8px 18px; margin-right: 8px; border: 1px solid #c9d1d9; border-bottom-color: #b8c0ca; background: #f4f6f8; color: #24292f; } QTabBar::tab:selected { background: #ffffff; border-color: #687785; border-bottom-color: #ffffff; font-weight: 600; } QTabBar::tab:hover:!selected { background: #eaf2ff; } """ if QT_IMPORT_ERROR is None: from . import accounts, appconfig from . import config as account_config 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(), } from .workers import BaseWorker, run_worker class AccountLoginCheckWorker(BaseWorker): def __init__(self, account, db_path=None, config=None, timeout=8): super().__init__() self.account = account self.db_path = db_path self.config = config self.timeout = timeout def execute(self): status = accounts.detect_login( self.account, timeout=self.timeout, path=self.db_path, config=self.config, ) self.row_updated.emit(self.account.id, status) return {"alias": self.account.alias, "status": status} class AccountsTab(QWidget): COLUMNS = ["账号名", "别名", "地区", "端口", "登录状态", "备注"] def __init__(self, parent=None, db_path=None, config=None, status_callback=None): super().__init__(parent) self.db_path = db_path self.config = appconfig.load_config() if config is None else 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.launch_button = QPushButton("启动登录") self.check_button = QPushButton("检测登录") toolbar = QHBoxLayout() for button in ( self.add_button, self.edit_button, self.delete_button, self.launch_button, self.check_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.table.itemSelectionChanged.connect(self._update_button_state) self.table.doubleClicked.connect(self.edit_account) self.refresh_accounts() def _set_status(self, message): if self.status_callback is not None: self.status_callback(message) 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, ): 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), status, account.note or "", ] for column, value in enumerate(values): item = QTableWidgetItem(value) 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 = AccountDialog(self, default_port=default_port, config=self.config) if dialog.exec() != QDialog.Accepted: return try: accounts.create_account( path=self.db_path, config=self.config, **dialog.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 = AccountDialog( self, account=account, default_port=account.debug_port, config=self.config, ) if dialog.exec() != QDialog.Accepted: return try: updated = accounts.update_account( account.alias, path=self.db_path, config=self.config, **dialog.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 try: accounts.launch_for_login(account, config=self.config) except Exception as exc: self._show_error(exc) return self.login_statuses[account.alias] = "已启动" self.refresh_accounts() 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, ) 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}") class MainWindow(QMainWindow): """Main application window with the fixed five-tab workflow.""" def __init__(self, db_path=None, config=None): super().__init__() self.db_path = db_path self.config = appconfig.load_config() if config is None else config self.setWindowTitle("cmshopee") self.resize(1180, 760) self.tabs = QTabWidget() self.tabs.setObjectName("mainTabs") self.tabs.setStyleSheet(TAB_STYLE) self.tabs.currentChanged.connect(self._on_tab_changed) for title in TAB_TITLES: self.tabs.addTab(self._build_tab(title), title) self.setCentralWidget(self.tabs) self.statusBar().showMessage("就绪") def _build_tab(self, title): if title == "④ 账号管理": return AccountsTab( db_path=self.db_path, config=self.config, status_callback=self.statusBar().showMessage, ) widget = QWidget() widget.setObjectName(title) layout = QVBoxLayout(widget) layout.setContentsMargins(18, 18, 18, 18) layout.addStretch(1) return widget def _on_tab_changed(self, index): self.statusBar().showMessage(f"当前：{self.tabs.tabText(index)}") else: class MainWindow(QMainWindow): def __init__(self): raise RuntimeError("PySide6 未安装，无法启动 GUI") def _ensure_offscreen_for_headless_tests(): if "PYTEST_CURRENT_TEST" in os.environ and "QT_QPA_PLATFORM" not in os.environ: os.environ["QT_QPA_PLATFORM"] = "offscreen" def main() -> int: if QT_IMPORT_ERROR is not None: print("cmshopee GUI 无法启动：当前 Python 环境未安装 PySide6。") return 1 _ensure_offscreen_for_headless_tests() app = QApplication.instance() or QApplication(sys.argv) window = MainWindow() window.show() return app.exec()