"""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)