1531 lines
61 KiB
Python
1531 lines
61 KiB
Python
"""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
|
||
from .cdp import CDP, close_tab, create_tab, create_tab_info, find_product_tab, http_get
|
||
|
||
|
||
DEFAULT_REGION_HOST = "seller.shopee.tw"
|
||
TITLE_XPATH = "//input[@class='eds-input__input' and string-length(@modelvalue)>24]"
|
||
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_READY = (
|
||
"(function(){"
|
||
f"var s=document.evaluate({json.dumps(ITEMBOX_XPATH)},document,null,"
|
||
"XPathResult.ORDERED_NODE_SNAPSHOT_TYPE,null);"
|
||
f"var r=document.evaluate({json.dumps(TITLE_XPATH)},document,null,"
|
||
"XPathResult.FIRST_ORDERED_NODE_TYPE,null);"
|
||
"var up=document.querySelector('.shopee-image-manager__upload input[type=file]');"
|
||
"return (s.snapshotLength>0 && !!r.singleNodeValue && !!up);})()"
|
||
)
|
||
|
||
JS_RECTS = (
|
||
"(function(){"
|
||
f"var s=document.evaluate({json.dumps(ITEMBOX_XPATH)},document,null,"
|
||
"XPathResult.ORDERED_NODE_SNAPSHOT_TYPE,null);"
|
||
"var a=[];for(var i=0;i<s.snapshotLength;i++){"
|
||
"var el=s.snapshotItem(i);var r=el.getBoundingClientRect();var im=el.querySelector('img');"
|
||
"a.push({i:i,x:r.left+r.width/2,y:r.top+r.height/2,left:r.left,top:r.top,"
|
||
"w:r.width,h:r.height,src:im?im.src:null});}"
|
||
"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=/(失敗|失败|錯誤|错误|不支援|不支持|格式|大小|尺寸|超過|超过|重複|重复|duplicate|error|fail|invalid|unsupported)/i;"
|
||
"var uploadRe=/(圖片|图片|封面|照片|相片|圖像|图像|image|photo|cover|upload|上傳|上传|檔案|文件|file|格式|大小|尺寸|像素|解析度|分辨率|超過|超过|重複|重复|duplicate)/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_CLICK_UPLOAD_TILE = (
|
||
"(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';}"
|
||
"var manager=document.querySelector('.shopee-image-manager');"
|
||
"var box=manager?manager.querySelector('.shopee-image-manager__upload,[class*=image-manager__upload]'):null;"
|
||
"if(!box)return JSON.stringify({clicked:false,reason:'NO_UPLOAD_TILE'});"
|
||
"if(!visible(box))return JSON.stringify({clicked:false,reason:'UPLOAD_TILE_HIDDEN'});"
|
||
"box.scrollIntoView({block:'center',inline:'center'});"
|
||
"box.click();"
|
||
"return JSON.stringify({clicked:true,reason:null});"
|
||
"})()"
|
||
)
|
||
JS_TITLE_STATE = (
|
||
"(function(){"
|
||
f"var r=document.evaluate({json.dumps(TITLE_XPATH)},document,null,"
|
||
"XPathResult.FIRST_ORDERED_NODE_TYPE,null);"
|
||
"var el=r.singleNodeValue;if(!el)return null;"
|
||
"return JSON.stringify({value:el.value,modelvalue:el.getAttribute('modelvalue')});})()"
|
||
)
|
||
|
||
JS_FIRST_COVER = (
|
||
"(function(){"
|
||
f"var s=document.evaluate({json.dumps(ITEMBOX_XPATH)},document,null,"
|
||
"XPathResult.ORDERED_NODE_SNAPSHOT_TYPE,null);"
|
||
"if(!s.snapshotLength)return null;"
|
||
"var img=s.snapshotItem(0).querySelector('img');return img?img.src:null;})()"
|
||
)
|
||
|
||
JS_FIND_UPDATE = (
|
||
"(function(){var bs=[].slice.call(document.querySelectorAll('button.eds-button'));"
|
||
"var b=bs.find(function(b){var sp=b.querySelector('span');"
|
||
"return (sp?sp.innerText:(b.innerText||'')).trim()==='更新' && b.offsetParent!==null;});"
|
||
"if(!b)return JSON.stringify({found:false});"
|
||
"var dis=b.disabled||/disabled/i.test(b.className);"
|
||
"return JSON.stringify({found:true,disabled:dis});})()"
|
||
)
|
||
|
||
JS_CLICK_UPDATE = (
|
||
"(function(){var bs=[].slice.call(document.querySelectorAll('button.eds-button'));"
|
||
"var b=bs.find(function(b){var sp=b.querySelector('span');"
|
||
"return (sp?sp.innerText:(b.innerText||'')).trim()==='更新' && b.offsetParent!==null;});"
|
||
"if(!b)return 'NO_BTN';if(b.disabled||/disabled/i.test(b.className))return 'DISABLED';"
|
||
"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]'));"
|
||
"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 seen={};var out=[];cached.concat(current).forEach(function(item){var key=[item.text,item.html,item.url].join('|');if(seen[key])return;seen[key]=true;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(){"
|
||
f"var s=document.evaluate({json.dumps(ITEMBOX_XPATH)},document,null,"
|
||
"XPathResult.ORDERED_NODE_SNAPSHOT_TYPE,null);"
|
||
"if(!s.snapshotLength)return JSON.stringify({clicked:false,reason:'NO_IMAGE'});"
|
||
"var item=s.snapshotItem(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;i<selectors.length&&!btn;i++){btn=item.querySelector(selectors[i]);}"
|
||
"if(!btn){var all=[].slice.call(item.querySelectorAll('*'));"
|
||
"btn=all.find(function(e){var t=[e.className,e.innerText,e.title,e.getAttribute('aria-label')].join(' ');"
|
||
"return /(delete|删除|刪除)/i.test(t);});}"
|
||
"if(!btn)return JSON.stringify({clicked:false,reason:'NO_DELETE_BUTTON'});"
|
||
"if(btn.scrollIntoView)btn.scrollIntoView({block:'center'});"
|
||
"btn.click();return JSON.stringify({clicked:true,reason:null});})()"
|
||
)
|
||
|
||
JS_CLICK_DELETE_CONFIRM = (
|
||
"(function(){"
|
||
"function visible(e){var r=e.getBoundingClientRect();var s=getComputedStyle(e);"
|
||
"return r.width>0&&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 _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 _open_product_ready_error(cdp, fallback):
|
||
toasts = read_page_toasts(cdp)
|
||
invalid_text = _product_unavailable_toast_text(toasts)
|
||
if invalid_text:
|
||
return product_unavailable_error_message(invalid_text)
|
||
for toast in reversed(toasts):
|
||
text = str(toast.get("text") or "").strip()
|
||
if text:
|
||
return f"{fallback}：{text}"
|
||
return fallback
|
||
|
||
|
||
def _title_state(cdp):
|
||
return _json_value(cdp, JS_TITLE_STATE, default={}) or {}
|
||
|
||
|
||
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
|
||
while time.time() < end:
|
||
try:
|
||
if cdp.val(JS_READY):
|
||
cdp.val(
|
||
"(function(){var m=document.querySelector('.shopee-image-manager');"
|
||
"if(m)m.scrollIntoView({block:'center'});return 1;})()"
|
||
)
|
||
time.sleep(0.5)
|
||
return True
|
||
except Exception:
|
||
pass
|
||
invalid_text = _product_unavailable_toast_text(read_page_toasts(cdp))
|
||
if invalid_text:
|
||
raise EditorError(product_unavailable_error_message(invalid_text))
|
||
time.sleep(1)
|
||
raise EditorError(_open_product_ready_error(cdp, "等待蝦皮商品编辑器就绪超时"))
|
||
|
||
|
||
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):
|
||
end = time.time() + timeout
|
||
last_url = ""
|
||
cookie_names = set()
|
||
while time.time() < end:
|
||
last_url = _current_url(cdp) or last_url
|
||
try:
|
||
cookies = cdp.send("Network.getAllCookies").get("cookies", [])
|
||
cookie_names = {
|
||
c.get("name")
|
||
for c in cookies
|
||
if "shopee" in (c.get("domain") or "")
|
||
}
|
||
except Exception:
|
||
cookie_names = set()
|
||
if _is_login_url(last_url) or "SPC_ST" in cookie_names or "SPC_U" in cookie_names:
|
||
break
|
||
time.sleep(0.5)
|
||
return last_url, cookie_names
|
||
|
||
|
||
def login_status(account, timeout=8) -> dict:
|
||
"""Return detailed Shopee login status for an account's CDP session."""
|
||
|
||
host = _cdp_host(account)
|
||
pages = [t for t in http_get("/json", host=host) if t.get("type") == "page"]
|
||
shopee_page = next((p for p in pages if "shopee" in (p.get("url") or "")), None)
|
||
if not shopee_page:
|
||
ws = create_tab(_seller_home_url(account), host=host)
|
||
initial_url = _seller_home_url(account)
|
||
cdp = CDP(ws)
|
||
else:
|
||
initial_url = shopee_page.get("url") or ""
|
||
if _is_login_url(initial_url):
|
||
return {
|
||
"logged_in": False,
|
||
"reason": "LOGIN_PAGE",
|
||
"url": initial_url,
|
||
"host": host,
|
||
"cookie_names": [],
|
||
}
|
||
cdp = CDP(shopee_page["webSocketDebuggerUrl"])
|
||
|
||
try:
|
||
_ensure_page_domains(cdp)
|
||
url, names = _wait_for_login_probe(cdp, timeout=timeout)
|
||
url = url or initial_url
|
||
if _is_login_url(url):
|
||
reason = "LOGIN_PAGE"
|
||
logged_in = False
|
||
elif "SPC_ST" in names or "SPC_U" in names:
|
||
reason = None
|
||
logged_in = True
|
||
else:
|
||
reason = "NO_SESSION_COOKIE"
|
||
logged_in = False
|
||
return {
|
||
"logged_in": logged_in,
|
||
"reason": reason,
|
||
"url": url,
|
||
"host": host,
|
||
"cookie_names": sorted(name for name in names if name),
|
||
}
|
||
finally:
|
||
cdp.close()
|
||
|
||
|
||
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):
|
||
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:
|
||
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)
|
||
raise
|
||
|
||
|
||
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 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)
|
||
try:
|
||
_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)
|
||
return {
|
||
"old_title": old_title,
|
||
"old_cover_src": old_cover_src,
|
||
"old_cover_path": old_cover_path,
|
||
}
|
||
finally:
|
||
_close_collected_product(cdp)
|
||
|
||
|
||
|
||
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)
|
||
try:
|
||
cdp.close()
|
||
finally:
|
||
if created_by_app and target_id:
|
||
try:
|
||
close_tab(target_id, host=host)
|
||
except Exception:
|
||
pass
|
||
|
||
|
||
def change_title(cdp, new_title) -> dict:
|
||
"""Write a new title with the verified native setter + input/change events."""
|
||
|
||
expr = (
|
||
"(function(){"
|
||
f"var r=document.evaluate({json.dumps(TITLE_XPATH)},document,null,"
|
||
"XPathResult.FIRST_ORDERED_NODE_TYPE,null);var el=r.singleNodeValue;"
|
||
"if(!el)return '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}
|
||
|
||
|
||
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(
|
||
"(function(){var m=document.querySelector('.shopee-image-manager');"
|
||
"if(m)m.scrollIntoView({block:'center'});return !!m;})()"
|
||
)
|
||
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("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,
|
||
"upload_click": upload_click,
|
||
"upload_state": _upload_state(cdp),
|
||
}
|
||
|
||
cdp.send("DOM.setFileInputFiles", {"objectId": oid, "files": [image_win_path]})
|
||
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_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
|