fix: handle Shopee update confirmation and upload edge cases

This commit is contained in:
chengma
2026-06-30 11:44:27 +08:00
parent cd1256bc5e
commit 580a7a570d
6 changed files with 877 additions and 33 deletions
+421 -15
View File
@@ -43,6 +43,47 @@ JS_RECTS = (
"return JSON.stringify(a);})()"
)
JS_UPLOAD_STATE = (
"(function(){"
"function visible(e){if(!e)return false;var r=e.getBoundingClientRect();var s=getComputedStyle(e);"
"return r.width>0&&r.height>0&&s.visibility!=='hidden'&&s.display!=='none';}"
"function text(e){return ((e&&((e.innerText||e.textContent)||''))||'').trim();}"
"var manager=document.querySelector('.shopee-image-manager');"
"var items=manager?[].slice.call(manager.querySelectorAll('.shopee-image-manager__itembox,[class*=image-manager__itembox]')):[];"
"var imgs=items.map(function(item){var img=item.querySelector('img');return img?img.src:null;}).filter(Boolean);"
"var upload=manager?manager.querySelector('.shopee-image-manager__upload input[type=file]'):null;"
"var uploadBox=manager?manager.querySelector('.shopee-image-manager__upload,[class*=image-manager__upload]'):null;"
"var busy=manager?[].slice.call(manager.querySelectorAll('[class*=loading],[class*=Loading],[class*=spinner],[class*=Spinner],[class*=progress],[class*=Progress],[class*=uploading],[class*=Uploading]')).filter(visible):[];"
"var errRe=/(失敗|失败|錯誤|错误|不支援|不支持|格式|大小|尺寸|超過|超过|error|fail|invalid|unsupported)/i;"
"var uploadRe=/(圖片|图片|封面|照片|相片|圖像|图像|image|photo|cover|upload|上傳|上传|檔案|文件|file|格式|大小|尺寸|像素|解析度|分辨率|超過|超过)/i;"
"var texts=manager?[].slice.call(manager.querySelectorAll('*')).filter(visible).map(text).filter(Boolean):[];"
"var errors=texts.filter(function(t){return errRe.test(t);}).slice(0,8);"
"var roots=[].slice.call(document.querySelectorAll('.eds-modal__content,.eds-modal__box,[role=dialog]')).filter(visible);"
"var modalTexts=roots.map(text).filter(Boolean);"
"var crop=modalTexts.find(function(t){return /(裁剪|裁切|剪裁|crop)/i.test(t);})||'';"
"var toastTexts=[].slice.call(document.querySelectorAll('[class*=toast],[class*=Toast],[class*=message],[class*=Message],[class*=notice]')).filter(visible).map(text).filter(Boolean).slice(0,8);"
"var pageErrorToasts=toastTexts.filter(function(t){return errRe.test(t);}).slice(0,8);"
"var errorToasts=pageErrorToasts.filter(function(t){return uploadRe.test(t);}).slice(0,8);"
"return JSON.stringify({"
"manager_exists:!!manager,"
"count:items.length,"
"srcs:imgs.slice(0,12),"
"blob_count:imgs.filter(function(s){return /^blob:/.test(s);}).length,"
"cdn_count:imgs.filter(function(s){return /susercontent/.test(s);}).length,"
"upload_input_exists:!!upload,"
"upload_input_disabled:!!(upload&&(upload.disabled||upload.getAttribute('aria-disabled')==='true')),"
"upload_tile_visible:visible(uploadBox),"
"busy_count:busy.length,"
"errors:errors,"
"crop_modal:!!crop,"
"crop_text:crop,"
"toasts:toastTexts,"
"error_toasts:errorToasts,"
"page_error_toasts:pageErrorToasts"
"});})()"
)
JS_TITLE_STATE = (
"(function(){"
f"var r=document.evaluate({json.dumps(TITLE_XPATH)},document,null,"
@@ -76,6 +117,59 @@ JS_CLICK_UPDATE = (
"b.click();return 'CLICKED';})()"
)
JS_FIND_UPDATE_CONFIRM = (
"(function(){"
"function visible(e){if(!e)return false;var r=e.getBoundingClientRect();var s=getComputedStyle(e);"
"return r.width>0&&r.height>0&&s.visibility!=='hidden'&&s.display!=='none';}"
"function text(e){return ((e&&((e.innerText||e.textContent)||''))||'').trim();}"
"function updateTitle(t){return /(確定|确定)/.test(t)&&/更新商品/.test(t);}"
"function title(root){return text(root.querySelector('.eds-modal__title,[class*=modal__title],[class*=Modal__title]'))||text(root);}"
"function buttons(root){var scope=root.querySelector('.eds-modal__footer,[class*=modal__footer],[class*=Modal__footer]')||root;"
"return [].slice.call(scope.querySelectorAll('button,[role=button]')).filter(visible);}"
"function hasUpdateButton(root){return buttons(root).some(function(b){return text(b)==='更新'&&!b.disabled&&!/disabled/i.test(b.className);});}"
"function uniq(items){var out=[];items.forEach(function(e){if(e&&out.indexOf(e)<0)out.push(e);});return out;}"
"var selector='.eds-modal__content,.eds-modal__box,[role=dialog]';"
"var roots=[].slice.call(document.querySelectorAll(selector));"
"var titles=[].slice.call(document.querySelectorAll('.eds-modal__title,[class*=modal__title],[class*=Modal__title]'))"
".filter(visible).filter(function(t){return updateTitle(text(t));})"
".map(function(t){return t.closest(selector);});"
"roots=uniq(roots.concat(titles)).filter(visible).filter(function(r){return updateTitle(title(r));}).filter(hasUpdateButton);"
"if(!roots.length)return JSON.stringify({present:false});"
"var root=roots[roots.length-1];"
"var seen=buttons(root).map(function(b){return text(b);}).filter(Boolean);"
"return JSON.stringify({present:true,title:title(root),buttons:seen.slice(0,8)});})()"
)
JS_CLICK_UPDATE_CONFIRM = (
"(function(){"
"function visible(e){if(!e)return false;var r=e.getBoundingClientRect();var s=getComputedStyle(e);"
"return r.width>0&&r.height>0&&s.visibility!=='hidden'&&s.display!=='none';}"
"function text(e){return ((e&&((e.innerText||e.textContent)||''))||'').trim();}"
"function updateTitle(t){return /(確定|确定)/.test(t)&&/更新商品/.test(t);}"
"function title(root){return text(root.querySelector('.eds-modal__title,[class*=modal__title],[class*=Modal__title]'))||text(root);}"
"function buttons(root){var scope=root.querySelector('.eds-modal__footer,[class*=modal__footer],[class*=Modal__footer]')||root;"
"return [].slice.call(scope.querySelectorAll('button,[role=button]')).filter(visible);}"
"function hasUpdateButton(root){return buttons(root).some(function(b){return text(b)==='更新'&&!b.disabled&&!/disabled/i.test(b.className);});}"
"function uniq(items){var out=[];items.forEach(function(e){if(e&&out.indexOf(e)<0)out.push(e);});return out;}"
"var selector='.eds-modal__content,.eds-modal__box,[role=dialog]';"
"var roots=[].slice.call(document.querySelectorAll(selector));"
"var titles=[].slice.call(document.querySelectorAll('.eds-modal__title,[class*=modal__title],[class*=Modal__title]'))"
".filter(visible).filter(function(t){return updateTitle(text(t));})"
".map(function(t){return t.closest(selector);});"
"roots=uniq(roots.concat(titles)).filter(visible).filter(function(r){return updateTitle(title(r));}).filter(hasUpdateButton);"
"if(!roots.length)return JSON.stringify({present:false,clicked:false,reason:'NO_UPDATE_CONFIRM_MODAL'});"
"var root=roots[roots.length-1];"
"var allButtons=buttons(root);"
"var seen=allButtons.map(function(b){return text(b);}).filter(Boolean);"
"var exact=allButtons.filter(function(b){var t=text(b);"
"return t==='更新'&&!b.disabled&&!/disabled/i.test(b.className);});"
"var primary=exact.filter(function(b){return /eds-button--primary/.test(b.className);});"
"var candidates=primary.length?primary:exact;"
"if(!candidates.length)return JSON.stringify({present:true,clicked:false,reason:'NO_UPDATE_CONFIRM_BUTTON',buttons:seen.slice(0,8)});"
"candidates[0].click();"
"return JSON.stringify({present:true,clicked:true,reason:null,text:text(candidates[0]),buttons:seen.slice(0,8)});})()"
)
JS_TOASTS = (
"(function(){var es=[].slice.call(document.querySelectorAll("
"'[class*=toast],[class*=Toast],[class*=message],[class*=Message],[class*=notice]'));"
@@ -84,6 +178,25 @@ JS_TOASTS = (
"return JSON.stringify(t.slice(0,5));})()"
)
JS_POST_UPDATE_STATE = (
"(function(){"
"function visible(e){if(!e)return false;var r=e.getBoundingClientRect();var s=getComputedStyle(e);"
"return r.width>0&&r.height>0&&s.visibility!=='hidden'&&s.display!=='none';}"
"function text(e){return ((e&&((e.innerText||e.textContent)||''))||'').trim();}"
"var toastTexts=[].slice.call(document.querySelectorAll('[class*=toast],[class*=Toast],[class*=message],[class*=Message],[class*=notice]'))"
".filter(visible).map(text).filter(Boolean).slice(0,8);"
"var errRe=/(失敗|失败|錯誤|错误|不支援|不支持|無法|无法|不允許|不允许|error|fail|invalid|請|请)/i;"
"var okRe=/(成功|已更新|更新完成|儲存成功|保存成功|success|saved)/i;"
"var url=location.href;"
"return JSON.stringify({"
"url:url,"
"redirected_to_list:url.indexOf('/portal/product/list/')>=0,"
"toasts:toastTexts,"
"error_toasts:toastTexts.filter(function(t){return errRe.test(t);}),"
"success_toasts:toastTexts.filter(function(t){return okRe.test(t);})"
"});})()"
)
JS_CLICK_FIRST_DELETE = (
"(function(){"
f"var s=document.evaluate({json.dumps(ITEMBOX_XPATH)},document,null,"
@@ -200,6 +313,147 @@ def _image_rects(cdp):
return _json_value(cdp, JS_RECTS, default=[]) or []
def _upload_state(cdp):
return _json_value(cdp, JS_UPLOAD_STATE, default={}) or {}
def _post_update_state(cdp):
return _json_value(cdp, JS_POST_UPDATE_STATE, default=None) or {}
def _wait_post_update(cdp, timeout=8):
end = time.time() + timeout
last_state = {}
last_error_state = None
while time.time() < end:
last_state = _post_update_state(cdp)
if not last_state:
return {"ok": True, "observed_success": False, "observed_failure": False}
error_toasts = last_state.get("error_toasts") or []
success_toasts = last_state.get("success_toasts") or []
redirected = bool(last_state.get("redirected_to_list"))
if redirected or success_toasts:
last_state.update({"ok": True, "reason": None, "observed_success": True, "observed_failure": False})
return last_state
if error_toasts:
last_error_state = dict(last_state)
time.sleep(0.5)
last_state = last_state or _post_update_state(cdp)
if not last_state:
return {"ok": True, "observed_success": False, "observed_failure": False}
success_toasts = last_state.get("success_toasts") or []
redirected = bool(last_state.get("redirected_to_list"))
if redirected or success_toasts:
last_state.update({"ok": True, "reason": None, "observed_success": True, "observed_failure": False})
return last_state
error_state = last_error_state or (last_state if last_state.get("error_toasts") else None)
if error_state:
error_state.update({"ok": False, "reason": "POST_UPDATE_ERROR", "observed_success": False, "observed_failure": True})
return error_state
last_state.update({"ok": True, "reason": None, "observed_success": False, "observed_failure": False})
return last_state
def _upload_input_ready(state):
return bool((state or {}).get("upload_input_exists")) and not bool((state or {}).get("upload_input_disabled"))
def _unstable_image_manager_reason(state, rects, expected_count, require_upload_input=True):
state = state or {}
if state.get("crop_modal"):
return "UPLOAD_CROP_REQUIRED"
if state.get("errors") or state.get("error_toasts"):
return "UPLOAD_PAGE_ERROR"
if expected_count is not None and len(rects) != expected_count:
return "IMAGE_COUNT_NOT_READY"
if state.get("busy_count") or any(str(r.get("src") or "").startswith("blob:") for r in rects):
return "IMAGE_MANAGER_BUSY"
if require_upload_input and not _upload_input_ready(state):
return "UPLOAD_INPUT_NOT_READY"
return "IMAGE_MANAGER_NOT_STABLE"
def _wait_image_manager_stable(
cdp,
expected_count=None,
timeout=20,
settle_seconds=1.0,
poll=0.5,
require_upload_input=True,
):
end = time.time() + timeout
last_rects = []
last_state = {}
last_signature = None
stable_hits = 0
while time.time() < end:
last_rects = _image_rects(cdp)
last_state = _upload_state(cdp)
signature = tuple(r.get("src") for r in last_rects)
count_ok = expected_count is None or len(last_rects) == expected_count
no_blob = not any(str(r.get("src") or "").startswith("blob:") for r in last_rects)
ready = (
bool(last_rects)
and count_ok
and no_blob
and not last_state.get("busy_count")
and not last_state.get("crop_modal")
and not last_state.get("errors")
and not last_state.get("error_toasts")
and (not require_upload_input or _upload_input_ready(last_state))
)
if ready and signature == last_signature:
stable_hits += 1
elif ready:
last_signature = signature
stable_hits = 1
else:
last_signature = signature
stable_hits = 0
if stable_hits >= 2:
if settle_seconds:
time.sleep(settle_seconds)
return {"ok": True, "rects": last_rects, "upload_state": last_state}
time.sleep(poll)
return {
"ok": False,
"reason": _unstable_image_manager_reason(
last_state,
last_rects,
expected_count,
require_upload_input=require_upload_input,
),
"rects": last_rects,
"upload_state": last_state,
"expected_count": expected_count,
"require_upload_input": require_upload_input,
"count_after": len(last_rects),
}
def _cover_upload_error_message(result):
reason = result.get("reason") or "COVER_UPDATE_FAILED"
state = result.get("upload_state") or {}
if reason == "UPLOAD_PAGE_ERROR":
details = state.get("errors") or state.get("error_toasts") or state.get("toasts") or []
return "新封面上传失败:" + ";".join(map(str, details[:3])) if details else "新封面上传失败"
if reason == "UPLOAD_CROP_REQUIRED":
return "新封面上传后出现裁剪确认框,当前版本未自动处理裁剪弹窗"
if reason == "UPLOAD_STILL_PROCESSING":
return "新封面上传仍在处理中,未取得 Shopee CDN 地址"
if reason == "UPLOAD_TIMEOUT":
return "新封面上传超时,未取得 Shopee CDN 地址"
if reason == "IMAGE_MANAGER_BUSY":
return "商品图片区域仍在加载,未开始上传新封面"
if reason == "IMAGE_COUNT_NOT_READY":
return "删除旧封面后图片数量未稳定,未开始上传新封面"
if reason == "UPLOAD_INPUT_NOT_READY":
return "商品图片上传入口未恢复可用,未开始上传新封面"
if reason == "IMAGE_MANAGER_NOT_STABLE":
return "商品图片区域未稳定,未开始上传新封面"
return str(reason)
def _wait_ready(cdp, timeout=60):
end = time.time() + timeout
while time.time() < end:
@@ -437,7 +691,7 @@ def change_title(cdp, new_title) -> dict:
return {"ok": ok, "written": written, "value": value, "modelvalue": modelvalue}
def replace_cover(cdp, image_win_path, old_cover_path=None, timeout=90) -> dict:
def replace_cover(cdp, image_win_path, old_cover_path=None, timeout=180) -> dict:
"""Upload an image and drag it to the first position.
When the image manager is full, only delete the current first image if the
@@ -453,7 +707,24 @@ def replace_cover(cdp, image_win_path, old_cover_path=None, timeout=90) -> dict:
"if(m)m.scrollIntoView({block:'center'});return !!m;})()"
)
time.sleep(0.5)
before = _image_rects(cdp)
stable = _wait_image_manager_stable(
cdp,
timeout=20,
settle_seconds=1.0,
require_upload_input=False,
)
if not stable.get("ok"):
rects = stable.get("rects") or []
return {
"ok": False,
"reason": stable.get("reason"),
"count_before": len(rects),
"count_after": len(rects),
"upload_state": stable.get("upload_state"),
"stable": stable,
}
before = stable.get("rects") or _image_rects(cdp)
count_before = len(before)
delete_result = None
if len(before) >= 9:
@@ -473,24 +744,78 @@ def replace_cover(cdp, image_win_path, old_cover_path=None, timeout=90) -> dict:
"count_before": count_before,
"delete": delete_result,
}
before = _image_rects(cdp)
stable = _wait_image_manager_stable(
cdp,
expected_count=delete_result.get("count_after"),
timeout=30,
settle_seconds=2.0,
require_upload_input=True,
)
if not stable.get("ok"):
return {
"ok": False,
"reason": stable.get("reason"),
"count_before": count_before,
"count_after": stable.get("count_after"),
"delete": delete_result,
"upload_state": stable.get("upload_state"),
"stable": stable,
}
before = stable.get("rects") or _image_rects(cdp)
else:
stable = _wait_image_manager_stable(
cdp,
expected_count=len(before),
timeout=20,
settle_seconds=1.0,
require_upload_input=True,
)
if not stable.get("ok"):
return {
"ok": False,
"reason": stable.get("reason"),
"count_before": count_before,
"count_after": stable.get("count_after"),
"upload_state": stable.get("upload_state"),
"stable": stable,
}
before = stable.get("rects") or _image_rects(cdp)
before_srcs = {r.get("src") for r in before}
oid = cdp.object_id("document.querySelector('.shopee-image-manager__upload input[type=file]')")
if not oid:
return {"ok": False, "reason": "NO_UPLOAD_INPUT", "count_before": count_before, "delete": delete_result}
return {
"ok": False,
"reason": "NO_UPLOAD_INPUT",
"count_before": count_before,
"delete": delete_result,
"upload_state": _upload_state(cdp),
}
cdp.send("DOM.setFileInputFiles", {"objectId": oid, "files": [image_win_path]})
cdp.val(
changed = cdp.val(
"(function(){var up=document.querySelector('.shopee-image-manager__upload input[type=file]');"
"if(!up)return false;"
"up.dispatchEvent(new Event('input',{bubbles:true}));"
"up.dispatchEvent(new Event('change',{bubbles:true}));return true;})()"
)
time.sleep(2)
new_src = None
last_state = _upload_state(cdp)
if not changed and not last_state.get("busy_count") and not last_state.get("blob_count"):
return {
"ok": False,
"reason": "UPLOAD_INPUT_NOT_READY",
"count_before": count_before,
"delete": delete_result,
"upload_state": last_state,
}
blob_seen = bool(last_state.get("blob_count"))
file_size = os.path.getsize(image_win_path) if os.path.exists(image_win_path) else None
end = time.time() + timeout
while time.time() < end:
time.sleep(1.5)
cur = _image_rects(cdp)
last_state = _upload_state(cdp)
blob_seen = blob_seen or any(str(r.get("src") or "").startswith("blob:") for r in cur)
ready = [
r for r in cur
if r.get("src") not in before_srcs
@@ -498,11 +823,43 @@ def replace_cover(cdp, image_win_path, old_cover_path=None, timeout=90) -> dict:
and "susercontent" in r.get("src")
and "blob:" not in r.get("src")
]
if len(cur) > len(before) and ready:
if ready and (len(cur) > len(before) or len(ready) == 1):
new_src = ready[-1]["src"]
break
if (last_state.get("errors") or last_state.get("error_toasts")) and not last_state.get("busy_count"):
return {
"ok": False,
"reason": "UPLOAD_PAGE_ERROR",
"count_before": count_before,
"count_after": len(cur),
"delete": delete_result,
"upload_state": last_state,
"file_size": file_size,
}
if last_state.get("crop_modal"):
return {
"ok": False,
"reason": "UPLOAD_CROP_REQUIRED",
"count_before": count_before,
"count_after": len(cur),
"delete": delete_result,
"upload_state": last_state,
"file_size": file_size,
}
time.sleep(1.5)
if not new_src:
return {"ok": False, "reason": "UPLOAD_TIMEOUT", "count_before": count_before, "delete": delete_result}
cur = _image_rects(cdp)
reason = "UPLOAD_STILL_PROCESSING" if blob_seen or (last_state or {}).get("busy_count") else "UPLOAD_TIMEOUT"
return {
"ok": False,
"reason": reason,
"count_before": count_before,
"count_after": len(cur),
"delete": delete_result,
"upload_state": last_state,
"blob_seen": blob_seen,
"file_size": file_size,
}
time.sleep(1)
cur = _image_rects(cdp)
@@ -595,8 +952,8 @@ def _delete_first_cover(cdp, before, timeout=15):
}
def click_update(cdp) -> dict:
"""Click the Shopee update button if it is present and enabled."""
def click_update(cdp, confirm_timeout=3, post_timeout=8) -> dict:
"""Click the Shopee update button, confirm Shopee's final modal, and observe submit result."""
info = _json_value(cdp, JS_FIND_UPDATE, default={"found": False})
if not info.get("found"):
@@ -604,15 +961,64 @@ def click_update(cdp) -> dict:
if info.get("disabled"):
return {"clicked": False, "reason": "UPDATE_DISABLED", "toasts": []}
result = cdp.val(JS_CLICK_UPDATE)
time.sleep(3 if result == "CLICKED" else 0)
toasts = _json_value(cdp, JS_TOASTS, default=[]) or []
if result != "CLICKED":
toasts = _json_value(cdp, JS_TOASTS, default=[]) or []
return {"clicked": False, "reason": result, "toasts": toasts}
confirm_result = _confirm_update_modal(cdp, timeout=confirm_timeout)
if confirm_result.get("present") and not confirm_result.get("clicked"):
toasts = _json_value(cdp, JS_TOASTS, default=[]) or []
return {
"clicked": False,
"reason": confirm_result.get("reason") or "UPDATE_CONFIRM_NOT_CLICKED",
"toasts": toasts,
"confirm": confirm_result,
}
post_update = _wait_post_update(cdp, timeout=post_timeout)
toasts = post_update.get("toasts") if post_update else None
if toasts is None:
toasts = _json_value(cdp, JS_TOASTS, default=[]) or []
if post_update.get("observed_failure"):
return {
"clicked": False,
"reason": post_update.get("reason") or "POST_UPDATE_ERROR",
"toasts": toasts,
"confirm": confirm_result,
"post_update": post_update,
}
return {
"clicked": result == "CLICKED",
"reason": None if result == "CLICKED" else result,
"clicked": True,
"reason": None,
"toasts": toasts,
"confirm": confirm_result,
"post_update": post_update,
}
def _confirm_update_modal(cdp, timeout=3):
end = time.time() + timeout
last_present = None
while time.time() < end:
state = _json_value(
cdp,
JS_FIND_UPDATE_CONFIRM,
default={"present": False},
) or {"present": False}
if state.get("present"):
confirm_attempt = _json_value(
cdp,
JS_CLICK_UPDATE_CONFIRM,
default={"present": True, "clicked": False, "reason": "NO_UPDATE_CONFIRM_BUTTON"},
) or {}
last_present = confirm_attempt
if confirm_attempt.get("clicked"):
time.sleep(1)
return confirm_attempt
time.sleep(0.5)
return last_present or {"present": False, "clicked": False, "reason": "NO_UPDATE_CONFIRM_MODAL"}
def apply_task(account, task, close_success_tab=False) -> dict:
"""Apply generated title/cover to Shopee.
@@ -638,7 +1044,7 @@ def apply_task(account, task, close_success_tab=False) -> dict:
old_cover_path=_get(task, "old_cover_path"),
)
if not cover_result.get("ok"):
return {"committed": False, "error": cover_result.get("reason"), "cover": cover_result}
return {"committed": False, "error": _cover_upload_error_message(cover_result), "cover": cover_result}
update_result = click_update(cdp)
committed = bool(update_result.get("clicked", False))
return {