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 .cdp import CDP, create_tab, 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']"
)

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


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 _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 _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 is_logged_in(account) -> bool:
    """Return whether the account's current Shopee session appears logged in."""

    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(f"https://{_region_host(account)}/portal/", host=host)
        cdp = CDP(ws)
    else:
        url = shopee_page.get("url") or ""
        if "/login" in url or "account/signin" in url:
            return False
        cdp = CDP(shopee_page["webSocketDebuggerUrl"])

    try:
        _ensure_page_domains(cdp)
        cookies = cdp.send("Network.getAllCookies").get("cookies", [])
        names = {
            c.get("name")
            for c in cookies
            if "shopee" in (c.get("domain") or "")
        }
        return "SPC_ST" in names or "SPC_U" in names
    finally:
        cdp.close()


def open_product(account, item_id) -> CDP:
    """Open or reuse a product edit tab, navigate to a clean edit URL, and wait ready."""

    host = _cdp_host(account)
    item_id = str(item_id)
    url = _product_url(account, item_id)
    tab = find_product_tab(item_id, host=host)
    ws = tab["webSocketDebuggerUrl"] if tab else create_tab(url, host=host)
    cdp = CDP(ws)
    _ensure_page_domains(cdp)
    try:
        cdp.send("Page.bringToFront")
    except Exception:
        pass
    cdp.send("Page.navigate", {"url": url})
    _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) -> dict:
    """Collect current title and cover snapshot before any edits."""

    item_id = _item_id(task)
    cdp = open_product(account, item_id)
    try:
        old_title = read_title(cdp)
        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")
        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:
        cdp.close()


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, timeout=90) -> dict:
    """Upload an image and drag it to the first position.

    Full 9-image deletion is intentionally left for T-502 because its confirm dialog
    selector is not verified yet.
    """

    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)
    if len(before) >= 9:
        return {"ok": False, "reason": "FULL_IMAGE_SLOTS", "count_before": len(before)}
    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": len(before)}

    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": len(before)}

    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}
    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": len(before),
        "count_after": len(after),
    }


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) -> 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)
    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)
            if not cover_result.get("ok"):
                return {"committed": False, "error": cover_result.get("reason"), "cover": cover_result}
        update_result = click_update(cdp)
        return {
            "committed": update_result.get("clicked", False),
            "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:
        cdp.close()