feat: 完成登录状态检测
- 增强 editor 登录检测,新增 login_status 详情结果 - 保留 is_logged_in bool 包装并处理登录页、缺 Cookie 等状态 - 新增登录检测单元测试覆盖 CDP 分支 - 更新任务看板、API 合约、当前状态和进度记录
This commit is contained in:
+80
-12
@@ -15,6 +15,13 @@ ITEMBOX_XPATH = (
|
||||
"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,"
|
||||
@@ -109,6 +116,10 @@ def _region_host(account=None):
|
||||
return value.strip("/") or DEFAULT_REGION_HOST
|
||||
|
||||
|
||||
def _portal_url(account):
|
||||
return f"https://{_region_host(account)}/portal/"
|
||||
|
||||
|
||||
def _product_url(account, item_id):
|
||||
return (
|
||||
f"https://{_region_host(account)}/portal/product/{item_id}"
|
||||
@@ -129,6 +140,11 @@ def _image_root(account):
|
||||
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:
|
||||
@@ -169,34 +185,86 @@ def _ensure_page_domains(cdp):
|
||||
pass
|
||||
|
||||
|
||||
def is_logged_in(account) -> bool:
|
||||
"""Return whether the account's current Shopee session appears logged in."""
|
||||
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(f"https://{_region_host(account)}/portal/", host=host)
|
||||
ws = create_tab(_portal_url(account), host=host)
|
||||
initial_url = _portal_url(account)
|
||||
cdp = CDP(ws)
|
||||
else:
|
||||
url = shopee_page.get("url") or ""
|
||||
if "/login" in url or "account/signin" in url:
|
||||
return False
|
||||
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)
|
||||
cookies = cdp.send("Network.getAllCookies").get("cookies", [])
|
||||
names = {
|
||||
c.get("name")
|
||||
for c in cookies
|
||||
if "shopee" in (c.get("domain") or "")
|
||||
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),
|
||||
}
|
||||
return "SPC_ST" in names or "SPC_U" in names
|
||||
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) -> CDP:
|
||||
"""Open or reuse a product edit tab, navigate to a clean edit URL, and wait ready."""
|
||||
|
||||
|
||||
Reference in New Issue
Block a user