645 lines
17 KiB
Python
645 lines
17 KiB
Python
"""PySide6 GUI entry point."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import sys
|
|
import threading
|
|
import time
|
|
from concurrent.futures import ThreadPoolExecutor, as_completed
|
|
|
|
try:
|
|
from PySide6.QtCore import QAbstractTableModel, QModelIndex, Qt, QTimer
|
|
from PySide6.QtGui import QColor, QIcon, QImage, QPainter, QPixmap
|
|
from PySide6.QtWidgets import (
|
|
QAbstractItemView,
|
|
QApplication,
|
|
QButtonGroup,
|
|
QCheckBox,
|
|
QComboBox,
|
|
QDialog,
|
|
QDialogButtonBox,
|
|
QFileDialog,
|
|
QFormLayout,
|
|
QGridLayout,
|
|
QHBoxLayout,
|
|
QHeaderView,
|
|
QInputDialog,
|
|
QLabel,
|
|
QLineEdit,
|
|
QMainWindow,
|
|
QMenu,
|
|
QMessageBox,
|
|
QPlainTextEdit,
|
|
QProgressBar,
|
|
QPushButton,
|
|
QRadioButton,
|
|
QScrollArea,
|
|
QSplitter,
|
|
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生成",
|
|
"③ 更新蝦皮",
|
|
"账号管理",
|
|
"设置",
|
|
"商品套图",
|
|
]
|
|
|
|
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;
|
|
}
|
|
"""
|
|
COLOR_SUCCESS = "#1a7f37"
|
|
COLOR_DANGER = "#cf222e"
|
|
COLOR_INFO = "#0969da"
|
|
COLOR_PENDING = "#9a6700"
|
|
COLOR_MUTED = "#6e7781"
|
|
COLOR_WARNING = "#bc4c00"
|
|
COLOR_DANGER_BG = "#ffebe9"
|
|
STATUS_LEVEL_COLORS = {
|
|
"muted": COLOR_MUTED,
|
|
"info": COLOR_INFO,
|
|
"success": COLOR_SUCCESS,
|
|
"warning": COLOR_WARNING,
|
|
"danger": COLOR_DANGER,
|
|
"pending": COLOR_PENDING,
|
|
}
|
|
BUTTON_RADIUS_PX = 4
|
|
BUTTON_BASE_STYLE = f"""
|
|
QPushButton {{
|
|
min-height: 24px;
|
|
min-width: 72px;
|
|
padding: 4px 10px;
|
|
border: 1px solid #d0d7de;
|
|
border-radius: {BUTTON_RADIUS_PX}px;
|
|
background-color: #f6f8fa;
|
|
color: #24292f;
|
|
}}
|
|
QPushButton:hover {{
|
|
background-color: #eef4ff;
|
|
border-color: #8c959f;
|
|
}}
|
|
QPushButton:pressed {{
|
|
background-color: #d8dee4;
|
|
border-color: #6e7781;
|
|
}}
|
|
QPushButton:focus {{
|
|
border-color: #0969da;
|
|
}}
|
|
QPushButton:disabled {{
|
|
background-color: #f6f8fa;
|
|
border-color: #d8dee4;
|
|
color: #8c959f;
|
|
}}
|
|
"""
|
|
|
|
|
|
|
|
_RawQMessageBox = QMessageBox if QT_IMPORT_ERROR is None else None
|
|
|
|
|
|
class _GuiAttributeProxy:
|
|
def __init__(self, name, fallback):
|
|
self._name = name
|
|
self._fallback = fallback
|
|
|
|
def _target(self):
|
|
package = sys.modules.get("app.gui")
|
|
if package is not None:
|
|
value = getattr(package, self._name, self._fallback)
|
|
if value is not self:
|
|
return value
|
|
return self._fallback
|
|
|
|
def __getattr__(self, item):
|
|
return getattr(self._target(), item)
|
|
|
|
def __call__(self, *args, **kwargs):
|
|
return self._target()(*args, **kwargs)
|
|
|
|
|
|
def _package_attr(name, fallback):
|
|
package = sys.modules.get("app.gui")
|
|
if package is not None:
|
|
value = getattr(package, name, fallback)
|
|
if value is not fallback:
|
|
return value
|
|
return fallback
|
|
|
|
|
|
def _call_package_attr(name, fallback, *args, **kwargs):
|
|
return _package_attr(name, fallback)(*args, **kwargs)
|
|
|
|
|
|
QMessageBox = _GuiAttributeProxy("QMessageBox", _RawQMessageBox)
|
|
from .. import accounts, ai, appconfig, chrome, db, diagnostics, editor, excel, image_paths, prompts
|
|
from .. import config as account_config
|
|
from ..workers import BaseWorker, run_worker as _raw_run_worker
|
|
|
|
|
|
def run_worker(*args, **kwargs):
|
|
return _package_attr("run_worker", _raw_run_worker)(*args, **kwargs)
|
|
|
|
|
|
PLAINTEXT_SECRET_TITLE = "本地明文保存提示"
|
|
PLAINTEXT_API_KEY_WARNING = (
|
|
"API Key 会以本地明文保存到 data/config/ai_models.json，仅供本机调用 AI 使用。"
|
|
"该文件已 gitignore，UI 打码显示，日志/导出不记录明文。"
|
|
)
|
|
PLAINTEXT_PASSWORD_WARNING = (
|
|
"密码会以本地明文保存到本地 SQLite，仅供人工参考，不会自动登录/自动填。"
|
|
"数据库文件已 gitignore，请勿提交或分享。"
|
|
)
|
|
|
|
|
|
def _qcolor(color):
|
|
return QColor(color)
|
|
|
|
|
|
def _status_base_color(status):
|
|
if status == "running":
|
|
return COLOR_INFO
|
|
if status == "failed":
|
|
return COLOR_DANGER
|
|
if status in {"skipped", "cancelled"}:
|
|
return COLOR_MUTED
|
|
return None
|
|
|
|
|
|
def _status_level_color(level):
|
|
return STATUS_LEVEL_COLORS.get(str(level or "muted"), COLOR_MUTED)
|
|
|
|
|
|
def _status_level_for_message(message, default="muted"):
|
|
text = str(message or "")
|
|
if any(keyword in text for keyword in ("失败", "错误", "异常", "未登录", "不可写", "点数不足", "阻止")):
|
|
return "danger"
|
|
if any(keyword in text for keyword in ("没有", "请先", "未配置", "不完整", "已取消", "不能", "无法")):
|
|
return "warning"
|
|
if any(
|
|
keyword in text
|
|
for keyword in (
|
|
"完成",
|
|
"成功",
|
|
"已保存",
|
|
"已回写",
|
|
"已新增",
|
|
"已删除",
|
|
"已软删除",
|
|
"已重置",
|
|
"已刷新",
|
|
"已修改",
|
|
"已启动",
|
|
"已复用",
|
|
"已生成",
|
|
"已新建",
|
|
"已另存为",
|
|
"已重命名",
|
|
)
|
|
):
|
|
return "success"
|
|
if any(keyword in text for keyword in ("正在", "开始", "进度", "检测中")):
|
|
return "info"
|
|
return default
|
|
|
|
|
|
def _emit_status(callback, message, level=None):
|
|
if callback is None:
|
|
return
|
|
actual_level = level or _status_level_for_message(message)
|
|
try:
|
|
callback(str(message), level=actual_level)
|
|
except TypeError:
|
|
callback(str(message))
|
|
|
|
|
|
def _danger_metric_text(text, active):
|
|
if not active:
|
|
return text
|
|
return f'<span style="color:{COLOR_DANGER}; font-weight:600;">{text}</span>'
|
|
|
|
|
|
def _outline_button_style(object_name, color, hover_bg="#f6f8fa", pressed_bg="#f6f8fa"):
|
|
return (
|
|
f"QPushButton#{object_name} {{ "
|
|
f"color: {color}; border-color: {color}; font-weight: 600; "
|
|
"}"
|
|
f"QPushButton#{object_name}:hover {{ "
|
|
f"background-color: {hover_bg}; border-color: {color}; "
|
|
"}"
|
|
f"QPushButton#{object_name}:pressed {{ "
|
|
f"background-color: {pressed_bg}; border-color: {color}; "
|
|
"}"
|
|
f"QPushButton#{object_name}:disabled {{ "
|
|
"color: #8c959f; border-color: #d8dee4; "
|
|
"}"
|
|
)
|
|
|
|
|
|
def _warning_outline_button_style(object_name):
|
|
return _outline_button_style(
|
|
object_name,
|
|
COLOR_WARNING,
|
|
hover_bg="#fff8f0",
|
|
pressed_bg="#ffedd5",
|
|
)
|
|
|
|
|
|
def _danger_outline_button_style(object_name):
|
|
return _outline_button_style(
|
|
object_name,
|
|
COLOR_DANGER,
|
|
hover_bg="#fff1f0",
|
|
pressed_bg=COLOR_DANGER_BG,
|
|
)
|
|
|
|
|
|
def _primary_button_style(object_name):
|
|
return (
|
|
f"QPushButton#{object_name} {{ "
|
|
f"background-color: {COLOR_INFO}; border-color: {COLOR_INFO}; "
|
|
"color: white; font-weight: 600; "
|
|
"}"
|
|
f"QPushButton#{object_name}:hover {{ "
|
|
"background-color: #0550ae; border-color: #0550ae; "
|
|
"}"
|
|
f"QPushButton#{object_name}:pressed {{ "
|
|
"background-color: #033d8b; border-color: #033d8b; "
|
|
"}"
|
|
)
|
|
|
|
|
|
class ProductStatusScopeDialog(QDialog):
|
|
"""供采集和生成复用的稳定中文范围选择框。"""
|
|
|
|
MINIMUM_WIDTH = 520
|
|
|
|
def __init__(
|
|
self,
|
|
title,
|
|
text,
|
|
informative_text,
|
|
normal_text,
|
|
all_text,
|
|
normal_value,
|
|
all_value,
|
|
normal_object_name,
|
|
all_object_name,
|
|
cancel_object_name,
|
|
all_enabled=True,
|
|
parent=None,
|
|
):
|
|
super().__init__(parent)
|
|
self._choice = None
|
|
self._informative_text = str(informative_text or "")
|
|
self.setWindowTitle(str(title or "选择范围"))
|
|
self.setModal(True)
|
|
self.setMinimumWidth(self.MINIMUM_WIDTH)
|
|
self.setStyleSheet(BUTTON_BASE_STYLE)
|
|
|
|
layout = QVBoxLayout(self)
|
|
layout.setContentsMargins(20, 18, 20, 18)
|
|
layout.setSpacing(10)
|
|
|
|
title_label = QLabel(str(text or ""))
|
|
title_label.setObjectName("productStatusScopeTitleLabel")
|
|
title_label.setWordWrap(True)
|
|
title_label.setStyleSheet("font-weight: 600; color: #24292f;")
|
|
layout.addWidget(title_label)
|
|
|
|
detail_label = QLabel(self._informative_text)
|
|
detail_label.setObjectName("productStatusScopeDetailLabel")
|
|
detail_label.setWordWrap(True)
|
|
detail_label.setStyleSheet("color: #57606a;")
|
|
layout.addWidget(detail_label)
|
|
|
|
layout.addSpacing(2)
|
|
self.normal_button = QPushButton(str(normal_text or ""))
|
|
self.normal_button.setObjectName(str(normal_object_name))
|
|
self.normal_button.setMinimumWidth(self.MINIMUM_WIDTH - 40)
|
|
self.normal_button.setMinimumHeight(34)
|
|
self.normal_button.setDefault(True)
|
|
self.normal_button.setFocus()
|
|
self.normal_button.setStyleSheet(_primary_button_style(normal_object_name))
|
|
self.normal_button.clicked.connect(lambda: self._accept_choice(normal_value))
|
|
layout.addWidget(self.normal_button)
|
|
|
|
self.all_button = QPushButton(str(all_text or ""))
|
|
self.all_button.setObjectName(str(all_object_name))
|
|
self.all_button.setMinimumWidth(self.MINIMUM_WIDTH - 40)
|
|
self.all_button.setMinimumHeight(34)
|
|
self.all_button.setEnabled(bool(all_enabled))
|
|
self.all_button.setStyleSheet(_warning_outline_button_style(all_object_name))
|
|
self.all_button.clicked.connect(lambda: self._accept_choice(all_value))
|
|
layout.addWidget(self.all_button)
|
|
|
|
self.cancel_button = QPushButton("取消")
|
|
self.cancel_button.setObjectName(str(cancel_object_name))
|
|
self.cancel_button.setMinimumWidth(self.MINIMUM_WIDTH - 40)
|
|
self.cancel_button.setMinimumHeight(32)
|
|
self.cancel_button.clicked.connect(self.reject)
|
|
layout.addWidget(self.cancel_button)
|
|
|
|
def defaultButton(self):
|
|
"""供调用方和 GUI 测试读取明确的默认按钮。"""
|
|
|
|
return self.normal_button
|
|
|
|
def informativeText(self):
|
|
"""按 QMessageBox 兼容接口返回说明文字。"""
|
|
|
|
return self._informative_text
|
|
|
|
def choice(self):
|
|
return self._choice
|
|
|
|
def _accept_choice(self, value):
|
|
self._choice = value
|
|
self.accept()
|
|
|
|
|
|
def _login_status_display(status):
|
|
return f"● {status or '未知'}"
|
|
|
|
|
|
def _login_status_color(status):
|
|
status_text = str(status or "")
|
|
if status_text == "已登录":
|
|
return _qcolor(COLOR_SUCCESS)
|
|
if status_text == "检测中":
|
|
return _qcolor(COLOR_INFO)
|
|
if "未登录" in status_text or "失败" in status_text:
|
|
return _qcolor(COLOR_DANGER)
|
|
return _qcolor(COLOR_MUTED)
|
|
|
|
|
|
def _build_empty_state_card(object_name, button_text="前往账号管理"):
|
|
card = QWidget()
|
|
card.setObjectName(object_name)
|
|
card.setStyleSheet(
|
|
f"QWidget#{object_name} {{ "
|
|
"background: #f6f8fa; border: 1px solid #d0d7de; "
|
|
"border-radius: 6px; padding: 10px; "
|
|
"}"
|
|
)
|
|
layout = QHBoxLayout(card)
|
|
layout.setContentsMargins(12, 10, 12, 10)
|
|
label = QLabel("")
|
|
label.setWordWrap(True)
|
|
button = QPushButton(button_text)
|
|
button.setObjectName(f"{object_name}Button")
|
|
button.setVisible(False)
|
|
layout.addWidget(label, 1)
|
|
layout.addWidget(button)
|
|
card.setVisible(False)
|
|
return card, label, button
|
|
|
|
|
|
def _set_empty_state(card, label, button, message=None, show_button=False):
|
|
active = bool(message)
|
|
card.setVisible(active)
|
|
label.setText(message or "")
|
|
button.setVisible(active and show_button)
|
|
|
|
|
|
BATCH_PROGRESS_FIELDS = [
|
|
("total", "总数"),
|
|
("imported", "导入"),
|
|
("collected", "已采集"),
|
|
("generated", "已生成"),
|
|
("applied", "已更新"),
|
|
("failed", "失败"),
|
|
("skipped", "略过"),
|
|
]
|
|
|
|
|
|
def _summarize_batch_progress(tasks):
|
|
summary = {key: 0 for key, _label in BATCH_PROGRESS_FIELDS}
|
|
task_rows = list(tasks or [])
|
|
summary["total"] = len(task_rows)
|
|
for task in task_rows:
|
|
status = getattr(task, "status", None)
|
|
stage = getattr(task, "stage", None)
|
|
if status == "failed":
|
|
summary["failed"] += 1
|
|
elif status == "skipped":
|
|
summary["skipped"] += 1
|
|
elif stage == "applied":
|
|
summary["applied"] += 1
|
|
elif stage == "generated":
|
|
summary["generated"] += 1
|
|
elif stage == "collected":
|
|
summary["collected"] += 1
|
|
else:
|
|
summary["imported"] += 1
|
|
return summary
|
|
|
|
|
|
def _format_batch_progress(summary):
|
|
parts = [f"{label}{summary.get(key, 0)}" for key, label in BATCH_PROGRESS_FIELDS]
|
|
return "批次进度：" + " · ".join(parts)
|
|
|
|
|
|
def _build_batch_progress_overview(object_name):
|
|
label = QLabel("")
|
|
label.setObjectName(object_name)
|
|
label.setWordWrap(True)
|
|
label.setStyleSheet(
|
|
f"QLabel#{object_name} {{ "
|
|
"background: #f6f8fa; border: 1px solid #d0d7de; "
|
|
"border-radius: 6px; padding: 8px 10px; color: #24292f; "
|
|
"}"
|
|
)
|
|
label.setVisible(False)
|
|
return label
|
|
|
|
|
|
def _set_batch_progress_overview(label, tasks):
|
|
summary = _summarize_batch_progress(tasks)
|
|
active = summary["total"] > 0
|
|
label.setVisible(active)
|
|
label.setText(_format_batch_progress(summary) if active else "")
|
|
|
|
|
|
def _batch_filter_current_index(combo, batches, selected_batch, had_previous_items):
|
|
batch_ids = {batch.id for batch in batches}
|
|
if selected_batch in batch_ids:
|
|
index = combo.findData(selected_batch)
|
|
if index >= 0:
|
|
return index
|
|
if selected_batch is None and had_previous_items:
|
|
index = combo.findData(None)
|
|
if index >= 0:
|
|
return index
|
|
return 1 if batches else 0
|
|
|
|
|
|
def _database_path(db_path=None, config=None) -> str:
|
|
return db_path or appconfig.db_path(config)
|
|
|
|
|
|
def _write_reset_run_log(db_path, task, action, message):
|
|
run_id = db.create_run_log(
|
|
"reset",
|
|
total=1,
|
|
options={
|
|
"action": action,
|
|
"task_id": getattr(task, "id", None),
|
|
"alias": getattr(task, "alias", None),
|
|
"item_id": getattr(task, "item_id", None),
|
|
},
|
|
path=db_path,
|
|
)
|
|
db.add_run_log_event(
|
|
run_id,
|
|
message,
|
|
task_id=getattr(task, "id", None),
|
|
alias=getattr(task, "alias", None),
|
|
item_id=getattr(task, "item_id", None),
|
|
path=db_path,
|
|
)
|
|
db.finish_run_log(
|
|
run_id,
|
|
status="done",
|
|
done=1,
|
|
success_count=1,
|
|
failed_count=0,
|
|
summary_json={
|
|
"action": action,
|
|
"task_id": getattr(task, "id", None),
|
|
"alias": getattr(task, "alias", None),
|
|
"item_id": getattr(task, "item_id", None),
|
|
},
|
|
path=db_path,
|
|
)
|
|
return run_id
|
|
|
|
|
|
def _safe_create_run_log(run_type, db_path=None, dry_run=False, total=0, options=None):
|
|
try:
|
|
return db.create_run_log(
|
|
run_type,
|
|
dry_run=dry_run,
|
|
total=total,
|
|
options=options or {},
|
|
path=db_path,
|
|
)
|
|
except Exception:
|
|
return None
|
|
|
|
|
|
def _safe_finish_run_log(run_id, db_path=None, **fields):
|
|
if run_id is None:
|
|
return
|
|
try:
|
|
db.finish_run_log(run_id, path=db_path, **fields)
|
|
except Exception:
|
|
return
|
|
|
|
|
|
def _safe_add_run_log_event(
|
|
run_id,
|
|
message,
|
|
db_path=None,
|
|
task=None,
|
|
account=None,
|
|
level="info",
|
|
task_id=None,
|
|
alias=None,
|
|
item_id=None,
|
|
):
|
|
safe_message = diagnostics.redact_log_text(message)
|
|
if task is not None:
|
|
task_id = getattr(task, "id", task_id)
|
|
alias = getattr(task, "alias", alias)
|
|
item_id = getattr(task, "item_id", item_id)
|
|
if account is not None:
|
|
alias = getattr(account, "alias", alias)
|
|
if run_id is None:
|
|
return safe_message
|
|
try:
|
|
db.add_run_log_event(
|
|
run_id,
|
|
safe_message,
|
|
task_id=task_id,
|
|
alias=alias,
|
|
item_id=item_id,
|
|
level=level,
|
|
path=db_path,
|
|
)
|
|
except Exception:
|
|
pass
|
|
return safe_message
|
|
|
|
|
|
def _safe_write_diagnostic_log(
|
|
message,
|
|
level="INFO",
|
|
step=None,
|
|
task=None,
|
|
account=None,
|
|
elapsed_ms=None,
|
|
payload=None,
|
|
exc=None,
|
|
log_dir=None,
|
|
):
|
|
try:
|
|
diagnostics.write_diagnostic_log(
|
|
message,
|
|
level=level,
|
|
step=step,
|
|
task_id=getattr(task, "id", None),
|
|
alias=getattr(task, "alias", None) or getattr(account, "alias", None),
|
|
item_id=getattr(task, "item_id", None),
|
|
elapsed_ms=elapsed_ms,
|
|
payload=payload,
|
|
exc=exc,
|
|
log_dir=log_dir,
|
|
)
|
|
except Exception:
|
|
return
|
|
|
|
|
|
def _elapsed_ms(started):
|
|
return int((time.monotonic() - started) * 1000)
|
|
|
|
__all__ = [name for name in globals() if not name.startswith("__")]
|