chengma/cmshoppe
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(ai-studio): add main workspace tab

Browse Source
This commit is contained in:
chengma
2026-07-11 14:16:11 +08:00
parent ab99737e03
commit 5451579073
14 changed files with 1369 additions and 27 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/gui/__init__.py
+4
View File
@@ -24,12 +24,16 @@ if QT_IMPORT_ERROR is None:
ApplyWorker,
CollectWorker,
GenerateWorker,
ImageStudioDownloadOriginalWorker,
ImageStudioGenerateJobsWorker,
ImageStudioPullImagesWorker,
WriteBackWorker,
)
from .tabs.accounts import AccountDialog, AccountsTab
from .tabs.apply import ApplyTab
from .tabs.collect import CollectTab
from .tabs.generate import GenerateTab
from .tabs.image_studio import ImageStudioPreviewDialog, ImageStudioTab
from .tabs.settings import SettingsTab
from .main_window import MainWindow
else:
app/gui/main_window.py
+12 -3
View File
@@ -7,6 +7,7 @@ from .tabs.accounts import AccountsTab
from .tabs.apply import ApplyTab
from .tabs.collect import CollectTab
from .tabs.generate import GenerateTab
from .tabs.image_studio import ImageStudioTab
from .tabs.settings import SettingsTab
from .widgets import *
@@ -72,7 +73,7 @@ def _fit_and_center_window(
class MainWindow(QMainWindow):
"""Main application window with the fixed five-tab workflow."""
"""Main application window with the fixed workflow tabs."""
def __init__(
self,
@@ -158,11 +159,19 @@ class MainWindow(QMainWindow):
config=self.config,
status_callback=self.show_status,
)
return SettingsTab(
if title == "⑤ 设置":
return SettingsTab(
config=self.config,
config_path=self.config_path,
ai_models_path=self.ai_models_path,
status_callback=self.show_status,
)
return ImageStudioTab(
db_path=self.db_path,
config=self.config,
config_path=self.config_path,
ai_models_path=self.ai_models_path,
status_callback=self.show_status,
prompts_dir=appconfig.image_studio_prompts_dir(self.config),
)
def refresh_task_tabs(self):
app/gui/tabs/image_studio.py
+984
View File
@@ -0,0 +1,984 @@
"""Tab 6: AI image studio UI."""
from __future__ import annotations
import os
from ... import accounts, appconfig, db, image_studio, prompts
from .. import file_manager
from ..widgets import *
from ..workers import (
ImageStudioDownloadOriginalWorker as _RealImageStudioDownloadOriginalWorker,
)
from ..workers import ImageStudioGenerateJobsWorker as _RealImageStudioGenerateJobsWorker
from ..workers import ImageStudioPullImagesWorker as _RealImageStudioPullImagesWorker
def ImageStudioPullImagesWorker(*args, **kwargs):
return _call_package_attr(
"ImageStudioPullImagesWorker",
_RealImageStudioPullImagesWorker,
*args,
**kwargs,
)
def ImageStudioDownloadOriginalWorker(*args, **kwargs):
return _call_package_attr(
"ImageStudioDownloadOriginalWorker",
_RealImageStudioDownloadOriginalWorker,
*args,
**kwargs,
)
def ImageStudioGenerateJobsWorker(*args, **kwargs):
return _call_package_attr(
"ImageStudioGenerateJobsWorker",
_RealImageStudioGenerateJobsWorker,
*args,
**kwargs,
)
class ImageStudioPreviewDialog(QDialog):
"""Simple large image preview used by original and pool tables."""
def __init__(self, asset, parent=None):
super().__init__(parent)
self.asset = asset
self.setWindowTitle(self._title_for_asset(asset))
layout = QVBoxLayout(self)
scroll = QScrollArea()
scroll.setWidgetResizable(False)
image_label = QLabel()
image_label.setAlignment(Qt.AlignCenter)
path = str(getattr(asset, "local_path", "") or "")
image = QImage(path) if path and os.path.isfile(path) else QImage()
if image.isNull():
image_label.setText("图片尚未下载或读取失败")
image_label.setMinimumSize(420, 260)
else:
image_label.setPixmap(QPixmap.fromImage(image))
image_label.resize(image.size())
self.setWindowTitle(
f"{self._title_for_asset(asset)} · {image.width()}x{image.height()}"
)
scroll.setWidget(image_label)
layout.addWidget(scroll, 1)
buttons = QHBoxLayout()
buttons.addStretch(1)
close_button = QPushButton("关闭")
close_button.clicked.connect(self.reject)
buttons.addWidget(close_button)
layout.addLayout(buttons)
self.resize(720, 520)
def _title_for_asset(self, asset):
badge = _asset_badge(getattr(asset, "kind", ""))
asset_id = getattr(asset, "id", "")
return f"AI工场图片预览：{badge} #{asset_id}"
class ImageStudioTab(QWidget):
"""Sixth tab: project-based AI image studio."""
PROJECT_COLUMNS = ["项目", "账号", "商品ID", "更新时间"]
ORIGINAL_COLUMNS = ["序号", "状态", "远程地址"]
POOL_COLUMNS = ["类型", "比例", "状态", "来源", "本地文件"]
JOB_STATUS_LABELS = {
"pending": "排队中",
"submitted": "已提交",
"running": "生成中",
"succeeded": "成功",
"failed": "失败",
"expired": "已过期",
"cancelled": "已停止",
}
def __init__(
self,
parent=None,
db_path=None,
config=None,
config_path=None,
status_callback=None,
prompts_dir=None,
):
super().__init__(parent)
self.setObjectName("imageStudioTab")
self.config = appconfig.load_config(config_path or appconfig.CONFIG_PATH) if config is None else config
self.db_path = db_path or appconfig.db_path(self.config)
self.config_path = config_path or self.config.get("config_path") or appconfig.CONFIG_PATH
self.prompts_dir = prompts_dir or appconfig.image_studio_prompts_dir(self.config)
self.cmhub_config_path = self.config.get("cmhub_config_path") or appconfig.cmhub_config_file_path(self.config)
self.status_callback = status_callback
self.projects = []
self.accounts = []
self.current_project = None
self.assets = []
self.jobs = []
self.selections = []
self.selected_source_asset_id = None
self._running_worker = None
self._running_thread = None
self._download_open_after = {}
self._build_ui()
self._connect_signals()
self.refresh_accounts()
self.refresh_templates()
self.refresh_projects()
def _build_ui(self):
root = QVBoxLayout(self)
root.setContentsMargins(10, 8, 10, 8)
root.setSpacing(8)
splitter = QSplitter(Qt.Horizontal)
splitter.setObjectName("imageStudioMainSplitter")
splitter.addWidget(self._build_project_panel())
splitter.addWidget(self._build_pool_panel())
splitter.addWidget(self._build_generation_panel())
splitter.setStretchFactor(0, 1)
splitter.setStretchFactor(1, 3)
splitter.setStretchFactor(2, 2)
root.addWidget(splitter, 1)
root.addWidget(self._build_final_panel(), 0)
def _build_project_panel(self):
panel = QWidget()
panel.setObjectName("imageStudioProjectPanel")
layout = QVBoxLayout(panel)
layout.setContentsMargins(0, 0, 0, 0)
layout.setSpacing(8)
project_form = QFormLayout()
project_form.setLabelAlignment(Qt.AlignRight)
self.account_combo = QComboBox()
self.account_combo.setObjectName("imageStudioAccountCombo")
self.item_id_edit = QLineEdit()
self.item_id_edit.setObjectName("imageStudioItemIdEdit")
self.item_id_edit.setPlaceholderText("商品ID")
project_form.addRow("账号", self.account_combo)
project_form.addRow("商品ID", self.item_id_edit)
layout.addLayout(project_form)
toolbar = QHBoxLayout()
self.open_project_button = QPushButton("打开项目")
self.open_project_button.setObjectName("imageStudioOpenProjectButton")
self.pull_images_button = QPushButton("拉取主图")
self.pull_images_button.setObjectName("imageStudioPullImagesButton")
self.open_folder_button = QPushButton("打开项目文件夹")
self.open_folder_button.setObjectName("imageStudioOpenFolderButton")
toolbar.addWidget(self.open_project_button)
toolbar.addWidget(self.pull_images_button)
toolbar.addWidget(self.open_folder_button)
layout.addLayout(toolbar)
self.project_table = QTableWidget(0, len(self.PROJECT_COLUMNS))
self.project_table.setObjectName("imageStudioProjectList")
self.project_table.setHorizontalHeaderLabels(self.PROJECT_COLUMNS)
self.project_table.setSelectionBehavior(QAbstractItemView.SelectRows)
self.project_table.setSelectionMode(QAbstractItemView.SingleSelection)
self.project_table.setEditTriggers(QAbstractItemView.NoEditTriggers)
self.project_table.horizontalHeader().setStretchLastSection(True)
self.project_table.verticalHeader().setVisible(False)
layout.addWidget(self.project_table, 1)
return panel
def _build_pool_panel(self):
panel = QWidget()
panel.setObjectName("imageStudioPoolPanel")
layout = QVBoxLayout(panel)
layout.setContentsMargins(0, 0, 0, 0)
layout.setSpacing(8)
original_header = QHBoxLayout()
original_header.addWidget(QLabel("蝦皮原主图"))
original_header.addStretch(1)
self.original_hint_label = QLabel("单击下载并加入照片池，双击查看大图")
self.original_hint_label.setObjectName("imageStudioOriginalHintLabel")
original_header.addWidget(self.original_hint_label)
layout.addLayout(original_header)
self.original_table = QTableWidget(0, len(self.ORIGINAL_COLUMNS))
self.original_table.setObjectName("imageStudioOriginalTable")
self.original_table.setHorizontalHeaderLabels(self.ORIGINAL_COLUMNS)
self.original_table.setSelectionBehavior(QAbstractItemView.SelectRows)
self.original_table.setSelectionMode(QAbstractItemView.SingleSelection)
self.original_table.setEditTriggers(QAbstractItemView.NoEditTriggers)
self.original_table.horizontalHeader().setStretchLastSection(True)
self.original_table.verticalHeader().setVisible(False)
layout.addWidget(self.original_table, 1)
pool_header = QHBoxLayout()
pool_header.addWidget(QLabel("照片池"))
pool_header.addStretch(1)
self.source_label = QLabel("源图：未选择")
self.source_label.setObjectName("imageStudioSourceLabel")
pool_header.addWidget(self.source_label)
layout.addLayout(pool_header)
self.pool_table = QTableWidget(0, len(self.POOL_COLUMNS))
self.pool_table.setObjectName("imageStudioPoolTable")
self.pool_table.setHorizontalHeaderLabels(self.POOL_COLUMNS)
self.pool_table.setSelectionBehavior(QAbstractItemView.SelectRows)
self.pool_table.setSelectionMode(QAbstractItemView.SingleSelection)
self.pool_table.setEditTriggers(QAbstractItemView.NoEditTriggers)
self.pool_table.setContextMenuPolicy(Qt.CustomContextMenu)
self.pool_table.horizontalHeader().setStretchLastSection(True)
self.pool_table.verticalHeader().setVisible(False)
layout.addWidget(self.pool_table, 2)
return panel
def _build_generation_panel(self):
panel = QWidget()
panel.setObjectName("imageStudioGenerationPanel")
layout = QVBoxLayout(panel)
layout.setContentsMargins(0, 0, 0, 0)
layout.setSpacing(8)
template_layout = QGridLayout()
self.template_combo = QComboBox()
self.template_combo.setObjectName("imageStudioTemplateCombo")
self.template_new_button = QPushButton("新建")
self.template_new_button.setObjectName("imageStudioTemplateNewButton")
self.template_rename_button = QPushButton("重命名")
self.template_rename_button.setObjectName("imageStudioTemplateRenameButton")
self.template_save_button = QPushButton("保存")
self.template_save_button.setObjectName("imageStudioTemplateSaveButton")
self.template_delete_button = QPushButton("删除")
self.template_delete_button.setObjectName("imageStudioTemplateDeleteButton")
template_layout.addWidget(self.template_combo, 0, 0, 1, 2)
template_layout.addWidget(self.template_new_button, 0, 2)
template_layout.addWidget(self.template_rename_button, 1, 0)
template_layout.addWidget(self.template_save_button, 1, 1)
template_layout.addWidget(self.template_delete_button, 1, 2)
layout.addLayout(template_layout)
self.prompt_edit = QPlainTextEdit()
self.prompt_edit.setObjectName("imageStudioPromptEdit")
self.prompt_edit.setPlaceholderText("输入完整图片生成提示词")
self.prompt_edit.setMinimumHeight(160)
layout.addWidget(self.prompt_edit, 2)
form = QFormLayout()
self.job_type_combo = QComboBox()
self.job_type_combo.setObjectName("imageStudioJobTypeCombo")
self.job_type_combo.addItem("主图", "main")
self.job_type_combo.addItem("详情图", "detail")
self.count_spin = QSpinBox()
self.count_spin.setObjectName("imageStudioCountSpin")
self.count_spin.setRange(1, 12)
self.count_spin.setValue(4)
self.aspect_combo = QComboBox()
self.aspect_combo.setObjectName("imageStudioAspectCombo")
for value in ("1:1", "3:4", "4:3", "9:16", "16:9"):
self.aspect_combo.addItem(value, value)
form.addRow("类型", self.job_type_combo)
form.addRow("数量", self.count_spin)
form.addRow("比例", self.aspect_combo)
layout.addLayout(form)
self.billing_label = QLabel("cmhub 托管模型：扣点以返回结果为准")
self.billing_label.setObjectName("imageStudioBillingLabel")
self.billing_label.setWordWrap(True)
layout.addWidget(self.billing_label)
action_layout = QHBoxLayout()
self.start_button = QPushButton("开始生成")
self.start_button.setObjectName("imageStudioStartButton")
self.stop_button = QPushButton("停止")
self.stop_button.setObjectName("imageStudioStopButton")
self.stop_button.setEnabled(False)
action_layout.addWidget(self.start_button)
action_layout.addWidget(self.stop_button)
layout.addLayout(action_layout)
self.progress_bar = QProgressBar()
self.progress_bar.setObjectName("imageStudioProgressBar")
self.progress_bar.setRange(0, 1)
self.progress_bar.setValue(0)
layout.addWidget(self.progress_bar)
self.log_view = QPlainTextEdit()
self.log_view.setObjectName("imageStudioLogView")
self.log_view.setReadOnly(True)
self.log_view.setPlaceholderText("运行日志会在开始后显示")
layout.addWidget(self.log_view, 1)
return panel
def _build_final_panel(self):
panel = QWidget()
panel.setObjectName("imageStudioFinalPanel")
layout = QHBoxLayout(panel)
layout.setContentsMargins(0, 0, 0, 0)
self.main_selection_label = QLabel("主图终选 0/9（拖放排序将在 T-592 接入）")
self.main_selection_label.setObjectName("imageStudioMainSelectionLabel")
self.detail_selection_label = QLabel("详情图终选 0/12（拖放排序将在 T-592 接入）")
self.detail_selection_label.setObjectName("imageStudioDetailSelectionLabel")
layout.addWidget(self.main_selection_label)
layout.addWidget(self.detail_selection_label)
layout.addStretch(1)
return panel
def _connect_signals(self):
self.open_project_button.clicked.connect(self.open_project)
self.pull_images_button.clicked.connect(self.pull_main_images)
self.open_folder_button.clicked.connect(self.open_project_folder)
self.project_table.itemSelectionChanged.connect(self._on_project_selection_changed)
self.original_table.cellClicked.connect(self._on_original_clicked)
self.original_table.cellDoubleClicked.connect(self._on_original_double_clicked)
self.pool_table.cellClicked.connect(self._on_pool_clicked)
self.pool_table.cellDoubleClicked.connect(self._on_pool_double_clicked)
self.pool_table.customContextMenuRequested.connect(self._show_pool_context_menu)
self.template_combo.currentIndexChanged.connect(self.load_selected_template)
self.template_new_button.clicked.connect(self.create_template)
self.template_rename_button.clicked.connect(self.rename_template)
self.template_save_button.clicked.connect(self.save_template)
self.template_delete_button.clicked.connect(self.delete_template)
self.prompt_edit.textChanged.connect(self._save_project_prompt)
self.start_button.clicked.connect(self.start_generation)
self.stop_button.clicked.connect(self.stop_generation)
def refresh_accounts(self):
self.account_combo.clear()
try:
self.accounts = accounts.list_accounts(path=self.db_path, config=self.config)
except Exception as exc:
self.accounts = []
self._status(f"账号读取失败：{exc}", "danger")
for account in self.accounts:
self.account_combo.addItem(
f"{account.account_name}（{account.alias}）",
account.alias,
)
if not self.accounts:
self.account_combo.addItem("暂无账号，请先到④账号管理添加", "")
def refresh_projects(self):
try:
db.init_db(self.db_path)
self.projects = image_studio.list_projects(path=self.db_path)
except Exception as exc:
self.projects = []
self._status(f"AI工场项目读取失败：{exc}", "danger")
self._fill_project_table()
if self.current_project is None and self.projects:
self._select_project(self.projects[0].id)
elif self.current_project is not None:
self._select_project(self.current_project.id, quiet=True)
def _fill_project_table(self):
self.project_table.setRowCount(len(self.projects))
for row, project in enumerate(self.projects):
values = [
f"{project.account_alias} / {project.item_id}",
project.account_name or project.account_alias,
project.item_id,
project.updated_at,
]
for column, value in enumerate(values):
item = QTableWidgetItem(str(value or ""))
item.setData(Qt.UserRole, int(project.id))
self.project_table.setItem(row, column, item)
self.project_table.resizeColumnsToContents()
def open_project(self, checked=False):
alias = str(self.account_combo.currentData() or "").strip()
item_id = self.item_id_edit.text().strip()
if not alias:
self._message("账号未选择", "请先在④账号管理添加并选择账号。")
return
if not item_id:
self._message("商品ID不能为空", "请输入要打开的蝦皮商品ID。")
return
account = accounts.get_account(alias, path=self.db_path, config=self.config)
try:
project = image_studio.create_or_get_project(
account,
item_id=item_id,
path=self.db_path,
)
except Exception as exc:
self._message("打开项目失败", str(exc))
self._status(f"打开AI工场项目失败：{exc}", "danger")
return
self.current_project = project
self.item_id_edit.setText(project.item_id)
self._set_account_combo(project.account_alias)
self.refresh_projects()
self._select_project(project.id)
self._status("AI工场项目已打开", "success")
def pull_main_images(self, checked=False):
alias = str(self.account_combo.currentData() or "").strip()
item_id = self.item_id_edit.text().strip()
if self.current_project is not None:
alias = self.current_project.account_alias
item_id = self.current_project.item_id
if not alias or not item_id:
self._message("项目未打开", "请先选择账号和商品ID并打开项目。")
return
worker = ImageStudioPullImagesWorker(
alias,
item_id,
db_path=self.db_path,
config=self.config,
)
worker.log.connect(self._append_log)
worker.finished.connect(self._on_pull_finished)
worker.failed.connect(self._on_worker_failed)
self._start_worker(worker, "AI工场拉取主图")
self._append_log("[AI工场] 拉取主图开始")
def _on_pull_finished(self, summary):
if summary.get("ok") is False:
if self._running_worker is not None:
self._on_worker_failed(-1, summary.get("error") or "拉取主图失败")
return
project = summary.get("project")
if project is not None:
self.current_project = project
self._set_account_combo(project.account_alias)
self.item_id_edit.setText(project.item_id)
self._finish_worker()
self.refresh_projects()
if project is not None:
self._select_project(project.id)
self._status(f"已拉取 {summary.get('count', 0)} 张蝦皮原主图", "success")
def open_project_folder(self, checked=False):
if self.current_project is None:
self._message("项目未打开", "请先打开一个AI工场项目。")
return
dirs = image_studio.default_project_image_dirs(self.current_project, self.config)
try:
os.makedirs(dirs["root"], exist_ok=True)
opened = file_manager.open_in_file_manager(dirs["root"])
except Exception as exc:
self._message("打开项目文件夹失败", str(exc))
self._status(f"打开项目文件夹失败：{exc}", "warning")
return
self._status(f"已打开项目文件夹：{opened}", "success")
def _on_project_selection_changed(self):
items = self.project_table.selectedItems()
if not items:
return
project_id = items[0].data(Qt.UserRole)
if project_id is not None:
self._select_project(project_id)
def _select_project(self, project_id, quiet=False):
try:
project = image_studio.get_project(project_id, path=self.db_path)
except Exception as exc:
self._status(f"读取AI工场项目失败：{exc}", "danger")
return
if project is None:
return
self.current_project = project
self._set_account_combo(project.account_alias)
self.item_id_edit.setText(project.item_id)
self.prompt_edit.blockSignals(True)
try:
self.prompt_edit.setPlainText(project.draft_prompt or "")
finally:
self.prompt_edit.blockSignals(False)
self.selected_source_asset_id = None
self.refresh_project_assets()
self._sync_project_selection(project.id)
if not quiet:
self._status(f"当前AI工场项目：{project.account_alias} / {project.item_id}", "muted")
def _sync_project_selection(self, project_id):
for row in range(self.project_table.rowCount()):
item = self.project_table.item(row, 0)
if item is not None and item.data(Qt.UserRole) == int(project_id):
if not self.project_table.item(row, 0).isSelected():
self.project_table.selectRow(row)
break
def refresh_project_assets(self):
if self.current_project is None:
self.assets = []
self.jobs = []
self.selections = []
else:
self.assets = image_studio.list_assets(self.current_project.id, path=self.db_path)
self.jobs = self._list_project_jobs(self.current_project.id)
self.selections = image_studio.list_selections(self.current_project.id, path=self.db_path)
self._fill_original_table()
self._fill_pool_table()
self._refresh_selection_labels()
self._refresh_source_label()
def _fill_original_table(self):
originals = [asset for asset in self.assets if asset.kind == image_studio.ASSET_KIND_ORIGINAL]
self.original_table.setRowCount(len(originals))
for row, asset in enumerate(originals):
values = [
str(asset.source_order or row + 1),
_asset_status_text(asset),
asset.remote_url or "",
]
for column, value in enumerate(values):
item = QTableWidgetItem(str(value or ""))
item.setData(Qt.UserRole, {"type": "asset", "asset_id": int(asset.id)})
self.original_table.setItem(row, column, item)
self.original_table.resizeColumnsToContents()
def _fill_pool_table(self):
rows = []
for asset in self.assets:
if asset.status == image_studio.ASSET_STATUS_MISSING:
continue
if asset.kind not in {"original", "generated_main", "generated_detail"}:
continue
rows.append(("asset", asset))
for job in self.jobs:
if job.status in {"pending", "submitted", "running", "failed", "expired", "cancelled"}:
rows.append(("job", job))
self.pool_table.setRowCount(len(rows))
for row, (row_type, obj) in enumerate(rows):
if row_type == "asset":
values = [
_asset_badge(obj.kind),
obj.aspect_ratio or "未知",
_asset_status_text(obj),
_source_text(obj, self.assets),
obj.local_path or "",
]
data = {"type": "asset", "asset_id": int(obj.id)}
else:
values = [
"任务",
"-",
self.JOB_STATUS_LABELS.get(obj.status, obj.status),
f"源图 #{obj.source_asset_id or '-'}",
obj.error or "",
]
data = {"type": "job", "job_id": int(obj.id)}
for column, value in enumerate(values):
item = QTableWidgetItem(str(value or ""))
item.setData(Qt.UserRole, data)
self.pool_table.setItem(row, column, item)
self.pool_table.resizeColumnsToContents()
def _on_original_clicked(self, row, column):
asset = self._asset_from_table_row(self.original_table, row)
if asset is not None:
self._ensure_original_in_pool(asset, open_after=False)
def _on_original_double_clicked(self, row, column):
asset = self._asset_from_table_row(self.original_table, row)
if asset is not None:
self._ensure_original_in_pool(asset, open_after=True)
def _on_pool_clicked(self, row, column):
data = self._row_data(self.pool_table, row)
if not data or data.get("type") != "asset":
return
asset = self._asset_by_id(data.get("asset_id"))
if asset is not None:
self._select_source_asset(asset)
def _on_pool_double_clicked(self, row, column):
data = self._row_data(self.pool_table, row)
if not data or data.get("type") != "asset":
return
asset = self._asset_by_id(data.get("asset_id"))
if asset is not None:
self._open_preview(asset)
def _ensure_original_in_pool(self, asset, open_after=False):
if str(asset.local_path or "").strip() and os.path.isfile(asset.local_path):
self._select_source_asset(asset)
if open_after:
self._open_preview(asset)
return
worker = ImageStudioDownloadOriginalWorker(
asset.id,
db_path=self.db_path,
config=self.config,
open_after=open_after,
)
worker.log.connect(self._append_log)
worker.finished.connect(self._on_download_finished)
worker.failed.connect(self._on_worker_failed)
self._start_worker(worker, "AI工场下载原图")
def _on_download_finished(self, summary):
if summary.get("ok") is False:
if self._running_worker is not None:
self._on_worker_failed(-1, summary.get("error") or "下载原图失败")
return
asset = summary.get("asset")
self._finish_worker()
self.refresh_project_assets()
if asset is not None:
refreshed = self._asset_by_id(asset.id) or asset
self._select_source_asset(refreshed)
if summary.get("open_after"):
self._open_preview(refreshed)
self._status("原图已加入照片池", "success")
def _select_source_asset(self, asset):
self.selected_source_asset_id = int(asset.id)
self._refresh_source_label()
self._status(f"已选择源图：{_asset_badge(asset.kind)} #{asset.id}", "success")
def _refresh_source_label(self):
asset = self._asset_by_id(self.selected_source_asset_id)
if asset is None:
self.source_label.setText("源图：未选择")
return
self.source_label.setText(f"源图：{_asset_badge(asset.kind)} #{asset.id}")
def _open_preview(self, asset):
dialog = ImageStudioPreviewDialog(asset, self)
dialog.exec()
def _show_pool_context_menu(self, position):
row = self.pool_table.rowAt(position.y())
data = self._row_data(self.pool_table, row)
if not data or data.get("type") != "asset":
return
asset_id = data.get("asset_id")
menu = QMenu(self)
try:
counts = image_studio.asset_reference_counts(asset_id, path=self.db_path)
referenced = bool(counts.get("total"))
except Exception:
referenced = True
remove_action = menu.addAction(
"移除照片" if not referenced else "移除照片（已被任务或终选引用）"
)
remove_action.setEnabled(not referenced)
action = menu.exec(self.pool_table.viewport().mapToGlobal(position))
if action is remove_action and not referenced:
self.remove_asset(asset_id)
def remove_asset(self, asset_id):
if not self._confirm("移除照片", "只从AI工场照片池移除记录，不删除本地图片文件。"):
return
try:
image_studio.remove_asset_if_unused(asset_id, path=self.db_path)
except Exception as exc:
self._message("不能移除照片", str(exc))
return
if self.selected_source_asset_id == int(asset_id):
self.selected_source_asset_id = None
self.refresh_project_assets()
self._status("照片已从池中移除", "success")
def refresh_templates(self, selected=None):
current = selected or self.template_combo.currentData()
self.template_combo.blockSignals(True)
try:
self.template_combo.clear()
self.template_combo.addItem("选择模板", "")
for name in prompts.list_image_studio_templates(self.prompts_dir):
self.template_combo.addItem(name, name)
if current:
index = self.template_combo.findData(current)
if index >= 0:
self.template_combo.setCurrentIndex(index)
finally:
self.template_combo.blockSignals(False)
def load_selected_template(self, index=None):
name = self.template_combo.currentData()
if not name:
return
try:
self.prompt_edit.setPlainText(prompts.load_image_studio_template(name, self.prompts_dir))
except Exception as exc:
self._message("加载模板失败", str(exc))
def create_template(self, checked=False):
name, ok = QInputDialog.getText(self, "新建模板", "模板名称")
if not ok:
return
try:
prompts.save_image_studio_template(name, self.prompt_edit.toPlainText(), self.prompts_dir)
except Exception as exc:
self._message("新建模板失败", str(exc))
return
self.refresh_templates(selected=name)
self._status("AI工场模板已新建", "success")
def rename_template(self, checked=False):
old = self.template_combo.currentData()
if not old:
self._message("未选择模板", "请先选择要重命名的模板。")
return
new, ok = QInputDialog.getText(self, "重命名模板", "新模板名称", text=old)
if not ok:
return
try:
prompts.rename_image_studio_template(old, new, self.prompts_dir)
except Exception as exc:
self._message("重命名模板失败", str(exc))
return
self.refresh_templates(selected=new)
self._status("AI工场模板已重命名", "success")
def save_template(self, checked=False):
name = self.template_combo.currentData()
if not name:
name, ok = QInputDialog.getText(self, "保存模板", "模板名称")
if not ok:
return
try:
prompts.save_image_studio_template(name, self.prompt_edit.toPlainText(), self.prompts_dir)
except Exception as exc:
self._message("保存模板失败", str(exc))
return
self.refresh_templates(selected=name)
self._status("AI工场模板已保存", "success")
def delete_template(self, checked=False):
name = self.template_combo.currentData()
if not name:
self._message("未选择模板", "请先选择要删除的模板。")
return
if not self._confirm("删除模板", f"确定删除模板「{name}」吗？"):
return
try:
prompts.delete_image_studio_template(name, self.prompts_dir)
except Exception as exc:
self._message("删除模板失败", str(exc))
return
self.refresh_templates()
self._status("AI工场模板已删除", "success")
def _save_project_prompt(self):
if self.current_project is None:
return
try:
self.current_project = image_studio.update_project_prompt(
self.current_project.id,
self.prompt_edit.toPlainText(),
path=self.db_path,
)
except Exception as exc:
self._status(f"保存AI工场草稿提示词失败：{exc}", "warning")
def start_generation(self, checked=False):
if self.current_project is None:
self._message("项目未打开", "请先打开一个AI工场项目。")
return
source = self._asset_by_id(self.selected_source_asset_id)
if source is None:
self._message("未选择源图", "请先在照片池单击选择一张源图。")
return
if not str(source.local_path or "").strip() or not os.path.isfile(source.local_path):
self._message("源图不可用", "请先单击蝦皮原主图下载到本地后再生成。")
return
prompt_text = self.prompt_edit.toPlainText().strip()
if not prompt_text:
self._message("提示词不能为空", "请输入完整图片生成提示词。")
return
count = self.count_spin.value()
self.progress_bar.setRange(0, count)
self.progress_bar.setValue(0)
self.log_view.clear()
self._append_log(f"[AI工场] 本轮生图开始：{count} 张，来源 cmhub 托管模型")
worker = ImageStudioGenerateJobsWorker(
self.current_project.id,
source.id,
prompt_text,
count,
job_type=self.job_type_combo.currentData(),
aspect_ratio=self.aspect_combo.currentData(),
db_path=self.db_path,
config=self.config,
cmhub_config_path=self.cmhub_config_path,
)
worker.progress.connect(self._on_generate_progress)
worker.log.connect(self._append_log)
worker.finished.connect(self._on_generation_finished)
worker.failed.connect(self._on_worker_failed)
self._start_worker(worker, "AI工场生成图片")
def stop_generation(self, checked=False):
if self._running_worker is not None and hasattr(self._running_worker, "cancel"):
self._running_worker.cancel()
self._append_log("[AI工场] 已请求停止，正在等待安全边界")
self._status("AI工场生成已请求停止", "warning")
def _on_generate_progress(self, payload):
total = max(1, int(payload.get("total") or self.progress_bar.maximum() or 1))
done = min(total, int(payload.get("done") or 0))
self.progress_bar.setRange(0, total)
self.progress_bar.setValue(done)
if payload.get("points_balance") is not None:
text = f"cmhub 托管模型：余额 {payload.get('points_balance')}"
if payload.get("points_cost") is not None:
text += f"，本张扣点 {payload.get('points_cost')}"
self.billing_label.setText(text)
def _on_generation_finished(self, summary):
if summary.get("ok") is False:
if self._running_worker is not None:
self._on_worker_failed(-1, summary.get("error") or "AI工场生成失败")
return
self._finish_worker()
self.refresh_project_assets()
total = int(summary.get("total") or 0)
success = int(summary.get("success") or 0)
failed = int(summary.get("failed") or 0)
cancelled = int(summary.get("cancelled") or 0)
self._append_log(f"[AI工场] 本轮完成：总数{total}，成功{success}，失败{failed}，停止{cancelled}")
level = "warning" if failed or cancelled else "success"
self._status(f"AI工场生成完成：成功{success}，失败{failed}，停止{cancelled}", level)
def _on_worker_failed(self, row, error):
self._finish_worker()
message = diagnostics.redact_log_text(str(error or "未知错误"))
self._append_log(f"[AI工场] 失败：{message}")
self._status(f"AI工场任务失败：{message}", "danger")
self._message("AI工场任务失败", message)
self.refresh_project_assets()
def _start_worker(self, worker, thread_name):
self._set_running(True)
self._running_worker = worker
self._running_thread = run_worker(worker, thread_name)
def _finish_worker(self):
self._running_worker = None
self._running_thread = None
self._set_running(False)
def _set_running(self, running):
self.open_project_button.setEnabled(not running)
self.pull_images_button.setEnabled(not running)
self.open_folder_button.setEnabled(not running)
self.project_table.setEnabled(not running)
self.original_table.setEnabled(not running)
self.pool_table.setEnabled(not running)
self.template_combo.setEnabled(not running)
self.template_new_button.setEnabled(not running)
self.template_rename_button.setEnabled(not running)
self.template_save_button.setEnabled(not running)
self.template_delete_button.setEnabled(not running)
self.prompt_edit.setEnabled(not running)
self.job_type_combo.setEnabled(not running)
self.count_spin.setEnabled(not running)
self.aspect_combo.setEnabled(not running)
self.start_button.setEnabled(not running)
self.stop_button.setEnabled(running)
def _set_account_combo(self, alias):
index = self.account_combo.findData(alias)
if index >= 0:
self.account_combo.setCurrentIndex(index)
def _list_project_jobs(self, project_id):
conn = db.connect(self.db_path)
try:
rows = conn.execute(
"""
SELECT * FROM image_studio_jobs
WHERE project_id = ?
ORDER BY updated_at DESC, id DESC
""",
(int(project_id),),
).fetchall()
return [image_studio.ImageStudioJob(**dict(row)) for row in rows]
finally:
conn.close()
def _refresh_selection_labels(self):
main_count = sum(1 for item in self.selections if item.selection_type == "main")
detail_count = sum(1 for item in self.selections if item.selection_type == "detail")
main_target = getattr(self.current_project, "target_main_count", 9) if self.current_project else 9
detail_target = getattr(self.current_project, "target_detail_count", 12) if self.current_project else 12
self.main_selection_label.setText(f"主图终选 {main_count}/{main_target}（拖放排序将在 T-592 接入）")
self.detail_selection_label.setText(f"详情图终选 {detail_count}/{detail_target}（拖放排序将在 T-592 接入）")
def _asset_from_table_row(self, table, row):
data = self._row_data(table, row)
if not data or data.get("type") != "asset":
return None
return self._asset_by_id(data.get("asset_id"))
def _row_data(self, table, row):
if row < 0 or row >= table.rowCount():
return None
item = table.item(row, 0)
if item is None:
return None
return item.data(Qt.UserRole)
def _asset_by_id(self, asset_id):
if asset_id is None:
return None
for asset in self.assets:
if int(asset.id) == int(asset_id):
return asset
try:
return image_studio.get_asset(asset_id, path=self.db_path)
except Exception:
return None
def _append_log(self, message):
text = diagnostics.redact_log_text(str(message or ""))
self.log_view.appendPlainText(text)
scrollbar = self.log_view.verticalScrollBar()
scrollbar.setValue(scrollbar.maximum())
def _message(self, title, text):
box = QMessageBox(self)
box.setWindowTitle(str(title or "提示"))
box.setText(str(text or ""))
ok_button = box.addButton("确定", QMessageBox.AcceptRole)
box.setDefaultButton(ok_button)
box.exec()
def _confirm(self, title, text):
box = QMessageBox(self)
box.setWindowTitle(str(title or "确认"))
box.setText(str(text or ""))
yes_button = box.addButton("确定", QMessageBox.AcceptRole)
box.addButton("取消", QMessageBox.RejectRole)
box.setDefaultButton(yes_button)
box.exec()
return box.clickedButton() is yes_button
def _status(self, message, level="muted"):
_emit_status(self.status_callback, message, level=level)
def _asset_badge(kind):
return {
"original": "原图",
"generated_main": "主图",
"generated_detail": "详情图",
}.get(str(kind or ""), str(kind or "图片"))
def _asset_status_text(asset):
status = str(getattr(asset, "status", "") or "")
local_path = str(getattr(asset, "local_path", "") or "")
if status == image_studio.ASSET_STATUS_MISSING:
return "文件缺失"
if local_path and os.path.isfile(local_path):
return "可用"
if getattr(asset, "remote_url", None):
return "远程待下载"
return "待生成"
def _source_text(asset, assets):
parent_id = getattr(asset, "parent_asset_id", None)
if not parent_id:
return "原始来源"
for item in assets:
if int(item.id) == int(parent_id):
return f"{_asset_badge(item.kind)} #{item.id}"
return f"源图 #{parent_id}"
app/gui/widgets.py
+1
View File
@@ -60,6 +60,7 @@ TAB_TITLES = [
"③ 更新蝦皮",
"④ 账号管理",
"⑤ 设置",
"⑥ AI工场",
]
TAB_STYLE = """
app/gui/workers.py
+178
View File
@@ -4,7 +4,9 @@ from __future__ import annotations
import datetime as _dt
import re
import threading
from .. import image_studio, image_studio_generation, image_studio_images
from .widgets import *
@@ -14,6 +16,182 @@ _USER_LOG_PATH_RE = re.compile(
)
def _image_studio_step_label(step):
return {
"ensure_chrome": "准备账号浏览器",
"login_check": "检测登录",
"open_product": "打开商品页",
"read_main_images": "读取蝦皮原主图",
"cover_submit": "提交生图任务",
"cover_poll": "查询生图结果",
"cover_download": "下载生成图片",
"job_done": "完成单张任务",
}.get(str(step or ""), str(step or "处理"))
def _image_studio_result_label(result):
return {
"start": "开始",
"success": "成功",
"failed": "失败",
"cancelled": "已停止",
"blocked": "已阻断",
"reused": "已复用",
"launched": "已启动",
"resume": "继续查询",
"queued": "排队中",
"running": "生成中",
"expired": "已过期",
}.get(str(result or ""), str(result or ""))
def _format_image_studio_event(event):
event = dict(event or {})
step = _image_studio_step_label(event.get("step"))
result = _image_studio_result_label(event.get("result"))
detail = str(event.get("detail") or "").strip()
job_id = event.get("job_id")
prefix = f"[AI工场] {step}"
if job_id is not None:
prefix += f" #{job_id}"
if result:
prefix += f"：{result}"
if detail:
prefix += f"，{diagnostics.redact_log_text(detail)}"
if event.get("points_cost") is not None:
prefix += f"，扣点 {event.get('points_cost')}"
if event.get("points_balance") is not None:
prefix += f"，余额 {event.get('points_balance')}"
return prefix
class ImageStudioPullImagesWorker(BaseWorker):
"""Read Shopee main image URLs for one AI studio project in background."""
def __init__(self, account_alias, item_id, *, db_path=None, config=None):
super().__init__()
self.account_alias = account_alias
self.item_id = item_id
self.db_path = db_path
self.config = config
def execute(self):
def on_step(payload):
self.log.emit(_format_image_studio_event(payload))
result = image_studio.pull_remote_main_image_urls(
self.account_alias,
self.item_id,
path=self.db_path,
config=self.config,
on_step=on_step,
)
project = result.get("project")
assets = result.get("assets") or []
return {
"project": project,
"assets": assets,
"count": len(assets),
"account": result.get("account"),
}
class ImageStudioDownloadOriginalWorker(BaseWorker):
"""Download one remote original image into the project pool."""
def __init__(self, asset_id, *, db_path=None, config=None, open_after=False):
super().__init__()
self.asset_id = int(asset_id)
self.db_path = db_path
self.config = config
self.open_after = bool(open_after)
def execute(self):
self.log.emit("[AI工场] 下载原图：开始")
asset = image_studio_images.download_original_asset(
self.asset_id,
path=self.db_path,
config=self.config,
)
self.log.emit("[AI工场] 下载原图：成功")
return {"asset": asset, "open_after": self.open_after}
class ImageStudioGenerateJobsWorker(BaseWorker):
"""Run cmhub hosted image generation jobs for the AI studio."""
def __init__(
self,
project_id,
source_asset_id,
prompt,
count,
*,
job_type="main",
aspect_ratio="1:1",
db_path=None,
config=None,
cmhub_config_path=None,
):
super().__init__()
self.project_id = int(project_id)
self.source_asset_id = int(source_asset_id)
self.prompt = str(prompt or "")
self.count = int(count or 0)
self.job_type = str(job_type or "main")
self.aspect_ratio = str(aspect_ratio or "1:1")
self.db_path = db_path
self.config = config
self.cmhub_config_path = cmhub_config_path
self._done = 0
self._failed = 0
self._lock = threading.Lock()
def execute(self):
self.progress.emit({"total": self.count, "done": 0, "failed": 0})
def on_event(payload):
event = dict(payload or {})
self.log.emit(_format_image_studio_event(event))
if event.get("points_cost") is not None or event.get("points_balance") is not None:
self.progress.emit(
{
"total": self.count,
"done": self._done,
"failed": self._failed,
"points_cost": event.get("points_cost"),
"points_balance": event.get("points_balance"),
}
)
if event.get("step") == "job_done":
with self._lock:
self._done += 1
if event.get("result") not in {"success"}:
self._failed += 1
progress = {
"total": self.count,
"done": self._done,
"failed": self._failed,
}
self.progress.emit(progress)
summary = image_studio_generation.generate_image_jobs(
self.project_id,
self.source_asset_id,
self.prompt,
self.count,
job_type=self.job_type,
aspect_ratio=self.aspect_ratio,
config=self.config,
cmhub_config_path=self.cmhub_config_path,
path=self.db_path,
should_stop=self.should_cancel,
on_event=on_event,
)
summary["project_id"] = self.project_id
return summary
def _generation_mode_label(mode):
mode = appconfig.normalize_generate_mode(mode)
return {
app/image_studio.py
+45
View File
@@ -452,6 +452,51 @@ def list_assets(project_id, kind=None, include_missing=True, path=None, conn=Non
return _fetch_all(database, sql, params, ImageStudioAsset)
def asset_reference_counts(asset_id, path=None, conn=None):
"""Return selection/job references for one asset before pool removal."""
asset_id = int(asset_id)
with _connection(conn, path) as database:
selection_count = database.execute(
"SELECT COUNT(*) FROM image_studio_selections WHERE asset_id = ?",
(asset_id,),
).fetchone()[0]
source_job_count = database.execute(
"SELECT COUNT(*) FROM image_studio_jobs WHERE source_asset_id = ?",
(asset_id,),
).fetchone()[0]
output_job_count = database.execute(
"SELECT COUNT(*) FROM image_studio_jobs WHERE output_asset_id = ?",
(asset_id,),
).fetchone()[0]
return {
"selection": int(selection_count),
"source_job": int(source_job_count),
"output_job": int(output_job_count),
"total": int(selection_count + source_job_count + output_job_count),
}
def remove_asset_if_unused(asset_id, path=None, conn=None):
"""Remove one pool asset row only when no job or selection references it.
The local image file is intentionally kept on disk. AI工场 removal is a
pool-level operation, not a destructive file cleanup.
"""
asset_id = int(asset_id)
with _connection(conn, path) as database:
asset = get_asset(asset_id, conn=database)
if asset is None:
raise db.DbError("照片不存在")
counts = asset_reference_counts(asset_id, conn=database)
if counts["total"]:
raise db.DbError("照片正在被生成任务或终选引用，不能移除")
with database:
database.execute("DELETE FROM image_studio_assets WHERE id = ?", (asset_id,))
return asset
def sync_original_asset_urls(project_id, image_urls, path=None, conn=None):
"""Store the read-only Shopee main image URL snapshot as remote-only assets."""
docs/00-ai-start-here.md
+5 -5
View File
@@ -4,10 +4,10 @@
## 一句话定位
蝦皮圈優化助手（代号 cmshopee）是一个 Windows 本地桌面自动化工具（PySide6，5 Tab），让运营管理多个 Shopee 账号，并用 CDP 驱动 Chrome + AI 批量改商品标题、换商品封面。
蝦皮圈優化助手（代号 cmshopee）是一个 Windows 本地桌面自动化工具（PySide6，6 Tab），让运营管理多个 Shopee 账号，并用 CDP 驱动 Chrome + AI 批量改商品标题、换商品封面，同时提供 AI工场图片候选生成工作区。
5 Tab 流水线（工作流优先顺序）：
**① 导入采集 → ② AI生成 → ③ 更新蝦皮 → ④ 账号管理 → ⑤ 设置**
主流水线（工作流优先顺序）：
**① 导入采集 → ② AI生成 → ③ 更新蝦皮 → ④ 账号管理 → ⑤ 设置**；**⑥ AI工场** 是图片项目/照片池/cmhub 托管生图工作区，不自动上传蝦皮。
目标闭环：④ 配账号并登录 → ① 导入 Excel（按“别名”列关联账号）、采集旧标题/旧封面并回写 → ② 用提示词 AI 生成新标题/新封面（不设逐条确认阶段）→ ③ 对已生成任务点击「开始更新」，弹窗确认后批量改标题+换封面并点「更新」提交 → 结果实时存 SQLite、批量回写原 Excel。
@@ -30,7 +30,7 @@
## 当前阶段
当前项目处于：**V0 单账号流程已验证，正在实现 V1 多账号 + Excel + AI + 5 Tab GUI 流水线**。
当前项目处于：**V0 单账号流程已验证，正在实现 V1 多账号 + Excel + AI + 6 Tab GUI 工作台**。
优先路径：
@@ -65,7 +65,7 @@
**V1 当前 coding 目标**：
- 5 Tab 流水线：① 导入采集 → ② AI生成 → ③ 更新蝦皮 → ④ 账号管理 → ⑤ 设置。
- 6 Tab 工作台：① 导入采集 → ② AI生成 → ③ 更新蝦皮 → ④ 账号管理 → ⑤ 设置；⑥ AI工场用于图片候选生成，不自动提交线上。
- GUI 固定为 PySide6；后台采集/生成/更新用 `QObject` worker + `QThread` + signal 回传进度。
- 多账号管理；账号以独立 user-data-dir 隔离。③ 更新默认串行，提供「检查本轮更新」按钮；⑤ 可开启按账号并行和设置每批最大更新条数。
- Excel 导入/回写 + SQLite 实时落库 + 本地图片目录。
docs/03-tech-stack.md
+2 -2
View File
@@ -13,7 +13,7 @@
| WebSocket | `websocket-client`（import `websocket`） | 已定 | 讲 CDP 协议；用 `suppress_origin=True` 绕过 403 |
| HTTP | `requests` | 已定 | 读 `/json` 拿 tab 列表；`trust_env=False` 忽略代理 |
| 浏览器 | Google Chrome（已安装） | 已定 | 带 `--remote-debugging-port` 启动 |
| GUI 框架 | PySide6（Qt for Python，`QTabWidget` 5 Tab） | 已定 | 当前环境已安装 PySide6；V1 需要表格、图片预览、后台任务进度、确认弹窗，Qt 的 signal/slot + QThread 更适合 |
| GUI 框架 | PySide6（Qt for Python，`QTabWidget` 6 Tab） | 已定 | 当前环境已安装 PySide6；V1 需要表格、图片预览、后台任务进度、确认弹窗，Qt 的 signal/slot + QThread 更适合 |
| 应用配置 | `data/config.json`（JSON，stdlib） | 已定 | 少量应用级设置：Chrome 路径、目录根、端口、DB 路径等 |
| 业务数据 | SQLite（stdlib `sqlite3`，`data/cmshopee.db`） | 已定 | 账号、任务、结果：成行增长、要查询/统计/导出 |
| Excel 读写 | `openpyxl` | 已定 | 导入任务、回写结果；stdlib 读不了 .xlsx |
@@ -30,7 +30,7 @@
## 二、决策记录与演进
- **CDP 自研而非 playwright**：当前已验证根目录 `cdp.py`，正式代码迁入 `app/cdp.py`；它零重依赖、完全可控，并已在开发环境绕开了代理（`*_proxy` 指向本地 :1080）和 Chrome 的 Origin 403 两个坑。未来若交互复杂度大幅上升，再评估 playwright。
- **GUI 选 PySide6**：V1 是 5 Tab 运营工作台，包含任务表格、筛选、图片预览、后台采集/生成/更新、进度与停止。当前环境已安装 PySide6，且 Tkinter 不可用；Qt 的 `QThread`/signal-slot 比 Tkinter 手动 queue/after 更适合长任务回传 UI。
- **GUI 选 PySide6**：V1 是 6 Tab 运营工作台，包含任务表格、筛选、图片预览、后台采集/生成/更新、AI工场照片池、进度与停止。当前环境已安装 PySide6，且 Tkinter 不可用；Qt 的 `QThread`/signal-slot 比 Tkinter 手动 queue/after 更适合长任务回传 UI。
- **存储拆两层**：应用设置进 `data/config.json`，账号/任务/结果进 SQLite。判据：少量人改无需查询 → 配置文件；成行增长要查询/导出 → DB。同一事实只存一处，不重复。取代早期的 `accounts.json` 方案。
- **Excel 用 openpyxl**：运营用真实 .xlsx；stdlib 无法读写 xlsx，引入一个轻依赖比改用 CSV 更贴合用户习惯。
- **多账号隔离用独立 user-data-dir，不用 Chrome profile**：profile 共享同一 user-data-dir/进程/调试端口，无法每账号独立 CDP 与并行；独立 user-data-dir 才契合自动化。详见 [架构 3.0](04-architecture.md)。
docs/04-architecture.md
+10 -7
View File
@@ -5,14 +5,14 @@
## 一、系统结构
Windows 本地桌面自动化工具，无后端服务，5 Tab GUI 驱动一条流水线。
Windows 本地桌面自动化工具，无后端服务，6 Tab GUI 驱动一条流水线与一个 AI工场图片工作区。
```text
运营（人）
|
v
GUI（PySide6 QTabWidget，5 Tab）
① 导入采集 ② AI生成 ③ 更新蝦皮 ④ 账号管理 ⑤ 设置
GUI（PySide6 QTabWidget，6 Tab）
① 导入采集 ② AI生成 ③ 更新蝦皮 ④ 账号管理 ⑤ 设置 ⑥ AI工场
|
v
核心模块（Python）
@@ -24,7 +24,10 @@ GUI（PySide6 QTabWidget，5 Tab）
├── chrome 按账号拼启动参数、启动/探测 Chrome、生成快捷方式
├── cdp CDP 客户端（连接、找/开 tab、执行 JS、拖拽）
├── editor 登录检测 / 采集旧标题旧封面 / 改标题 / 换封面 / 点更新
└── ai 文本生成（提示词+旧标题→新标题）/ 图像生成（提示词+旧封面→新封面）
├── ai 文本生成（提示词+旧标题→新标题）/ 图像生成（提示词+旧封面→新封面）
├── image_studio AI工场项目/资产/job/终选顺序数据服务
├── image_studio_images 远程原图安全下载、缩略图、原图落盘
└── image_studio_generation cmhub 托管多图异步 submit/poll/download 编排
|
v
Google Chrome（每账号独立 --user-data-dir + --remote-debugging-port） + AI 服务（默认 cmhub 网关；direct 仅内部兼容/回滚）
@@ -35,8 +38,8 @@ Shopee 卖家中心页面 / 本地图片目录
真实组件：
- GUI 入口：根目录 `main.py` 调用 `app/gui/` 包（PySide6 + `QMainWindow` + `QTabWidget`，5 Tab）；包入口 `app/gui/__init__.py` 提供 `main()` 并兼容 `from app import gui` / `from app.gui import MainWindow`；也支持 `python -m app`。
- 核心模块统一放在正式代码包 `app/`：`appconfig.py`、`db.py`、`excel.py`、`config.py`、`accounts.py`、`chrome.py`、`editor.py`、`workers.py`、`ai.py`、`prompts.py`；CDP 底座迁入 `app/cdp.py`（当前根目录 `cdp.py` 为已验证来源）。
- GUI 入口：根目录 `main.py` 调用 `app/gui/` 包（PySide6 + `QMainWindow` + `QTabWidget`，6 Tab）；包入口 `app/gui/__init__.py` 提供 `main()` 并兼容 `from app import gui` / `from app.gui import MainWindow`；也支持 `python -m app`。
- 核心模块统一放在正式代码包 `app/`：`appconfig.py`、`db.py`、`excel.py`、`config.py`、`accounts.py`、`chrome.py`、`editor.py`、`workers.py`、`ai.py`、`prompts.py`、`image_studio.py`、`image_studio_images.py`、`image_studio_generation.py`；CDP 底座迁入 `app/cdp.py`（当前根目录 `cdp.py` 为已验证来源）。
- 已验证脚本（重构进模块）：`prototypes/demo.py`、`prototypes/set_title.py`、`prototypes/set_cover.py`、`prototypes/get_title.py`、`prototypes/cookies.py`、`prototypes/inspect_images.py`、`prototypes/grab.py`。
- 外部依赖：本机 Google Chrome；Shopee；AI 服务（文本+图像；普通产品默认 cmhub 网关，由 `data/config.json` 的 `ai.cmhub` + `data/config/cmhub.json` 配置；direct 直连模型清单仅作为内部兼容/手工回滚路径保留）；`openpyxl`。
@@ -59,7 +62,7 @@ imported → collected → generated → applied
## 三、职责划分
**GUI（5 Tab）**：见 [routes.md](routes.md)。只做交互与预览，不写业务逻辑；耗时操作走 PySide6 `QObject` worker + `QThread`，用 signal 回主线程刷新 UI。**① 采集点击后会为本轮匹配到的账号自动确保 Chrome 就绪：已打开则复用，未打开才启动；随后只检测登录态，未登录账号的任务跳过并汇总提示去 ④人工登录。③ 更新蝦皮仍是线上提交高风险链路：执行前只检测账号 Chrome/CDP/登录态，不自动启动缺失账号 Chrome。**
**GUI（6 Tab）**：见 [routes.md](routes.md)。只做交互与预览，不写业务逻辑；耗时操作走 PySide6 `QObject` worker + `QThread`，用 signal 回主线程刷新 UI。**① 采集点击后会为本轮匹配到的账号自动确保 Chrome 就绪：已打开则复用，未打开才启动；随后只检测登录态，未登录账号的任务跳过并汇总提示去 ④人工登录。③ 更新蝦皮仍是线上提交高风险链路：执行前只检测账号 Chrome/CDP/登录态，不自动启动缺失账号 Chrome。⑥ AI工场只读拉图、下载图片和 cmhub 托管生成都走 worker，不自动上传蝦皮。**
**核心模块**
docs/README.md
+2 -2
View File
@@ -4,7 +4,7 @@
## 一句话定位
蝦皮圈優化助手（代号 cmshopee）是一个给**电商运营**使用的 Windows PySide6 桌面自动化工具，用于**管理多个 Shopee 卖家账号、并用 CDP 驱动 Chrome 批量修改商品标题、替换商品封面图**。V0 已验证单账号「改标题 + 换封面」闭环；当前 V1 目标是 5 Tab 流水线：导入采集 → AI 生成 → 点击「开始更新」并确认后批量提交 → 回写结果。
蝦皮圈優化助手（代号 cmshopee）是一个给**电商运营**使用的 Windows PySide6 桌面自动化工具，用于**管理多个 Shopee 卖家账号、并用 CDP 驱动 Chrome 批量修改商品标题、替换商品封面图**。V0 已验证单账号「改标题 + 换封面」闭环；当前 V1 工作台为 6 Tab：①导入采集 → ②AI生成 → ③点击「开始更新」并确认后批量提交 → 回写结果，④账号管理，⑤设置，⑥AI工场图片候选生成。
## 文档导航
@@ -23,7 +23,7 @@
- [界面与流程结构](routes.md)：GUI 窗口、操作流程、按钮职责（无前端路由，用 GUI 流程替代）。
- [当前实现状态](current-state.md)：当前代码现实、可运行命令、下一步可做任务。
- [常见问题排查](troubleshooting.md)：本地配置、启动报错、敏感文件修复等排障记录。
- [产品与 UI 评估](ux-review.md)：以 PM + UI 设计视角评估 5 Tab 模块 / 组件合理性，含优化方案与优先级清单。
- [产品与 UI 评估](ux-review.md)：以 PM + UI 设计视角评估主流程模块 / 组件合理性，含优化方案与优先级清单。
- [界面配色设计](ui-color-design.md)：语义色板与组件配色映射规范，指导给状态 / 按钮 / 校验 / 登录状态上色。
- [工程评估](engineering-review.md)：全栈视角评估工程基础设施与可维护性（依赖清单 / CI / 打包 / gui.py 拆分 / lint），含 P0-P2 与优先级。
- [打包与分发](packaging.md)：PyInstaller 免安装 exe 打包命令、排除本地数据规则和用户手动更新方式。
docs/routes.md
+31 -3
View File
@@ -1,11 +1,11 @@
# 界面与流程结构
> 桌面工具，无前端路由。用 **5 Tab GUI（PySide6 `QMainWindow` + `QTabWidget`）+ 流水线** 约定界面职责与导航。
> 桌面工具，无前端路由。用 **6 Tab GUI（PySide6 `QMainWindow` + `QTabWidget`）+ 流水线** 约定界面职责与导航。
## Tab 顺序与职责（工作流优先）
```
① 导入采集 │ ② AI生成 │ ③ 更新蝦皮 │ ④ 账号管理 │ ⑤ 设置
① 导入采集 │ ② AI生成 │ ③ 更新蝦皮 │ ④ 账号管理 │ ⑤ 设置 │ ⑥ AI工场
```
| Tab | 职责 | 风险 |
@@ -15,12 +15,13 @@
| ③ 更新蝦皮 | 对**已生成**任务点击「开始更新」后弹窗确认；确认后打开编辑页换标题+封面并逐条点「更新」提交；结果回写 Excel | **写线上，高** |
| ④ 账号管理 | Shopee 账号（账号名/别名/数据目录/端口/密码本地明文仅参考/登录状态）；启动登录、检测登录、生成快捷方式；启动登录必须复用已打开的同账号 Chrome，避免重复开窗口；检测登录遇到 `accounts.shopee.tw/seller/login` 必须显示未登录 | 中 |
| ⑤ 设置 | cmhub 网关/API Key、生文/生图别名、生成参数、Chrome 路径、默认端口、蝦皮更新执行参数等 | — |
| ⑥ AI工场 | 按账号+商品建立图片项目；只读拉取蝦皮原主图 URL；单击原图下载进入照片池；选择源图后用 cmhub 托管模型生成多张主图/详情图候选；管理 AI工场完整提示词模板；终选拖放与导出由后续任务接入 | 不触线上，中 |
任务的**阶段状态**贯穿各 Tab：`imported → collected → generated → applied`（或 `failed/skipped`）。② 不设逐条人工确认阶段；③ 无常驻提交开关，点击「开始更新」后必须弹窗确认当前筛选范围、任务数量和线上提交风险。各 Tab 聚焦各自阶段的列与按钮，但操作同一批任务（同一 batch）。
## 全局 Tab 栏可用性
5 个主 Tab 是高频导航入口，不能使用 Qt 默认的紧凑宽度。`MainWindow` 必须为 `QTabWidget/QTabBar` 设置基础样式：
6 个主 Tab 是高频导航入口，不能使用 Qt 默认的紧凑宽度。`MainWindow` 必须为 `QTabWidget/QTabBar` 设置基础样式：
- 每个 Tab 设置稳定最小宽度和足够左右 padding，避免文字贴边或窄到误点。
- Tab 之间保留明显间距，当前 Tab 有清晰背景/边框高亮。
@@ -173,6 +174,31 @@
- ③「更新内容」默认只更新标题，每批最大更新条数默认 1，同时更新蝦皮账号默认 1。
- ③ 点击「开始更新」会先按「更新内容」校验缺失内容，再弹确认框。
## ⑥ AI工场
```
┌ AI工场 ──────────────────────────────────────────────────────┐
│ 左轨：账号[▼] 商品ID[____] [打开项目] [拉取主图] [打开项目文件夹] │
│ 项目列表：账号 / 商品ID / 更新时间 │
│ 中区：蝦皮原主图（单击下载进池，双击预览） │
│ 照片池：原图/主图/详情图角标、比例、排队/生成/失败状态 │
│ 右侧：模板[▼] [新建][重命名][保存][删除] │
│ [完整提示词输入框] │
│ 类型[主图▼] 数量[4] 比例[1:1▼] cmhub扣点/余额提示 │
│ [开始生成][停止] 进度条 运行日志 │
│ 底部：主图终选 / 详情图终选占位（T-592 接拖放排序） │
└───────────────────────────────────────────────────────────────┘
```
- 项目以 `账号别名 + 商品ID` 唯一；打开项目只创建/切换本地项目，不修改蝦皮。
- 「拉取主图」复用已验证只读 CDP：后台打开商品详情页读取主图 URL，写入 `image_studio_assets(kind=original)`；不下载图片、不改标题/封面、不点击更新。
- 原主图抽屉单击时才下载对应远程原图到项目 `originals/` 并设为源图；双击远程原图会先下载再打开大图预览。
- 照片池展示原图、生成主图、生成详情图和在途/失败任务状态；单击可用图片设为源图，双击打开大图；右键移除只删除未被任务或终选引用的照片池记录，不删除本地图片文件。
- 右侧只有一个完整提示词框；模板目录固定为 `data/prompts/image_studio/`，与②标题/封面模板隔离。界面不显示“主提示词 / 每张动作词”。
- 生图固定走 cmhub 托管模型，使用⑤设置里的 cmhub Base URL/API Key/生图别名和图片并发；界面只显示扣点、余额、进度、失败，不展示自定义 Provider、API Key、生成来源选择或“导入本地图片”入口。
- 拉主图、下载原图、生图 submit/poll/download 均通过 worker 执行，主线程只刷新 UI；运行中禁用项目切换、模板编辑、源图选择和生成设置，停止为协作式停止。
- 本小节只覆盖 T-591 第一版：终选拖放排序由 T-592 接入，导出 JPEG 由 T-593 接入，故底部终选盘当前是只读占位。
## 流程导航
```text
@@ -201,10 +227,12 @@
| `ApplyTab(QWidget)` | ③ | 已生成任务筛选 +「更新内容」下拉 + 缺失内容阻断 +「检查本轮更新」+ 分批开始更新确认 + 检查/真实更新运行日志 + 结果回写与结束汇总 |
| `AccountsTab(QWidget)` | ④ | 账号增删改、启动登录、检测登录、生成快捷方式；登录检测把 Shopee accounts 登录页判为未登录 |
| `SettingsTab(QWidget)` | ⑤ | cmhub 网关配置 + 响应式三列设置表单 + 生成参数 + Chrome/端口配置 + 蝦皮更新执行；数据路径字段隐藏但保留配置兼容 |
| `ImageStudioTab(QWidget)` | ⑥ | AI工场项目列表、只读拉蝦皮主图、原图下载进池、照片池、大图预览、完整提示词模板 CRUD、cmhub 托管多图生成控制 |
| `TaskTableModel(QAbstractTableModel)` | ①②③ | 任务表格数据模型，供 `QTableView` 使用 |
| `BaseWorker(QObject)` | 后台 | 定义 `progress/log/row_updated/failed/finished/cancelled` signals |
| `ApplyWorker(BaseWorker)` | ③ | 账号就绪预检、检查本轮更新、按每批最大条数分批、按账号并行或串行调用 `editor.apply_task(...)`、逐条 `set_applied()`，失败继续，写运行日志 |
| `AIModelTestWorker(BaseWorker)` | ⑤ | 后台调用 `appconfig.test_ai_model()` 测试模型连接 |
| `WriteBackWorker(BaseWorker)` | ①③ | ①回写旧字段；③回写新标题/新封面/更新状态 |
| `ImageStudioPullImagesWorker / ImageStudioDownloadOriginalWorker / ImageStudioGenerateJobsWorker` | ⑥ | 后台执行只读拉主图、远程原图下载、cmhub 托管生图 submit/poll/download；不直接操作 QWidget |
> 采集、生成、更新都是耗时操作，使用 `QObject` worker + `QThread`。Worker 不直接操作 QWidget，只通过 signal 通知主线程刷新 UI。
docs/tasks/T-591.md
+8 -2
View File
@@ -3,7 +3,7 @@ id: T-591
title: ⑥ AI工场主界面、照片池、大图预览与生成控制
phase: 7
deps: [T-588, T-589, T-590]
status: TODO
status: DONE
created: 2026-07-11
---
@@ -39,4 +39,10 @@ created: 2026-07-11
## 执行记录
（完成后记录 GUI 文件、视口截图和测试。）
- 2026-07-11：完成 ⑥ AI工场主界面第一版接入。
- 新增 `ImageStudioTab`：主窗口出现第六个 Tab `⑥ AI工场`，包含账号+商品项目入口、项目列表、只读拉主图、原主图抽屉、照片池、大图预览、单一完整提示词框、AI工场模板 CRUD、类型/数量/比例/cmhub 扣点提示、开始生成/停止/进度/日志和底部终选占位。
- 新增 `ImageStudioPullImagesWorker`、`ImageStudioDownloadOriginalWorker`、`ImageStudioGenerateJobsWorker`：拉蝦皮主图、远程原图下载、cmhub 托管多图 submit/poll/download 均走 worker；运行中禁用关键控件，停止为协作式停止。
- `image_studio` 增加照片池移除保护：未被任务/终选引用的资产可从池记录移除但不删除本地文件；被 job/selection 引用时阻断。
- 更新 `docs/routes.md`、`docs/04-architecture.md`、入口/技术栈/README 文档，把主窗口事实改为 6 Tab，并明确 T-591 只做照片池/生成，终选拖放和导出留给 T-592/T-593。
- 补充 GUI/offscreen 与 service 测试：主窗口 6 Tab 顺序、AI工场控件/objectName、无“导入本地图片”入口、模板读取、项目创建、原图/照片池状态展示、照片池移除引用保护。
- 验证：当前主工作区因无关未提交 `app/default_prompts/cover/*` 变动导致全量 unittest 的默认封面模板断言失败；已在干净 worktree 仅套用 T-591 diff 后运行 `python -m ruff check app tests main.py`、`py -3.10 -m compileall app main.py`、`py -3.10 -m unittest discover -s tests`（362 tests）和 `git diff --check`，全部通过。
tests/test_gui.py
+58 -3
View File
@@ -12,7 +12,7 @@ sys.path.insert(0, os.path.dirname(__file__))
from _helpers import TempDirMixin
from app import gui
from app import accounts, ai, appconfig, db, image_paths, prompts, update_check
from app import accounts, ai, appconfig, db, image_paths, image_studio, prompts, update_check
if gui.QT_IMPORT_ERROR is not None:
raise unittest.SkipTest("PySide6 未安装")
@@ -33,6 +33,7 @@ from app.gui import (
CollectTab,
GenerateWorker,
GenerateTab,
ImageStudioTab,
MainWindow,
SettingsTab,
TAB_STYLE,
@@ -436,13 +437,13 @@ class GuiTests(TempDirMixin, unittest.TestCase):
self.assert_foreground(model, 0, 4, gui.COLOR_SUCCESS)
self.assert_foreground(model, 0, 5, gui.COLOR_SUCCESS)
def test_main_window_has_five_tabs_in_workflow_order(self):
def test_main_window_has_workflow_tabs_in_order(self):
with self.make_temp_dir() as temp_dir:
window = MainWindow(config=self.make_config(temp_dir))
self.addCleanup(window.close)
self.assertEqual(gui.display_name(), window.windowTitle())
self.assertEqual(5, window.tabs.count())
self.assertEqual(6, window.tabs.count())
self.assertEqual(
TAB_TITLES,
[window.tabs.tabText(index) for index in range(window.tabs.count())],
@@ -460,6 +461,7 @@ class GuiTests(TempDirMixin, unittest.TestCase):
self.assertIsInstance(window.tabs.widget(1), GenerateTab)
self.assertIsInstance(window.tabs.widget(2), ApplyTab)
self.assertIsInstance(window.tabs.widget(4), SettingsTab)
self.assertIsInstance(window.tabs.widget(5), ImageStudioTab)
self.assertEqual(
"回写旧数据到 Excel",
window.tabs.widget(0).write_back_button.text(),
@@ -487,6 +489,59 @@ class GuiTests(TempDirMixin, unittest.TestCase):
self.assert_removed(temp_dir)
def test_image_studio_tab_builds_project_pool_and_template_controls(self):
with self.make_temp_dir() as temp_dir:
cfg = self.make_config(temp_dir)
prompts_dir = os.path.join(temp_dir, "prompts", "image_studio")
accounts.create_account("主店", "alias-a", debug_port=9222, config=cfg)
prompts.save_image_studio_template("工场模板", "完整提示词", prompts_dir)
tab = ImageStudioTab(
config=cfg,
db_path=cfg["db_path"],
prompts_dir=prompts_dir,
)
self.addCleanup(tab.close)
self.assertEqual("imageStudioTab", tab.objectName())
self.assertEqual("⑥ AI工场", TAB_TITLES[5])
self.assertGreaterEqual(tab.account_combo.count(), 1)
self.assertEqual("alias-a", tab.account_combo.itemData(0))
self.assertNotIn(
"导入本地图片",
" ".join(button.text() for button in tab.findChildren(gui.QPushButton)),
)
self.assertEqual("完整提示词", prompts.load_image_studio_template("工场模板", prompts_dir))
template_index = tab.template_combo.findData("工场模板")
self.assertGreaterEqual(template_index, 0)
tab.template_combo.setCurrentIndex(template_index)
tab.load_selected_template()
self.assertEqual("完整提示词", tab.prompt_edit.toPlainText())
tab.item_id_edit.setText("51100639510")
tab.open_project()
self.assertIsNotNone(tab.current_project)
self.assertEqual("alias-a", tab.current_project.account_alias)
self.assertEqual("51100639510", tab.current_project.item_id)
self.assertEqual(1, tab.project_table.rowCount())
original = image_studio.sync_original_asset_urls(
tab.current_project.id,
[{"index": 1, "src": "https://susercontent.com/main-1.jpg"}],
path=cfg["db_path"],
)[0]
tab.refresh_project_assets()
self.assertEqual(1, tab.original_table.rowCount())
self.assertEqual("1", tab.original_table.item(0, 0).text())
self.assertEqual("远程待下载", tab.original_table.item(0, 1).text())
self.assertEqual(1, tab.pool_table.rowCount())
self.assertEqual("原图", tab.pool_table.item(0, 0).text())
self.assertEqual("远程待下载", tab.pool_table.item(0, 2).text())
self.assertEqual(original.id, tab.pool_table.item(0, 0).data(gui.Qt.UserRole)["asset_id"])
self.assert_removed(temp_dir)
def test_startup_update_gate_forced_blocks_and_opens_download(self):
boxes = []
tests/test_image_studio.py
+29
View File
@@ -210,6 +210,35 @@ class ImageStudioTests(TempDirMixin, unittest.TestCase):
self.assert_removed(temp_dir)
def test_remove_asset_only_when_not_referenced(self):
with self.make_temp_dir() as temp_dir:
db_path = os.path.join(temp_dir, "cmshopee.db")
db.init_db(db_path)
project = image_studio.create_or_get_project(
account_alias="alias",
account_slug="alias_slug",
item_id="51100639510",
path=db_path,
)
free_asset = image_studio.add_asset(project.id, "original", path=db_path)
referenced_asset = image_studio.add_asset(project.id, "original", path=db_path)
image_studio.create_job(
project.id,
source_asset_id=referenced_asset.id,
path=db_path,
)
counts = image_studio.asset_reference_counts(referenced_asset.id, path=db_path)
self.assertEqual(1, counts["source_job"])
with self.assertRaisesRegex(db.DbError, "引用"):
image_studio.remove_asset_if_unused(referenced_asset.id, path=db_path)
removed = image_studio.remove_asset_if_unused(free_asset.id, path=db_path)
self.assertEqual(free_asset.id, removed.id)
self.assertIsNone(image_studio.get_asset(free_asset.id, path=db_path))
self.assert_removed(temp_dir)
def test_sync_original_asset_urls_is_idempotent_and_marks_missing(self):
with self.make_temp_dir() as temp_dir:
db_path = os.path.join(temp_dir, "cmshopee.db")