3103 lines
117 KiB
Python
3103 lines
117 KiB
Python
"""Concrete PySide6 workers used by GUI tabs."""
|
||
|
||
from __future__ import annotations
|
||
|
||
import datetime as _dt
|
||
import re
|
||
import threading
|
||
import time
|
||
|
||
try:
|
||
from PySide6.QtCore import Signal
|
||
except ModuleNotFoundError: # pragma: no cover - GUI import guard
|
||
Signal = None
|
||
|
||
from .. import ai, image_studio, image_studio_export, image_studio_generation, image_studio_images
|
||
from ..collect_skip import ALIAS_UNMATCHED, LOGIN_REQUIRED, empty_skip_reason_counts
|
||
from .widgets import *
|
||
|
||
|
||
_USER_LOG_URL_RE = re.compile(r"https?://[^\s，,；;）)\]]+", re.IGNORECASE)
|
||
_USER_LOG_PATH_RE = re.compile(
|
||
r"(?i)(/api(?:/v\d+)?/[^\s，,；;）)\]]*|/generated/images/[^\s，,；;）)\]]*)"
|
||
)
|
||
|
||
|
||
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"，{_image_studio_user_detail(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
|
||
|
||
|
||
def _format_product_suite_event(event):
|
||
return _format_image_studio_event(event).replace("[AI工场]", "[商品套图]", 1)
|
||
|
||
|
||
def _image_studio_user_detail(detail):
|
||
text = diagnostics.redact_log_text(str(detail or "")).replace("\r", " ").replace("\n", " ").strip()
|
||
text = _USER_LOG_URL_RE.sub("[链接已隐藏]", text)
|
||
text = _USER_LOG_PATH_RE.sub("[接口路径已隐藏]", text)
|
||
text = text.replace("GET [链接已隐藏]", "请求 cmhub")
|
||
text = text.replace("POST [链接已隐藏]", "请求 cmhub")
|
||
if len(text) > 180:
|
||
return text[:177] + "..."
|
||
return text
|
||
|
||
|
||
class ImageStudioPullImagesWorker(BaseWorker):
|
||
"""Read Shopee main image URLs for one AI studio project in background."""
|
||
|
||
def __init__(
|
||
self,
|
||
account_alias,
|
||
item_id,
|
||
*,
|
||
pull_run_token="",
|
||
db_path=None,
|
||
config=None,
|
||
):
|
||
super().__init__()
|
||
self.account_alias = account_alias
|
||
self.item_id = item_id
|
||
self.pull_run_token = str(pull_run_token or "")
|
||
self.db_path = db_path
|
||
self.config = config
|
||
|
||
def execute(self):
|
||
if self.should_cancel():
|
||
return {
|
||
"pull_run_token": self.pull_run_token,
|
||
"cancelled": True,
|
||
"assets": [],
|
||
}
|
||
|
||
def on_step(payload):
|
||
self.log.emit(_format_image_studio_event(payload))
|
||
|
||
try:
|
||
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,
|
||
should_stop=self.should_cancel,
|
||
)
|
||
except image_studio.ImageStudioPullCancelled as exc:
|
||
return {
|
||
"pull_run_token": self.pull_run_token,
|
||
"cancelled": True,
|
||
"project": exc.project,
|
||
"assets": list(exc.assets or []),
|
||
}
|
||
project = result.get("project")
|
||
assets = result.get("assets") or []
|
||
return {
|
||
"pull_run_token": self.pull_run_token,
|
||
"project": project,
|
||
"assets": assets,
|
||
"count": len(assets),
|
||
"account": result.get("account"),
|
||
"cancelled": self.should_cancel(),
|
||
}
|
||
|
||
|
||
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,
|
||
max_retries=2,
|
||
retry_delays=(1, 2),
|
||
):
|
||
super().__init__()
|
||
self.asset_id = int(asset_id)
|
||
self.db_path = db_path
|
||
self.config = config
|
||
self.open_after = bool(open_after)
|
||
self.max_retries = max(0, int(max_retries or 0))
|
||
self.retry_delays = tuple(float(delay) for delay in (retry_delays or ()))
|
||
|
||
def execute(self):
|
||
attempts = self.max_retries + 1
|
||
for attempt in range(1, attempts + 1):
|
||
if self.should_cancel():
|
||
return {"asset_id": self.asset_id, "cancelled": True}
|
||
self.progress.emit(
|
||
{
|
||
"asset_id": self.asset_id,
|
||
"state": "start",
|
||
"attempt": attempt,
|
||
"attempts": attempts,
|
||
}
|
||
)
|
||
self.log.emit(f"[AI工场] 下载蝦皮原主图 #{self.asset_id}：开始")
|
||
try:
|
||
asset = image_studio_images.download_original_asset(
|
||
self.asset_id,
|
||
path=self.db_path,
|
||
config=self.config,
|
||
should_stop=self.should_cancel,
|
||
)
|
||
except image_studio_images.ImageStudioImageCancelled:
|
||
return {"asset_id": self.asset_id, "cancelled": True}
|
||
except Exception:
|
||
retry = attempt
|
||
if attempt >= attempts:
|
||
self.progress.emit(
|
||
{
|
||
"asset_id": self.asset_id,
|
||
"state": "failed",
|
||
"attempt": attempt,
|
||
"attempts": attempts,
|
||
}
|
||
)
|
||
self.log.emit(f"[AI工场] 下载蝦皮原主图 #{self.asset_id}：最终失败")
|
||
return {
|
||
"ok": False,
|
||
"asset_id": self.asset_id,
|
||
"open_after": self.open_after,
|
||
"error": "蝦皮原主图下载失败，请稍后再次点击图片重试。",
|
||
}
|
||
delay = self._retry_delay(retry)
|
||
self.progress.emit(
|
||
{
|
||
"asset_id": self.asset_id,
|
||
"state": "retry",
|
||
"retry": retry,
|
||
"max_retries": self.max_retries,
|
||
"delay_seconds": delay,
|
||
}
|
||
)
|
||
self.log.emit(
|
||
f"[AI工场] 下载蝦皮原主图 #{self.asset_id}：失败，准备重试 {retry}/{self.max_retries}"
|
||
)
|
||
if not self._wait_for_retry(delay):
|
||
return {"asset_id": self.asset_id, "cancelled": True}
|
||
continue
|
||
self.progress.emit(
|
||
{
|
||
"asset_id": self.asset_id,
|
||
"state": "success",
|
||
"attempt": attempt,
|
||
"attempts": attempts,
|
||
}
|
||
)
|
||
self.log.emit(f"[AI工场] 下载蝦皮原主图 #{self.asset_id}：成功")
|
||
return {"asset": asset, "asset_id": self.asset_id, "open_after": self.open_after}
|
||
return {"asset_id": self.asset_id, "cancelled": True}
|
||
|
||
def _retry_delay(self, retry):
|
||
if retry <= 0:
|
||
return 0.0
|
||
index = min(retry - 1, len(self.retry_delays) - 1)
|
||
return self.retry_delays[index] if index >= 0 else 0.0
|
||
|
||
def _wait_for_retry(self, delay_seconds):
|
||
deadline = time.monotonic() + max(0.0, float(delay_seconds or 0))
|
||
while time.monotonic() < deadline:
|
||
if self.should_cancel():
|
||
return False
|
||
time.sleep(min(0.1, max(0.0, deadline - time.monotonic())))
|
||
return not self.should_cancel()
|
||
|
||
|
||
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
|
||
|
||
|
||
class ProductSuiteGenerateWorker(BaseWorker):
|
||
"""Create and run independently configured product-suite jobs."""
|
||
|
||
def __init__(
|
||
self,
|
||
project_id,
|
||
job_specs,
|
||
*,
|
||
run_token="",
|
||
generation_round_key=None,
|
||
aspect_ratio="1:1",
|
||
db_path=None,
|
||
config=None,
|
||
cmhub_config_path=None,
|
||
):
|
||
super().__init__()
|
||
self.project_id = int(project_id)
|
||
self.job_specs = [dict(spec) for spec in (job_specs or [])]
|
||
self.run_token = str(run_token or "")
|
||
self.generation_round_key = str(generation_round_key or "").strip()
|
||
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.job_ids = []
|
||
self._done = 0
|
||
self._failed = 0
|
||
self._lock = threading.Lock()
|
||
|
||
def execute(self):
|
||
total = len(self.job_specs)
|
||
if total <= 0:
|
||
raise ValueError("商品套图生成任务不能为空")
|
||
jobs = []
|
||
for spec in self.job_specs:
|
||
jobs.append(
|
||
image_studio.create_job(
|
||
self.project_id,
|
||
source_asset_id=spec.get("source_asset_id"),
|
||
job_type=spec.get("job_type") or "套图",
|
||
prompt=spec.get("prompt") or "",
|
||
generation_source="cmhub",
|
||
provider="cmhub",
|
||
generation_round_key=spec.get("generation_round_key") or self.generation_round_key or None,
|
||
generation_slot_index=spec.get("generation_slot_index"),
|
||
path=self.db_path,
|
||
)
|
||
)
|
||
self.job_ids = [job.id for job in jobs]
|
||
self.progress.emit(
|
||
{
|
||
"run_token": self.run_token,
|
||
"total": len(jobs),
|
||
"done": 0,
|
||
"failed": 0,
|
||
"job_ids": list(self.job_ids),
|
||
}
|
||
)
|
||
|
||
def on_event(payload):
|
||
event = dict(payload or {})
|
||
self.log.emit(_format_product_suite_event(event))
|
||
if event.get("step") == "job_done":
|
||
with self._lock:
|
||
self._done += 1
|
||
if event.get("result") != "success":
|
||
self._failed += 1
|
||
progress = {
|
||
"run_token": self.run_token,
|
||
"total": len(jobs),
|
||
"done": self._done,
|
||
"failed": self._failed,
|
||
"job_ids": list(self.job_ids),
|
||
}
|
||
self.progress.emit(progress)
|
||
|
||
summary = image_studio_generation.run_jobs(
|
||
jobs,
|
||
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
|
||
summary["job_ids"] = list(self.job_ids)
|
||
summary["cancelled_count"] = int(summary.get("cancelled", 0) or 0)
|
||
summary["run_token"] = self.run_token
|
||
summary["generation_round_key"] = self.generation_round_key or None
|
||
return summary
|
||
|
||
|
||
class ProductSuiteHistoryExportWorker(BaseWorker):
|
||
"""Copy one product-suite generation round outside the GUI thread."""
|
||
|
||
def __init__(self, project_id, generation_round_key, parent_dir, *, db_path=None):
|
||
super().__init__()
|
||
self.project_id = int(project_id)
|
||
self.generation_round_key = generation_round_key
|
||
self.parent_dir = str(parent_dir or "")
|
||
self.db_path = db_path
|
||
|
||
def execute(self):
|
||
self.log.emit("[商品套图] 导出历史套图：开始")
|
||
result = image_studio_export.export_generation_round(
|
||
self.project_id,
|
||
self.generation_round_key,
|
||
self.parent_dir,
|
||
path=self.db_path,
|
||
should_stop=self.should_cancel,
|
||
)
|
||
self.log.emit("[商品套图] 导出历史套图：完成")
|
||
return {
|
||
"target_dir": result.target_dir,
|
||
"file_count": len(result.files),
|
||
"skipped_count": int(result.skipped_count),
|
||
"cancelled": bool(result.cancelled),
|
||
}
|
||
|
||
|
||
class ProductSuiteAiWriteWorker(BaseWorker):
|
||
"""Analyze local product images without blocking the suite workspace."""
|
||
|
||
def __init__(
|
||
self,
|
||
instruction,
|
||
context,
|
||
*,
|
||
image_paths=None,
|
||
config=None,
|
||
cmhub_config_path=None,
|
||
):
|
||
super().__init__()
|
||
self.instruction = str(instruction or "")
|
||
self.context = str(context or "")
|
||
self.image_paths = [str(path or "") for path in list(image_paths or [])]
|
||
self.config = config
|
||
self.cmhub_config_path = cmhub_config_path
|
||
|
||
def execute(self):
|
||
if self.should_cancel():
|
||
return {"cancelled": True}
|
||
result = ai.analyze_product_images(
|
||
self.instruction,
|
||
self.context,
|
||
self.image_paths,
|
||
config=self.config,
|
||
cmhub_config_path=self.cmhub_config_path,
|
||
)
|
||
if self.should_cancel():
|
||
return {"cancelled": True}
|
||
return {
|
||
"text": str(result.get("text") or "").strip(),
|
||
"image_count": int(result.get("image_count", 0) or 0),
|
||
"metadata": dict(result.get("metadata") or {}),
|
||
}
|
||
|
||
|
||
class ProductSuiteImportImagesWorker(BaseWorker):
|
||
"""Validate and copy local product images outside the GUI thread."""
|
||
|
||
def __init__(
|
||
self,
|
||
project_id,
|
||
*,
|
||
file_paths=None,
|
||
image_bytes=None,
|
||
filename_hint="clipboard.png",
|
||
db_path=None,
|
||
config=None,
|
||
):
|
||
super().__init__()
|
||
self.project_id = int(project_id)
|
||
self.file_paths = list(file_paths or [])
|
||
self.image_bytes = bytes(image_bytes) if image_bytes is not None else None
|
||
self.filename_hint = str(filename_hint or "clipboard.png")
|
||
self.db_path = db_path
|
||
self.config = config
|
||
|
||
def execute(self):
|
||
if self.image_bytes is not None:
|
||
asset = image_studio_images.import_original_bytes(
|
||
self.project_id,
|
||
self.image_bytes,
|
||
filename_hint=self.filename_hint,
|
||
path=self.db_path,
|
||
config=self.config,
|
||
)
|
||
return {"assets": [asset], "errors": [], "limit": 16}
|
||
return image_studio_images.import_original_files(
|
||
self.project_id,
|
||
self.file_paths,
|
||
path=self.db_path,
|
||
config=self.config,
|
||
)
|
||
|
||
|
||
class ImageStudioResumeJobsWorker(BaseWorker):
|
||
"""Resume submitted/running or failed-download AI studio jobs."""
|
||
|
||
def __init__(
|
||
self,
|
||
*,
|
||
project_id=None,
|
||
aspect_ratio="1:1",
|
||
db_path=None,
|
||
config=None,
|
||
cmhub_config_path=None,
|
||
):
|
||
super().__init__()
|
||
self.project_id = int(project_id) if project_id is not None else None
|
||
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):
|
||
def on_event(payload):
|
||
event = dict(payload or {})
|
||
self.log.emit(_format_image_studio_event(event))
|
||
if event.get("step") == "job_done":
|
||
with self._lock:
|
||
self._done += 1
|
||
if event.get("result") not in {"success"}:
|
||
self._failed += 1
|
||
progress = {"done": self._done, "failed": self._failed}
|
||
self.progress.emit(progress)
|
||
|
||
summary = image_studio_generation.resume_image_jobs(
|
||
project_id=self.project_id,
|
||
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
|
||
|
||
|
||
class ImageStudioExportWorker(BaseWorker):
|
||
"""Export AI studio final selections to local JPEG files."""
|
||
|
||
def __init__(
|
||
self,
|
||
project_id,
|
||
parent_dir,
|
||
*,
|
||
existing_mode=image_studio_export.EXISTING_FAIL,
|
||
db_path=None,
|
||
config=None,
|
||
):
|
||
super().__init__()
|
||
self.project_id = int(project_id)
|
||
self.parent_dir = parent_dir
|
||
self.existing_mode = existing_mode
|
||
self.db_path = db_path
|
||
self.config = config
|
||
|
||
def execute(self):
|
||
self.log.emit("[AI工场] 导出终选：开始")
|
||
result = image_studio_export.export_project_selection(
|
||
self.project_id,
|
||
self.parent_dir,
|
||
existing_mode=self.existing_mode,
|
||
path=self.db_path,
|
||
config=self.config,
|
||
)
|
||
self.log.emit("[AI工场] 导出终选：成功")
|
||
return {
|
||
"target_dir": result.target_dir,
|
||
"main_count": result.main_count,
|
||
"detail_count": result.detail_count,
|
||
"file_count": len(result.files),
|
||
"existing_mode": result.existing_mode,
|
||
}
|
||
|
||
|
||
def _generation_mode_label(mode):
|
||
mode = appconfig.normalize_generate_mode(mode)
|
||
return {
|
||
"title": "只生成标题",
|
||
"cover": "只生成封面",
|
||
"title_cover": "生成标题和封面",
|
||
}.get(mode, "只生成标题")
|
||
|
||
|
||
def _update_mode_label(mode):
|
||
mode = appconfig.normalize_update_mode(mode)
|
||
return {
|
||
"title": "只更新标题",
|
||
"cover": "只更新封面",
|
||
"title_cover": "更新标题和封面",
|
||
}.get(mode, "只更新标题")
|
||
|
||
|
||
class GenerateWorker(BaseWorker):
|
||
"""Generate titles and covers for eligible collected or failed generation tasks."""
|
||
|
||
def __init__(
|
||
self,
|
||
tasks,
|
||
prompt_values,
|
||
db_path=None,
|
||
config=None,
|
||
diagnostic_log_dir=None,
|
||
):
|
||
super().__init__()
|
||
self.tasks = list(tasks)
|
||
self.prompt_values = dict(prompt_values or {})
|
||
self.db_path = db_path
|
||
self.config = config
|
||
self.diagnostic_log_dir = diagnostic_log_dir
|
||
self._run_id = None
|
||
self._account_by_alias = {}
|
||
self._task_positions = {}
|
||
self._eligible_total = 0
|
||
self._last_progress_payload = {}
|
||
self._cmhub_points_balance = None
|
||
self._billing_error = None
|
||
self._billing_stop_requested = False
|
||
self._run_started_at_text = ""
|
||
self._run_started_monotonic = None
|
||
|
||
def execute(self):
|
||
self._run_started_at_text = self._format_local_time()
|
||
self._run_started_monotonic = time.monotonic()
|
||
account_rows = accounts.list_accounts(path=self.db_path, config=self.config)
|
||
account_by_alias = {
|
||
str(account.alias).strip(): account
|
||
for account in account_rows
|
||
if str(account.alias).strip()
|
||
}
|
||
self._account_by_alias = account_by_alias
|
||
ai_cfg = appconfig.ai_config(self.config)
|
||
generate_mode = appconfig.ai_generate_mode(self.config)
|
||
generate_cover = appconfig.generate_mode_includes_cover(generate_mode)
|
||
eligible = [
|
||
task for task in self.tasks
|
||
if ai.is_generatable_task(task, generate_mode=generate_mode)
|
||
]
|
||
component_totals = ai.generation_component_totals(
|
||
eligible,
|
||
generate_mode=generate_mode,
|
||
)
|
||
self._eligible_total = len(eligible)
|
||
self._task_positions = {
|
||
getattr(task, "id", None): index
|
||
for index, task in enumerate(eligible, start=1)
|
||
}
|
||
batch_ids = self._batch_ids(eligible)
|
||
self._run_id = self._create_run_log(eligible, batch_ids)
|
||
mode_text = _generation_mode_label(generate_mode)
|
||
if generate_cover:
|
||
if appconfig.ai_backend(self.config) == "cmhub":
|
||
concurrency = ai.cmhub_image_concurrency_plan(ai_cfg)
|
||
start_message = "[开始] 本轮生成 {total} 条：本轮生成内容：{mode_text}；标题{title_total}，图片{cover_total}；标题并发{title_concurrency}，图片并发{image_concurrency}，cmhub实际生图并发{request_concurrency}，下载并发{download_concurrency}；开始时间 {started_at}".format(
|
||
total=len(eligible),
|
||
mode_text=mode_text,
|
||
title_total=component_totals["title_total"],
|
||
cover_total=component_totals["cover_total"],
|
||
title_concurrency=ai_cfg.get("title_concurrency", 1),
|
||
image_concurrency=concurrency["configured_image_concurrency"],
|
||
request_concurrency=concurrency["request_concurrency"],
|
||
download_concurrency=concurrency["download_concurrency"],
|
||
started_at=self._run_started_at_text,
|
||
)
|
||
else:
|
||
start_message = "[开始] 本轮生成 {total} 条：本轮生成内容：{mode_text}；标题{title_total}，图片{cover_total}；标题并发{title_concurrency}，图片并发{image_concurrency}；开始时间 {started_at}".format(
|
||
total=len(eligible),
|
||
mode_text=mode_text,
|
||
title_total=component_totals["title_total"],
|
||
cover_total=component_totals["cover_total"],
|
||
title_concurrency=ai_cfg.get("title_concurrency", 1),
|
||
image_concurrency=ai_cfg.get("image_concurrency", 1),
|
||
started_at=self._run_started_at_text,
|
||
)
|
||
else:
|
||
start_message = "[开始] 本轮生成 {total} 条：本轮生成内容：{mode_text}；标题{title_total}；标题并发{title_concurrency}；开始时间 {started_at}".format(
|
||
total=len(eligible),
|
||
mode_text=mode_text,
|
||
title_total=component_totals["title_total"],
|
||
title_concurrency=ai_cfg.get("title_concurrency", 1),
|
||
started_at=self._run_started_at_text,
|
||
)
|
||
self._log_run_event(start_message)
|
||
try:
|
||
summary = ai.generate_batch(
|
||
self.tasks,
|
||
self.prompt_values,
|
||
ai_cfg={
|
||
"config": self.config,
|
||
"db_path": self.db_path,
|
||
"image_dir": appconfig.image_dir(self.config),
|
||
"account_by_alias": account_by_alias,
|
||
"on_task_update": self._emit_row_update,
|
||
"on_event": self._on_generation_event,
|
||
"on_error": self._on_generation_error,
|
||
"generate_cover": generate_cover,
|
||
"generate_mode": generate_mode,
|
||
},
|
||
on_progress=self._emit_generate_progress,
|
||
should_stop=self._should_stop_generation,
|
||
)
|
||
except Exception as exc:
|
||
error = diagnostics.redact_log_text(str(exc) or exc.__class__.__name__)
|
||
summary = {
|
||
"ok": False,
|
||
"error": error,
|
||
"total": len(eligible),
|
||
"title_total": component_totals["title_total"],
|
||
"title_done": 0,
|
||
"cover_done": 0,
|
||
"cover_total": component_totals["cover_total"] if generate_cover else 0,
|
||
"generated_done": 0,
|
||
"failed": len(eligible),
|
||
"cancelled": self.should_cancel(),
|
||
"generate_cover": generate_cover,
|
||
"generate_mode": generate_mode,
|
||
}
|
||
user_error = self._user_log_detail(error) or "未知错误"
|
||
self._log_run_event(
|
||
f"[失败] AI 生成运行失败：{user_error}",
|
||
level="error",
|
||
)
|
||
self._write_diagnostic_log(
|
||
"AI生成运行失败",
|
||
level="ERROR",
|
||
step="execute",
|
||
payload={"error": error},
|
||
exc=exc,
|
||
)
|
||
if self._cmhub_points_balance is not None:
|
||
summary["points_balance"] = self._cmhub_points_balance
|
||
if self._billing_error is not None:
|
||
summary["billing_error"] = dict(self._billing_error)
|
||
summary["ok"] = False
|
||
summary["cancelled"] = True
|
||
summary["run_id"] = self._run_id
|
||
summary["batch_ids"] = batch_ids
|
||
status = "failed" if summary.get("billing_error") or summary.get("error") else ("cancelled" if summary.get("cancelled") else "done")
|
||
level = "error" if summary.get("billing_error") or summary.get("error") else ("warning" if summary.get("cancelled") else "info")
|
||
self._log_run_event(self._format_generate_completion(summary), level=level)
|
||
self._finish_run_log(status, summary)
|
||
return summary
|
||
|
||
def _emit_generate_progress(self, payload):
|
||
progress = dict(payload or {})
|
||
if self._cmhub_points_balance is not None:
|
||
progress["points_balance"] = self._cmhub_points_balance
|
||
if self._billing_error is not None:
|
||
progress["billing_error"] = dict(self._billing_error)
|
||
self._last_progress_payload = dict(progress)
|
||
self.progress.emit(progress)
|
||
|
||
def _should_stop_generation(self):
|
||
return self.should_cancel() or self._billing_stop_requested
|
||
|
||
def _emit_row_update(self, task_id, fields):
|
||
self.row_updated.emit(int(task_id), dict(fields or {}))
|
||
|
||
def _on_generation_event(self, payload):
|
||
task = payload.get("task")
|
||
self._remember_cmhub_metadata(payload)
|
||
message = self._format_generation_event(payload)
|
||
if not message:
|
||
return
|
||
self._log_run_event(
|
||
message,
|
||
task=task,
|
||
level=payload.get("level") or "info",
|
||
persist=not bool(payload.get("debug_only")),
|
||
)
|
||
|
||
def _remember_cmhub_metadata(self, payload):
|
||
metadata = payload.get("metadata")
|
||
if not isinstance(metadata, dict):
|
||
return
|
||
if metadata.get("points_balance") is not None:
|
||
self._cmhub_points_balance = metadata.get("points_balance")
|
||
self._emit_generate_progress(self._last_progress_payload)
|
||
|
||
def _format_generation_event(self, payload):
|
||
task = payload.get("task")
|
||
phase = payload.get("phase") or "generate"
|
||
step = payload.get("step") or "unknown"
|
||
result = payload.get("result") or "start"
|
||
detail = self._user_log_detail(
|
||
payload.get("detail"),
|
||
phase=phase,
|
||
step=step,
|
||
code=payload.get("code"),
|
||
status=payload.get("status"),
|
||
)
|
||
if isinstance(payload.get("metadata"), dict):
|
||
return self._format_cmhub_billing_event(task, phase, payload.get("metadata"))
|
||
if phase == "title":
|
||
if result == "start" and step == "title_submit":
|
||
return f"[标题] {self._task_progress_label(task)} 开始生成"
|
||
if result == "skipped":
|
||
if detail and "旧标题作为封面参考" in detail:
|
||
return f"[图片] {self._task_progress_label(task)} {detail}"
|
||
return f"[标题] {self._task_progress_label(task)} 已有标题，跳过生文"
|
||
if result == "success" and step == "title_done":
|
||
return f"[标题] {self._task_progress_label(task)} 成功"
|
||
if result == "success" and step == "db_write":
|
||
suffix = f"，{detail}" if detail else ""
|
||
return f"[标题] {self._task_progress_label(task)} 已保存{suffix}"
|
||
if result == "retry":
|
||
return self._retry_message("标题", task, payload, detail)
|
||
if result == "failed":
|
||
return f"[失败] {self._task_plain_label(task)} 标题生成失败：{detail or '未知错误'}"
|
||
if result == "cancelled":
|
||
return f"[停止] {self._task_plain_label(task)} 标题生成已取消"
|
||
return None
|
||
if phase == "cover":
|
||
if result == "debug" and step == "cover_image_url":
|
||
debug_detail = self._debug_detail(payload.get("detail"))
|
||
return f"[调试] {self._task_progress_label(task)} {debug_detail}"
|
||
if result == "start" and step == "cover_submit":
|
||
return f"[图片] {self._task_progress_label(task)} 开始生成"
|
||
if result == "success" and step == "cover_request":
|
||
return f"[图片] {self._task_progress_label(task)} {detail or 'cmhub 已返回图片，耗时未知'}"
|
||
if result == "warning" and step == "cover_download":
|
||
return f"[图片] {self._task_progress_label(task)} {detail or '图片下载较慢'}"
|
||
if result == "success" and step == "cover_download":
|
||
return f"[图片] {self._task_progress_label(task)} {detail or '图片下载完成，耗时未知'}"
|
||
if result == "success" and step == "cover_save":
|
||
return f"[图片] {self._task_progress_label(task)} 本地保存完成，{detail or '耗时未知'}"
|
||
if result == "success" and step == "db_write":
|
||
suffix = "，已保存到本地" if detail else ""
|
||
return f"[图片] {self._task_progress_label(task)} 成功{suffix}"
|
||
if result == "retry":
|
||
return self._retry_message("图片", task, payload, detail)
|
||
if result == "failed":
|
||
return f"[失败] {self._task_plain_label(task)} 图片生成失败：{detail or '未知错误'}"
|
||
if result == "cancelled":
|
||
return f"[停止] {self._task_plain_label(task)} 图片生成已取消"
|
||
return None
|
||
return None
|
||
|
||
def _format_cmhub_billing_event(self, task, phase, metadata):
|
||
label = "标题" if phase == "title" else ("图片" if phase == "cover" else "AI")
|
||
parts = []
|
||
alias = metadata.get("alias") or metadata.get("model_used")
|
||
if alias:
|
||
parts.append(f"别名 {alias}")
|
||
if metadata.get("points_cost") is not None:
|
||
parts.append(f"扣点 {metadata.get('points_cost')}")
|
||
if metadata.get("points_balance") is not None:
|
||
parts.append(f"余额 {metadata.get('points_balance')}")
|
||
if metadata.get("call_id"):
|
||
parts.append(f"call_id={metadata.get('call_id')}")
|
||
if not parts:
|
||
return None
|
||
return f"[计费] {self._task_plain_label(task)} {label}生成：" + "，".join(str(part) for part in parts)
|
||
|
||
def _retry_message(self, label, task, payload, detail):
|
||
attempt = int(payload.get("attempt", 0) or 0)
|
||
attempts = int(payload.get("attempts", 0) or 0)
|
||
max_retries = max(0, attempts - 1)
|
||
retry_text = f"准备重试 {attempt}/{max_retries}" if max_retries else "准备重试"
|
||
reason = f"：{detail}" if detail else ""
|
||
return f"[{label}] {self._task_progress_label(task)} 调用失败，{retry_text}{reason}"
|
||
|
||
def _task_progress_label(self, task):
|
||
index = self._task_positions.get(getattr(task, "id", None), 0)
|
||
total = self._eligible_total or 0
|
||
item_id = getattr(task, "item_id", "") or "未知商品"
|
||
shop = self._task_shop_label(task)
|
||
shop_text = f"（{shop}）" if shop else ""
|
||
return f"{index}/{total} 商品 {item_id}{shop_text}"
|
||
|
||
def _task_plain_label(self, task):
|
||
item_id = getattr(task, "item_id", "") or "未知商品"
|
||
shop = self._task_shop_label(task)
|
||
return f"商品 {item_id}（{shop}）" if shop else f"商品 {item_id}"
|
||
|
||
def _task_shop_label(self, task):
|
||
alias = str(getattr(task, "alias", "") or "").strip()
|
||
account = self._account_by_alias.get(alias)
|
||
if account is not None:
|
||
return getattr(account, "account_name", None) or getattr(account, "alias", None) or alias
|
||
return getattr(task, "account_name", None) or alias
|
||
|
||
def _short_detail(self, detail):
|
||
if detail is None:
|
||
return ""
|
||
text = diagnostics.redact_log_text(str(detail)).replace("\r", " ").replace("\n", " ").strip()
|
||
if len(text) > 180:
|
||
return text[:177] + "..."
|
||
return text
|
||
|
||
def _debug_detail(self, detail):
|
||
if detail is None:
|
||
return ""
|
||
text = diagnostics.redact_log_text(str(detail)).replace("\r", " ").replace("\n", " ").strip()
|
||
if len(text) > 2000:
|
||
return text[:1997] + "..."
|
||
return text
|
||
|
||
def _user_log_detail(self, detail, phase=None, step=None, code=None, status=None):
|
||
if detail is None:
|
||
return ""
|
||
raw = diagnostics.redact_log_text(str(detail)).replace("\r", " ").replace("\n", " ").strip()
|
||
lowered = raw.lower()
|
||
code_text = str(code or "").strip().lower()
|
||
if code_text == "insufficient_points" or "点数不足" in raw:
|
||
return "点数不足，请先充值"
|
||
if code_text == "unauthorized" or "api key 无效" in raw.lower():
|
||
return "cmhub API Key 无效，请去⑤设置重填"
|
||
if code_text in {"model_not_allowed", "no_pricing_rule"} or "模型别名" in raw or "模型配置不可用" in raw:
|
||
return "cmhub 模型别名不可用，请去⑤设置刷新别名并保存"
|
||
if code_text == "content_blocked" or "内容安全" in raw:
|
||
return "cmhub 内容安全策略拒绝本次生成"
|
||
if code_text == "rate_limited" or "rate_limited" in lowered or "请求过于频繁" in raw:
|
||
return "cmhub 请求过于频繁，请稍后重试"
|
||
if code_text == "connect_timeout" or "连接 cmhub 超时" in raw:
|
||
return "连接 cmhub 超时，请检查网络或稍后重试"
|
||
if code_text == "read_timeout" or "等待 cmhub 返回超时" in raw:
|
||
return "等待 cmhub 返回超时，本条已失败；可稍后重试"
|
||
if (
|
||
code_text == "not_found"
|
||
or "not_found" in lowered
|
||
or "接口不存在" in raw
|
||
or str(status or "") == "404"
|
||
):
|
||
return "cmhub 网关接口不可用，请检查⑤设置中的 Base URL，或联系服务方确认网关版本"
|
||
if code_text == "upstream_error" or "upstream_error" in lowered or "上游" in raw:
|
||
return "cmhub 上游生成失败，请稍后重试"
|
||
if "下载 cmhub 图片失败" in raw:
|
||
return "下载 cmhub 图片失败，请检查网络后稍后重试"
|
||
|
||
text = raw.replace("image_url", "图片")
|
||
text = text.replace("返回 图片", "返回图片")
|
||
text = _USER_LOG_URL_RE.sub("[链接已隐藏]", text)
|
||
text = _USER_LOG_PATH_RE.sub("[接口路径已隐藏]", text)
|
||
text = text.replace("GET [链接已隐藏]", "请求 cmhub")
|
||
text = text.replace("POST [链接已隐藏]", "请求 cmhub")
|
||
if len(text) > 180:
|
||
return text[:177] + "..."
|
||
return text
|
||
|
||
def _format_generate_completion(self, summary):
|
||
progress = self._summary_text(summary)
|
||
billing_error = summary.get("billing_error") or {}
|
||
finished_at = self._format_local_time()
|
||
elapsed = self._format_run_elapsed()
|
||
suffix = f"；{self._completion_time_label(summary)} {finished_at}，总用时 {elapsed}"
|
||
if billing_error:
|
||
message = self._user_log_detail(billing_error.get("message") or "点数不足，请先充值") or "点数不足，请先充值"
|
||
return f"[失败] AI 生成已中止：{message}，{progress}{suffix}"
|
||
if summary.get("cancelled"):
|
||
return f"[停止] AI 生成已停止：{progress}{suffix}"
|
||
if summary.get("error"):
|
||
error = self._user_log_detail(summary.get("error")) or "未知错误"
|
||
return f"[失败] AI 生成失败：{error}，{progress}{suffix}"
|
||
return f"[完成] AI 生成完成：{progress}{suffix}"
|
||
|
||
def _completion_time_label(self, summary):
|
||
if summary.get("billing_error") or summary.get("error"):
|
||
return "失败时间"
|
||
if summary.get("cancelled"):
|
||
return "停止时间"
|
||
return "完成时间"
|
||
|
||
def _format_local_time(self):
|
||
return _dt.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
||
|
||
def _format_run_elapsed(self):
|
||
if self._run_started_monotonic is None:
|
||
return "0秒"
|
||
return self._format_duration(time.monotonic() - self._run_started_monotonic)
|
||
|
||
def _format_duration(self, seconds):
|
||
total = max(0, int(seconds or 0))
|
||
hours, remainder = divmod(total, 3600)
|
||
minutes, seconds = divmod(remainder, 60)
|
||
if hours:
|
||
return f"{hours}小时{minutes:02d}分{seconds:02d}秒"
|
||
if minutes:
|
||
return f"{minutes}分{seconds:02d}秒"
|
||
return f"{seconds}秒"
|
||
|
||
def _summary_text(self, summary):
|
||
title_total = summary.get("title_total", summary.get("total", 0))
|
||
cover_total = summary.get("cover_total", summary.get("total", 0))
|
||
return "标题{title}/{total}，图片{cover}/{cover_total}，失败{failed}".format(
|
||
title=summary.get("title_done", 0),
|
||
cover=summary.get("cover_done", 0),
|
||
cover_total=cover_total,
|
||
total=title_total,
|
||
failed=summary.get("failed", 0),
|
||
)
|
||
|
||
def _on_generation_error(self, payload):
|
||
task = payload.get("task")
|
||
phase = payload.get("phase") or "generate"
|
||
step = payload.get("step") or "unknown"
|
||
exception = payload.get("exception")
|
||
code = payload.get("code") or getattr(exception, "code", None)
|
||
status = payload.get("status") or getattr(exception, "status", None)
|
||
error = diagnostics.redact_log_text(payload.get("error") or "未知错误")
|
||
if str(code or "") == "insufficient_points":
|
||
self._billing_stop_requested = True
|
||
message = "点数不足，请先充值。本轮未开始任务将停止。"
|
||
self._billing_error = {
|
||
"code": "insufficient_points",
|
||
"message": message,
|
||
"phase": phase,
|
||
"task_id": getattr(task, "id", None),
|
||
"item_id": getattr(task, "item_id", None),
|
||
}
|
||
if status is not None:
|
||
self._billing_error["status"] = status
|
||
self._log_run_event(
|
||
f"[计费] {self._task_plain_label(task)} 点数不足，请先充值；本轮未开始任务将停止",
|
||
task=task,
|
||
level="error",
|
||
)
|
||
self._emit_generate_progress(self._last_progress_payload)
|
||
diagnostic_payload = {"phase": phase, "error": error}
|
||
if code is not None:
|
||
diagnostic_payload["code"] = str(code)
|
||
if status is not None:
|
||
diagnostic_payload["status"] = status
|
||
self._write_diagnostic_log(
|
||
"AI生成任务失败",
|
||
level="ERROR",
|
||
step=step,
|
||
task=task,
|
||
payload=diagnostic_payload,
|
||
exc=exception,
|
||
)
|
||
|
||
def _batch_ids(self, tasks):
|
||
batch_ids = []
|
||
for task in tasks:
|
||
batch_id = getattr(task, "batch_id", None)
|
||
if batch_id and batch_id not in batch_ids:
|
||
batch_ids.append(batch_id)
|
||
return batch_ids
|
||
|
||
def _create_run_log(self, eligible, batch_ids):
|
||
try:
|
||
ai_cfg = appconfig.ai_config(self.config)
|
||
return db.create_run_log(
|
||
"generate",
|
||
dry_run=False,
|
||
total=len(eligible),
|
||
options={
|
||
"batch_ids": batch_ids,
|
||
"default_text_model": ai_cfg.get("default_text_model"),
|
||
"default_image_model": ai_cfg.get("default_image_model"),
|
||
"resolution": ai_cfg.get("resolution"),
|
||
"title_concurrency": ai_cfg.get("title_concurrency"),
|
||
"image_concurrency": ai_cfg.get("image_concurrency"),
|
||
"generate_cover": ai_cfg.get("generate_cover", False),
|
||
"backend": ai_cfg.get("backend", "direct"),
|
||
},
|
||
path=self.db_path,
|
||
)
|
||
except Exception:
|
||
return None
|
||
|
||
def _finish_run_log(self, status, summary):
|
||
if self._run_id is None:
|
||
return
|
||
try:
|
||
generated_done = summary.get("generated_done")
|
||
if generated_done is None:
|
||
generated_done = summary.get("cover_done", 0)
|
||
if not summary.get("generate_cover", True) and not generated_done:
|
||
generated_done = summary.get("title_done", 0)
|
||
done = int(generated_done or 0) + int(summary.get("failed", 0) or 0)
|
||
db.finish_run_log(
|
||
self._run_id,
|
||
status=status,
|
||
done=done,
|
||
success_count=generated_done,
|
||
skipped_count=0,
|
||
failed_count=summary.get("failed", 0),
|
||
summary_json=summary,
|
||
path=self.db_path,
|
||
)
|
||
except Exception:
|
||
return
|
||
|
||
def _log_run_event(self, message, task=None, level="info", persist=True):
|
||
safe_message = diagnostics.redact_log_text(message)
|
||
self.log.emit(str(safe_message))
|
||
if self._run_id is None or not persist:
|
||
return
|
||
try:
|
||
db.add_run_log_event(
|
||
self._run_id,
|
||
safe_message,
|
||
task_id=getattr(task, "id", None),
|
||
alias=getattr(task, "alias", None),
|
||
item_id=getattr(task, "item_id", None),
|
||
level=level,
|
||
path=self.db_path,
|
||
)
|
||
except Exception:
|
||
return
|
||
|
||
def _write_diagnostic_log(
|
||
self,
|
||
message,
|
||
level="INFO",
|
||
step=None,
|
||
task=None,
|
||
payload=None,
|
||
exc=None,
|
||
):
|
||
try:
|
||
diagnostics.write_diagnostic_log(
|
||
message,
|
||
level=level,
|
||
step=step,
|
||
task_id=getattr(task, "id", None),
|
||
alias=getattr(task, "alias", None),
|
||
item_id=getattr(task, "item_id", None),
|
||
payload=payload,
|
||
exc=exc,
|
||
log_dir=self.diagnostic_log_dir,
|
||
)
|
||
except Exception:
|
||
return
|
||
|
||
class ApplyWorker(BaseWorker):
|
||
"""Apply generated title/cover changes, optionally previewing or grouping by account."""
|
||
|
||
def __init__(
|
||
self,
|
||
tasks,
|
||
db_path=None,
|
||
config=None,
|
||
preflight=True,
|
||
dry_run=False,
|
||
update_mode=None,
|
||
max_parallel_accounts=1,
|
||
batch_size=None,
|
||
diagnostic_log_dir=None,
|
||
):
|
||
super().__init__()
|
||
self.tasks = list(tasks)
|
||
self.db_path = db_path
|
||
self.config = config
|
||
self.preflight = preflight
|
||
self.dry_run = bool(dry_run)
|
||
self.update_mode = appconfig.normalize_update_mode(
|
||
update_mode,
|
||
allow_cover_update=appconfig.shopee_update_config(config).get("allow_cover_update", False),
|
||
)
|
||
self.max_parallel_accounts = max(
|
||
appconfig.SHOPEE_PARALLEL_ACCOUNTS_MIN,
|
||
min(appconfig.SHOPEE_PARALLEL_ACCOUNTS_MAX, int(max_parallel_accounts or 1)),
|
||
)
|
||
self.batch_size = None if batch_size is None else max(1, int(batch_size or 1))
|
||
self._current_batch_size = None
|
||
self._batch_count = 0
|
||
self._progress_lock = threading.Lock()
|
||
self.diagnostic_log_dir = diagnostic_log_dir
|
||
self._run_id = None
|
||
|
||
def execute(self):
|
||
account_rows = accounts.list_accounts(path=self.db_path, config=self.config)
|
||
account_by_alias = {
|
||
str(account.alias).strip(): account
|
||
for account in account_rows
|
||
if str(account.alias).strip()
|
||
}
|
||
eligible = [task for task in self.tasks if self._is_actionable_task(task)]
|
||
batch_ids = self._batch_ids(eligible)
|
||
total = len(eligible)
|
||
batch_size = self._effective_batch_size(total)
|
||
batches = self._task_batches(eligible, batch_size)
|
||
self._current_batch_size = batch_size
|
||
self._batch_count = len(batches)
|
||
counters = {
|
||
"done": 0,
|
||
"applied": 0,
|
||
"skipped": 0,
|
||
"failed": 0,
|
||
}
|
||
self._run_id = self._create_run_log(eligible, batch_ids)
|
||
self._log_run_event(
|
||
"step=start result=start detail=运行开始：{mode}，更新内容{update_mode}，任务{total}，每批最多{batch_size}，批次{batch_count}，{parallel}".format(
|
||
mode="检查本轮更新" if self.dry_run else "真实更新",
|
||
update_mode=_update_mode_label(self.update_mode),
|
||
total=total,
|
||
batch_size=batch_size,
|
||
batch_count=len(batches),
|
||
parallel=(
|
||
f"多账号并行最多{self.max_parallel_accounts}"
|
||
if self.max_parallel_accounts > 1
|
||
else "串行"
|
||
),
|
||
)
|
||
)
|
||
|
||
if self.preflight and not self.dry_run:
|
||
self._log_run_event("step=preflight result=start detail=账号就绪检查")
|
||
blocked = self._preflight_block(eligible, account_rows, account_by_alias)
|
||
if blocked:
|
||
self._log_preflight_blocked(blocked)
|
||
summary = self._summary(
|
||
ok=False,
|
||
total=total,
|
||
counters=counters,
|
||
batch_ids=batch_ids,
|
||
blocked=True,
|
||
extra=blocked,
|
||
)
|
||
self._finish_run_log("blocked", summary)
|
||
return summary
|
||
self._log_run_event("step=preflight result=success detail=账号检查通过")
|
||
elif not self.preflight:
|
||
self._log_run_event(
|
||
"step=preflight result=skipped detail=测试模式跳过更新前检查",
|
||
level="warning",
|
||
)
|
||
|
||
for batch_index, batch_tasks in enumerate(batches, start=1):
|
||
if self.should_cancel():
|
||
break
|
||
self._log_batch_start(batch_index, len(batches), batch_tasks, counters, total)
|
||
if self.dry_run:
|
||
for task in batch_tasks:
|
||
if self.should_cancel():
|
||
break
|
||
outcome = self._preview_task(task, account_by_alias)
|
||
self._record_outcome(counters, total, outcome)
|
||
elif self.max_parallel_accounts > 1:
|
||
self._run_parallel_by_account(batch_tasks, account_by_alias, counters, total)
|
||
else:
|
||
for task in batch_tasks:
|
||
if self.should_cancel():
|
||
break
|
||
outcome = self._apply_one_task(task, account_by_alias)
|
||
self._record_outcome(counters, total, outcome)
|
||
|
||
summary = self._summary(
|
||
ok=counters["failed"] == 0,
|
||
total=total,
|
||
counters=counters,
|
||
batch_ids=batch_ids,
|
||
)
|
||
self._finish_run_log("cancelled" if self.should_cancel() else "done", summary)
|
||
return summary
|
||
|
||
def _is_actionable_task(self, task):
|
||
return (
|
||
getattr(task, "stage", None) == "generated"
|
||
and getattr(task, "status", None) in {"success", "pending", "failed"}
|
||
and (
|
||
(
|
||
appconfig.update_mode_includes_title(self.update_mode)
|
||
and bool(getattr(task, "new_title", None))
|
||
)
|
||
or (
|
||
appconfig.update_mode_includes_cover(self.update_mode)
|
||
and bool(getattr(task, "new_cover_path", None))
|
||
)
|
||
)
|
||
)
|
||
|
||
def _preflight_block(self, eligible, account_rows, account_by_alias):
|
||
if not account_rows:
|
||
return {
|
||
"reason": "NO_ACCOUNTS",
|
||
"no_accounts": True,
|
||
}
|
||
duplicate_ports = self._duplicate_debug_ports(account_rows, eligible, account_by_alias)
|
||
if duplicate_ports:
|
||
return {
|
||
"reason": "DUPLICATE_DEBUG_PORT",
|
||
"duplicate_ports": duplicate_ports,
|
||
}
|
||
required_accounts = []
|
||
seen_aliases = set()
|
||
for task in eligible:
|
||
alias = str(task.alias).strip()
|
||
account = account_by_alias.get(alias)
|
||
if account is not None and alias not in seen_aliases:
|
||
required_accounts.append(account)
|
||
seen_aliases.add(alias)
|
||
not_running = []
|
||
logged_out = []
|
||
for account in required_accounts:
|
||
self._log_run_event(
|
||
f"step=check_chrome result=start detail=账号 {account.alias} debug_port={account.debug_port}",
|
||
level="info",
|
||
)
|
||
if not chrome.is_running(account.debug_port):
|
||
self._log_run_event(
|
||
f"step=check_chrome result=blocked detail=账号 {account.alias} CDP 端口未响应 debug_port={account.debug_port}",
|
||
level="warning",
|
||
)
|
||
not_running.append(self._account_payload(account, "CDP 端口未响应"))
|
||
continue
|
||
self._log_run_event(
|
||
f"step=check_chrome result=success detail=账号 {account.alias} debug_port={account.debug_port}",
|
||
level="info",
|
||
)
|
||
self._log_run_event(
|
||
f"step=login_check result=start detail=账号 {account.alias}",
|
||
level="info",
|
||
)
|
||
status = self._login_status(account)
|
||
if not status.get("logged_in"):
|
||
reason = self._login_skip_reason(status)
|
||
self._log_run_event(
|
||
f"step=login_check result=blocked detail=账号 {account.alias} {reason}",
|
||
level="warning",
|
||
)
|
||
logged_out.append(
|
||
self._account_payload(account, reason)
|
||
)
|
||
else:
|
||
self._log_run_event(
|
||
f"step=login_check result=success detail=账号 {account.alias}",
|
||
level="info",
|
||
)
|
||
if not_running or logged_out:
|
||
return {
|
||
"reason": "ACCOUNT_NOT_READY",
|
||
"not_running": not_running,
|
||
"logged_out": logged_out,
|
||
}
|
||
return None
|
||
|
||
def _duplicate_debug_ports(self, account_rows, eligible, account_by_alias):
|
||
required_aliases = {
|
||
str(task.alias).strip()
|
||
for task in eligible
|
||
if account_by_alias.get(str(task.alias).strip()) is not None
|
||
}
|
||
by_port = {}
|
||
for account in account_rows:
|
||
if account.alias not in required_aliases:
|
||
continue
|
||
by_port.setdefault(int(account.debug_port), []).append(account)
|
||
duplicates = []
|
||
for port, rows in by_port.items():
|
||
if len(rows) > 1:
|
||
duplicates.append(
|
||
{
|
||
"debug_port": port,
|
||
"aliases": [row.alias for row in rows],
|
||
}
|
||
)
|
||
return duplicates
|
||
|
||
def _effective_batch_size(self, total):
|
||
if self.batch_size is None:
|
||
return max(1, int(total or 1))
|
||
return self.batch_size
|
||
|
||
def _task_batches(self, tasks, batch_size):
|
||
if not tasks:
|
||
return []
|
||
return [
|
||
tasks[index:index + batch_size]
|
||
for index in range(0, len(tasks), batch_size)
|
||
]
|
||
|
||
def _log_batch_start(self, batch_index, batch_count, batch_tasks, counters, total):
|
||
first = counters["done"] + 1
|
||
last = min(first + len(batch_tasks) - 1, total)
|
||
label = "检查批次" if self.dry_run else "更新批次"
|
||
self._log_run_event(
|
||
f"step=batch result=start detail={label} {batch_index}/{batch_count} 开始：任务 {first}-{last}/{total}"
|
||
)
|
||
|
||
def _run_parallel_by_account(self, eligible, account_by_alias, counters, total):
|
||
groups = self._group_tasks_by_alias(eligible)
|
||
max_workers = min(self.max_parallel_accounts, len(groups))
|
||
if max_workers <= 1:
|
||
for group_tasks in groups:
|
||
self._run_task_group(group_tasks, account_by_alias, counters, total)
|
||
return
|
||
with ThreadPoolExecutor(max_workers=max_workers) as executor:
|
||
futures = [
|
||
executor.submit(
|
||
self._run_task_group,
|
||
group_tasks,
|
||
account_by_alias,
|
||
counters,
|
||
total,
|
||
)
|
||
for group_tasks in groups
|
||
]
|
||
for future in as_completed(futures):
|
||
future.result()
|
||
|
||
def _group_tasks_by_alias(self, tasks):
|
||
groups = []
|
||
index_by_alias = {}
|
||
for task in tasks:
|
||
alias = str(task.alias).strip()
|
||
if alias not in index_by_alias:
|
||
index_by_alias[alias] = len(groups)
|
||
groups.append([])
|
||
groups[index_by_alias[alias]].append(task)
|
||
return groups
|
||
|
||
def _run_task_group(self, tasks, account_by_alias, counters, total):
|
||
for task in tasks:
|
||
if self.should_cancel():
|
||
break
|
||
outcome = self._apply_one_task(task, account_by_alias)
|
||
self._record_outcome(counters, total, outcome)
|
||
|
||
def _preview_task(self, task, account_by_alias):
|
||
account = account_by_alias.get(str(task.alias).strip())
|
||
if account is None:
|
||
reason = "别名未匹配账号"
|
||
self._log_run_event(
|
||
f"step=preview result=skipped detail=检查：任务 {task.id} 商品 {task.item_id} 将略过：{reason}",
|
||
task=task,
|
||
level="warning",
|
||
)
|
||
return "skipped"
|
||
action_parts = []
|
||
if appconfig.update_mode_includes_title(self.update_mode) and getattr(task, "new_title", None):
|
||
action_parts.append("标题")
|
||
if appconfig.update_mode_includes_cover(self.update_mode) and getattr(task, "new_cover_path", None):
|
||
action_parts.append("封面")
|
||
action_text = "+".join(action_parts) or "无变更"
|
||
self._log_run_event(
|
||
"step=preview result=success detail=检查：任务 {task_id} 商品 {item_id} 账号 {alias} 将更新 {action}".format(
|
||
task_id=task.id,
|
||
item_id=task.item_id,
|
||
alias=account.alias,
|
||
action=action_text,
|
||
),
|
||
task=task,
|
||
)
|
||
return "applied"
|
||
|
||
def _apply_one_task(self, task, account_by_alias):
|
||
account = account_by_alias.get(str(task.alias).strip())
|
||
if account is None:
|
||
reason = "别名未匹配账号"
|
||
db.mark_skipped(task.id, reason, path=self.db_path)
|
||
self.row_updated.emit(task.id, {"status": "skipped", "last_error": reason})
|
||
self._log_run_event(
|
||
f"step=preflight result=skipped detail=任务 {task.id} 商品 {task.item_id} 已略过：{reason}",
|
||
task=task,
|
||
level="warning",
|
||
)
|
||
return "skipped"
|
||
|
||
started = time.monotonic()
|
||
current_step = "db_write"
|
||
|
||
def on_step(event):
|
||
nonlocal current_step
|
||
if isinstance(event, dict):
|
||
step = str(event.get("step") or "apply_task")
|
||
result = str(event.get("result") or "start")
|
||
detail = event.get("detail")
|
||
else:
|
||
step = str(event)
|
||
result = "start"
|
||
detail = None
|
||
current_step = step
|
||
level = "error" if result == "failed" else "info"
|
||
detail_text = "任务 {task_id} 商品 {item_id}".format(
|
||
task_id=task.id,
|
||
item_id=task.item_id,
|
||
)
|
||
if detail:
|
||
detail_text = f"{detail_text} {detail}"
|
||
self._log_run_event(
|
||
f"step={step} result={result} detail={detail_text}",
|
||
task=task,
|
||
level=level,
|
||
)
|
||
|
||
try:
|
||
self._log_run_event(
|
||
f"step=apply_task result=start detail=任务 {task.id} 商品 {task.item_id} 开始更新，账号 {account.alias}",
|
||
task=task,
|
||
)
|
||
current_step = "db_write"
|
||
self._log_run_event(
|
||
f"step=db_write result=start detail=任务 {task.id} 商品 {task.item_id} 标记更新运行",
|
||
task=task,
|
||
)
|
||
db.mark_running(task.id, "apply", path=self.db_path)
|
||
self.row_updated.emit(task.id, {"status": "running", "last_error": None})
|
||
current_step = "apply_task"
|
||
result = editor.apply_task(
|
||
account,
|
||
task,
|
||
on_step=on_step,
|
||
bring_to_front=True,
|
||
update_mode=self.update_mode,
|
||
)
|
||
committed = bool(result.get("committed")) and not result.get("error")
|
||
error = result.get("error")
|
||
failed_step = self._failed_apply_step(result, current_step)
|
||
current_step = "db_write"
|
||
self._log_run_event(
|
||
f"step=db_write result=start detail=任务 {task.id} 商品 {task.item_id} 保存更新结果",
|
||
task=task,
|
||
)
|
||
if committed:
|
||
db.set_applied(task.id, True, path=self.db_path)
|
||
elapsed_ms = self._elapsed_ms(started)
|
||
self.row_updated.emit(
|
||
task.id,
|
||
{
|
||
"stage": "applied",
|
||
"status": "success",
|
||
"committed": 1,
|
||
"last_error": None,
|
||
},
|
||
)
|
||
self._log_run_event(
|
||
f"step=db_write result=success detail=任务 {task.id} 商品 {task.item_id} 更新成功 elapsed_ms={elapsed_ms}",
|
||
task=task,
|
||
)
|
||
return "applied"
|
||
|
||
error = diagnostics.redact_log_text(error or "更新未提交")
|
||
display_error = db.format_failure_error(error, failed_step)
|
||
db.set_applied(task.id, False, error, path=self.db_path, step=failed_step)
|
||
elapsed_ms = self._elapsed_ms(started)
|
||
self.failed.emit(task.id, str(display_error))
|
||
self.row_updated.emit(
|
||
task.id,
|
||
{"status": "failed", "last_error": str(display_error), "committed": 0},
|
||
)
|
||
self._log_run_event(
|
||
f"step={failed_step} result=failed detail={error} elapsed_ms={elapsed_ms}",
|
||
task=task,
|
||
level="error",
|
||
)
|
||
self._log_run_event(
|
||
f"step=db_write result=success detail=任务 {task.id} 商品 {task.item_id} 保存失败状态 elapsed_ms={elapsed_ms}",
|
||
task=task,
|
||
)
|
||
self._write_diagnostic_log(
|
||
"蝦皮更新任务失败",
|
||
level="ERROR",
|
||
step=failed_step,
|
||
task=task,
|
||
elapsed_ms=elapsed_ms,
|
||
payload={"error": error, "result": result},
|
||
)
|
||
return "failed"
|
||
except Exception as exc:
|
||
error = diagnostics.redact_log_text(str(exc) or exc.__class__.__name__)
|
||
display_error = db.format_failure_error(error, current_step)
|
||
db.set_applied(task.id, False, error, path=self.db_path, step=current_step)
|
||
elapsed_ms = self._elapsed_ms(started)
|
||
self.failed.emit(task.id, display_error)
|
||
self.row_updated.emit(
|
||
task.id,
|
||
{"status": "failed", "last_error": display_error, "committed": 0},
|
||
)
|
||
self._log_run_event(
|
||
f"step={current_step} result=failed detail={error} elapsed_ms={elapsed_ms}",
|
||
task=task,
|
||
level="error",
|
||
)
|
||
self._write_diagnostic_log(
|
||
"蝦皮更新任务异常",
|
||
level="ERROR",
|
||
step=current_step,
|
||
task=task,
|
||
elapsed_ms=elapsed_ms,
|
||
payload={"error": error},
|
||
exc=exc,
|
||
)
|
||
return "failed"
|
||
|
||
def _record_outcome(self, counters, total, outcome):
|
||
with self._progress_lock:
|
||
counters["done"] += 1
|
||
if outcome == "applied":
|
||
counters["applied"] += 1
|
||
elif outcome == "skipped":
|
||
counters["skipped"] += 1
|
||
else:
|
||
counters["failed"] += 1
|
||
self._emit_progress(
|
||
counters["done"],
|
||
total,
|
||
counters["applied"],
|
||
counters["skipped"],
|
||
counters["failed"],
|
||
)
|
||
|
||
def _account_payload(self, account, reason=None):
|
||
payload = {
|
||
"account_name": account.account_name,
|
||
"alias": account.alias,
|
||
"debug_port": account.debug_port,
|
||
}
|
||
if reason:
|
||
payload["reason"] = reason
|
||
return payload
|
||
|
||
def _emit_progress(self, done, total, applied, skipped, failed):
|
||
self.progress.emit(
|
||
{
|
||
"done": done,
|
||
"total": total,
|
||
"applied": applied,
|
||
"skipped": skipped,
|
||
"failed": failed,
|
||
"dry_run": self.dry_run,
|
||
"batch_size": self._current_batch_size,
|
||
"batch_count": self._batch_count,
|
||
"update_mode": self.update_mode,
|
||
}
|
||
)
|
||
|
||
def _login_status(self, account):
|
||
try:
|
||
return accounts.detect_login(account, path=self.db_path, config=self.config)
|
||
except Exception as exc:
|
||
return {
|
||
"logged_in": False,
|
||
"reason": f"LOGIN_CHECK_FAILED: {exc}",
|
||
}
|
||
|
||
def _login_skip_reason(self, status):
|
||
reason = status.get("reason")
|
||
return f"账号未登录: {reason}" if reason else "账号未登录"
|
||
|
||
def _batch_ids(self, tasks):
|
||
batch_ids = []
|
||
for task in tasks:
|
||
batch_id = getattr(task, "batch_id", None)
|
||
if batch_id and batch_id not in batch_ids:
|
||
batch_ids.append(batch_id)
|
||
return batch_ids
|
||
|
||
def _summary(self, ok, total, counters, batch_ids, blocked=False, extra=None):
|
||
summary = {
|
||
"ok": ok,
|
||
"total": total,
|
||
"done": counters["done"],
|
||
"applied": counters["applied"],
|
||
"skipped": counters["skipped"],
|
||
"failed": counters["failed"],
|
||
"batch_ids": batch_ids,
|
||
"dry_run": self.dry_run,
|
||
"account_parallel": self.max_parallel_accounts > 1,
|
||
"batch_size": self._current_batch_size,
|
||
"batch_count": self._batch_count,
|
||
"update_mode": self.update_mode,
|
||
"run_id": self._run_id,
|
||
}
|
||
if blocked:
|
||
summary["blocked"] = True
|
||
if extra:
|
||
summary.update(extra)
|
||
return summary
|
||
|
||
def _create_run_log(self, eligible, batch_ids):
|
||
try:
|
||
return db.create_run_log(
|
||
"apply",
|
||
dry_run=self.dry_run,
|
||
total=len(eligible),
|
||
options={
|
||
"batch_ids": batch_ids,
|
||
"dry_run": self.dry_run,
|
||
"account_parallel": self.max_parallel_accounts > 1,
|
||
"max_parallel_accounts": self.max_parallel_accounts,
|
||
"batch_size": self._current_batch_size,
|
||
"batch_count": self._batch_count,
|
||
"update_mode": self.update_mode,
|
||
},
|
||
path=self.db_path,
|
||
)
|
||
except Exception:
|
||
return None
|
||
|
||
def _finish_run_log(self, status, summary):
|
||
if self._run_id is None:
|
||
return
|
||
try:
|
||
db.finish_run_log(
|
||
self._run_id,
|
||
status=status,
|
||
done=summary.get("done", 0),
|
||
success_count=summary.get("applied", 0),
|
||
skipped_count=summary.get("skipped", 0),
|
||
failed_count=summary.get("failed", 0),
|
||
summary_json=summary,
|
||
path=self.db_path,
|
||
)
|
||
except Exception:
|
||
return
|
||
|
||
def _log_run_event(self, message, task=None, level="info"):
|
||
safe_message = diagnostics.redact_log_text(message)
|
||
self.log.emit(str(safe_message))
|
||
if self._run_id is None:
|
||
return
|
||
try:
|
||
db.add_run_log_event(
|
||
self._run_id,
|
||
safe_message,
|
||
task_id=getattr(task, "id", None),
|
||
alias=getattr(task, "alias", None),
|
||
item_id=getattr(task, "item_id", None),
|
||
level=level,
|
||
path=self.db_path,
|
||
)
|
||
except Exception:
|
||
return
|
||
|
||
|
||
def _log_preflight_blocked(self, blocked):
|
||
if blocked.get("no_accounts"):
|
||
self._log_run_event(
|
||
"step=preflight result=blocked detail=当前没有配置账号",
|
||
level="warning",
|
||
)
|
||
for item in blocked.get("duplicate_ports") or []:
|
||
self._log_run_event(
|
||
"step=preflight result=blocked detail=调试端口重复 debug_port={port} aliases={aliases}".format(
|
||
port=item.get("debug_port") or "",
|
||
aliases=",".join(item.get("aliases") or []),
|
||
),
|
||
level="warning",
|
||
)
|
||
for item in blocked.get("not_running") or []:
|
||
self._log_run_event(
|
||
"step=check_chrome result=blocked detail=账号 {alias} Chrome 未启动或调试端口不可访问: {reason}".format(
|
||
alias=item.get("alias") or "",
|
||
reason=item.get("reason") or "",
|
||
),
|
||
level="warning",
|
||
)
|
||
for item in blocked.get("logged_out") or []:
|
||
self._log_run_event(
|
||
"step=login_check result=blocked detail=账号 {alias} 未登录蝦皮: {reason}".format(
|
||
alias=item.get("alias") or "",
|
||
reason=item.get("reason") or "",
|
||
),
|
||
level="warning",
|
||
)
|
||
|
||
def _failed_apply_step(self, result, fallback):
|
||
if not isinstance(result, dict):
|
||
return fallback or "apply_task"
|
||
title = result.get("title")
|
||
if isinstance(title, dict) and not title.get("ok", True):
|
||
return "change_title"
|
||
cover = result.get("cover")
|
||
if isinstance(cover, dict) and not cover.get("ok", True):
|
||
return "replace_cover"
|
||
update = result.get("update")
|
||
if isinstance(update, dict):
|
||
return "click_update"
|
||
return fallback or "apply_task"
|
||
|
||
def _write_diagnostic_log(
|
||
self,
|
||
message,
|
||
level="INFO",
|
||
step=None,
|
||
task=None,
|
||
elapsed_ms=None,
|
||
payload=None,
|
||
exc=None,
|
||
):
|
||
_safe_write_diagnostic_log(
|
||
message,
|
||
level=level,
|
||
step=step,
|
||
task=task,
|
||
elapsed_ms=elapsed_ms,
|
||
payload=payload,
|
||
exc=exc,
|
||
log_dir=self.diagnostic_log_dir,
|
||
)
|
||
|
||
def _elapsed_ms(self, started):
|
||
return _elapsed_ms(started)
|
||
|
||
class CollectWorker(BaseWorker):
|
||
"""Collect old title and cover for imported tasks."""
|
||
|
||
if Signal is not None:
|
||
activity = Signal(dict)
|
||
|
||
LOGIN_CHECK_ATTEMPTS = 3
|
||
LOGIN_CHECK_RETRY_DELAY_SECONDS = 2.0
|
||
|
||
def __init__(
|
||
self,
|
||
tasks,
|
||
db_path=None,
|
||
config=None,
|
||
preflight=True,
|
||
diagnostic_log_dir=None,
|
||
):
|
||
super().__init__()
|
||
self.tasks = list(tasks)
|
||
self.db_path = db_path
|
||
self.config = config
|
||
self.preflight = preflight
|
||
self.diagnostic_log_dir = diagnostic_log_dir
|
||
self._run_id = None
|
||
|
||
def execute(self):
|
||
account_rows = accounts.list_accounts(path=self.db_path, config=self.config)
|
||
account_by_alias = {
|
||
str(account.alias).strip(): account
|
||
for account in account_rows
|
||
if str(account.alias).strip()
|
||
}
|
||
eligible = [
|
||
task for task in self.tasks
|
||
if getattr(task, "stage", None) == "imported"
|
||
]
|
||
batch_ids = self._batch_ids(eligible)
|
||
total = len(eligible)
|
||
collected = 0
|
||
skipped = 0
|
||
failed = 0
|
||
done = 0
|
||
login_skip_reasons = {}
|
||
login_required_accounts = {}
|
||
preflight_info = {}
|
||
skip_reason_counts = empty_skip_reason_counts()
|
||
|
||
self._run_id = self._create_run_log(eligible, batch_ids)
|
||
self._emit_activity(
|
||
"preflight_started",
|
||
total=total,
|
||
step="preflight",
|
||
)
|
||
self._log_run_event(
|
||
f"step=preflight result=start detail=采集运行开始 total={total}"
|
||
)
|
||
|
||
if self.preflight:
|
||
blocked, preflight_info = self._preflight_prepare(eligible, account_rows, account_by_alias)
|
||
if blocked:
|
||
self._log_preflight_blocked(blocked)
|
||
summary = self._summary(
|
||
ok=False,
|
||
total=total,
|
||
done=done,
|
||
collected=collected,
|
||
skipped=skipped,
|
||
failed=failed,
|
||
batch_ids=batch_ids,
|
||
blocked=True,
|
||
extra={
|
||
**blocked,
|
||
"skip_reason_counts": dict(skip_reason_counts),
|
||
},
|
||
)
|
||
self._finish_run_log("blocked", summary)
|
||
return summary
|
||
for item in preflight_info.get("logged_out") or []:
|
||
alias = str(item.get("alias") or "").strip()
|
||
reason = item.get("reason") or "账号未登录"
|
||
if alias:
|
||
login_skip_reasons[alias] = reason
|
||
login_required_accounts[alias] = item
|
||
self._log_run_event("step=preflight result=success detail=账号就绪检查完成")
|
||
else:
|
||
self._log_run_event(
|
||
"step=preflight result=skipped detail=测试模式跳过采集前检查",
|
||
level="warning",
|
||
)
|
||
|
||
for index, task in enumerate(eligible, start=1):
|
||
if self.should_cancel():
|
||
break
|
||
self._emit_activity(
|
||
"task_started",
|
||
task=task,
|
||
index=index,
|
||
total=total,
|
||
step="match_account",
|
||
)
|
||
account = account_by_alias.get(str(task.alias).strip())
|
||
if account is None:
|
||
skipped += 1
|
||
skip_reason_counts[ALIAS_UNMATCHED] += 1
|
||
done += 1
|
||
reason = "别名未匹配账号"
|
||
db.mark_skipped(task.id, reason, path=self.db_path)
|
||
self.row_updated.emit(task.id, {"status": "skipped", "last_error": reason})
|
||
self._log_run_event(
|
||
"step=preflight result=skipped detail=任务 {task_id} 商品 {item_id} {reason}".format(
|
||
task_id=task.id,
|
||
item_id=task.item_id,
|
||
reason=reason,
|
||
),
|
||
task=task,
|
||
level="warning",
|
||
)
|
||
self._emit_activity(
|
||
"task_finished",
|
||
task=task,
|
||
index=index,
|
||
total=total,
|
||
step="match_account",
|
||
result="skipped",
|
||
)
|
||
self._emit_progress(done, total, collected, skipped, failed)
|
||
continue
|
||
|
||
alias = str(task.alias).strip()
|
||
if alias in login_skip_reasons:
|
||
skipped += 1
|
||
skip_reason_counts[LOGIN_REQUIRED] += 1
|
||
done += 1
|
||
reason = login_skip_reasons[alias]
|
||
db.mark_skipped(task.id, reason, path=self.db_path)
|
||
self.row_updated.emit(task.id, {"status": "skipped", "last_error": reason})
|
||
self._log_run_event(
|
||
"step=login_check result=skipped detail=任务 {task_id} 商品 {item_id} {reason}".format(
|
||
task_id=task.id,
|
||
item_id=task.item_id,
|
||
reason=reason,
|
||
),
|
||
task=task,
|
||
level="warning",
|
||
)
|
||
self._emit_activity(
|
||
"task_finished",
|
||
task=task,
|
||
index=index,
|
||
total=total,
|
||
step="check_login",
|
||
result="skipped",
|
||
)
|
||
self._emit_progress(done, total, collected, skipped, failed)
|
||
continue
|
||
|
||
self._emit_activity(
|
||
"task_step",
|
||
task=task,
|
||
index=index,
|
||
total=total,
|
||
step="check_login",
|
||
)
|
||
status = self._confirmed_login_status(account, context="midrun", task=task)
|
||
if self._is_definitive_logged_out(status):
|
||
alias = str(task.alias).strip()
|
||
skipped += 1
|
||
skip_reason_counts[LOGIN_REQUIRED] += 1
|
||
done += 1
|
||
reason = self._midrun_login_skip_reason(status)
|
||
login_skip_reasons[alias] = reason
|
||
login_required_accounts[alias] = self._account_payload(account, reason)
|
||
db.mark_skipped(task.id, reason, path=self.db_path)
|
||
self.row_updated.emit(task.id, {"status": "skipped", "last_error": reason})
|
||
self._log_run_event(
|
||
"step=login_check result=skipped detail=任务 {task_id} 商品 {item_id} {reason}".format(
|
||
task_id=task.id,
|
||
item_id=task.item_id,
|
||
reason=reason,
|
||
),
|
||
task=task,
|
||
level="warning",
|
||
)
|
||
self._emit_activity(
|
||
"task_finished",
|
||
task=task,
|
||
index=index,
|
||
total=total,
|
||
step="check_login",
|
||
result="skipped",
|
||
)
|
||
self._emit_progress(done, total, collected, skipped, failed)
|
||
continue
|
||
if not status.get("logged_in"):
|
||
self._log_run_event(
|
||
"step=login_check result=uncertain detail=任务 {task_id} 商品 {item_id} 登录状态检测暂时不稳定，继续尝试采集当前商品: {detail}".format(
|
||
task_id=task.id,
|
||
item_id=task.item_id,
|
||
detail=self._login_status_detail(status),
|
||
),
|
||
task=task,
|
||
level="warning",
|
||
)
|
||
|
||
started = time.monotonic()
|
||
current_step = "db_write"
|
||
activity_result = "success"
|
||
|
||
def on_step(step):
|
||
nonlocal current_step
|
||
current_step = str(step)
|
||
self._emit_activity(
|
||
"task_step",
|
||
task=task,
|
||
index=index,
|
||
total=total,
|
||
step=current_step,
|
||
)
|
||
self._log_run_event(
|
||
"step={step} result=start detail=任务 {task_id} 商品 {item_id}".format(
|
||
step=current_step,
|
||
task_id=task.id,
|
||
item_id=task.item_id,
|
||
),
|
||
task=task,
|
||
)
|
||
|
||
try:
|
||
self._emit_activity(
|
||
"task_step",
|
||
task=task,
|
||
index=index,
|
||
total=total,
|
||
step="prepare_task",
|
||
)
|
||
self._log_run_event(
|
||
"step=db_write result=start detail=任务 {task_id} 商品 {item_id} 标记采集运行".format(
|
||
task_id=task.id,
|
||
item_id=task.item_id,
|
||
),
|
||
task=task,
|
||
)
|
||
db.mark_running(task.id, "collect", path=self.db_path)
|
||
self.row_updated.emit(task.id, {"status": "running"})
|
||
result = editor.collect(
|
||
account,
|
||
{
|
||
"item_id": task.item_id,
|
||
"old_cover_path": self._old_cover_path(account, task),
|
||
},
|
||
on_step=on_step,
|
||
)
|
||
if result.get("close_target_confirmed") is False:
|
||
self._log_run_event(
|
||
"step=close_product result=uncertain detail=任务 {task_id} 商品 {item_id} 商品页已请求关闭，但未在短时间内确认关闭；采集结果已保留，继续处理后续任务".format(
|
||
task_id=task.id,
|
||
item_id=task.item_id,
|
||
),
|
||
task=task,
|
||
level="warning",
|
||
)
|
||
self._write_diagnostic_log(
|
||
"采集商品页关闭确认超时",
|
||
level="WARNING",
|
||
step="close_product",
|
||
task=task,
|
||
payload={
|
||
"alias": getattr(account, "alias", None),
|
||
"close_target_confirmed": False,
|
||
},
|
||
)
|
||
current_step = "db_write"
|
||
self._emit_activity(
|
||
"task_step",
|
||
task=task,
|
||
index=index,
|
||
total=total,
|
||
step="save_result",
|
||
)
|
||
self._log_run_event(
|
||
"step=db_write result=start detail=任务 {task_id} 商品 {item_id} 保存采集结果".format(
|
||
task_id=task.id,
|
||
item_id=task.item_id,
|
||
),
|
||
task=task,
|
||
)
|
||
db.set_collected(
|
||
task.id,
|
||
result.get("old_title", ""),
|
||
result.get("old_cover_path", ""),
|
||
path=self.db_path,
|
||
)
|
||
collected += 1
|
||
elapsed_ms = self._elapsed_ms(started)
|
||
self.row_updated.emit(
|
||
task.id,
|
||
{
|
||
"stage": "collected",
|
||
"status": "success",
|
||
"old_title": result.get("old_title", ""),
|
||
"old_cover_path": result.get("old_cover_path", ""),
|
||
},
|
||
)
|
||
self._log_run_event(
|
||
"step=db_write result=success detail=任务 {task_id} 商品 {item_id} 采集成功 elapsed_ms={elapsed_ms}".format(
|
||
task_id=task.id,
|
||
item_id=task.item_id,
|
||
elapsed_ms=elapsed_ms,
|
||
),
|
||
task=task,
|
||
)
|
||
except Exception as exc:
|
||
activity_result = "failed"
|
||
failed += 1
|
||
error = str(exc) or exc.__class__.__name__
|
||
safe_error = diagnostics.redact_log_text(error)
|
||
display_error = db.format_failure_error(safe_error, current_step)
|
||
elapsed_ms = self._elapsed_ms(started)
|
||
db.mark_failed(task.id, "collect", safe_error, path=self.db_path, step=current_step)
|
||
self.failed.emit(task.id, display_error)
|
||
self.row_updated.emit(task.id, {"status": "failed", "last_error": display_error})
|
||
self._log_run_event(
|
||
"step={step} result=failed detail={error} elapsed_ms={elapsed_ms}".format(
|
||
step=current_step,
|
||
error=safe_error,
|
||
elapsed_ms=elapsed_ms,
|
||
),
|
||
task=task,
|
||
level="error",
|
||
)
|
||
self._write_diagnostic_log(
|
||
"采集任务失败",
|
||
level="ERROR",
|
||
step=current_step,
|
||
task=task,
|
||
elapsed_ms=elapsed_ms,
|
||
payload={"error": safe_error},
|
||
exc=exc,
|
||
)
|
||
finally:
|
||
done += 1
|
||
self._emit_activity(
|
||
"task_finished",
|
||
task=task,
|
||
index=index,
|
||
total=total,
|
||
step=current_step,
|
||
result=activity_result,
|
||
)
|
||
self._emit_progress(done, total, collected, skipped, failed)
|
||
|
||
summary = self._summary(
|
||
ok=failed == 0,
|
||
total=total,
|
||
done=done,
|
||
collected=collected,
|
||
skipped=skipped,
|
||
failed=failed,
|
||
batch_ids=batch_ids,
|
||
extra={
|
||
**preflight_info,
|
||
"login_required_accounts": list(login_required_accounts.values()),
|
||
"skip_reason_counts": dict(skip_reason_counts),
|
||
},
|
||
)
|
||
self._finish_run_log("cancelled" if self.should_cancel() else "done", summary)
|
||
return summary
|
||
|
||
def _preflight_prepare(self, eligible, account_rows, account_by_alias):
|
||
if not account_rows:
|
||
return (
|
||
{
|
||
"reason": "NO_ACCOUNTS",
|
||
"no_accounts": True,
|
||
},
|
||
{},
|
||
)
|
||
required_accounts = []
|
||
seen_aliases = set()
|
||
for task in eligible:
|
||
alias = str(task.alias).strip()
|
||
account = account_by_alias.get(alias)
|
||
if account is not None and alias not in seen_aliases:
|
||
required_accounts.append(account)
|
||
seen_aliases.add(alias)
|
||
launch_failed = []
|
||
logged_out = []
|
||
launched = []
|
||
reused = []
|
||
for account in required_accounts:
|
||
self._log_run_event(
|
||
f"step=ensure_chrome result=start detail=账号 {account.alias} debug_port={account.debug_port}",
|
||
level="info",
|
||
)
|
||
if chrome.is_running(account.debug_port):
|
||
self._log_run_event(
|
||
f"step=ensure_chrome result=reused detail=账号 {account.alias} Chrome 已打开，复用现有窗口 debug_port={account.debug_port}",
|
||
level="info",
|
||
)
|
||
reused.append(self._account_payload(account, "已复用"))
|
||
else:
|
||
try:
|
||
result = accounts.launch_for_login(account, path=self.db_path, config=self.config)
|
||
except Exception as exc:
|
||
reason = diagnostics.redact_log_text(str(exc) or exc.__class__.__name__)
|
||
self._log_run_event(
|
||
f"step=ensure_chrome result=blocked detail=账号 {account.alias} Chrome 启动失败: {reason}",
|
||
level="error",
|
||
)
|
||
launch_failed.append(self._account_payload(account, f"Chrome 启动失败: {reason}"))
|
||
continue
|
||
action = "launched" if result.get("launched") else "reused"
|
||
detail = "已启动" if action == "launched" else "已复用"
|
||
self._log_run_event(
|
||
f"step=ensure_chrome result={action} detail=账号 {account.alias} {detail} debug_port={account.debug_port}",
|
||
level="info",
|
||
)
|
||
target = launched if result.get("launched") else reused
|
||
target.append(self._account_payload(account, detail))
|
||
self._log_run_event(
|
||
f"step=login_check result=start detail=账号 {account.alias}",
|
||
level="info",
|
||
)
|
||
status = self._confirmed_login_status(account, context="preflight")
|
||
if status.get("logged_in"):
|
||
self._log_run_event(
|
||
f"step=login_check result=success detail=账号 {account.alias}",
|
||
level="info",
|
||
)
|
||
elif self._is_definitive_logged_out(status):
|
||
reason = self._login_skip_reason(status)
|
||
self._log_run_event(
|
||
f"step=login_check result=blocked detail=账号 {account.alias} {reason}",
|
||
level="warning",
|
||
)
|
||
logged_out.append(
|
||
self._account_payload(account, reason)
|
||
)
|
||
else:
|
||
self._log_run_event(
|
||
"step=login_check result=uncertain detail=账号 {alias} 登录状态检测暂时不稳定，继续进入采集流程: {detail}".format(
|
||
alias=account.alias,
|
||
detail=self._login_status_detail(status),
|
||
),
|
||
level="warning",
|
||
)
|
||
info = {
|
||
"launched_accounts": launched,
|
||
"reused_accounts": reused,
|
||
"logged_out": logged_out,
|
||
}
|
||
if launch_failed:
|
||
return (
|
||
{
|
||
"reason": "CHROME_LAUNCH_FAILED",
|
||
"launch_failed": launch_failed,
|
||
},
|
||
info,
|
||
)
|
||
return None, info
|
||
|
||
def _account_payload(self, account, reason=None):
|
||
payload = {
|
||
"account_name": account.account_name,
|
||
"alias": account.alias,
|
||
"debug_port": account.debug_port,
|
||
}
|
||
if reason:
|
||
payload["reason"] = reason
|
||
return payload
|
||
|
||
def _emit_activity(
|
||
self,
|
||
state,
|
||
*,
|
||
task=None,
|
||
index=0,
|
||
total=0,
|
||
step=None,
|
||
result=None,
|
||
):
|
||
signal = getattr(self, "activity", None)
|
||
if signal is None:
|
||
return
|
||
payload = {
|
||
"state": str(state),
|
||
"index": int(index or 0),
|
||
"total": int(total or 0),
|
||
}
|
||
if task is not None:
|
||
payload.update(
|
||
{
|
||
"task_id": getattr(task, "id", None),
|
||
"item_id": str(getattr(task, "item_id", "") or ""),
|
||
"alias": str(getattr(task, "alias", "") or ""),
|
||
}
|
||
)
|
||
if step:
|
||
payload["step"] = str(step)
|
||
if result:
|
||
payload["result"] = str(result)
|
||
signal.emit(payload)
|
||
|
||
def _emit_progress(self, done, total, collected, skipped, failed):
|
||
self.progress.emit(
|
||
{
|
||
"done": done,
|
||
"total": total,
|
||
"collected": collected,
|
||
"skipped": skipped,
|
||
"failed": failed,
|
||
}
|
||
)
|
||
|
||
def _login_status(self, account):
|
||
try:
|
||
return accounts.detect_login(account, path=self.db_path, config=self.config)
|
||
except Exception as exc:
|
||
return {
|
||
"logged_in": False,
|
||
"reason": f"LOGIN_CHECK_FAILED: {exc}",
|
||
}
|
||
|
||
def _confirmed_login_status(self, account, context, task=None):
|
||
started = time.monotonic()
|
||
subject = self._login_check_subject(account, task)
|
||
last_status = {}
|
||
for attempt in range(1, self.LOGIN_CHECK_ATTEMPTS + 1):
|
||
status = dict(self._login_status(account) or {})
|
||
status["login_check_attempts"] = attempt
|
||
last_status = status
|
||
if status.get("logged_in"):
|
||
if attempt > 1:
|
||
self._log_run_event(
|
||
"step=login_check result=recovered detail={subject} 登录检测已恢复，第{attempt}/{total}次确认已登录 elapsed_ms={elapsed_ms}".format(
|
||
subject=subject,
|
||
attempt=attempt,
|
||
total=self.LOGIN_CHECK_ATTEMPTS,
|
||
elapsed_ms=self._elapsed_ms(started),
|
||
),
|
||
task=task,
|
||
)
|
||
return status
|
||
if self._is_definitive_logged_out(status):
|
||
return status
|
||
if attempt < self.LOGIN_CHECK_ATTEMPTS:
|
||
self._log_run_event(
|
||
"step=login_check result=retry detail={subject} 登录状态暂时无法读取，第{attempt}/{total}次检测后将在{delay:g}秒后重试：{detail}".format(
|
||
subject=subject,
|
||
attempt=attempt,
|
||
total=self.LOGIN_CHECK_ATTEMPTS,
|
||
delay=self.LOGIN_CHECK_RETRY_DELAY_SECONDS,
|
||
detail=self._login_status_detail(status),
|
||
),
|
||
task=task,
|
||
level="warning",
|
||
)
|
||
time.sleep(self.LOGIN_CHECK_RETRY_DELAY_SECONDS)
|
||
last_status["login_check_uncertain"] = True
|
||
self._write_diagnostic_log(
|
||
"采集登录检测暂不确定",
|
||
level="WARNING",
|
||
step="login_check",
|
||
task=task,
|
||
payload={
|
||
"context": context,
|
||
"alias": getattr(account, "alias", None),
|
||
"reason": last_status.get("reason"),
|
||
"url": last_status.get("url"),
|
||
"cookie_names": list(last_status.get("cookie_names") or []),
|
||
"cookie_read_succeeded": bool(
|
||
last_status.get("cookie_read_succeeded")
|
||
),
|
||
"probe_error": last_status.get("probe_error"),
|
||
"probe_attempts": last_status.get("probe_attempts"),
|
||
"attempts": last_status.get("login_check_attempts"),
|
||
},
|
||
)
|
||
return last_status
|
||
|
||
def _login_check_subject(self, account, task=None):
|
||
alias = str(getattr(account, "alias", "") or "未知账号")
|
||
if task is None:
|
||
return f"账号 {alias}"
|
||
return f"任务 {task.id} 商品 {task.item_id} 账号 {alias}"
|
||
|
||
def _is_definitive_logged_out(self, status):
|
||
reason = str((status or {}).get("reason") or "").strip()
|
||
url = str((status or {}).get("url") or "").lower()
|
||
return reason.startswith("LOGIN_PAGE") or (
|
||
"accounts.shopee." in url and "/seller/login" in url
|
||
)
|
||
|
||
def _login_status_detail(self, status):
|
||
status = status or {}
|
||
raw_reason = str(status.get("reason") or "").strip()
|
||
if raw_reason == "LOGIN_CHECK_TARGET_UNAVAILABLE":
|
||
reason = "CDP页面暂时不可用"
|
||
elif raw_reason.startswith("LOGIN_CHECK_FAILED"):
|
||
reason = "登录检测调用失败"
|
||
elif raw_reason == "NO_SESSION_COOKIE":
|
||
reason = "暂未读取到登录会话"
|
||
elif raw_reason == "LOGIN_PAGE":
|
||
reason = "检测到登录页面"
|
||
else:
|
||
reason = raw_reason or "未知原因"
|
||
url = status.get("url") or "未知URL"
|
||
cookie_names = [str(name) for name in (status.get("cookie_names") or []) if name]
|
||
cookie_text = ",".join(sorted(cookie_names)) if cookie_names else "未读到登录Cookie"
|
||
return f"原因={reason}，URL={url}，Cookie名称={cookie_text}"
|
||
|
||
def _login_skip_reason(self, status):
|
||
reason = status.get("reason")
|
||
return f"账号未登录: {reason}" if reason else "账号未登录"
|
||
|
||
def _midrun_login_skip_reason(self, status):
|
||
reason = self._login_skip_reason(status)
|
||
return f"采集中途掉登录: {reason}"
|
||
|
||
def _old_cover_path(self, account, task):
|
||
image_root = appconfig.image_dir(self.config)
|
||
return image_paths.task_image_path(image_root, task, account, "old")
|
||
|
||
def _batch_ids(self, tasks):
|
||
batch_ids = []
|
||
for task in tasks:
|
||
batch_id = getattr(task, "batch_id", None)
|
||
if batch_id and batch_id not in batch_ids:
|
||
batch_ids.append(batch_id)
|
||
return batch_ids
|
||
|
||
def _summary(
|
||
self,
|
||
ok,
|
||
total,
|
||
done,
|
||
collected,
|
||
skipped,
|
||
failed,
|
||
batch_ids,
|
||
blocked=False,
|
||
extra=None,
|
||
):
|
||
summary = {
|
||
"ok": ok,
|
||
"total": total,
|
||
"done": done,
|
||
"collected": collected,
|
||
"skipped": skipped,
|
||
"failed": failed,
|
||
"batch_ids": batch_ids,
|
||
"run_id": self._run_id,
|
||
}
|
||
if blocked:
|
||
summary["blocked"] = True
|
||
if extra:
|
||
summary.update(extra)
|
||
return summary
|
||
|
||
def _create_run_log(self, eligible, batch_ids):
|
||
try:
|
||
return db.create_run_log(
|
||
"collect",
|
||
dry_run=False,
|
||
total=len(eligible),
|
||
options={
|
||
"batch_ids": batch_ids,
|
||
"preflight": self.preflight,
|
||
},
|
||
path=self.db_path,
|
||
)
|
||
except Exception:
|
||
return None
|
||
|
||
def _finish_run_log(self, status, summary):
|
||
if self._run_id is None:
|
||
return
|
||
try:
|
||
db.finish_run_log(
|
||
self._run_id,
|
||
status=status,
|
||
done=summary.get("done", 0),
|
||
success_count=summary.get("collected", 0),
|
||
skipped_count=summary.get("skipped", 0),
|
||
failed_count=summary.get("failed", 0),
|
||
summary_json=summary,
|
||
path=self.db_path,
|
||
)
|
||
except Exception:
|
||
return
|
||
|
||
def _log_run_event(self, message, task=None, level="info"):
|
||
safe_message = diagnostics.redact_log_text(message)
|
||
self.log.emit(str(safe_message))
|
||
if self._run_id is None:
|
||
return
|
||
try:
|
||
db.add_run_log_event(
|
||
self._run_id,
|
||
safe_message,
|
||
task_id=getattr(task, "id", None),
|
||
alias=getattr(task, "alias", None),
|
||
item_id=getattr(task, "item_id", None),
|
||
level=level,
|
||
path=self.db_path,
|
||
)
|
||
except Exception:
|
||
return
|
||
|
||
def _log_preflight_blocked(self, blocked):
|
||
if blocked.get("no_accounts"):
|
||
self._log_run_event(
|
||
"step=preflight result=blocked detail=当前没有配置账号",
|
||
level="warning",
|
||
)
|
||
for item in blocked.get("not_running") or []:
|
||
self._log_run_event(
|
||
"step=preflight result=blocked detail=账号 {alias} Chrome 未启动或调试端口不可访问: {reason}".format(
|
||
alias=item.get("alias") or "",
|
||
reason=item.get("reason") or "",
|
||
),
|
||
level="warning",
|
||
)
|
||
for item in blocked.get("logged_out") or []:
|
||
self._log_run_event(
|
||
"step=preflight result=blocked detail=账号 {alias} 未登录蝦皮: {reason}".format(
|
||
alias=item.get("alias") or "",
|
||
reason=item.get("reason") or "",
|
||
),
|
||
level="warning",
|
||
)
|
||
|
||
def _write_diagnostic_log(
|
||
self,
|
||
message,
|
||
level="INFO",
|
||
step=None,
|
||
task=None,
|
||
elapsed_ms=None,
|
||
payload=None,
|
||
exc=None,
|
||
):
|
||
try:
|
||
diagnostics.write_diagnostic_log(
|
||
message,
|
||
level=level,
|
||
step=step,
|
||
task_id=getattr(task, "id", None),
|
||
alias=getattr(task, "alias", None),
|
||
item_id=getattr(task, "item_id", None),
|
||
elapsed_ms=elapsed_ms,
|
||
payload=payload,
|
||
exc=exc,
|
||
log_dir=self.diagnostic_log_dir,
|
||
)
|
||
except Exception:
|
||
return
|
||
|
||
def _elapsed_ms(self, started):
|
||
return int((time.monotonic() - started) * 1000)
|
||
|
||
class WriteBackWorker(BaseWorker):
|
||
"""Write Excel fields back in a background thread."""
|
||
|
||
def __init__(self, batch_id, db_path=None, excel_path=None, mode="old", diagnostic_log_dir=None):
|
||
super().__init__()
|
||
self.batch_id = batch_id
|
||
self.db_path = db_path
|
||
self.excel_path = excel_path
|
||
self.mode = mode
|
||
self.diagnostic_log_dir = diagnostic_log_dir
|
||
self._run_id = None
|
||
|
||
def execute(self):
|
||
batch_ids = self._batch_ids()
|
||
self._run_id = _safe_create_run_log(
|
||
"write_back",
|
||
db_path=self.db_path,
|
||
total=len(batch_ids),
|
||
options={
|
||
"batch_ids": batch_ids,
|
||
"mode": self.mode,
|
||
"excel_path": self.excel_path,
|
||
},
|
||
)
|
||
self._log_run_event(
|
||
f"step=start result=start detail=Excel 回写开始 mode={self.mode} batch_count={len(batch_ids)}"
|
||
)
|
||
results = []
|
||
try:
|
||
for batch_id in batch_ids:
|
||
started = time.monotonic()
|
||
self._log_run_event(
|
||
f"step=write_excel result=start detail=batch_id={batch_id} mode={self.mode}"
|
||
)
|
||
result = self._write_one(batch_id)
|
||
results.append(result)
|
||
self._log_run_event(
|
||
"step=write_excel result=success detail=batch_id={batch_id} files={files} rows={rows} elapsed_ms={elapsed_ms}".format(
|
||
batch_id=batch_id,
|
||
files=result.get("files", 0),
|
||
rows=result.get("rows", 0),
|
||
elapsed_ms=self._elapsed_ms(started),
|
||
)
|
||
)
|
||
except Exception as exc:
|
||
error = diagnostics.redact_log_text(str(exc) or exc.__class__.__name__)
|
||
self._log_run_event(
|
||
f"step=write_excel result=failed detail={error}",
|
||
level="error",
|
||
)
|
||
self._write_diagnostic_log(
|
||
"Excel回写失败",
|
||
level="ERROR",
|
||
step="write_excel",
|
||
payload={"batch_ids": batch_ids, "mode": self.mode, "error": error},
|
||
exc=exc,
|
||
)
|
||
_safe_finish_run_log(
|
||
self._run_id,
|
||
db_path=self.db_path,
|
||
status="failed",
|
||
done=len(results),
|
||
success_count=sum(result.get("rows", 0) for result in results),
|
||
failed_count=1,
|
||
summary_json={"ok": False, "error": error, "mode": self.mode},
|
||
)
|
||
raise
|
||
result = results[0] if len(results) == 1 else self._combined_result(results)
|
||
self.progress.emit(
|
||
{
|
||
"done": result.get("rows", 0),
|
||
"total": result.get("rows", 0),
|
||
"files": result.get("files", 0),
|
||
}
|
||
)
|
||
_safe_finish_run_log(
|
||
self._run_id,
|
||
db_path=self.db_path,
|
||
status="done",
|
||
done=len(batch_ids),
|
||
success_count=result.get("rows", 0),
|
||
failed_count=0,
|
||
summary_json={"ok": result.get("ok", False), "mode": self.mode, "result": result},
|
||
)
|
||
return result
|
||
|
||
def _batch_ids(self):
|
||
if isinstance(self.batch_id, (list, tuple, set)):
|
||
return list(self.batch_id)
|
||
return [self.batch_id]
|
||
|
||
def _write_one(self, batch_id):
|
||
if self.mode == "results":
|
||
return excel.write_back_results(
|
||
batch_id,
|
||
excel_path=self.excel_path,
|
||
path=self.db_path,
|
||
)
|
||
return excel.write_back(
|
||
batch_id,
|
||
excel_path=self.excel_path,
|
||
path=self.db_path,
|
||
)
|
||
|
||
def _combined_result(self, results):
|
||
written_files = []
|
||
for result in results:
|
||
for file_path in result.get("written_files", []):
|
||
if file_path not in written_files:
|
||
written_files.append(file_path)
|
||
return {
|
||
"ok": all(result.get("ok", False) for result in results),
|
||
"batch_id": [result.get("batch_id") for result in results],
|
||
"files": sum(result.get("files", 0) for result in results),
|
||
"rows": sum(result.get("rows", 0) for result in results),
|
||
"written_files": written_files,
|
||
}
|
||
|
||
def _log_run_event(self, message, level="info"):
|
||
safe_message = _safe_add_run_log_event(
|
||
self._run_id,
|
||
message,
|
||
db_path=self.db_path,
|
||
level=level,
|
||
)
|
||
self.log.emit(str(safe_message))
|
||
|
||
def _write_diagnostic_log(self, message, level="INFO", step=None, payload=None, exc=None):
|
||
_safe_write_diagnostic_log(
|
||
message,
|
||
level=level,
|
||
step=step,
|
||
payload=payload,
|
||
exc=exc,
|
||
log_dir=self.diagnostic_log_dir,
|
||
)
|
||
|
||
def _elapsed_ms(self, started):
|
||
return _elapsed_ms(started)
|
||
|
||
class AccountLoginCheckWorker(BaseWorker):
|
||
def __init__(self, account, db_path=None, config=None, timeout=8, diagnostic_log_dir=None):
|
||
super().__init__()
|
||
self.account = account
|
||
self.db_path = db_path
|
||
self.config = config
|
||
self.timeout = timeout
|
||
self.diagnostic_log_dir = diagnostic_log_dir
|
||
self._run_id = None
|
||
|
||
def execute(self):
|
||
self._run_id = _safe_create_run_log(
|
||
"login_check",
|
||
db_path=self.db_path,
|
||
total=1,
|
||
options={
|
||
"alias": self.account.alias,
|
||
"debug_port": self.account.debug_port,
|
||
"timeout": self.timeout,
|
||
},
|
||
)
|
||
started = time.monotonic()
|
||
self._log_run_event(
|
||
f"step=detect_login result=start detail=账号 {self.account.alias} debug_port={self.account.debug_port}"
|
||
)
|
||
try:
|
||
status = accounts.detect_login(
|
||
self.account,
|
||
timeout=self.timeout,
|
||
path=self.db_path,
|
||
config=self.config,
|
||
)
|
||
except Exception as exc:
|
||
error = diagnostics.redact_log_text(str(exc) or exc.__class__.__name__)
|
||
elapsed_ms = self._elapsed_ms(started)
|
||
self._log_run_event(
|
||
f"step=detect_login result=failed detail={error} elapsed_ms={elapsed_ms}",
|
||
level="error",
|
||
)
|
||
self._write_diagnostic_log(
|
||
"登录检测失败",
|
||
level="ERROR",
|
||
step="detect_login",
|
||
elapsed_ms=elapsed_ms,
|
||
payload={"alias": self.account.alias, "error": error},
|
||
exc=exc,
|
||
)
|
||
_safe_finish_run_log(
|
||
self._run_id,
|
||
db_path=self.db_path,
|
||
status="failed",
|
||
done=0,
|
||
failed_count=1,
|
||
summary_json={"ok": False, "alias": self.account.alias, "error": error},
|
||
)
|
||
raise
|
||
elapsed_ms = self._elapsed_ms(started)
|
||
logged_in = bool(status.get("logged_in"))
|
||
result_text = "success" if logged_in else "failed"
|
||
level = "info" if logged_in else "warning"
|
||
self._log_run_event(
|
||
"step=detect_login result={result} detail=账号 {alias} logged_in={logged_in} reason={reason} elapsed_ms={elapsed_ms}".format(
|
||
result=result_text,
|
||
alias=self.account.alias,
|
||
logged_in=logged_in,
|
||
reason=status.get("reason") or "",
|
||
elapsed_ms=elapsed_ms,
|
||
),
|
||
level=level,
|
||
)
|
||
_safe_finish_run_log(
|
||
self._run_id,
|
||
db_path=self.db_path,
|
||
status="done",
|
||
done=1,
|
||
success_count=1 if logged_in else 0,
|
||
failed_count=0 if logged_in else 1,
|
||
summary_json={"ok": logged_in, "alias": self.account.alias, "status": status},
|
||
)
|
||
self.row_updated.emit(self.account.id, status)
|
||
return {"alias": self.account.alias, "status": status}
|
||
|
||
def _log_run_event(self, message, level="info"):
|
||
safe_message = _safe_add_run_log_event(
|
||
self._run_id,
|
||
message,
|
||
db_path=self.db_path,
|
||
account=self.account,
|
||
level=level,
|
||
)
|
||
self.log.emit(str(safe_message))
|
||
|
||
def _write_diagnostic_log(
|
||
self,
|
||
message,
|
||
level="INFO",
|
||
step=None,
|
||
elapsed_ms=None,
|
||
payload=None,
|
||
exc=None,
|
||
):
|
||
_safe_write_diagnostic_log(
|
||
message,
|
||
level=level,
|
||
step=step,
|
||
account=self.account,
|
||
elapsed_ms=elapsed_ms,
|
||
payload=payload,
|
||
exc=exc,
|
||
log_dir=self.diagnostic_log_dir,
|
||
)
|
||
|
||
def _elapsed_ms(self, started):
|
||
return _elapsed_ms(started)
|
||
|
||
class CMHubSettingsWorker(BaseWorker):
|
||
"""Fetch cmhub aliases and optional balance without blocking the GUI."""
|
||
|
||
def __init__(
|
||
self,
|
||
base_url,
|
||
api_key,
|
||
connect_timeout=10,
|
||
include_balance=True,
|
||
db_path=None,
|
||
diagnostic_log_dir=None,
|
||
):
|
||
super().__init__()
|
||
self.base_url = str(base_url or "").strip()
|
||
self.api_key = str(api_key or "")
|
||
self.connect_timeout = max(1, int(connect_timeout or 10))
|
||
self.include_balance = bool(include_balance)
|
||
self.db_path = db_path
|
||
self.diagnostic_log_dir = diagnostic_log_dir
|
||
self._run_id = None
|
||
|
||
def execute(self):
|
||
self._run_id = self._create_run_log()
|
||
started = time.monotonic()
|
||
action = "测试连接/查余额" if self.include_balance else "刷新别名"
|
||
self._log_run_event(f"step=cmhub_settings result=start detail={action}")
|
||
try:
|
||
models = ai.fetch_cmhub_models(
|
||
self.base_url,
|
||
self.api_key,
|
||
connect_timeout=self.connect_timeout,
|
||
)
|
||
balance = None
|
||
if self.include_balance:
|
||
balance = ai.fetch_cmhub_balance(
|
||
self.base_url,
|
||
self.api_key,
|
||
connect_timeout=self.connect_timeout,
|
||
)
|
||
except Exception as exc:
|
||
error = self._safe_error(exc)
|
||
elapsed_ms = self._elapsed_ms(started)
|
||
self._log_run_event(
|
||
f"step=cmhub_settings result=failed detail={error} elapsed_ms={elapsed_ms}",
|
||
level="error",
|
||
)
|
||
self._write_diagnostic_log(
|
||
"cmhub 设置检测失败",
|
||
level="ERROR",
|
||
step="cmhub_settings",
|
||
elapsed_ms=elapsed_ms,
|
||
payload={"base_url": self.base_url, "error": error},
|
||
exc=exc,
|
||
)
|
||
_safe_finish_run_log(
|
||
self._run_id,
|
||
db_path=self.db_path,
|
||
status="failed",
|
||
done=0,
|
||
failed_count=1,
|
||
summary_json={"ok": False, "error": error},
|
||
)
|
||
raise RuntimeError(error) from exc
|
||
|
||
elapsed_ms = self._elapsed_ms(started)
|
||
payload = {
|
||
"ok": True,
|
||
"models": appconfig.sanitize_for_log(models),
|
||
"balance": appconfig.sanitize_for_log(balance or {}),
|
||
"points_balance": (balance or {}).get("points_balance"),
|
||
}
|
||
title_count = self._priced_count(models, "title")
|
||
image_count = self._priced_count(models, "image")
|
||
vision_count = self._priced_count(models, "vision")
|
||
self._log_run_event(
|
||
"step=cmhub_settings result=success detail=title_aliases={title_count} image_aliases={image_count} vision_aliases={vision_count} points_balance={points_balance} elapsed_ms={elapsed_ms}".format(
|
||
title_count=title_count,
|
||
image_count=image_count,
|
||
vision_count=vision_count,
|
||
points_balance=payload.get("points_balance") if payload.get("points_balance") is not None else "",
|
||
elapsed_ms=elapsed_ms,
|
||
)
|
||
)
|
||
_safe_finish_run_log(
|
||
self._run_id,
|
||
db_path=self.db_path,
|
||
status="done",
|
||
done=1,
|
||
success_count=1,
|
||
summary_json=payload,
|
||
)
|
||
return payload
|
||
|
||
def _priced_count(self, models, operation):
|
||
return sum(
|
||
1
|
||
for model in models or []
|
||
if str(model.get("operation_type") or "").lower() == operation
|
||
and str(model.get("pricing_status") or "").lower() != "unpriced"
|
||
and str(model.get("alias") or "").strip()
|
||
and (
|
||
operation != "vision"
|
||
or (
|
||
str(model.get("pricing_status") or "").lower() == "priced"
|
||
and bool(model.get("requires_image"))
|
||
)
|
||
)
|
||
)
|
||
|
||
def _create_run_log(self):
|
||
if not self.db_path:
|
||
return None
|
||
return _safe_create_run_log(
|
||
"cmhub_settings_test",
|
||
db_path=self.db_path,
|
||
total=1,
|
||
options={"base_url": self.base_url, "include_balance": self.include_balance},
|
||
)
|
||
|
||
def _log_run_event(self, message, level="info"):
|
||
safe_message = _safe_add_run_log_event(
|
||
self._run_id,
|
||
message,
|
||
db_path=self.db_path,
|
||
level=level,
|
||
)
|
||
self.log.emit(str(safe_message))
|
||
|
||
def _write_diagnostic_log(
|
||
self,
|
||
message,
|
||
level="INFO",
|
||
step=None,
|
||
elapsed_ms=None,
|
||
payload=None,
|
||
exc=None,
|
||
):
|
||
_safe_write_diagnostic_log(
|
||
message,
|
||
level=level,
|
||
step=step,
|
||
elapsed_ms=elapsed_ms,
|
||
payload=payload,
|
||
exc=exc,
|
||
log_dir=self.diagnostic_log_dir,
|
||
)
|
||
|
||
def _safe_error(self, exc):
|
||
raw = str(exc) or exc.__class__.__name__
|
||
redacted = appconfig.redact_secrets(raw, [self.api_key])
|
||
return diagnostics.redact_log_text(redacted)
|
||
|
||
def _elapsed_ms(self, started):
|
||
return _elapsed_ms(started)
|
||
|
||
class AIModelTestWorker(BaseWorker):
|
||
"""Test one AI model connection without blocking the GUI thread."""
|
||
|
||
def __init__(self, model_name, ai_models_path=None, db_path=None, diagnostic_log_dir=None):
|
||
super().__init__()
|
||
self.model_name = model_name
|
||
self.ai_models_path = ai_models_path or appconfig.AI_MODELS_PATH
|
||
self.db_path = db_path
|
||
self.diagnostic_log_dir = diagnostic_log_dir
|
||
self._run_id = None
|
||
|
||
def execute(self):
|
||
self._run_id = self._create_run_log()
|
||
started = time.monotonic()
|
||
self._log_run_event(
|
||
f"step=test_connection result=start detail=AI模型 {self.model_name}"
|
||
)
|
||
try:
|
||
result = appconfig.test_ai_model(self.model_name, path=self.ai_models_path)
|
||
except Exception as exc:
|
||
error = diagnostics.redact_log_text(str(exc) or exc.__class__.__name__)
|
||
elapsed_ms = self._elapsed_ms(started)
|
||
self._log_run_event(
|
||
f"step=test_connection result=failed detail={error} elapsed_ms={elapsed_ms}",
|
||
level="error",
|
||
)
|
||
self._write_diagnostic_log(
|
||
"AI模型测试连接异常",
|
||
level="ERROR",
|
||
step="test_connection",
|
||
elapsed_ms=elapsed_ms,
|
||
payload={"model_name": self.model_name, "error": error},
|
||
exc=exc,
|
||
)
|
||
_safe_finish_run_log(
|
||
self._run_id,
|
||
db_path=self.db_path,
|
||
status="failed",
|
||
done=0,
|
||
failed_count=1,
|
||
summary_json={"ok": False, "name": self.model_name, "error": error},
|
||
)
|
||
raise
|
||
elapsed_ms = self._elapsed_ms(started)
|
||
payload = dict(appconfig.sanitize_for_log(result or {}))
|
||
payload["name"] = self.model_name
|
||
ok = bool(payload.get("ok"))
|
||
self._log_run_event(
|
||
"step=test_connection result={result} detail=AI模型 {name} status={status} error={error} elapsed_ms={elapsed_ms}".format(
|
||
result="success" if ok else "failed",
|
||
name=self.model_name,
|
||
status=payload.get("status") or "",
|
||
error=payload.get("error") or "",
|
||
elapsed_ms=elapsed_ms,
|
||
),
|
||
level="info" if ok else "warning",
|
||
)
|
||
_safe_finish_run_log(
|
||
self._run_id,
|
||
db_path=self.db_path,
|
||
status="done",
|
||
done=1,
|
||
success_count=1 if ok else 0,
|
||
failed_count=0 if ok else 1,
|
||
summary_json=payload,
|
||
)
|
||
return payload
|
||
|
||
def _create_run_log(self):
|
||
if not self.db_path:
|
||
return None
|
||
return _safe_create_run_log(
|
||
"ai_model_test",
|
||
db_path=self.db_path,
|
||
total=1,
|
||
options={"model_name": self.model_name},
|
||
)
|
||
|
||
def _log_run_event(self, message, level="info"):
|
||
safe_message = _safe_add_run_log_event(
|
||
self._run_id,
|
||
message,
|
||
db_path=self.db_path,
|
||
level=level,
|
||
)
|
||
self.log.emit(str(safe_message))
|
||
|
||
def _write_diagnostic_log(
|
||
self,
|
||
message,
|
||
level="INFO",
|
||
step=None,
|
||
elapsed_ms=None,
|
||
payload=None,
|
||
exc=None,
|
||
):
|
||
_safe_write_diagnostic_log(
|
||
message,
|
||
level=level,
|
||
step=step,
|
||
elapsed_ms=elapsed_ms,
|
||
payload=payload,
|
||
exc=exc,
|
||
log_dir=self.diagnostic_log_dir,
|
||
)
|
||
|
||
def _elapsed_ms(self, started):
|
||
return _elapsed_ms(started)
|