diff --git a/app/editor.py b/app/editor.py index e9e3b4c..1cbda41 100644 --- a/app/editor.py +++ b/app/editor.py @@ -803,6 +803,22 @@ def read_cover_src(cdp) -> str: return cdp.val(JS_FIRST_COVER) or "" +def read_product_image_urls(cdp): + """Read all current Shopee product image URLs in page order.""" + + images = [] + for position, rect in enumerate(_image_rects(cdp), start=1): + src = str((rect or {}).get("src") or "").strip() + if not src: + continue + try: + index = int((rect or {}).get("i", position - 1)) + 1 + except (TypeError, ValueError): + index = position + images.append({"index": index, "src": src}) + return images + + def download_cover(src, out_path) -> str: """Download a cover image to a local path and return the absolute path.""" @@ -891,6 +907,12 @@ def _close_collected_product(cdp): pass +def close_readonly_product(cdp): + """Close a read-only product CDP session using the collection tab cleanup rules.""" + + _close_collected_product(cdp) + + def change_title(cdp, new_title) -> dict: """Write a new title with the verified native setter + input/change events.""" diff --git a/app/image_studio.py b/app/image_studio.py index ce0c7d2..2bf34b2 100644 --- a/app/image_studio.py +++ b/app/image_studio.py @@ -9,11 +9,12 @@ from dataclasses import dataclass from datetime import datetime from typing import Iterable, Optional -from . import appconfig, db +from . import accounts, appconfig, chrome, db, editor from .config import make_slug PROJECT_STATUS_ACTIVE = "active" +ASSET_KIND_ORIGINAL = "original" ASSET_STATUS_AVAILABLE = "available" ASSET_STATUS_MISSING = "missing" ASSET_STATUSES = {ASSET_STATUS_AVAILABLE, ASSET_STATUS_MISSING} @@ -90,6 +91,10 @@ class ImageStudioSelection: updated_at: str +class ImageStudioError(RuntimeError): + """Raised when the AI image studio service cannot complete an operation.""" + + def _now() -> str: return datetime.now().isoformat(timespec="seconds") @@ -141,6 +146,83 @@ def _task_key(project_id): return f"image-studio-{int(project_id)}-{uuid.uuid4().hex}" +def _db_path(path=None, config=None) -> str: + return path or appconfig.db_path(config) + + +def _normalize_item_id(item_id) -> str: + text = str(item_id or "").strip() + if not text: + raise ImageStudioError("商品ID不能为空") + return text + + +def _notify_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 _is_definitive_logged_out(status): + reason = str((status or {}).get("reason") or "").strip() + url = str((status or {}).get("url") or "").lower() + return reason.startswith("LOGIN_PAGE") or ( + "accounts.shopee." in url and "/seller/login" in url + ) + + +def _login_status_detail(status): + status = status or {} + reason = status.get("reason") or "未知原因" + url = status.get("url") or "未知URL" + cookie_names = [str(name) for name in (status.get("cookie_names") or []) if name] + cookie_text = ",".join(sorted(cookie_names)) if cookie_names else "未读到登录Cookie" + return f"原因={reason},URL={url},Cookie名称={cookie_text}" + + +def _ensure_account_ready_for_read(account, *, path=None, config=None, login_timeout=8, on_step=None): + port = accounts.normalize_debug_port(_get(account, "debug_port")) + alias = str(_get(account, "alias") or "").strip() + _notify_step(on_step, "ensure_chrome", "start", f"账号 {alias} debug_port={port}") + if chrome.is_running(port): + chrome_info = { + "action": "reused", + "reused": True, + "launched": False, + "debug_port": port, + } + _notify_step(on_step, "ensure_chrome", "reused", f"账号 {alias} Chrome 已打开") + else: + try: + chrome_info = accounts.launch_for_login(account, path=path, config=config) + except Exception as exc: + raise ImageStudioError(f"账号 {alias} Chrome 启动失败:{exc}") from exc + action = "launched" if chrome_info.get("launched") else "reused" + _notify_step(on_step, "ensure_chrome", action, f"账号 {alias} Chrome 已{ '启动' if action == 'launched' else '复用' }") + + _notify_step(on_step, "login_check", "start", f"账号 {alias}") + try: + status = accounts.detect_login(account, timeout=login_timeout, path=path, config=config) + except Exception as exc: + status = {"logged_in": False, "reason": f"LOGIN_CHECK_FAILED: {exc}"} + if status.get("logged_in"): + _notify_step(on_step, "login_check", "success", f"账号 {alias} 已登录") + return {**chrome_info, "login_status": status, "login_uncertain": False} + if _is_definitive_logged_out(status): + detail = _login_status_detail(status) + _notify_step(on_step, "login_check", "blocked", detail) + raise ImageStudioError(f"账号 {alias} 未登录:{detail}") + detail = _login_status_detail(status) + _notify_step(on_step, "login_check", "uncertain", f"账号 {alias} 登录状态暂不稳定,继续尝试读取商品:{detail}") + return {**chrome_info, "login_status": status, "login_uncertain": True} + + def _ensure_assets_belong_to_project(database, project_id, asset_ids): ids = [int(asset_id) for asset_id in asset_ids if asset_id is not None] if not ids: @@ -370,6 +452,91 @@ def list_assets(project_id, kind=None, include_missing=True, path=None, conn=Non return _fetch_all(database, sql, params, ImageStudioAsset) +def sync_original_asset_urls(project_id, image_urls, path=None, conn=None): + """Store the read-only Shopee main image URL snapshot as remote-only assets.""" + + normalized = [] + seen_urls = set() + for position, item in enumerate(image_urls or [], start=1): + if isinstance(item, dict): + src = str(item.get("src") or "").strip() + source_order = int(item.get("index") or position) + else: + src = str(item or "").strip() + source_order = position + if not src or src in seen_urls: + continue + seen_urls.add(src) + normalized.append({"src": src, "source_order": source_order}) + + now = _now() + with _connection(conn, path) as database: + with database: + existing_rows = database.execute( + """ + SELECT * FROM image_studio_assets + WHERE project_id = ? AND kind = ? + """, + (int(project_id), ASSET_KIND_ORIGINAL), + ).fetchall() + by_url = { + str(row["remote_url"]): row + for row in existing_rows + if row["remote_url"] + } + active_ids = set() + for item in normalized: + row = by_url.get(item["src"]) + if row is not None: + active_ids.add(int(row["id"])) + database.execute( + """ + UPDATE image_studio_assets + SET source_order = ?, + status = ?, + updated_at = ? + WHERE id = ? + """, + ( + int(item["source_order"]), + ASSET_STATUS_AVAILABLE, + now, + int(row["id"]), + ), + ) + else: + cursor = database.execute( + """ + INSERT INTO image_studio_assets + (project_id, kind, remote_url, local_path, status, + source_order, created_at, updated_at) + VALUES (?, ?, ?, NULL, ?, ?, ?, ?) + """, + ( + int(project_id), + ASSET_KIND_ORIGINAL, + item["src"], + ASSET_STATUS_AVAILABLE, + int(item["source_order"]), + now, + now, + ), + ) + active_ids.add(int(cursor.lastrowid)) + missing_ids = [int(row["id"]) for row in existing_rows if int(row["id"]) not in active_ids] + if missing_ids: + placeholders = ",".join("?" for _ in missing_ids) + database.execute( + f""" + UPDATE image_studio_assets + SET status = ?, updated_at = ? + WHERE id IN ({placeholders}) + """, + [ASSET_STATUS_MISSING, now, *missing_ids], + ) + return list_assets(project_id, kind=ASSET_KIND_ORIGINAL, conn=database) + + def mark_asset_status(asset_id, status, path=None, conn=None): if str(status) not in ASSET_STATUSES: raise db.DbError("AI工场资产状态必须是 available 或 missing") @@ -563,3 +730,61 @@ def list_selections(project_id, selection_type=None, path=None, conn=None): sql += " ORDER BY selection_type, position" with _connection(conn, path) as database: return _fetch_all(database, sql, params, ImageStudioSelection) + + +def pull_remote_main_image_urls( + account_or_alias, + item_id, + *, + path=None, + config=None, + login_timeout=8, + on_step=None, +): + """Create/open an AI studio project and read Shopee main image URLs without editing.""" + + cfg = appconfig.load_config() if config is None else config + database_path = _db_path(path, cfg) + db.init_db(database_path) + item = _normalize_item_id(item_id) + try: + account = accounts.resolve_account(account_or_alias, path=database_path, config=cfg) + except Exception as exc: + raise ImageStudioError(f"AI工场账号不可用:{exc}") from exc + + project = create_or_get_project(account, item_id=item, path=database_path) + readiness = _ensure_account_ready_for_read( + account, + path=database_path, + config=cfg, + login_timeout=login_timeout, + on_step=on_step, + ) + cdp = None + try: + _notify_step(on_step, "open_product", "start", f"商品 {item}") + cdp = editor.open_product(account, item, on_step=on_step, bring_to_front=False) + _notify_step(on_step, "open_product", "success", f"商品 {item}") + _notify_step(on_step, "read_main_images", "start", f"商品 {item}") + images = editor.read_product_image_urls(cdp) + if not images: + raise ImageStudioError("未读取到蝦皮商品主图 URL") + assets = sync_original_asset_urls(project.id, images, path=database_path) + _notify_step(on_step, "read_main_images", "success", f"读取 {len(images)} 张主图 URL") + project = get_project(project.id, path=database_path) + return { + "project": project, + "images": images, + "assets": assets, + "account": account, + "readiness": readiness, + } + except ImageStudioError: + raise + except editor.EditorError as exc: + raise ImageStudioError(str(exc)) from exc + except Exception as exc: + raise ImageStudioError(f"读取蝦皮原主图失败:{exc}") from exc + finally: + if cdp is not None: + editor.close_readonly_product(cdp) diff --git a/docs/04-architecture.md b/docs/04-architecture.md index e5d80eb..ebe6530 100644 --- a/docs/04-architecture.md +++ b/docs/04-architecture.md @@ -462,6 +462,7 @@ data/images///__new. # AI 生成的新 | 标题输入框 | XPath `//input[@class='eds-input__input' and string-length(@modelvalue)>24]` | | 写标题 | 原生 setter + 派发 `input`/`change`;`value`==`modelvalue`==新值 | | 读旧封面 | 第一张 itembox 的 `img.src`(`susercontent` CDN),下载到本地 | +| AI工场原主图读取 | 复用 `open_product(..., bring_to_front=False)` 后台只读打开商品详情页,使用已验证 itembox 顺序读取全部主图 `img.src` 并返回 `{index, src}`;不要求上传 input 之外的新选择器、不下载图片、不改标题/封面、不拖拽、不点击更新;本轮自动新建 tab 按采集规则关闭,复用用户已有 tab 不关闭 | | 上传输入框 | `.shopee-image-manager__upload input[type=file]`;上传前先点击 `.shopee-image-manager__upload` 上传块以模拟人工选择图片入口,短暂等待后重新获取 input,再用 `DOM.setFileInputFiles` 传 Windows 路径并派发 `input`/`change` | | 上传成功 | 上传前先等图片管理器稳定。注意分两种状态:未满 9 张时,上传前要求图片 src 连续稳定、无 loading/blob、上传 input 存在且未禁用;满 9 张时,删除第一张之前只要求现有图片列表稳定,不得要求上传 input 可用,因为 Shopee 可能因满格隐藏/禁用上传入口;删除成功后再要求上传 input 恢复可用。上传后等新图 src 为 `susercontent`。若手动上传成功但自动上传一直转圈,优先检查是否绕过了上传块点击导致 Shopee 前端上传队列未完整初始化;代码应走“点击上传块 → 等待 → 重新取 input → `DOM.setFileInputFiles`”的人工等价路径。T-404 补丁后超时失败会返回 `upload_state`,区分仍在转圈(`UPLOAD_STILL_PROCESSING`)、图片上传错误(`UPLOAD_PAGE_ERROR`)、裁剪弹窗(`UPLOAD_CROP_REQUIRED`)和上传入口未恢复(`UPLOAD_INPUT_NOT_READY`);上传阶段只能把图片管理器内错误或图片/文件/上传相关 toast 归为封面上传错误,物流/备货等页面级校验错误不能阻断封面上传,应留到点击「更新」提交阶段处理;`有1張重複的圖片` / `重複` / `重复` / `duplicate` 属于封面上传错误,必须立即失败并提示新封面与现有商品图片重复 | | 封面=第一位 | `Input.dispatchMouseEvent` 拖到第一位,落点 `第一张.left - 0.30*宽` | diff --git a/docs/tasks/T-587.md b/docs/tasks/T-587.md index 4087b65..a8ea600 100644 --- a/docs/tasks/T-587.md +++ b/docs/tasks/T-587.md @@ -3,7 +3,7 @@ id: T-587 title: AI工场商品项目入口与 CDP 只读拉取蝦皮原主图 URL phase: 7 deps: [T-586, T-563] -status: TODO +status: DONE created: 2026-07-11 --- @@ -36,4 +36,11 @@ AI工场以“账号 + 商品 ID”建立项目,并从该账号 Chrome 的商 ## 执行记录 -(完成后记录 mock、只读实跑与 tab 清理结果。) +- 2026-07-11:完成 AI工场商品项目入口与原主图 URL 只读拉取代码。 + - `app/editor.py` 新增 `read_product_image_urls()`,复用已验证 itembox rects 顺序返回 `{index, src}`;新增 `close_readonly_product()` 复用①采集 tab 清理规则。 + - `app/image_studio.py` 新增项目入口 `pull_remote_main_image_urls()`:按账号+商品创建/打开项目,复用/启动账号 Chrome,只检测登录;明确登录页阻断,不确定登录继续尝试打开商品页;商品页用 `open_product(..., bring_to_front=False)` 后台只读打开,读取 URL 后同步为 remote-only `original` assets,不下载图片、不修改蝦皮、不点击更新。 + - 原图 URL 刷新走 `sync_original_asset_urls()`:相同 URL 复用原 asset 并更新顺序,不再出现的 URL 标记 `missing`,不物理删除历史记录。 + - `docs/04-architecture.md` 已补充 AI工场原主图读取的 CDP 约束。 + - 测试覆盖:已开 Chrome 复用、未开 Chrome 启动、明确未登录阻断、`NO_SESSION_COOKIE` 不确定登录继续、商品失效 toast 文案上浮、成功读取 1..9 张 URL、重复刷新幂等、自动新建/复用 tab 清理接口,以及成功路径不调用标题修改/封面替换/点击更新。 + - 验证:在只套用 T-587 diff 的 clean worktree 中运行 `python -m ruff check app tests main.py`、`py -3.10 -m compileall app main.py`、`py -3.10 -m unittest discover -s tests`(350 tests)和 `git diff --check`,全部通过。 + - 真实蝦皮只读实跑未执行:本轮未指定可用的已登录测试账号别名;为避免擅自打开真实浏览器,只做 mock/CDP 合约验证,后续接 GUI 前可用测试账号对 ITEM_ID 51100639510 做一次只读确认。 diff --git a/tests/test_editor_login.py b/tests/test_editor_login.py index 0f28e80..35a1f43 100644 --- a/tests/test_editor_login.py +++ b/tests/test_editor_login.py @@ -39,10 +39,11 @@ class FakeCDP: class FakeProductCDP: - def __init__(self, ws, ready=True, toasts=None): + def __init__(self, ws, ready=True, toasts=None, rects=None): self.ws = ws self.ready = ready self.toasts = list(toasts or []) + self.rects = list(rects or []) self.closed = False self.sent = [] self.toast_observer_installed = False @@ -55,6 +56,8 @@ class FakeProductCDP: return True if expr == editor.JS_PAGE_TOASTS: return json.dumps(self.toasts) + if expr == editor.JS_RECTS: + return json.dumps(self.rects) return None def send(self, method, params=None): @@ -655,6 +658,20 @@ class EditorLoginTests(unittest.TestCase): self.assertTrue(cdp.closed) close_tab.assert_not_called() + def test_read_product_image_urls_returns_all_images_in_page_order(self): + cdp = FakeProductCDP("ws-existing", rects=cover_rects(3, prefix="main")) + + images = editor.read_product_image_urls(cdp) + + self.assertEqual( + [ + {"index": 1, "src": "https://susercontent.com/main-0.jpg"}, + {"index": 2, "src": "https://susercontent.com/main-1.jpg"}, + {"index": 3, "src": "https://susercontent.com/main-2.jpg"}, + ], + images, + ) + def test_click_update_confirms_shopee_update_modal(self): cdp = FakeUpdateCDP(confirm_present=True, confirm_click=True) diff --git a/tests/test_image_studio.py b/tests/test_image_studio.py index 63a0aaf..23ae295 100644 --- a/tests/test_image_studio.py +++ b/tests/test_image_studio.py @@ -2,15 +2,27 @@ import os import sys import unittest from types import SimpleNamespace +from unittest import mock sys.path.insert(0, os.path.dirname(__file__)) from _helpers import TempDirMixin -from app import db, image_studio +from app import accounts, db, editor, image_studio class ImageStudioTests(TempDirMixin, unittest.TestCase): + def _config(self, temp_dir): + return { + "data_dir": temp_dir, + "db_path": os.path.join(temp_dir, "cmshopee.db"), + "user_data_root": os.path.join(temp_dir, "chrome_user_data_dir"), + "chrome_path": "chrome.exe", + "default_debug_port": 9222, + "debug_port_range": [9222, 9230], + "cdp_ready_timeout": 1, + } + def test_init_db_adds_image_studio_tables_without_breaking_existing_tables(self): with self.make_temp_dir() as temp_dir: db_path = os.path.join(temp_dir, "cmshopee.db") @@ -198,6 +210,49 @@ class ImageStudioTests(TempDirMixin, unittest.TestCase): self.assert_removed(temp_dir) + def test_sync_original_asset_urls_is_idempotent_and_marks_missing(self): + with self.make_temp_dir() as temp_dir: + db_path = os.path.join(temp_dir, "cmshopee.db") + db.init_db(db_path) + project = image_studio.create_or_get_project( + account_alias="alias", + account_slug="alias_slug", + item_id="51100639510", + path=db_path, + ) + + first_sync = image_studio.sync_original_asset_urls( + project.id, + [ + {"index": 1, "src": "https://susercontent.com/a.jpg"}, + {"index": 2, "src": "https://susercontent.com/b.jpg"}, + ], + path=db_path, + ) + second_sync = image_studio.sync_original_asset_urls( + project.id, + [ + {"index": 1, "src": "https://susercontent.com/a.jpg"}, + {"index": 2, "src": "https://susercontent.com/b.jpg"}, + ], + path=db_path, + ) + third_sync = image_studio.sync_original_asset_urls( + project.id, + [{"index": 1, "src": "https://susercontent.com/b.jpg"}], + path=db_path, + ) + + self.assertEqual([asset.id for asset in first_sync], [asset.id for asset in second_sync]) + self.assertEqual(2, len(third_sync)) + by_url = {asset.remote_url: asset for asset in third_sync} + self.assertEqual(image_studio.ASSET_STATUS_MISSING, by_url["https://susercontent.com/a.jpg"].status) + self.assertEqual(image_studio.ASSET_STATUS_AVAILABLE, by_url["https://susercontent.com/b.jpg"].status) + self.assertEqual(1, by_url["https://susercontent.com/b.jpg"].source_order) + self.assertIsNone(by_url["https://susercontent.com/b.jpg"].local_path) + + self.assert_removed(temp_dir) + def test_job_lifecycle_and_resumable_query(self): with self.make_temp_dir() as temp_dir: db_path = os.path.join(temp_dir, "cmshopee.db") @@ -326,6 +381,192 @@ class ImageStudioTests(TempDirMixin, unittest.TestCase): self.assert_removed(temp_dir) + def test_pull_remote_main_image_urls_reuses_running_chrome_and_writes_snapshot(self): + with self.make_temp_dir() as temp_dir: + cfg = self._config(temp_dir) + account = accounts.create_account("主店", "alias-a", debug_port=9222, config=cfg) + images = [ + {"index": index, "src": f"https://susercontent.com/main-{index}.jpg"} + for index in range(1, 10) + ] + cdp = SimpleNamespace() + + with mock.patch("app.image_studio.chrome.is_running", return_value=True) as is_running, \ + mock.patch("app.image_studio.accounts.launch_for_login") as launch_for_login, \ + mock.patch( + "app.image_studio.accounts.detect_login", + return_value={"logged_in": True, "cookie_names": ["SPC_ST"]}, + ) as detect_login, \ + mock.patch("app.image_studio.editor.open_product", return_value=cdp) as open_product, \ + mock.patch( + "app.image_studio.editor.read_product_image_urls", + return_value=images, + ) as read_urls, \ + mock.patch("app.image_studio.editor.close_readonly_product") as close_readonly, \ + mock.patch("app.image_studio.editor.change_title") as change_title, \ + mock.patch("app.image_studio.editor.replace_cover") as replace_cover, \ + mock.patch("app.image_studio.editor.click_update") as click_update: + result = image_studio.pull_remote_main_image_urls( + "alias-a", + "51100639510", + path=cfg["db_path"], + config=cfg, + ) + second = image_studio.pull_remote_main_image_urls( + "alias-a", + "51100639510", + path=cfg["db_path"], + config=cfg, + ) + + self.assertEqual("alias-a", result["project"].account_alias) + self.assertEqual("51100639510", result["project"].item_id) + self.assertTrue(result["readiness"]["reused"]) + self.assertFalse(result["readiness"]["login_uncertain"]) + self.assertEqual(images, result["images"]) + self.assertEqual([asset.id for asset in result["assets"]], [asset.id for asset in second["assets"]]) + self.assertEqual(9, len(result["assets"])) + self.assertEqual( + list(range(1, 10)), + [asset.source_order for asset in result["assets"]], + ) + self.assertTrue(all(asset.kind == image_studio.ASSET_KIND_ORIGINAL for asset in result["assets"])) + self.assertTrue(all(asset.local_path is None for asset in result["assets"])) + is_running.assert_called_with(9222) + launch_for_login.assert_not_called() + detect_login.assert_called() + open_product.assert_called_with(account, "51100639510", on_step=None, bring_to_front=False) + self.assertEqual(2, read_urls.call_count) + self.assertEqual(2, close_readonly.call_count) + change_title.assert_not_called() + replace_cover.assert_not_called() + click_update.assert_not_called() + + self.assert_removed(temp_dir) + + def test_pull_remote_main_image_urls_launches_chrome_when_needed(self): + with self.make_temp_dir() as temp_dir: + cfg = self._config(temp_dir) + accounts.create_account("主店", "alias-a", debug_port=9222, config=cfg) + cdp = SimpleNamespace() + + with mock.patch("app.image_studio.chrome.is_running", return_value=False) as is_running, \ + mock.patch( + "app.image_studio.accounts.launch_for_login", + return_value={"launched": True, "reused": False, "debug_port": 9222}, + ) as launch_for_login, \ + mock.patch( + "app.image_studio.accounts.detect_login", + return_value={"logged_in": True}, + ), \ + mock.patch("app.image_studio.editor.open_product", return_value=cdp), \ + mock.patch( + "app.image_studio.editor.read_product_image_urls", + return_value=[{"index": 1, "src": "https://susercontent.com/main.jpg"}], + ), \ + mock.patch("app.image_studio.editor.close_readonly_product"): + result = image_studio.pull_remote_main_image_urls( + "alias-a", + "51100639510", + path=cfg["db_path"], + config=cfg, + ) + + self.assertTrue(result["readiness"]["launched"]) + is_running.assert_called_once_with(9222) + launch_for_login.assert_called_once() + + self.assert_removed(temp_dir) + + def test_pull_remote_main_image_urls_blocks_definitive_logged_out(self): + with self.make_temp_dir() as temp_dir: + cfg = self._config(temp_dir) + accounts.create_account("主店", "alias-a", debug_port=9222, config=cfg) + + with mock.patch("app.image_studio.chrome.is_running", return_value=True), \ + mock.patch( + "app.image_studio.accounts.detect_login", + return_value={ + "logged_in": False, + "reason": "LOGIN_PAGE", + "url": "https://accounts.shopee.tw/seller/login", + "cookie_names": [], + }, + ), \ + mock.patch("app.image_studio.editor.open_product") as open_product: + with self.assertRaisesRegex(image_studio.ImageStudioError, "未登录"): + image_studio.pull_remote_main_image_urls( + "alias-a", + "51100639510", + path=cfg["db_path"], + config=cfg, + ) + + open_product.assert_not_called() + + self.assert_removed(temp_dir) + + def test_pull_remote_main_image_urls_continues_when_login_status_uncertain(self): + with self.make_temp_dir() as temp_dir: + cfg = self._config(temp_dir) + accounts.create_account("主店", "alias-a", debug_port=9222, config=cfg) + cdp = SimpleNamespace() + + with mock.patch("app.image_studio.chrome.is_running", return_value=True), \ + mock.patch( + "app.image_studio.accounts.detect_login", + return_value={ + "logged_in": False, + "reason": "NO_SESSION_COOKIE", + "url": "https://seller.shopee.tw/portal/", + "cookie_names": [], + }, + ), \ + mock.patch("app.image_studio.editor.open_product", return_value=cdp) as open_product, \ + mock.patch( + "app.image_studio.editor.read_product_image_urls", + return_value=[{"index": 1, "src": "https://susercontent.com/main.jpg"}], + ), \ + mock.patch("app.image_studio.editor.close_readonly_product"): + result = image_studio.pull_remote_main_image_urls( + "alias-a", + "51100639510", + path=cfg["db_path"], + config=cfg, + ) + + self.assertTrue(result["readiness"]["login_uncertain"]) + open_product.assert_called_once() + + self.assert_removed(temp_dir) + + def test_pull_remote_main_image_urls_preserves_product_unavailable_toast_error(self): + with self.make_temp_dir() as temp_dir: + cfg = self._config(temp_dir) + accounts.create_account("主店", "alias-a", debug_port=9222, config=cfg) + + with mock.patch("app.image_studio.chrome.is_running", return_value=True), \ + mock.patch( + "app.image_studio.accounts.detect_login", + return_value={"logged_in": True}, + ), \ + mock.patch( + "app.image_studio.editor.open_product", + side_effect=editor.EditorError("商品失效:please input correct product id"), + ), \ + mock.patch("app.image_studio.editor.close_readonly_product") as close_readonly: + with self.assertRaisesRegex(image_studio.ImageStudioError, "please input correct product id"): + image_studio.pull_remote_main_image_urls( + "alias-a", + "bad-item", + path=cfg["db_path"], + config=cfg, + ) + + close_readonly.assert_not_called() + + self.assert_removed(temp_dir) + if __name__ == "__main__": unittest.main()