diff --git a/app/editor.py b/app/editor.py index 3e556c1..9901205 100644 --- a/app/editor.py +++ b/app/editor.py @@ -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.""" diff --git a/docs/06-tasks.md b/docs/06-tasks.md index 23d1008..ff36a08 100644 --- a/docs/06-tasks.md +++ b/docs/06-tasks.md @@ -37,7 +37,7 @@ | --- | --- | --- | --- | --- | | T-101 | `config` 生成 slug + 创建 `chrome_user_data_dir/` | T-003 | 别名→唯一 slug；目录按需建；路径绝对化 | DONE | | T-102 | `app/chrome.py` 启动器：拼参数并启动、探测端口 | T-101, T-002 | 含三参数；端口就绪可探测 | DONE | -| T-103 | 首次登录保活 + 登录检测 `is_logged_in` | T-102, T-001 | 关闭再启动免重登；登录/未登录判断准确 | TODO | +| T-103 | 首次登录保活 + 登录检测 `is_logged_in` | T-102, T-001 | 关闭再启动免重登；登录/未登录判断准确 | DONE | | T-104 | PySide6 五 Tab 主窗口骨架（`QMainWindow` + `QTabWidget`，5 Tab 空壳） | T-002 | 五个 Tab 按顺序可切换；启动不阻塞；基础状态栏可用 | TODO | | T-104b | PySide6 worker 基类与线程启动工具（`BaseWorker` + `QThread` 包装） | T-104 | signals: progress/log/row_updated/failed/finished/cancelled；取消标记可用；worker 不直接操作 QWidget | TODO | | T-105 | Tab④ 账号增删改（账号名/别名/端口/密码本地明文仅参考）+ 启动登录 + 检测登录 | T-104, T-101, T-103 | 增删改入库、建目录；密码字段 UI 打码；状态列刷新；不自动登录/填密码 | TODO | diff --git a/docs/api.md b/docs/api.md index d5f5a5b..341d590 100644 --- a/docs/api.md +++ b/docs/api.md @@ -145,7 +145,8 @@ class CDP: send/ev/val/object_id/drag/close # suppress_origin、trust_env=Fals ## editor 模块（`app/editor.py`，已建，重构自现有脚本） ```python -is_logged_in(account) -> bool # 重定向登录页或缺 SPC_ST → False +login_status(account, timeout=8) -> dict # {logged_in, reason, url, host, cookie_names} +is_logged_in(account) -> bool # login_status(...).logged_in；重定向登录页或缺 SPC_ST/SPC_U → False open_product(account, item_id) -> CDP # 连端口、导航商品页、等就绪 # 采集（只读） @@ -162,6 +163,8 @@ apply_task(account, task) -> dict # 对已生成任务：换标题+ # -> {committed, error} ``` +`login_status()` 不自动登录；无 Shopee tab 时只打开卖家中心用于检测/人工登录。判断规则：最终 URL 是登录页 → `LOGIN_PAGE`；缺少 `SPC_ST`/`SPC_U` → `NO_SESSION_COOKIE`；有会话 Cookie → 已登录。 + ## ai 模块（`app/ai.py`，待建，外部 AI，服务商待定） ```python diff --git a/docs/current-state.md b/docs/current-state.md index 5197c66..0d53d83 100644 --- a/docs/current-state.md +++ b/docs/current-state.md @@ -6,10 +6,10 @@ ## 当前快照 - 日期：2026-06-27 -- 阶段：V0 单账号 CDP 流程已验证；V1 已完成 T-000 正式代码包结构、T-001 `app/editor.py` 模块化、T-002 `app/appconfig.py` 应用配置、T-003 SQLite 持久化地基、T-004 本地数据忽略规则、T-005 AI 模型清单后端、T-006 单元测试基座、T-101 账号 user-data-dir 工具、T-102 Chrome 启动器。 +- 阶段：V0 单账号 CDP 流程已验证；V1 已完成 T-000 正式代码包结构、T-001 `app/editor.py` 模块化、T-002 `app/appconfig.py` 应用配置、T-003 SQLite 持久化地基、T-004 本地数据忽略规则、T-005 AI 模型清单后端、T-006 单元测试基座、T-101 账号 user-data-dir 工具、T-102 Chrome 启动器、T-103 登录保活与检测。 - 技术栈：Python 3.10+，自研 CDP（websocket-client + requests），SQLite（sqlite3）+ `config.json` + openpyxl + AI（服务商待定），GUI PySide6 5 Tab（已定）。 -- 生产代码：已建立 `app/` 包 + 根入口 `main.py`；`app/cdp.py` 为已验证 CDP 底座；`app/editor.py` 已封装标题/封面/采集/更新按钮能力；`app/appconfig.py` 已实现 `config.json` 默认值/读写/更新、AI 参数与端口读取，以及 `config/ai_models.json` 模型清单 CRUD/过滤/打码/测试连接；`app/db.py` 已实现 SQLite schema、连接 PRAGMA、批次/账号/任务与阶段写库函数；`app/config.py` 已实现账号 slug 与 user-data-dir 创建；`app/chrome.py` 已实现 Chrome 参数拼装、启动与 CDP 端口探测；`app/gui.py` 目前是入口占位，完整 PySide6 主窗口待 T-104。 -- 测试：`tests/` 已建立；T-006 后纯逻辑改动必须运行 `python -m unittest discover -s tests`，当前覆盖 appconfig/db/config/chrome，并对尚未实现的 app.excel/app.prompts 做契约占位 skip；CDP/Shopee 改动仍需测试商品手动验证。 +- 生产代码：已建立 `app/` 包 + 根入口 `main.py`；`app/cdp.py` 为已验证 CDP 底座；`app/editor.py` 已封装登录状态检测、标题/封面/采集/更新按钮能力；`app/appconfig.py` 已实现 `config.json` 默认值/读写/更新、AI 参数与端口读取，以及 `config/ai_models.json` 模型清单 CRUD/过滤/打码/测试连接；`app/db.py` 已实现 SQLite schema、连接 PRAGMA、批次/账号/任务与阶段写库函数；`app/config.py` 已实现账号 slug 与 user-data-dir 创建；`app/chrome.py` 已实现 Chrome 参数拼装、启动与 CDP 端口探测；`app/gui.py` 目前是入口占位，完整 PySide6 主窗口待 T-104。 +- 测试：`tests/` 已建立；T-006 后纯逻辑改动必须运行 `python -m unittest discover -s tests`，当前覆盖 appconfig/db/config/chrome/editor 登录检测，并对尚未实现的 app.excel/app.prompts 做契约占位 skip；CDP/Shopee 改动仍需测试商品手动验证。 - 数据：`config.json`、`config/ai_models.json`、`cmshopee.db`、`chrome_user_data_dir/`、`images/` 已由 `.gitignore` 排除；`app/appconfig.py` 首次读取缺失的 `config.json` 时会在本地写默认配置，`app/db.py` 调用 `init_db()` 时会在本地创建 SQLite DB。 ## 既定设计要点（文档已定） @@ -33,7 +33,7 @@ | `chrome-remote-debug-lan.md` | 已有 | WSL→Windows CDP 转发排查记录 | | `app/__init__.py` / `app/__main__.py` / `main.py` | 已有 | 正式包与启动入口；`python main.py` / `python -m app` 可运行占位入口 | | `app/gui.py` | 已有 | GUI 占位入口；完整 PySide6 主窗口待 T-104 | -| `app/editor.py` | 已有 | T-001 产出：登录检测、打开商品页、读/写标题、读/下载封面、上传拖封面、更新按钮、apply_task | +| `app/editor.py` | 已有 | T-001/T-103 产出：登录状态检测、打开商品页、读/写标题、读/下载封面、上传拖封面、更新按钮、apply_task | | `app/appconfig.py` | 已有 | T-002 产出：`config.json` 默认值、读写、更新、路径/端口/AI 参数读取；拒绝敏感字段写入 | | `app/db.py` | 已有 | T-003 产出：batches/accounts/tasks schema；WAL/busy_timeout/foreign_keys；账号/批次/任务与 set_* 阶段写库 | | `app/config.py` | 已有 | T-101 产出：别名→稳定 slug；创建并返回绝对 user-data-dir | @@ -54,9 +54,9 @@ 任务状态以 [`06-tasks.md`](06-tasks.md) 为准，历史记录见 [`../progress.md`](../progress.md)。 -- 已完成：T-000（正式代码包结构）、T-001（`app/editor.py` 模块化）、T-002（`app/appconfig.py` + `config.json`）、T-003（`app/db.py` + SQLite 建表）、T-004（本地数据 gitignore）、T-005（AI 模型清单后端）、T-006（单元测试基座）、T-101（账号 slug/user-data-dir）、T-102（Chrome 启动器）。 +- 已完成：T-000（正式代码包结构）、T-001（`app/editor.py` 模块化）、T-002（`app/appconfig.py` + `config.json`）、T-003（`app/db.py` + SQLite 建表）、T-004（本地数据 gitignore）、T-005（AI 模型清单后端）、T-006（单元测试基座）、T-101（账号 slug/user-data-dir）、T-102（Chrome 启动器）、T-103（登录保活与检测）。 - 正在进行：无。 -- 下一个可领取任务：**T-103（首次登录保活 + 登录检测）**。 +- 下一个可领取任务：**T-104（PySide6 五 Tab 主窗口骨架）**。 ## 当前可运行内容 diff --git a/progress.md b/progress.md index 379b99a..5fb390e 100644 --- a/progress.md +++ b/progress.md @@ -312,3 +312,11 @@ - 范围：本轮不实现 `.lnk` 快捷方式，留给 T-106。 - 验证：`py -3 -m compileall app main.py tests` 通过；`py -3 -m unittest discover -s tests` 通过（17 tests，skipped=2）；`python -m unittest discover -s tests` 通过（17 tests，skipped=2）。 - 下一步：按任务看板领取 T-103。 + +## 【2026-06-27】T-103 登录保活与检测 + +- 状态：DONE +- 变更：增强 `app/editor.py` 登录检测，新增 `login_status(account, timeout=8)` 返回 `logged_in/reason/url/host/cookie_names`，`is_logged_in()` 保持 bool 包装；无 Shopee tab 时打开卖家中心用于检测/人工登录；新增 `tests/test_editor_login.py` 覆盖登录页、会话 Cookie、缺 Cookie、无 Shopee tab 创建检测页；同步 `docs/06-tasks.md`、`docs/current-state.md`、`docs/api.md`。 +- 规则：不自动登录、不填密码；登录态依赖 T-101/T-102 的稳定 user-data-dir 保留；URL 命中登录页返回 `LOGIN_PAGE`，缺 `SPC_ST/SPC_U` 返回 `NO_SESSION_COOKIE`。 +- 验证：`py -3 -m compileall app main.py tests` 通过；`py -3 -m unittest discover -s tests` 通过（22 tests，skipped=2）；`python -m unittest discover -s tests` 通过（22 tests，skipped=2）。 +- 下一步：按任务看板领取 T-104。 diff --git a/tests/test_editor_login.py b/tests/test_editor_login.py new file mode 100644 index 0000000..12dba6f --- /dev/null +++ b/tests/test_editor_login.py @@ -0,0 +1,133 @@ +import os +import sys +import unittest +from unittest import mock + +sys.path.insert(0, os.path.dirname(__file__)) + +from app import editor + + +class FakeCDP: + instances = [] + + def __init__(self, ws, url="", cookies=None): + self.ws = ws + self.url = url + self.cookies = cookies or [] + self.closed = False + FakeCDP.instances.append(self) + + def val(self, expr): + if expr == "location.href": + return self.url + return None + + def send(self, method, params=None): + if method == "Network.getAllCookies": + return {"cookies": self.cookies} + return {} + + def close(self): + self.closed = True + + +def cookie(name, domain=".shopee.tw"): + return {"name": name, "domain": domain} + + +class EditorLoginTests(unittest.TestCase): + def setUp(self): + FakeCDP.instances = [] + + def patch_cdp(self, url="", cookies=None): + def factory(ws): + return FakeCDP(ws, url=url, cookies=cookies) + + return mock.patch("app.editor.CDP", side_effect=factory) + + def test_login_status_false_for_login_page_url(self): + with mock.patch( + "app.editor.http_get", + return_value=[ + { + "type": "page", + "url": "https://seller.shopee.tw/account/signin", + "webSocketDebuggerUrl": "ws-login", + } + ], + ): + status = editor.login_status({"debug_port": 9222}, timeout=0) + + self.assertFalse(status["logged_in"]) + self.assertEqual("LOGIN_PAGE", status["reason"]) + self.assertEqual([], FakeCDP.instances) + + def test_login_status_true_with_session_cookie(self): + with mock.patch( + "app.editor.http_get", + return_value=[ + { + "type": "page", + "url": "https://seller.shopee.tw/portal/", + "webSocketDebuggerUrl": "ws-portal", + } + ], + ), self.patch_cdp( + url="https://seller.shopee.tw/portal/", + cookies=[cookie("SPC_ST"), cookie("csrftoken")], + ): + status = editor.login_status({"debug_port": 9222}, timeout=1) + + self.assertTrue(status["logged_in"]) + self.assertIsNone(status["reason"]) + self.assertIn("SPC_ST", status["cookie_names"]) + self.assertTrue(FakeCDP.instances[0].closed) + + def test_login_status_false_without_session_cookie(self): + with mock.patch( + "app.editor.http_get", + return_value=[ + { + "type": "page", + "url": "https://seller.shopee.tw/portal/", + "webSocketDebuggerUrl": "ws-portal", + } + ], + ), self.patch_cdp( + url="https://seller.shopee.tw/portal/", + cookies=[cookie("csrftoken")], + ): + status = editor.login_status({"debug_port": 9222}, timeout=0) + + self.assertFalse(status["logged_in"]) + self.assertEqual("NO_SESSION_COOKIE", status["reason"]) + + def test_login_status_opens_portal_when_no_shopee_page(self): + with mock.patch("app.editor.http_get", return_value=[]), mock.patch( + "app.editor.create_tab", return_value="ws-new" + ) as create_tab, self.patch_cdp( + url="https://seller.shopee.tw/portal/", + cookies=[cookie("SPC_U")], + ): + status = editor.login_status( + {"debug_port": 9222, "region_host": "seller.shopee.tw"}, + timeout=1, + ) + + self.assertTrue(status["logged_in"]) + create_tab.assert_called_once_with( + "https://seller.shopee.tw/portal/", + host="127.0.0.1:9222", + ) + + def test_is_logged_in_returns_boolean(self): + with mock.patch( + "app.editor.login_status", + return_value={"logged_in": True}, + ): + self.assertTrue(editor.is_logged_in({"debug_port": 9222})) + + +if __name__ == "__main__": + unittest.main()