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

T-571 降低更新前台抢焦点

Browse Source
This commit is contained in:
chengma
2026-07-09 15:04:20 +08:00
parent d595e7455c
commit 5400145147
7 changed files with 403 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/editor.py
+210 -4
View File
@@ -15,6 +15,9 @@ ITEMBOX_XPATH = (
"//div[@class='container']/div[@class='can-drag shopee-image-manager__itembox' "
"and @data-draggable='true']"
)
COVER_FOREGROUND_RECOVERY_TIMEOUT_SECONDS = 120
COVER_FOREGROUND_UPLOAD_RECOVERY_REASONS = {"UPLOAD_STILL_PROCESSING", "UPLOAD_TIMEOUT"}
COVER_FOREGROUND_DRAG_RECOVERY_REASONS = {"DRAG_NOT_FIRST", "NEW_IMAGE_NOT_FOUND"}
LOGIN_PATH_MARKERS = (
"/login",
@@ -980,6 +983,7 @@ def replace_cover(cdp, image_win_path, old_cover_path=None, timeout=180) -> dict
}
before = stable.get("rects") or _image_rects(cdp)
before_srcs = {r.get("src") for r in before}
before_src_list = _src_snapshot(before_srcs)
upload_click = _click_upload_tile(cdp)
if not upload_click.get("clicked"):
return {
@@ -1075,13 +1079,20 @@ def replace_cover(cdp, image_win_path, old_cover_path=None, timeout=180) -> dict
"upload_state": last_state,
"blob_seen": blob_seen,
"file_size": file_size,
"before_srcs": before_src_list,
}
time.sleep(1)
cur = _image_rects(cdp)
new_rect = next((r for r in cur if r.get("src") == new_src), None)
if not new_rect:
return {"ok": False, "reason": "NEW_IMAGE_NOT_FOUND", "new_src": new_src, "delete": delete_result}
return {
"ok": False,
"reason": "NEW_IMAGE_NOT_FOUND",
"new_src": new_src,
"delete": delete_result,
"count_after": len(cur),
}
first = cur[0]
cdp.drag(new_rect["x"], new_rect["y"], first["left"] - first["w"] * 0.30, first["y"])
time.sleep(1.2)
@@ -1101,6 +1112,170 @@ def replace_cover(cdp, image_win_path, old_cover_path=None, timeout=180) -> dict
}
def _src_snapshot(values):
return sorted(str(value) for value in values if value)
def _recover_cover_after_foreground(cdp, cover_result, timeout=COVER_FOREGROUND_RECOVERY_TIMEOUT_SECONDS):
reason = str((cover_result or {}).get("reason") or "")
if reason in COVER_FOREGROUND_UPLOAD_RECOVERY_REASONS:
return _recover_cover_upload_after_foreground(cdp, cover_result, timeout=timeout)
if reason in COVER_FOREGROUND_DRAG_RECOVERY_REASONS:
return _recover_cover_drag_after_foreground(cdp, cover_result, timeout=timeout)
recovered = dict(cover_result or {})
recovered["foreground_recovery"] = {"attempted": False, "reason": "NOT_RECOVERABLE"}
return recovered
def _recover_cover_upload_after_foreground(cdp, cover_result, timeout=COVER_FOREGROUND_RECOVERY_TIMEOUT_SECONDS):
before_srcs = set(str(src) for src in (cover_result or {}).get("before_srcs") or [] if src)
if not before_srcs:
recovered = dict(cover_result or {})
recovered["foreground_recovery"] = {
"attempted": True,
"ok": False,
"reason": "MISSING_BEFORE_SRCS",
}
return recovered
original_reason = str((cover_result or {}).get("reason") or "UPLOAD_TIMEOUT")
end = time.time() + max(0.0, float(timeout))
last_state = (cover_result or {}).get("upload_state") or {}
blob_seen = bool((cover_result or {}).get("blob_seen"))
last_rects = []
while True:
cur = _image_rects(cdp)
last_rects = cur
last_state = _upload_state(cdp)
blob_seen = blob_seen or any(str(r.get("src") or "").startswith("blob:") for r in cur)
if (last_state.get("errors") or last_state.get("error_toasts")) and not last_state.get("busy_count"):
reason = "UPLOAD_DUPLICATE_IMAGE" if _has_duplicate_upload_error(last_state) else "UPLOAD_PAGE_ERROR"
return {
"ok": False,
"reason": reason,
"original_reason": original_reason,
"foreground_recovery": {"attempted": True, "ok": False, "reason": reason},
"count_after": len(cur),
"upload_state": last_state,
"before_srcs": _src_snapshot(before_srcs),
"blob_seen": blob_seen,
}
ready = [
r for r in cur
if r.get("src") not in before_srcs
and r.get("src")
and "susercontent" in r.get("src")
and "blob:" not in r.get("src")
]
if ready:
return _drag_existing_cover_to_first(
cdp,
ready[-1]["src"],
original_reason=original_reason,
foreground_recovery_reason="UPLOAD_READY_AFTER_FOREGROUND",
timeout=min(max(0.0, float(timeout)), 30.0),
)
if last_state.get("crop_modal"):
return {
"ok": False,
"reason": "UPLOAD_CROP_REQUIRED",
"original_reason": original_reason,
"foreground_recovery": {"attempted": True, "ok": False, "reason": "UPLOAD_CROP_REQUIRED"},
"count_after": len(cur),
"upload_state": last_state,
"before_srcs": _src_snapshot(before_srcs),
"blob_seen": blob_seen,
}
if time.time() >= end:
break
time.sleep(1.5)
reason = "UPLOAD_STILL_PROCESSING" if blob_seen or (last_state or {}).get("busy_count") else "UPLOAD_TIMEOUT"
return {
"ok": False,
"reason": reason,
"original_reason": original_reason,
"foreground_recovery": {"attempted": True, "ok": False, "reason": "RECOVERY_TIMEOUT"},
"count_after": len(last_rects),
"upload_state": last_state,
"before_srcs": _src_snapshot(before_srcs),
"blob_seen": blob_seen,
}
def _recover_cover_drag_after_foreground(cdp, cover_result, timeout=COVER_FOREGROUND_RECOVERY_TIMEOUT_SECONDS):
new_src = str((cover_result or {}).get("new_src") or "")
original_reason = str((cover_result or {}).get("reason") or "DRAG_NOT_FIRST")
return _drag_existing_cover_to_first(
cdp,
new_src,
original_reason=original_reason,
foreground_recovery_reason="REDRAG_AFTER_FOREGROUND",
timeout=timeout,
)
def _drag_existing_cover_to_first(
cdp,
new_src,
original_reason,
foreground_recovery_reason,
timeout=COVER_FOREGROUND_RECOVERY_TIMEOUT_SECONDS,
):
if not new_src:
return {
"ok": False,
"reason": original_reason,
"original_reason": original_reason,
"foreground_recovery": {
"attempted": True,
"ok": False,
"reason": "NEW_SRC_MISSING",
},
}
end = time.time() + max(0.0, float(timeout))
last_rects = []
while True:
cur = _image_rects(cdp)
last_rects = cur
new_rect = next((r for r in cur if r.get("src") == new_src), None)
if new_rect:
if not cur:
break
first = cur[0]
cdp.drag(new_rect["x"], new_rect["y"], first["left"] - first["w"] * 0.30, first["y"])
time.sleep(1.2)
after = _image_rects(cdp)
cover_ok = bool(after and after[0].get("src") == new_src)
index = next((r["i"] for r in after if r.get("src") == new_src), None)
return {
"ok": cover_ok,
"reason": None if cover_ok else "DRAG_NOT_FIRST",
"original_reason": original_reason,
"foreground_recovery": {
"attempted": True,
"ok": cover_ok,
"reason": foreground_recovery_reason,
},
"new_src": new_src,
"index": index,
"count_after": len(after),
}
if time.time() >= end:
break
time.sleep(1.5)
return {
"ok": False,
"reason": "NEW_IMAGE_NOT_FOUND",
"original_reason": original_reason,
"foreground_recovery": {"attempted": True, "ok": False, "reason": "NEW_IMAGE_NOT_FOUND"},
"new_src": new_src,
"count_after": len(last_rects),
}
def _validated_old_cover_backup(old_cover_path):
if not old_cover_path:
return None
@@ -1236,7 +1411,7 @@ def _confirm_update_modal(cdp, timeout=3):
return last_present or {"present": False, "clicked": False, "reason": "NO_UPDATE_CONFIRM_MODAL"}
def apply_task(account, task, close_success_tab=False, on_step=None) -> dict:
def apply_task(account, task, close_success_tab=False, on_step=None, bring_to_front=True) -> dict:
"""Apply generated title/cover to Shopee.
The caller must perform the batch confirmation before calling this function.
@@ -1245,7 +1420,7 @@ def apply_task(account, task, close_success_tab=False, on_step=None) -> dict:
item_id = _item_id(task)
current_step = "open_product"
_notify_apply_step(on_step, current_step, "start")
cdp = open_product(account, item_id)
cdp = open_product(account, item_id, bring_to_front=bring_to_front)
_notify_apply_step(on_step, current_step, "success")
committed = False
try:
@@ -1269,11 +1444,42 @@ def apply_task(account, task, close_success_tab=False, on_step=None) -> dict:
new_cover_path,
old_cover_path=_get(task, "old_cover_path"),
)
if not cover_result.get("ok"):
recoverable_reasons = (
COVER_FOREGROUND_UPLOAD_RECOVERY_REASONS
| COVER_FOREGROUND_DRAG_RECOVERY_REASONS
)
if not bring_to_front and str(cover_result.get("reason") or "") in recoverable_reasons:
current_step = "cover_retry_foreground"
_notify_apply_step(
on_step,
current_step,
"start",
"后台上传/拖拽疑似受限，已提前台恢复一次",
)
try:
cdp.send("Page.bringToFront")
cover_result = _recover_cover_after_foreground(cdp, cover_result)
except Exception as exc:
cover_result = dict(cover_result)
cover_result["foreground_recovery"] = {
"attempted": True,
"ok": False,
"reason": str(exc),
}
if cover_result.get("ok"):
_notify_apply_step(on_step, current_step, "success", "前台恢复成功")
current_step = "replace_cover"
_notify_apply_step(on_step, current_step, "success")
else:
detail = _cover_upload_error_message(cover_result)
_notify_apply_step(on_step, current_step, "failed", detail)
current_step = "replace_cover"
if not cover_result.get("ok"):
error = _cover_upload_error_message(cover_result)
_notify_apply_step(on_step, current_step, "failed", error)
return {"committed": False, "error": error, "cover": cover_result}
_notify_apply_step(on_step, current_step, "success")
_notify_apply_step(on_step, "replace_cover", "success")
current_step = "click_update"
_notify_apply_step(on_step, current_step, "start")
update_result = click_update(cdp)
app/gui/workers.py
+15
View File
@@ -568,6 +568,8 @@ class ApplyWorker(BaseWorker):
self._current_batch_size = None
self._batch_count = 0
self._progress_lock = threading.Lock()
self._foreground_lock = threading.Lock()
self._foregrounded_aliases = set()
self.diagnostic_log_dir = diagnostic_log_dir
self._run_id = None
@@ -890,11 +892,13 @@ class ApplyWorker(BaseWorker):
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"
bring_to_front = self._should_bring_account_to_front(account)
result = editor.apply_task(
account,
task,
close_success_tab=self.close_success_tab,
on_step=on_step,
bring_to_front=bring_to_front,
)
committed = bool(result.get("committed")) and not result.get("error")
error = result.get("error")
@@ -974,6 +978,17 @@ class ApplyWorker(BaseWorker):
exc=exc,
)
return "failed"
def _should_bring_account_to_front(self, account):
alias = str(getattr(account, "alias", "") or "").strip()
if not alias:
return True
with self._foreground_lock:
if alias in self._foregrounded_aliases:
return False
self._foregrounded_aliases.add(alias)
return True
def _record_outcome(self, counters, total, outcome):
with self._progress_lock:
counters["done"] += 1