669 lines
23 KiB
Python
669 lines
23 KiB
Python
"""Shopee editor operations built on the verified CDP primitives."""
|
|
|
|
import json
|
|
import os
|
|
import time
|
|
from urllib.parse import urlparse
|
|
|
|
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",
|
|
"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_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_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_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()
|
|
return any(marker in value for marker in LOGIN_PATH_MARKERS)
|
|
|
|
|
|
def _json_value(cdp, expr, default=None):
|
|
raw = cdp.val(expr)
|
|
if not raw:
|
|
return default
|
|
return json.loads(raw)
|
|
|
|
|
|
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 _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
|
|
time.sleep(1)
|
|
raise TimeoutError("等待 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 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
|
|
_ensure_page_domains(cdp)
|
|
try:
|
|
cdp.send("Page.bringToFront")
|
|
except Exception:
|
|
pass
|
|
cdp.send("Page.navigate", {"url": url})
|
|
_notify_collect_step(on_step, "wait_ready")
|
|
_wait_ready(cdp)
|
|
return cdp
|
|
|
|
|
|
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 = os.path.join(_image_root(account), f"{item_id}_old.jpg")
|
|
_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 _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=90) -> dict:
|
|
"""Upload an image and drag it to the first position.
|
|
|
|
When the image manager is full, only delete the current first image if the
|
|
old cover backup from the collect stage exists locally.
|
|
"""
|
|
|
|
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)
|
|
before = _image_rects(cdp)
|
|
count_before = len(before)
|
|
delete_result = None
|
|
if len(before) >= 9:
|
|
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,
|
|
}
|
|
before = _image_rects(cdp)
|
|
before_srcs = {r.get("src") for r in before}
|
|
oid = cdp.object_id("document.querySelector('.shopee-image-manager__upload input[type=file]')")
|
|
if not oid:
|
|
return {"ok": False, "reason": "NO_UPLOAD_INPUT", "count_before": count_before, "delete": delete_result}
|
|
|
|
cdp.send("DOM.setFileInputFiles", {"objectId": oid, "files": [image_win_path]})
|
|
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;})()"
|
|
)
|
|
new_src = None
|
|
end = time.time() + timeout
|
|
while time.time() < end:
|
|
time.sleep(1.5)
|
|
cur = _image_rects(cdp)
|
|
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 len(cur) > len(before) and ready:
|
|
new_src = ready[-1]["src"]
|
|
break
|
|
if not new_src:
|
|
return {"ok": False, "reason": "UPLOAD_TIMEOUT", "count_before": count_before, "delete": delete_result}
|
|
|
|
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,
|
|
}
|
|
|
|
|
|
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) -> dict:
|
|
"""Click the Shopee update button if it is present and enabled."""
|
|
|
|
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)
|
|
time.sleep(3 if result == "CLICKED" else 0)
|
|
toasts = _json_value(cdp, JS_TOASTS, default=[]) or []
|
|
return {
|
|
"clicked": result == "CLICKED",
|
|
"reason": None if result == "CLICKED" else result,
|
|
"toasts": toasts,
|
|
}
|
|
|
|
|
|
def apply_task(account, task, close_success_tab=False) -> dict:
|
|
"""Apply generated title/cover to Shopee.
|
|
|
|
The caller must perform the batch confirmation before calling this function.
|
|
"""
|
|
|
|
item_id = _item_id(task)
|
|
cdp = open_product(account, item_id)
|
|
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:
|
|
title_result = change_title(cdp, new_title)
|
|
if not title_result.get("ok"):
|
|
return {"committed": False, "error": "标题写入后 value/modelvalue 未同步", "title": title_result}
|
|
if new_cover_path:
|
|
cover_result = replace_cover(
|
|
cdp,
|
|
new_cover_path,
|
|
old_cover_path=_get(task, "old_cover_path"),
|
|
)
|
|
if not cover_result.get("ok"):
|
|
return {"committed": False, "error": cover_result.get("reason"), "cover": cover_result}
|
|
update_result = click_update(cdp)
|
|
committed = bool(update_result.get("clicked", False))
|
|
return {
|
|
"committed": committed,
|
|
"error": update_result.get("reason"),
|
|
"title": title_result,
|
|
"cover": cover_result,
|
|
"update": update_result,
|
|
}
|
|
except Exception as 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:
|
|
close_tab(target_id, host=host)
|
|
except Exception:
|
|
pass
|