"""PySide6 GUI entry point.""" from __future__ import annotations import os import sys try: from PySide6.QtCore import QAbstractTableModel, QModelIndex, Qt from PySide6.QtWidgets import ( QAbstractItemView, QApplication, QDialog, QDialogButtonBox, QFileDialog, QFormLayout, QHBoxLayout, QHeaderView, QLabel, QLineEdit, QMainWindow, QMessageBox, QPlainTextEdit, QPushButton, QTableView, 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, db, excel from . import config as account_config def _database_path(db_path=None, config=None) -> str: return db_path or appconfig.db_path(config) class TaskTableModel(QAbstractTableModel): """Table model for task rows shared by workflow tabs.""" HEADERS = ["账号", "别名", "商品ID", "阶段"] STAGE_TEXT = { "imported": "待采集", "collected": "已采集", "generated": "已生成", "applied": "已更新", } STATUS_TEXT = { "running": "处理中", "failed": "失败", "skipped": "略过", "cancelled": "已取消", } def __init__(self, parent=None): super().__init__(parent) self.tasks = [] self.account_by_alias = {} def set_tasks(self, tasks, accounts): self.beginResetModel() self.tasks = list(tasks) self.account_by_alias = { str(account.alias).strip(): account for account in accounts if str(account.alias).strip() } self.endResetModel() def rowCount(self, parent=QModelIndex()): return 0 if parent.isValid() else len(self.tasks) def columnCount(self, parent=QModelIndex()): return 0 if parent.isValid() else len(self.HEADERS) def headerData(self, section, orientation, role=Qt.DisplayRole): if role != Qt.DisplayRole: return None if orientation == Qt.Horizontal and 0 <= section < len(self.HEADERS): return self.HEADERS[section] return section + 1 if orientation == Qt.Vertical else None def data(self, index, role=Qt.DisplayRole): if not index.isValid(): return None task = self.tasks[index.row()] if role == Qt.DisplayRole: return self._display_value(task, index.column()) if role == Qt.ToolTipRole and self.is_unmatched(task): return "别名未匹配账号，采集时将略过" return None def flags(self, index): if not index.isValid(): return Qt.NoItemFlags return Qt.ItemIsEnabled | Qt.ItemIsSelectable def task_at(self, row): if row < 0 or row >= len(self.tasks): return None return self.tasks[row] def is_unmatched(self, task) -> bool: return str(task.alias).strip() not in self.account_by_alias def unmatched_count(self) -> int: return sum(1 for task in self.tasks if self.is_unmatched(task)) def _account_name(self, task) -> str: account = self.account_by_alias.get(str(task.alias).strip()) if account is not None: return account.account_name return task.account_name or "" def _stage_text(self, task) -> str: if self.is_unmatched(task): return "略过" if task.status in self.STATUS_TEXT and task.status != "pending": return self.STATUS_TEXT[task.status] return self.STAGE_TEXT.get(task.stage, task.stage) def _display_value(self, task, column): values = [ self._account_name(task), task.alias, task.item_id, self._stage_text(task), ] return values[column] if 0 <= column < len(values) else None class CollectTab(QWidget): """Tab 1: import Excel files and list imported tasks.""" 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.current_batch_id = None self.import_button = QPushButton("导入 Excel...") self.refresh_button = QPushButton("刷新") toolbar = QHBoxLayout() toolbar.addWidget(self.import_button) toolbar.addWidget(self.refresh_button) toolbar.addStretch(1) self.model = TaskTableModel(self) self.table = QTableView() self.table.setModel(self.model) 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.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.import_button.clicked.connect(self.import_excel) self.refresh_button.clicked.connect(self.refresh_tasks) self.refresh_tasks() def _set_status(self, message): if self.status_callback is not None: self.status_callback(message) def _show_error(self, message): QMessageBox.warning(self, "导入采集", str(message)) self._set_status(str(message)) def _choose_excel_files(self): files, _selected_filter = QFileDialog.getOpenFileNames( self, "选择 Excel 文件", "", "Excel 文件 (*.xlsx *.xlsm)", ) return files def import_excel(self, checked=False): file_paths = self._choose_excel_files() if not file_paths: return try: result = excel.import_tasks(file_paths, path=self.db_path) except Exception as exc: self._show_error(exc) return if result.get("batch_id"): self.current_batch_id = result["batch_id"] self.refresh_tasks() stats = result.get("stats") or {} self._set_status( "导入完成：有效{valid}，无效{invalid}，入库{inserted}，未匹配{unmatched}".format( valid=stats.get("valid", 0), invalid=stats.get("invalid", 0), inserted=stats.get("inserted", 0), unmatched=self.model.unmatched_count(), ) ) def refresh_tasks(self, checked=False): try: db.init_db(self.db_path) task_rows = db.list_tasks(batch_id=self.current_batch_id, path=self.db_path) account_rows = accounts.list_accounts(path=self.db_path, config=self.config) except Exception as exc: self.model.set_tasks([], []) self.empty_label.setText("任务读取失败") self._set_status(f"任务读取失败：{exc}") return self.model.set_tasks(task_rows, account_rows) if task_rows: unmatched = self.model.unmatched_count() self.empty_label.setText( "" if unmatched == 0 else f"{unmatched} 条任务别名未匹配账号，阶段显示为“略过”" ) else: self.empty_label.setText("暂无任务") 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("检测登录") 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): 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, 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), 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}") 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}") class MainWindow(QMainWindow): """Main application window with the fixed five-tab workflow.""" def __init__(self, db_path=None, config=None): super().__init__() self.config = appconfig.load_config() if config is None else config self.db_path = _database_path(db_path, self.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 CollectTab( db_path=self.db_path, config=self.config, status_callback=self.statusBar().showMessage, ) 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()