555 lines
20 KiB
Python
555 lines
20 KiB
Python
"""Task table models for GUI tabs."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from .. import ai, appconfig, diagnostics
|
|
from ..collect_skip import skipped_stage_text
|
|
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": "已取消",
|
|
}
|
|
PRODUCT_UNAVAILABLE_MARKERS = (
|
|
"商品失效",
|
|
"please input correct product id",
|
|
"商品不存在",
|
|
"商品已删除",
|
|
"商品已下架",
|
|
"無效商品",
|
|
"无效商品",
|
|
"無權限",
|
|
"无权限",
|
|
"product not found",
|
|
"item not found",
|
|
"invalid product",
|
|
"invalid item",
|
|
)
|
|
|
|
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:
|
|
error = getattr(task, "last_error", "") or ""
|
|
if self.is_unmatched(task) or getattr(task, "status", "") == "skipped":
|
|
if error:
|
|
return diagnostics.redact_log_text(str(error))
|
|
if self.is_unmatched(task):
|
|
return "别名未匹配账号，采集时将略过"
|
|
return "未记录具体略过原因"
|
|
if error:
|
|
return diagnostics.redact_log_text(str(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 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 _is_product_unavailable(self, task) -> bool:
|
|
if getattr(task, "status", "") != "failed":
|
|
return False
|
|
error = str(getattr(task, "last_error", "") or "").lower()
|
|
return bool(error) and any(marker.lower() in error for marker in self.PRODUCT_UNAVAILABLE_MARKERS)
|
|
|
|
def _stage_text(self, task) -> str:
|
|
unmatched = self.is_unmatched(task)
|
|
if unmatched or getattr(task, "status", "") == "skipped":
|
|
return skipped_stage_text(
|
|
getattr(task, "last_error", ""),
|
|
alias_matched=not unmatched,
|
|
)
|
|
if self._is_product_unavailable(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
|
|
self.generate_mode = "title"
|
|
self.generate_cover_enabled = False
|
|
|
|
def set_generate_cover_enabled(self, enabled):
|
|
self.set_generate_mode("title_cover" if enabled else "title")
|
|
|
|
def set_generate_mode(self, mode):
|
|
self.generate_mode = appconfig.normalize_generate_mode(mode)
|
|
self.generate_cover_enabled = appconfig.generate_mode_includes_cover(self.generate_mode)
|
|
if not self.tasks:
|
|
return
|
|
top_left = self.index(0, 4)
|
|
bottom_right = self.index(len(self.tasks) - 1, 5)
|
|
self.dataChanged.emit(top_left, bottom_right, [Qt.DisplayRole, Qt.ForegroundRole])
|
|
|
|
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() == 1:
|
|
if self._cover_reset_count(task) > 0:
|
|
return _qcolor(COLOR_WARNING)
|
|
if role == Qt.ForegroundRole and index.column() in {4, 5}:
|
|
return self._component_status_color(self._display_value(task, index.column()))
|
|
if role == Qt.ToolTipRole:
|
|
if index.column() == 1 and self._cover_reset_count(task) > 0:
|
|
return self._cover_reset_tooltip(task)
|
|
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, level=None):
|
|
_emit_status(self.status_callback, message, level=level)
|
|
|
|
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 _failed_status_text(self, task):
|
|
collect_attempts = self._attempt_count(task, "collect_attempts")
|
|
generate_attempts = self._attempt_count(task, "generate_attempts")
|
|
apply_attempts = self._attempt_count(task, "apply_attempts")
|
|
if apply_attempts > 0:
|
|
return "更新失败"
|
|
if generate_attempts > 0:
|
|
return "生成失败"
|
|
if collect_attempts > 0 or getattr(task, "stage", None) == "imported":
|
|
return "采集失败"
|
|
return "失败"
|
|
|
|
def _attempt_count(self, task, field):
|
|
try:
|
|
return int(getattr(task, field, 0) or 0)
|
|
except (TypeError, ValueError):
|
|
return 0
|
|
|
|
def _cover_reset_count(self, task):
|
|
try:
|
|
return int(getattr(task, "cover_reset_count", 0) or 0)
|
|
except (TypeError, ValueError):
|
|
return 0
|
|
|
|
def _cover_reset_tooltip(self, task):
|
|
count = self._cover_reset_count(task)
|
|
return f"该商品封面已重置 {count} 次，可能有多张候选图，双击可查看封面画廊"
|
|
|
|
def _has_new_title(self, task):
|
|
return bool(str(getattr(task, "new_title", "") or "").strip())
|
|
|
|
def _has_new_cover(self, task):
|
|
return bool(str(getattr(task, "new_cover_path", "") or "").strip())
|
|
|
|
def _generate_failed(self, task):
|
|
return (
|
|
getattr(task, "status", None) == "failed"
|
|
and self._attempt_count(task, "generate_attempts") > 0
|
|
)
|
|
|
|
def _last_error_text(self, task):
|
|
return str(getattr(task, "last_error", "") or "").lower()
|
|
|
|
def _is_title_failure(self, task):
|
|
if not self._generate_failed(task):
|
|
return False
|
|
error = self._last_error_text(task)
|
|
if any(marker in error for marker in ("title", "标题", "生文", "文本")):
|
|
return True
|
|
return not self._has_new_title(task) and not self._is_cover_failure(task)
|
|
|
|
def _is_cover_failure(self, task):
|
|
if not self._generate_failed(task):
|
|
return False
|
|
error = self._last_error_text(task)
|
|
return any(
|
|
marker in error
|
|
for marker in ("cover", "封面", "图片", "图像", "新封面")
|
|
)
|
|
|
|
def _is_collect_incomplete(self, task):
|
|
return (
|
|
getattr(task, "stage", None) == "imported"
|
|
and self._attempt_count(task, "generate_attempts") == 0
|
|
and not self._is_collect_failure(task)
|
|
)
|
|
|
|
def _is_collect_failure(self, task):
|
|
return (
|
|
getattr(task, "status", None) == "failed"
|
|
and self._attempt_count(task, "collect_attempts") > 0
|
|
and self._attempt_count(task, "generate_attempts") == 0
|
|
)
|
|
|
|
def _title_status_text(self, task):
|
|
if getattr(task, "status", None) == "skipped":
|
|
return "略过"
|
|
if self._is_collect_failure(task):
|
|
return "采集失败"
|
|
if self._is_collect_incomplete(task):
|
|
return "未采集"
|
|
if self._is_title_failure(task):
|
|
return "失败"
|
|
if self._has_new_title(task):
|
|
return "已生成"
|
|
if getattr(task, "status", None) == "running":
|
|
return "处理中"
|
|
if getattr(task, "stage", None) == "applied":
|
|
return "已生成" if self._has_new_title(task) else "未生成"
|
|
return "待生成"
|
|
|
|
def _cover_status_text(self, task):
|
|
if getattr(task, "status", None) == "skipped":
|
|
return "略过"
|
|
if self._is_collect_failure(task):
|
|
return "采集失败"
|
|
if self._is_collect_incomplete(task):
|
|
return "未采集"
|
|
if self._has_new_cover(task):
|
|
return "已生成"
|
|
if self._is_cover_failure(task):
|
|
return "失败"
|
|
if not self.generate_cover_enabled:
|
|
return "未启用"
|
|
if self.generate_mode == "cover":
|
|
if not ai.cover_title_context(task):
|
|
return "未开始"
|
|
elif self._is_title_failure(task) or not self._has_new_title(task):
|
|
return "未开始"
|
|
if getattr(task, "status", None) == "running":
|
|
return "处理中"
|
|
return "待生成"
|
|
|
|
def _component_status_color(self, text):
|
|
if text in {"失败", "采集失败"}:
|
|
return _qcolor(COLOR_DANGER)
|
|
if text == "处理中":
|
|
return _qcolor(COLOR_INFO)
|
|
if text == "已生成":
|
|
return _qcolor(COLOR_SUCCESS)
|
|
if text in {"未启用", "未采集", "未开始", "略过", "已取消"}:
|
|
return _qcolor(COLOR_MUTED)
|
|
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._title_status_text(task),
|
|
self._cover_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)
|