"""Shopee editor operations built on the verified CDP primitives.""" import json import os import time from urllib.parse import urlparse from . import appconfig, image_paths, product_status from .cdp import ( CDP, close_tab, close_tab_and_wait, create_tab, create_tab_info, find_product_tab, http_get, ) DEFAULT_REGION_HOST = "seller.shopee.tw" TITLE_FIELD_SELECTOR = '[data-product-edit-field-unique-id="name"]' IMAGE_FIELD_SELECTOR = '[data-product-edit-field-unique-id="images"]' TITLE_XPATH = ( "//*[@data-product-edit-field-unique-id='name']" "//input[contains(concat(' ', normalize-space(@class), ' '), ' eds-input__input ')]" ) LEGACY_TITLE_XPATH = "//input[@class='eds-input__input' and string-length(@modelvalue)>24]" ITEMBOX_XPATH = ( "//*[@data-product-edit-field-unique-id='images']" "//div[contains(concat(' ', normalize-space(@class), ' '), " "' shopee-image-manager__itembox ') and @data-draggable='true']" ) LEGACY_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", "account/signin", "accounts.shopee.tw/seller/login", "seller/login", "seller/accounts/signin", ) JS_EDITOR_FIELD_HELPERS = "".join( ( "function cmVisible(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 cmTitleCandidate(){", f"var root=document.querySelector({json.dumps(TITLE_FIELD_SELECTOR)});", "if(root){var primary=[].slice.call(root.querySelectorAll('input.eds-input__input'))" ".filter(cmVisible);return {node:primary.length===1?primary[0]:null,count:primary.length," "source:'business',root_exists:true};}", f"var snap=document.evaluate({json.dumps(LEGACY_TITLE_XPATH)},document,null," "XPathResult.ORDERED_NODE_SNAPSHOT_TYPE,null);", "var legacy=[];for(var i=0;i0&&uploads.length===1;" "return JSON.stringify({ready:ready,title_count:title.count,title_source:title.source," "title_field_exists:title.root_exists,title_ambiguous:title.count>1," "image_field_exists:image.root_exists,image_manager_exists:!!image.node," "image_count:items.length,cdn_count:srcs.filter(function(s){return /susercontent/.test(s);}).length," "blob_count:srcs.filter(function(s){return /^blob:/.test(s);}).length," "upload_input_exists:!!up,upload_input_count:uploads.length,current_url:location.href});})()" ) JS_RECTS = ( "(function(){" + JS_EDITOR_FIELD_HELPERS + "var image=cmMainImageContext();var items=cmMainImageItems(image);" "var a=[];for(var i=0;i0&&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]'));" "var t=es.filter(function(e){return e.offsetParent!==null;})" ".map(function(e){return (e.innerText||'').trim();}).filter(Boolean);" "return JSON.stringify(t.slice(0,5));})()" ) JS_TOAST_OBSERVER_SOURCE = ( "(function(){" "if(window.__cmshopee_toast_observer_installed)return true;" "window.__cmshopee_toast_observer_installed=true;" "window.__cmshopee_toasts=window.__cmshopee_toasts||[];" "function text(e){return ((e&&e.innerText)||(e&&e.textContent)||'').trim();}" "function visible(e){if(!e)return false;var s=getComputedStyle(e);var r=e.getBoundingClientRect();return s.display!=='none'&&s.visibility!=='hidden'&&r.width>0&&r.height>0;}" "function rootOf(e){return e.closest?e.closest('.eds-toast,.eds-toasts,[role=alert],[class*=toast],[class*=Toast],[class*=message],[class*=Message],[class*=notice]')||e:e;}" "function record(e){var root=rootOf(e);var value=text(root)||text(e);if(!value)return;window.__cmshopee_toasts.push({text:value,html:((root.outerHTML||e.outerHTML)||'').slice(0,2000),url:location.href,visible:visible(root)||visible(e),created_at:(new Date()).toISOString()});}" "function scan(){try{Array.from(document.querySelectorAll('.eds-toasts .eds-toast,.eds-toast,.eds-toast__content,[role=alert],[class*=toast],[class*=Toast],[class*=message],[class*=Message],[class*=notice]')).forEach(record);window.__cmshopee_toasts=window.__cmshopee_toasts.slice(-20);}catch(e){}}" "try{if(document.documentElement||document.body){var mo=new MutationObserver(function(){scan();});mo.observe(document.documentElement||document.body,{childList:true,subtree:true,characterData:true,attributes:true,attributeFilter:['style','class']});window.__cmshopee_toast_observer=mo;}}catch(e){}" "scan();return true;" "})()" ) JS_INSTALL_TOAST_OBSERVER = JS_TOAST_OBSERVER_SOURCE JS_PAGE_TOASTS = ( "(function(){" "function text(e){return ((e&&e.innerText)||(e&&e.textContent)||'').trim();}" "function visible(e){if(!e)return false;var s=getComputedStyle(e);var r=e.getBoundingClientRect();return s.display!=='none'&&s.visibility!=='hidden'&&r.width>0&&r.height>0;}" "function rootOf(e){return e.closest?e.closest('.eds-toast,.eds-toasts,[role=alert],[class*=toast],[class*=Toast],[class*=message],[class*=Message],[class*=notice]')||e:e;}" "function record(e){var root=rootOf(e);var value=text(root)||text(e);if(!value)return null;return {text:value,html:((root.outerHTML||e.outerHTML)||'').slice(0,2000),url:location.href,visible:visible(root)||visible(e),created_at:(new Date()).toISOString()};}" "var cached=(window.__cmshopee_toasts||[]).slice(-10);" "var current=Array.from(document.querySelectorAll('.eds-toasts .eds-toast,.eds-toast,.eds-toast__content,[role=alert],[class*=toast],[class*=Toast],[class*=message],[class*=Message],[class*=notice]')).map(record).filter(Boolean);" "var positions={};var out=[];" "cached.forEach(function(item){var copy=Object.assign({},item,{visible:false});var key=[copy.text,copy.url].join('|');" "if(positions[key]!==undefined){out[positions[key]]=copy;return;}positions[key]=out.length;out.push(copy);});" "current.forEach(function(item){var key=[item.text,item.url].join('|');" "if(positions[key]!==undefined){out[positions[key]]=item;return;}positions[key]=out.length;out.push(item);});" "return JSON.stringify(out.slice(-10));" "})()" ) 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(){" + JS_EDITOR_FIELD_HELPERS + "var image=cmMainImageContext();var items=cmMainImageItems(image);" "if(!items.length)return JSON.stringify({clicked:false,reason:'NO_IMAGE'});" "var item=items[0];" "['mouseover','mouseenter','mousemove'].forEach(function(n){" "item.dispatchEvent(new MouseEvent(n,{bubbles:true,view:window}));});" "var selectors=['.shopee-image-manager__icon--delete','[class*=\"icon--delete\"]','[class*=\"delete\"]'];" "var btn=null;" "for(var i=0;i0&&r.height>0&&s.visibility!=='hidden'&&s.display!=='none';}" "var rootSelectors='[role=dialog],[class*=dialog],[class*=Dialog],[class*=modal]," "[class*=Modal],[class*=popover],[class*=Popover],[class*=popup],[class*=Popup]';" "var roots=[].slice.call(document.querySelectorAll(rootSelectors)).filter(visible);" "var yes=/(删除|刪除|確認|确认|確定|确定|OK|Yes)/i;" "var no=/(取消|cancel|否|No)/i;" "var seen=[];var candidates=[];" "roots.forEach(function(root){[].slice.call(root.querySelectorAll('button,[role=button]')).forEach(function(b){" "var text=((b.innerText||b.textContent||b.getAttribute('aria-label')||'')+'').trim();" "if(!text||!visible(b))return;seen.push(text);" "if(b.disabled||/disabled/i.test(b.className)||no.test(text)||!yes.test(text))return;" "candidates.push({button:b,text:text,score:/(删除|刪除)/i.test(text)?0:1});});});" "if(!candidates.length)return JSON.stringify({clicked:false,reason:'NO_CONFIRM_BUTTON',buttons:seen.slice(0,8)});" "candidates.sort(function(a,b){return a.score-b.score;});" "candidates[0].button.click();" "return JSON.stringify({clicked:true,reason:null,text:candidates[0].text});})()" ) class EditorError(RuntimeError): """Raised for expected editor automation failures with user-readable messages.""" def _get(obj, *names, default=None): for name in names: if isinstance(obj, dict) and name in obj: return obj[name] if hasattr(obj, name): return getattr(obj, name) return default def _cdp_host(account=None): explicit = _get(account, "cdp_host", "debug_host") if explicit: return str(explicit).replace("http://", "").replace("https://", "").rstrip("/") port = _get(account, "debug_port", "port") if port: return f"127.0.0.1:{port}" return os.environ.get("CDP_HOST", "127.0.0.1:9222") def _region_host(account=None): value = _get(account, "region_host", "host", default=DEFAULT_REGION_HOST) value = str(value or DEFAULT_REGION_HOST).strip() if "://" in value: value = urlparse(value).netloc return value.strip("/") or DEFAULT_REGION_HOST def _seller_home_url(account): return f"https://{_region_host(account)}/" def _product_url(account, item_id): return ( f"https://{_region_host(account)}/portal/product/{item_id}" "?pageEntry=product_list&ignore-html-cache=1" ) def _find_exact_product_tab(account, item_id, host): """只返回该账号已打开且商品路径精确匹配的 target。""" expected = urlparse(_product_url(account, item_id)) expected_host = expected.netloc.lower() expected_path = expected.path.rstrip("/") for target in http_get("/json", host=host): if target.get("type") != "page" or not target.get("webSocketDebuggerUrl"): continue try: current = urlparse(str(target.get("url") or "")) except Exception: continue if ( current.netloc.lower() == expected_host and current.path.rstrip("/") == expected_path ): return target return None def _item_id(task): value = _get(task, "item_id", "itemid", "product_id") if not value: raise EditorError("缺少商品 id，无法打开商品页") return str(value) def _image_root(account): alias = _get(account, "slug", "alias", "account_name", default="default") safe = "".join(ch.lower() if ch.isalnum() else "_" for ch in str(alias)).strip("_") return os.path.join("images", safe or "default") def _is_login_url(url): value = (url or "").lower() try: parsed = urlparse(value) if parsed.netloc.startswith("accounts.shopee.") and parsed.path.startswith("/seller/login"): return True except Exception: pass return any(marker in value for marker in LOGIN_PATH_MARKERS) def _json_value(cdp, expr, default=None): raw = cdp.val(expr) if raw is None or raw == "": return default if isinstance(raw, (dict, list)): return raw return json.loads(raw) PRODUCT_UNAVAILABLE_TEXT_MARKERS = ( "please input correct product id", "商品失效", "商品不存在", "商品已删除", "商品已下架", "無效商品", "无效商品", "無效的商品", "无效的商品", "無權限", "无权限", "沒有權限", "没有权限", "無法查看商品", "无法查看商品", "product not found", "item not found", "not exist", "not found", "invalid product", "invalid item", ) def install_toast_observer(cdp) -> None: try: cdp.send("Page.addScriptToEvaluateOnNewDocument", {"source": JS_TOAST_OBSERVER_SOURCE}) except Exception: pass try: cdp.val(JS_INSTALL_TOAST_OBSERVER) except Exception: pass def read_page_toasts(cdp): try: values = _json_value(cdp, JS_PAGE_TOASTS, default=[]) except Exception: return [] if not isinstance(values, list): return [] toasts = [] for entry in values: if isinstance(entry, str): text = entry.strip() if text: toasts.append({"text": text, "html": "", "url": "", "visible": False, "created_at": ""}) continue if not isinstance(entry, dict): continue text = str(entry.get("text") or "").strip() if not text: continue toasts.append( { "text": text, "html": str(entry.get("html") or "")[:2000], "url": str(entry.get("url") or ""), "visible": bool(entry.get("visible")), "created_at": str(entry.get("created_at") or ""), } ) return toasts[-10:] def is_product_unavailable_error(text): value = str(text or "").strip().lower() return bool(value) and any(marker.lower() in value for marker in PRODUCT_UNAVAILABLE_TEXT_MARKERS) def product_unavailable_error_message(text): value = str(text or "").strip() if value.startswith("商品失效"): return value return f"商品失效：{value or '商品详情页无法加载'}" def _product_unavailable_toast_text(toasts): for toast in reversed(toasts or []): text = str(toast.get("text") or "").strip() if is_product_unavailable_error(text): return text return None def _same_page_url(left, right): try: left_url = urlparse(str(left or "")) right_url = urlparse(str(right or "")) except Exception: return False return bool( left_url.netloc and left_url.netloc == right_url.netloc and left_url.path.rstrip("/") == right_url.path.rstrip("/") ) def _open_product_ready_error(cdp, fallback, current_url=""): toasts = read_page_toasts(cdp) invalid_text = _product_unavailable_toast_text(toasts) if invalid_text: return product_unavailable_error_message(invalid_text) page_url = str(current_url or _current_url(cdp) or "") for toast in reversed(toasts): text = str(toast.get("text") or "").strip() if text and toast.get("visible") and _same_page_url(toast.get("url"), page_url): return f"{fallback}；页面提示（可能无关）：{text}" return fallback def _title_state(cdp): return _json_value(cdp, JS_TITLE_STATE, default={}) or {} def _ready_state(cdp): raw = cdp.val(JS_READY) if isinstance(raw, bool): return {"ready": raw} if isinstance(raw, dict): return raw if raw in (None, ""): return {} parsed = json.loads(raw) return parsed if isinstance(parsed, dict) else {} def _ready_error_message(fallback, state): if not state: return fallback details = [] title_count = state.get("title_count") if title_count == 1: details.append("商品名称输入框正常") elif isinstance(title_count, int) and title_count > 1: details.append(f"商品名称输入框匹配到{title_count}个，无法确定唯一字段") elif title_count is not None: details.append("未识别商品名称输入框") image_count = state.get("image_count") if not state.get("image_manager_exists") and "image_manager_exists" in state: details.append("未识别商品主图区域") elif isinstance(image_count, int) and image_count > 0: details.append(f"商品图片{image_count}张") elif image_count is not None: details.append("商品图片尚未加载") upload_count = state.get("upload_input_count") if upload_count == 1: details.append("主图上传入口正常") elif isinstance(upload_count, int) and upload_count > 1: details.append(f"主图上传入口匹配到{upload_count}个，无法确定唯一入口") elif "upload_input_exists" in state or upload_count is not None: details.append("主图上传入口尚未加载") current_url = str(state.get("current_url") or "") if current_url: try: parsed = urlparse(current_url) current_url = f"{parsed.scheme}://{parsed.netloc}{parsed.path}" except Exception: pass details.append(f"当前页面：{current_url}") return f"{fallback}：{'；'.join(details)}" if details else fallback 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 _click_upload_tile(cdp): return _json_value( cdp, JS_CLICK_UPLOAD_TILE, default={"clicked": False, "reason": "NO_UPLOAD_TILE"}, ) or {"clicked": False, "reason": "NO_UPLOAD_TILE"} 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 _has_duplicate_upload_error(state): texts = [] for key in ("errors", "error_toasts", "toasts", "page_error_toasts"): values = (state or {}).get(key) or [] texts.extend(str(value) for value in values) return any( token in text.lower() for text in texts for token in ("重複", "重复", "duplicate") ) def _cover_upload_error_message(result): reason = result.get("reason") or "COVER_UPDATE_FAILED" state = result.get("upload_state") or {} if reason == "UPLOAD_DUPLICATE_IMAGE": details = state.get("errors") or state.get("error_toasts") or state.get("toasts") or state.get("page_error_toasts") or [] return "新封面与现有商品图片重复：" + "；".join(map(str, details[:3])) if details else "新封面与现有商品图片重复" 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 "新封面上传仍在处理中，未取得蝦皮 CDN 地址" if reason == "UPLOAD_TIMEOUT": return "新封面上传超时，未取得蝦皮 CDN 地址" if reason == "UPLOAD_TILE_NOT_READY": return "新封面上传入口未可点击，未开始上传新封面" 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 last_state = {} first_probe = True while first_probe or time.time() < end: first_probe = False try: last_state = _ready_state(cdp) if last_state.get("ready"): cdp.val(JS_SCROLL_MAIN_IMAGE_MANAGER) time.sleep(0.5) return True except Exception: pass current_url = str(last_state.get("current_url") or _current_url(cdp) or "") if _is_login_url(current_url): raise EditorError("账号未登录，请先到④账号管理人工登录蝦皮") invalid_text = _product_unavailable_toast_text(read_page_toasts(cdp)) if invalid_text: raise EditorError(product_unavailable_error_message(invalid_text)) if time.time() >= end: break time.sleep(1) fallback = _ready_error_message("等待蝦皮商品编辑器就绪超时", last_state) raise EditorError( _open_product_ready_error( cdp, fallback, current_url=last_state.get("current_url") if last_state else "", ) ) def _ensure_page_domains(cdp): for domain in ("Page", "Runtime", "DOM", "Network"): try: cdp.send(f"{domain}.enable") except Exception: pass def _current_url(cdp): try: return cdp.val("location.href") or "" except Exception: return "" def _wait_for_login_probe(cdp, timeout=8, initial_url=""): deadline = time.monotonic() + max(0.0, float(timeout)) last_url = str(initial_url or "") cookie_names = set() cookie_read_succeeded = False probe_error = None first_probe = True while first_probe or time.monotonic() < deadline: first_probe = False try: last_url = cdp.val("location.href") or last_url except Exception as exc: return { "url": last_url, "cookie_names": cookie_names, "cookie_read_succeeded": cookie_read_succeeded, "target_available": False, "probe_error": exc.__class__.__name__, } if _is_login_url(last_url): return { "url": last_url, "cookie_names": cookie_names, "cookie_read_succeeded": cookie_read_succeeded, "target_available": True, "probe_error": probe_error, } try: cookies = cdp.send("Network.getAllCookies").get("cookies", []) except Exception as exc: return { "url": last_url, "cookie_names": cookie_names, "cookie_read_succeeded": cookie_read_succeeded, "target_available": False, "probe_error": exc.__class__.__name__, } cookie_read_succeeded = True cookie_names = { cookie.get("name") for cookie in cookies if "shopee" in (cookie.get("domain") or "") } if "SPC_ST" in cookie_names or "SPC_U" in cookie_names: break remaining = deadline - time.monotonic() if remaining <= 0: break time.sleep(min(0.5, remaining)) return { "url": last_url, "cookie_names": cookie_names, "cookie_read_succeeded": cookie_read_succeeded, "target_available": cookie_read_succeeded, "probe_error": probe_error, } def _login_target_priority(target): url = str((target or {}).get("url") or "").lower() if _is_login_url(url): return 3 if "/portal/product/" in url: return 2 if "/portal/" in url: return 0 return 1 def _login_target_key(target): target = target or {} return str(target.get("webSocketDebuggerUrl") or target.get("id") or target.get("url") or "") def _shopee_page_targets(host): pages = [ target for target in http_get("/json", host=host) if target.get("type") == "page" and "shopee" in str(target.get("url") or "").lower() and target.get("webSocketDebuggerUrl") ] return sorted(pages, key=_login_target_priority) def login_status(account, timeout=8) -> dict: """Return detailed Shopee login status for an account's CDP session.""" host = _cdp_host(account) deadline = time.monotonic() + max(0.0, float(timeout)) attempted_targets = set() opened_home = False probe_attempts = 0 last_probe = { "url": "", "cookie_names": set(), "cookie_read_succeeded": False, "target_available": False, "probe_error": None, } while True: candidates = [ target for target in _shopee_page_targets(host) if _login_target_key(target) not in attempted_targets ] if not candidates and not opened_home: home_url = _seller_home_url(account) ws = create_tab(home_url, host=host) candidates = [ { "type": "page", "url": home_url, "webSocketDebuggerUrl": ws, } ] opened_home = True if not candidates: break for target in candidates: key = _login_target_key(target) if key: attempted_targets.add(key) probe_attempts += 1 initial_url = str(target.get("url") or "") if _is_login_url(initial_url): return { "logged_in": False, "reason": "LOGIN_PAGE", "url": initial_url, "host": host, "cookie_names": [], "cookie_read_succeeded": False, "probe_error": None, "probe_attempts": probe_attempts, } cdp = None try: cdp = CDP(target["webSocketDebuggerUrl"]) _ensure_page_domains(cdp) remaining = max(0.0, deadline - time.monotonic()) probe = _wait_for_login_probe( cdp, timeout=remaining, initial_url=initial_url, ) except Exception as exc: probe = { "url": initial_url, "cookie_names": set(), "cookie_read_succeeded": False, "target_available": False, "probe_error": exc.__class__.__name__, } finally: if cdp is not None: cdp.close() last_probe = probe url = str(probe.get("url") or initial_url) names = set(probe.get("cookie_names") or []) if _is_login_url(url): return { "logged_in": False, "reason": "LOGIN_PAGE", "url": url, "host": host, "cookie_names": sorted(name for name in names if name), "cookie_read_succeeded": bool(probe.get("cookie_read_succeeded")), "probe_error": probe.get("probe_error"), "probe_attempts": probe_attempts, } if not probe.get("cookie_read_succeeded") or not probe.get( "target_available" ): continue logged_in = "SPC_ST" in names or "SPC_U" in names return { "logged_in": logged_in, "reason": None if logged_in else "NO_SESSION_COOKIE", "url": url, "host": host, "cookie_names": sorted(name for name in names if name), "cookie_read_succeeded": True, "probe_error": probe.get("probe_error"), "probe_attempts": probe_attempts, } if time.monotonic() >= deadline: break return { "logged_in": False, "reason": "LOGIN_CHECK_TARGET_UNAVAILABLE", "url": str(last_probe.get("url") or ""), "host": host, "cookie_names": sorted( name for name in (last_probe.get("cookie_names") or []) if name ), "cookie_read_succeeded": False, "probe_error": last_probe.get("probe_error"), "probe_attempts": probe_attempts, } def is_logged_in(account) -> bool: """Return whether the account's current Shopee session appears logged in.""" return bool(login_status(account).get("logged_in")) def _close_open_product_failure(cdp, confirm_target_closed=False): target_id = getattr(cdp, "target_id", None) created_by_app = bool(getattr(cdp, "created_by_app", False)) host = getattr(cdp, "cdp_host", None) try: cdp.close() finally: if created_by_app and target_id: try: if confirm_target_closed: close_tab_and_wait(target_id, host=host, timeout=2.0) else: close_tab(target_id, host=host) except Exception: pass def open_product(account, item_id, on_step=None, bring_to_front=True) -> CDP: """Open or reuse a product edit tab, navigate to a clean edit URL, and wait ready.""" _notify_collect_step(on_step, "open_product") host = _cdp_host(account) item_id = str(item_id) url = _product_url(account, item_id) tab = find_product_tab(item_id, host=host) created_by_app = tab is None if tab is None: tab = create_tab_info(url, host=host, background=not bring_to_front) ws = tab["webSocketDebuggerUrl"] cdp = CDP(ws) cdp.target_id = tab.get("id") cdp.created_by_app = created_by_app cdp.cdp_host = host try: _ensure_page_domains(cdp) install_toast_observer(cdp) if bring_to_front: try: cdp.send("Page.bringToFront") except Exception: pass cdp.send("Page.navigate", {"url": url}) install_toast_observer(cdp) _notify_collect_step(on_step, "wait_ready") _wait_ready(cdp) return cdp except Exception: _close_open_product_failure(cdp, confirm_target_closed=not bring_to_front) raise def open_or_focus_product_tab(account, item_id): """聚焦现有精确商品页，或新建商品页且不刷新用户已有页面。 此方法不能复用 ``open_product``：真实更新需要重新导航，人工查看不能 刷新用户正在编辑、尚未保存的商品页。 """ host = _cdp_host(account) item_id = str(item_id) url = _product_url(account, item_id) target = _find_exact_product_tab(account, item_id, host) created_by_app = target is None if target is None: target = create_tab_info(url, host=host, background=False) cdp = CDP(target["webSocketDebuggerUrl"]) cdp.target_id = target.get("id") cdp.created_by_app = created_by_app cdp.cdp_host = host try: _ensure_page_domains(cdp) if created_by_app: install_toast_observer(cdp) cdp.send("Page.bringToFront") if created_by_app: _wait_ready(cdp) return { "item_id": item_id, "created": created_by_app, "target_id": target.get("id"), } except Exception: if created_by_app: _close_open_product_failure(cdp) raise finally: cdp.close() def read_title(cdp) -> str: """Read the current Shopee title input value.""" state = _title_state(cdp) return state.get("value") or state.get("modelvalue") or "" def read_cover_src(cdp) -> str: """Read the first product image URL, which is the current cover.""" return cdp.val(JS_FIRST_COVER) or "" def read_product_image_urls(cdp): """Read all current Shopee product image URLs in page order.""" images = [] for position, rect in enumerate(_image_rects(cdp), start=1): src = str((rect or {}).get("src") or "").strip() if not src: continue try: index = int((rect or {}).get("i", position - 1)) + 1 except (TypeError, ValueError): index = position images.append({"index": index, "src": src}) return images def download_cover(src, out_path) -> str: """Download a cover image to a local path and return the absolute path.""" if not src: raise EditorError("旧封面链接为空，无法下载") import requests out_path = os.path.abspath(str(out_path)) root, ext = os.path.splitext(out_path) if not ext: path = urlparse(src).path _, guessed = os.path.splitext(path) out_path = root + (guessed if guessed and len(guessed) <= 5 else ".jpg") os.makedirs(os.path.dirname(out_path), exist_ok=True) session = requests.Session() session.trust_env = False resp = session.get(src, timeout=30) resp.raise_for_status() with open(out_path, "wb") as fh: fh.write(resp.content) return out_path def collect(account, task, on_step=None) -> dict: """Collect current title and cover snapshot before any edits.""" item_id = _item_id(task) cdp = open_product(account, item_id, on_step=on_step, bring_to_front=False) result = None try: _notify_collect_step(on_step, "read_product_status") status_snapshot = read_product_status(cdp) collect_scope = product_status.normalize_collect_scope( _get(task, "collection_scope") ) if not product_status.should_collect_content( status_snapshot.get("product_status"), collect_scope, ): result = { **status_snapshot, "collection_skipped": True, "collection_skip_reason": product_status.collect_skip_reason( status_snapshot.get("product_status") ), } else: _notify_collect_step(on_step, "read_title") old_title = read_title(cdp) _notify_collect_step(on_step, "read_cover") old_cover_src = read_cover_src(cdp) out_path = _get(task, "old_cover_path") if not out_path: out_path = image_paths.task_image_path( appconfig.image_dir(), task, account, "old", ) _notify_collect_step(on_step, "download_cover") old_cover_path = download_cover(old_cover_src, out_path) result = { "old_title": old_title, "old_cover_src": old_cover_src, "old_cover_path": old_cover_path, **status_snapshot, } finally: close_target_confirmed = _close_collected_product(cdp) result["close_target_confirmed"] = close_target_confirmed return result def read_product_status(cdp) -> dict: """Read the current product's warning state without retaining page HTML.""" try: alerts = _json_value(cdp, JS_PRODUCT_STATUS_ALERTS, default=None) if not isinstance(alerts, list): raise ValueError("商品状态提示读取结果无效") snapshot = product_status.classify_alerts(alerts) snapshot["product_status_error"] = None return snapshot except Exception as exc: snapshot = product_status.classify_alerts(None) snapshot["product_status_error"] = str(exc) or exc.__class__.__name__ return snapshot def _notify_collect_step(callback, step): if callback is None: return try: callback(step) except Exception: pass def _notify_apply_step(callback, step, result="start", detail=None): if callback is None: return payload = {"step": step, "result": result} if detail: payload["detail"] = str(detail) try: callback(payload) except Exception: pass def _close_collected_product(cdp): target_id = getattr(cdp, "target_id", None) created_by_app = bool(getattr(cdp, "created_by_app", False)) host = getattr(cdp, "cdp_host", None) close_target_confirmed = None try: cdp.close() finally: if created_by_app and target_id: try: close_target_confirmed = close_tab_and_wait( target_id, host=host, timeout=2.0, ) except Exception: close_target_confirmed = False return close_target_confirmed def close_readonly_product(cdp): """Close a read-only product CDP session using the collection tab cleanup rules.""" return _close_collected_product(cdp) def change_title(cdp, new_title) -> dict: """Write a new title with the verified native setter + input/change events.""" expr = ( "(function(){" + JS_EDITOR_FIELD_HELPERS + "var title=cmTitleCandidate();var el=title.node;" "if(!el)return title.count>1?'AMBIGUOUS_INPUT':'NO_INPUT';" "var s=Object.getOwnPropertyDescriptor(window.HTMLInputElement.prototype,'value').set;" f"el.focus();s.call(el,{json.dumps(str(new_title))});" "el.dispatchEvent(new Event('input',{bubbles:true}));" "el.dispatchEvent(new Event('change',{bubbles:true}));el.blur();return el.value;})()" ) written = cdp.val(expr) time.sleep(0.4) state = _title_state(cdp) value = state.get("value") modelvalue = state.get("modelvalue") ok = value == str(new_title) and modelvalue == str(new_title) return { "ok": ok, "written": written, "value": value, "modelvalue": modelvalue, "candidate_count": state.get("candidate_count"), "source": state.get("source"), } def replace_cover(cdp, image_win_path, old_cover_path=None, timeout=180) -> dict: """Upload an image and drag it to the first position. Cover replacement always deletes the current first Shopee image first, and only proceeds when the old cover backup from the collect stage exists. """ image_win_path = os.path.abspath(str(image_win_path)) if not os.path.exists(image_win_path): raise FileNotFoundError(f"封面图片不存在: {image_win_path}") cdp.val(JS_SCROLL_MAIN_IMAGE_MANAGER) time.sleep(0.5) 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) backup_path = _validated_old_cover_backup(old_cover_path) if not backup_path: return { "ok": False, "reason": "OLD_COVER_BACKUP_MISSING", "count_before": count_before, "old_cover_path": old_cover_path, } delete_result = _delete_first_cover(cdp, before) if not delete_result.get("ok"): return { "ok": False, "reason": delete_result.get("reason"), "count_before": count_before, "delete": delete_result, } 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) 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 { "ok": False, "reason": "UPLOAD_TILE_NOT_READY", "count_before": count_before, "delete": delete_result, "upload_click": upload_click, "upload_state": _upload_state(cdp), } time.sleep(1) oid = cdp.object_id(JS_MAIN_UPLOAD_INPUT) if not oid: return { "ok": False, "reason": "NO_UPLOAD_INPUT", "count_before": count_before, "delete": delete_result, "upload_click": upload_click, "upload_state": _upload_state(cdp), } cdp.send("DOM.setFileInputFiles", {"objectId": oid, "files": [image_win_path]}) changed = cdp.val(JS_DISPATCH_MAIN_UPLOAD_INPUT_EVENTS) 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_click": upload_click, "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: 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) 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, "count_before": count_before, "count_after": len(cur), "delete": delete_result, "upload_click": upload_click, "upload_state": last_state, "file_size": file_size, } 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 and (len(cur) > len(before) or len(ready) == 1): new_src = ready[-1]["src"] break 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_click": upload_click, "upload_state": last_state, "file_size": file_size, } time.sleep(1.5) if not new_src: 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_click": upload_click, "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, "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) 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", "new_src": new_src, "index": index, "count_before": count_before, "count_after": len(after), "delete": delete_result, "upload_click": upload_click, } 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 path = os.path.abspath(str(old_cover_path)) if not os.path.exists(path): return None return path def _delete_first_cover(cdp, before, timeout=15): count_before = len(before) if not before: return {"ok": False, "reason": "NO_IMAGE", "count_before": 0} first = before[0] try: cdp.send( "Input.dispatchMouseEvent", {"type": "mouseMoved", "x": first["x"], "y": first["y"]}, ) time.sleep(0.2) except Exception: pass click_result = _json_value( cdp, JS_CLICK_FIRST_DELETE, default={"clicked": False, "reason": "NO_DELETE_BUTTON"}, ) or {} if not click_result.get("clicked"): return { "ok": False, "reason": click_result.get("reason") or "NO_DELETE_BUTTON", "count_before": count_before, "click": click_result, } confirm_result = None end = time.time() + timeout while time.time() < end: cur = _image_rects(cdp) if len(cur) < count_before: return { "ok": True, "reason": None, "count_before": count_before, "count_after": len(cur), "click": click_result, "confirm": confirm_result, } confirm_attempt = _json_value( cdp, JS_CLICK_DELETE_CONFIRM, default={"clicked": False, "reason": "NO_CONFIRM_BUTTON"}, ) or {} if confirm_attempt.get("clicked"): confirm_result = confirm_attempt time.sleep(0.5) cur = _image_rects(cdp) return { "ok": False, "reason": "DELETE_TIMEOUT", "count_before": count_before, "count_after": len(cur), "click": click_result, "confirm": confirm_result, } 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"): return {"clicked": False, "reason": "NO_UPDATE_BUTTON", "toasts": []} if info.get("disabled"): return {"clicked": False, "reason": "UPDATE_DISABLED", "toasts": []} result = cdp.val(JS_CLICK_UPDATE) 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": 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, on_step=None, bring_to_front=True, update_mode=None) -> dict: """Apply generated title/cover to Shopee. The caller must perform the batch confirmation before calling this function. """ item_id = _item_id(task) current_step = "open_product" _notify_apply_step(on_step, current_step, "start") cdp = open_product(account, item_id, bring_to_front=bring_to_front) _notify_apply_step(on_step, current_step, "success") committed = False try: title_result = None cover_result = None mode = appconfig.normalize_update_mode( update_mode, allow_cover_update=bool(_get(task, "new_cover_path")), ) new_title = _get(task, "new_title") new_cover_path = _get(task, "new_cover_path") if not appconfig.update_mode_includes_title(mode): new_title = None if not appconfig.update_mode_includes_cover(mode): new_cover_path = None if new_title: current_step = "change_title" _notify_apply_step(on_step, current_step, "start") title_result = change_title(cdp, new_title) if not title_result.get("ok"): _notify_apply_step(on_step, current_step, "failed", "标题写入后 value/modelvalue 未同步") return {"committed": False, "error": "标题写入后 value/modelvalue 未同步", "title": title_result} _notify_apply_step(on_step, current_step, "success") if new_cover_path: current_step = "replace_cover" _notify_apply_step(on_step, current_step, "start") cover_result = replace_cover( cdp, 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, "replace_cover", "success") current_step = "click_update" _notify_apply_step(on_step, current_step, "start") update_result = click_update(cdp) committed = bool(update_result.get("clicked", False)) _notify_apply_step( on_step, current_step, "success" if committed else "failed", update_result.get("reason"), ) return { "committed": committed, "error": update_result.get("reason"), "title": title_result, "cover": cover_result, "update": update_result, } except Exception as exc: _notify_apply_step(on_step, current_step, "failed", str(exc)) return {"committed": False, "error": str(exc)} finally: _close_applied_product(cdp, committed=committed) def _close_applied_product(cdp, committed=False): target_id = getattr(cdp, "target_id", None) created_by_app = bool(getattr(cdp, "created_by_app", False)) host = getattr(cdp, "cdp_host", None) should_close_tab = bool(created_by_app and target_id) should_wait_before_close = bool(committed) try: cdp.close() finally: if should_close_tab: try: if should_wait_before_close: time.sleep(2) close_tab(target_id, host=host) except Exception: pass