387 lines
13 KiB
Python
387 lines
13 KiB
Python
"""Task table models for GUI tabs."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from .widgets import *
|
|
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.all_tasks = []
|
|
self.account_by_alias = {}
|
|
self.filter_mode = "all"
|
|
|
|
def set_tasks(self, tasks, accounts):
|
|
self.beginResetModel()
|
|
self.all_tasks = list(tasks)
|
|
self.account_by_alias = {
|
|
str(account.alias).strip(): account
|
|
for account in accounts
|
|
if str(account.alias).strip()
|
|
}
|
|
self.tasks = self._filtered_tasks()
|
|
self.endResetModel()
|
|
|
|
def set_filter_mode(self, mode):
|
|
self.beginResetModel()
|
|
self.filter_mode = mode if mode in {"all", "unmatched"} else "all"
|
|
self.tasks = self._filtered_tasks()
|
|
self.endResetModel()
|
|
|
|
def _filtered_tasks(self):
|
|
if self.filter_mode == "unmatched":
|
|
return [task for task in self.all_tasks if self.is_unmatched(task)]
|
|
return list(self.all_tasks)
|
|
|
|
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.ForegroundRole and index.column() == 3:
|
|
return self._stage_color(task)
|
|
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.all_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 _stage_color(self, task):
|
|
if self.is_unmatched(task):
|
|
return _qcolor(COLOR_MUTED)
|
|
base_color = _status_base_color(getattr(task, "status", None))
|
|
if base_color is not None:
|
|
return _qcolor(base_color)
|
|
if getattr(task, "stage", None) in {"collected", "generated", "applied"}:
|
|
return _qcolor(COLOR_SUCCESS)
|
|
return _qcolor(COLOR_PENDING)
|
|
|
|
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 GenerateTaskTableModel(QAbstractTableModel):
|
|
"""Table model for Tab 2 generation candidates."""
|
|
|
|
HEADERS = ["店铺", "商品ID", "旧标题", "新标题", "状态"]
|
|
|
|
STATUS_TEXT = {
|
|
"running": "处理中",
|
|
"failed": "失败",
|
|
"skipped": "略过",
|
|
"cancelled": "已取消",
|
|
}
|
|
|
|
STAGE_TEXT = {
|
|
"imported": "未采集",
|
|
"collected": "待生成",
|
|
"generated": "已生成",
|
|
"applied": "已更新",
|
|
}
|
|
|
|
def __init__(self, parent=None, db_path=None, status_callback=None):
|
|
super().__init__(parent)
|
|
self.tasks = []
|
|
self.account_by_alias = {}
|
|
self.db_path = db_path
|
|
self.status_callback = status_callback
|
|
self.last_edit_error = None
|
|
|
|
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 in (Qt.DisplayRole, Qt.EditRole):
|
|
return self._display_value(task, index.column())
|
|
if role == Qt.ForegroundRole and index.column() == 4:
|
|
return self._status_color(task)
|
|
if role == Qt.ToolTipRole:
|
|
if index.column() == 3 and self._can_edit_title(task):
|
|
return "双击可微调新标题，只修改本地待更新内容"
|
|
if task.last_error:
|
|
return task.last_error
|
|
return None
|
|
|
|
def setData(self, index, value, role=Qt.EditRole):
|
|
if role != Qt.EditRole or not index.isValid() or index.column() != 3:
|
|
return False
|
|
task = self.tasks[index.row()]
|
|
if not self._can_edit_title(task):
|
|
self._set_status("该任务不能修改新标题")
|
|
return False
|
|
title = str(value or "").strip()
|
|
if title == str(task.new_title or ""):
|
|
return True
|
|
try:
|
|
db.update_generated_title(task.id, title, path=self.db_path)
|
|
updated = db.get_task(task.id, path=self.db_path)
|
|
except Exception as exc:
|
|
self.last_edit_error = str(exc)
|
|
self._set_status(f"新标题修改失败：{exc}")
|
|
return False
|
|
self.tasks[index.row()] = updated
|
|
self.last_edit_error = None
|
|
self.dataChanged.emit(index, index, [Qt.DisplayRole, Qt.EditRole, Qt.ToolTipRole])
|
|
self._set_status(f"已修改新标题：商品 {task.item_id}")
|
|
return True
|
|
|
|
def flags(self, index):
|
|
if not index.isValid():
|
|
return Qt.NoItemFlags
|
|
flags = Qt.ItemIsEnabled | Qt.ItemIsSelectable
|
|
if index.column() == 3 and self._can_edit_title(self.tasks[index.row()]):
|
|
flags |= Qt.ItemIsEditable
|
|
return flags
|
|
|
|
def _can_edit_title(self, task):
|
|
return (
|
|
getattr(task, "stage", None) == "generated"
|
|
and getattr(task, "status", None) != "running"
|
|
and int(getattr(task, "committed", 0) or 0) == 0
|
|
and bool(getattr(task, "new_title", None))
|
|
)
|
|
|
|
def _set_status(self, message):
|
|
if self.status_callback is not None:
|
|
self.status_callback(message)
|
|
|
|
def task_at(self, row):
|
|
if row < 0 or row >= len(self.tasks):
|
|
return None
|
|
return self.tasks[row]
|
|
|
|
def account_name_for(self, task):
|
|
return self._account_name(task)
|
|
|
|
def _account_name(self, task):
|
|
account = self.account_by_alias.get(str(task.alias).strip())
|
|
if account is not None:
|
|
return account.account_name
|
|
return task.account_name or task.alias or ""
|
|
|
|
def _status_text(self, task):
|
|
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 _status_color(self, task):
|
|
base_color = _status_base_color(getattr(task, "status", None))
|
|
if base_color is not None:
|
|
return _qcolor(base_color)
|
|
if getattr(task, "stage", None) in {"generated", "applied"}:
|
|
return _qcolor(COLOR_SUCCESS)
|
|
return _qcolor(COLOR_PENDING)
|
|
|
|
def _display_value(self, task, column):
|
|
values = [
|
|
self._account_name(task),
|
|
task.item_id,
|
|
task.old_title or "",
|
|
task.new_title or "",
|
|
self._status_text(task),
|
|
]
|
|
return values[column] if 0 <= column < len(values) else None
|
|
|
|
|
|
class ApplyTaskTableModel(QAbstractTableModel):
|
|
"""Table model for Tab 3 update candidates."""
|
|
|
|
HEADERS = ["店铺", "商品ID", "新标题", "新封面", "阶段", "结果"]
|
|
|
|
STATUS_TEXT = {
|
|
"running": "处理中",
|
|
"failed": "失败",
|
|
"skipped": "略过",
|
|
"cancelled": "已取消",
|
|
"pending": "待更新",
|
|
"success": "成功",
|
|
}
|
|
|
|
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.ForegroundRole:
|
|
if index.column() == 4:
|
|
return self._stage_color(task)
|
|
if index.column() == 5:
|
|
return self._result_color(task)
|
|
if role == Qt.ToolTipRole and task.last_error:
|
|
return task.last_error
|
|
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 account_name_for(self, task):
|
|
account = self.account_by_alias.get(str(task.alias).strip())
|
|
if account is not None:
|
|
return account.account_name
|
|
return task.account_name or task.alias or ""
|
|
|
|
def _display_value(self, task, column):
|
|
values = [
|
|
self.account_name_for(task),
|
|
task.item_id,
|
|
task.new_title or "",
|
|
os.path.basename(task.new_cover_path or ""),
|
|
self._stage_text(task),
|
|
self._result_text(task),
|
|
]
|
|
return values[column] if 0 <= column < len(values) else None
|
|
|
|
def _stage_text(self, task):
|
|
if task.stage == "generated":
|
|
return "待更新"
|
|
if task.stage == "applied":
|
|
return "已更新"
|
|
return task.stage
|
|
|
|
def _result_text(self, task):
|
|
if task.status == "success" and task.stage == "generated":
|
|
return "待更新"
|
|
return self.STATUS_TEXT.get(task.status, task.status)
|
|
|
|
def _stage_color(self, task):
|
|
if getattr(task, "stage", None) == "applied":
|
|
return _qcolor(COLOR_SUCCESS)
|
|
return _qcolor(COLOR_PENDING)
|
|
|
|
def _result_color(self, task):
|
|
base_color = _status_base_color(getattr(task, "status", None))
|
|
if base_color is not None:
|
|
return _qcolor(base_color)
|
|
if getattr(task, "status", None) == "pending":
|
|
return _qcolor(COLOR_PENDING)
|
|
if getattr(task, "status", None) == "success" and getattr(task, "stage", None) == "generated":
|
|
return _qcolor(COLOR_PENDING)
|
|
if getattr(task, "stage", None) == "applied" or getattr(task, "status", None) == "success":
|
|
return _qcolor(COLOR_SUCCESS)
|
|
return _qcolor(COLOR_PENDING)
|
|
|