feat(ai-studio): read shopee source image urls
This commit is contained in:
chengma
+242
-1
@@ -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()
|
||||
|
||||
Reference in New Issue
Block a user