cmshoppe/app/editor.py

"""Shopee editor operations built on the verified CDP primitives."""

import json
import os
import time
from urllib.parse import urlparse

from . import 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']"
)

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 "新封面上传仍在处理中，未取得 Shopee CDN 地址"
    if reason == "UPLOAD_TIMEOUT":
        return "新封面上传超时，未取得 Shopee 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, "等待 Shopee 商品编辑器就绪超时"))


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) -> 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)
    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)
        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)
    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("images", 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}
    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,
        }

    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}
    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 _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) -> 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)
    _notify_apply_step(on_step, current_step, "success")
    committed = False
    try:
        title_result = None
        cover_result = None
        new_title = _get(task, "new_title")
        new_cover_path = _get(task, "new_cover_path")
        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"):
                error = _cover_upload_error_message(cover_result)
                _notify_apply_step(on_step, current_step, "failed", error)
                return {"committed": False, "error": error, "cover": cover_result}
            _notify_apply_step(on_step, current_step, "success")
        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, close_success_tab=close_success_tab and committed)


def _close_applied_product(cdp, close_success_tab=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 close_success_tab and created_by_app and target_id:
            try:
                time.sleep(2)
                close_tab(target_id, host=host)
            except Exception:
                pass