diff --git a/src/gyxx_flow/modules/product_commerce/collect_douyin_qianchuan_ads.py b/src/gyxx_flow/modules/product_commerce/collect_douyin_qianchuan_ads.py index 97086ae..ee958e8 100644 --- a/src/gyxx_flow/modules/product_commerce/collect_douyin_qianchuan_ads.py +++ b/src/gyxx_flow/modules/product_commerce/collect_douyin_qianchuan_ads.py @@ -47,6 +47,10 @@ TASK_CENTER_URL = ( "https://qianchuan.jinritemai.com/tools/task-center/download-task" "?aavid={advertiser_id}" ) +TASK_CENTER_PAGE_URL = TASK_CENTER_URL.format(advertiser_id=quote(ADVERTISER_ID)) +# 任务中心深链缺账户上下文时，千川会把页面重定向到登录域的账户选择页； +# 该页每个账户是一张 .account-list-item 卡片，user-id 含唯一广告主 ID。 +ACCOUNT_ITEM_SELECTOR = ".account-list-item" TASK_SCENE_MARKER = "全域推广数据-推商品" TASK_READY_MARKER = "下载任务列表" TASK_COMPLETE_MARKER = "生成完成" @@ -188,6 +192,70 @@ def _validate_task_center_page(page) -> None: raise DouyinQianchuanDownloadError("巨量千川下载任务中心未达到可用状态") +def _on_account_selection_page(page) -> bool: + try: + return page.locator(ACCOUNT_ITEM_SELECTOR).count() > 0 + except Exception: + return False + + +def _select_advertiser_account(page) -> None: + if not ADVERTISER_ID: + raise DouyinQianchuanDownloadError( + "巨量千川任务中心跳转到账户选择页，且未配置目标广告主 ID" + ) + account = page.locator(ACCOUNT_ITEM_SELECTOR).filter(has_text=ADVERTISER_ID) + if account.count() == 0: + raise DouyinQianchuanDownloadError( + f"巨量千川账户选择页上未找到目标广告主：{ADVERTISER_ID}" + ) + account.first.click() + + +def _wait_task_center_ready(page, *, timeout_ms: int = 120_000) -> None: + """等待任务中心可用；被重定向到账户选择页时点击目标广告主后继续等待。""" + + deadline = time.monotonic() + timeout_ms / 1000 + selections = 0 + regotos = 0 + while True: + if _on_account_selection_page(page): + if selections >= 2: + raise DouyinQianchuanDownloadError( + "巨量千川选择目标广告主后仍停留在账户选择页" + ) + selections += 1 + _select_advertiser_account(page) + page.wait_for_timeout(3_000) + continue + remaining_ms = int((deadline - time.monotonic()) * 1000) + if remaining_ms <= 0: + raise DouyinQianchuanDownloadError( + "巨量千川下载任务中心在等待期间未达到可用状态" + ) + try: + page.wait_for_function( + "(marker) => (document.body?.innerText || '').includes(marker)", + arg=TASK_READY_MARKER, + timeout=min(15_000, remaining_ms), + ) + return + except Exception as exc: + message = str(exc).casefold() + if ( + not is_browser_timeout_error(exc) + and "execution context was destroyed" not in message + ): + raise + # 等待落空且已不在账户选择页：账户选择后若未自动回到任务中心， + # 重新打开一次深链；后续轮次仅继续等待直至超时。 + if regotos < 1: + regotos += 1 + page.goto( + TASK_CENTER_PAGE_URL, wait_until="domcontentloaded", timeout=90_000 + ) + + def _download_task_from_page( page, start_date: date, @@ -206,15 +274,11 @@ def _download_task_from_page( except Exception as exc: if ( "execution context was destroyed" not in str(exc).casefold() - or navigation_retries >= 1 + or navigation_retries >= 2 ): raise navigation_retries += 1 - page.wait_for_function( - "(marker) => (document.body?.innerText || '').includes(marker)", - arg=TASK_READY_MARKER, - timeout=15_000, - ) + _wait_task_center_ready(page) _validate_task_center_page(page) continue if has_rows: @@ -248,15 +312,11 @@ def _download_from_task_center( task_page = context.new_page() try: task_page.goto( - TASK_CENTER_URL.format(advertiser_id=quote(ADVERTISER_ID)), + TASK_CENTER_PAGE_URL, wait_until="domcontentloaded", timeout=90_000, ) - task_page.wait_for_function( - "(marker) => (document.body?.innerText || '').includes(marker)", - arg=TASK_READY_MARKER, - timeout=90_000, - ) + _wait_task_center_ready(task_page) _validate_task_center_page(task_page) return _download_task_from_page( task_page, diff --git a/src/gyxx_flow/modules/product_commerce/jd_video_metadata.py b/src/gyxx_flow/modules/product_commerce/jd_video_metadata.py index 2166c55..6fcb715 100644 --- a/src/gyxx_flow/modules/product_commerce/jd_video_metadata.py +++ b/src/gyxx_flow/modules/product_commerce/jd_video_metadata.py @@ -1437,7 +1437,8 @@ def load_jd_product_references( """ SELECT style_name, jd_spus FROM dim_style - WHERE jd_spus IS NOT NULL AND cardinality(jd_spus) > 0 + WHERE jsonb_typeof(jd_spus) = 'array' + AND jsonb_array_length(jd_spus) > 0 """ ) rows = cursor.fetchall() diff --git a/src/gyxx_flow/modules/product_commerce/upload_video_to_guanghe.py b/src/gyxx_flow/modules/product_commerce/upload_video_to_guanghe.py index ab4f0d8..92e8d85 100644 --- a/src/gyxx_flow/modules/product_commerce/upload_video_to_guanghe.py +++ b/src/gyxx_flow/modules/product_commerce/upload_video_to_guanghe.py @@ -503,7 +503,8 @@ def load_tm_item_ids_for_styles(style_names: List[str]) -> Dict[str, List[str]]: """ SELECT style_name, tm_spus FROM dim_style - WHERE tm_spus IS NOT NULL AND cardinality(tm_spus) > 0 + WHERE jsonb_typeof(tm_spus) = 'array' + AND jsonb_array_length(tm_spus) > 0 """ ) rows = cur.fetchall() diff --git a/tests/modules/product_commerce/test_collect_douyin_qianchuan_ads.py b/tests/modules/product_commerce/test_collect_douyin_qianchuan_ads.py index e2f8e4d..ce9cfa2 100644 --- a/tests/modules/product_commerce/test_collect_douyin_qianchuan_ads.py +++ b/tests/modules/product_commerce/test_collect_douyin_qianchuan_ads.py @@ -4,6 +4,7 @@ from datetime import date from pathlib import Path import collect_douyin_qianchuan_ads as collector +import pytest def test_qianchuan_browser_uses_scrapling_stealth_and_persistent_state( @@ -216,3 +217,174 @@ def test_download_task_from_page_downloads_completed_matching_row() -> None: collector._matching_task_rows = original assert actual is expected_download + + +def test_wait_task_center_ready_selects_advertiser_after_redirect() -> None: + clicked: list[str] = [] + state = {"on_selection": True} + + class _AccountCard: + def click(self) -> None: + clicked.append("account") + state["on_selection"] = False + + class _FilteredLocator: + first = _AccountCard() + + def count(self) -> int: + return 1 + + class _AccountListLocator: + def count(self) -> int: + return 1 if state["on_selection"] else 0 + + def filter(self, *, has_text: str): + assert has_text == collector.ADVERTISER_ID + return _FilteredLocator() + + class _Page: + def locator(self, selector: str) -> _AccountListLocator: + assert selector == collector.ACCOUNT_ITEM_SELECTOR + return _AccountListLocator() + + def wait_for_timeout(self, ms: int) -> None: + pass + + def wait_for_function(self, expression: str, *, arg, timeout: int) -> None: + assert not state["on_selection"] + assert arg == collector.TASK_READY_MARKER + + collector._wait_task_center_ready(_Page()) + + assert clicked == ["account"] + + +def test_wait_task_center_ready_raises_when_selection_page_persists() -> None: + class _AccountCard: + def click(self) -> None: + pass + + class _FilteredLocator: + first = _AccountCard() + + def count(self) -> int: + return 1 + + class _AccountListLocator: + def count(self) -> int: + return 1 + + def filter(self, *, has_text: str): + assert has_text == collector.ADVERTISER_ID + return _FilteredLocator() + + class _Page: + def locator(self, selector: str) -> _AccountListLocator: + assert selector == collector.ACCOUNT_ITEM_SELECTOR + return _AccountListLocator() + + def wait_for_timeout(self, ms: int) -> None: + pass + + def wait_for_function(self, expression: str, *, arg, timeout: int) -> None: + raise AssertionError("不应在账户选择页上等待任务中心标记") + + with pytest.raises( + collector.DouyinQianchuanDownloadError, match="仍停留在账户选择页" + ): + collector._wait_task_center_ready(_Page()) + + +def test_wait_task_center_ready_raises_when_never_ready(monkeypatch) -> None: + class _AccountListLocator: + def count(self) -> int: + return 0 + + class _Page: + def locator(self, selector: str) -> _AccountListLocator: + assert selector == collector.ACCOUNT_ITEM_SELECTOR + return _AccountListLocator() + + def wait_for_function(self, expression: str, *, arg, timeout: int) -> None: + raise RuntimeError("Page.wait_for_function: Timeout 15000ms exceeded.") + + def goto(self, url: str, **options: object) -> None: + assert url == collector.TASK_CENTER_PAGE_URL + + monkeypatch.setattr(collector, "is_browser_timeout_error", lambda exc: True) + + with pytest.raises( + collector.DouyinQianchuanDownloadError, match="未达到可用状态" + ): + collector._wait_task_center_ready(_Page(), timeout_ms=50) + + +def test_download_task_from_page_recovers_from_context_destroyed( + monkeypatch, +) -> None: + expected_download = object() + state = {"count_calls": 0} + + class _DownloadInfo: + value = expected_download + + def __enter__(self): + return self + + def __exit__(self, exc_type, exc, traceback) -> None: + pass + + class _DownloadButton: + def click(self) -> None: + pass + + class _Row: + def inner_text(self) -> str: + return "全域推广数据-推商品\n生成完成\n下载" + + def get_by_text(self, text: str, *, exact: bool) -> _DownloadButton: + assert (text, exact) == ("下载", True) + return _DownloadButton() + + class _Rows: + first = _Row() + + def count(self) -> int: + state["count_calls"] += 1 + if state["count_calls"] == 1: + raise RuntimeError( + "Locator.count: Execution context was destroyed, " + "most likely because of a navigation." + ) + return 1 + + class _Body: + def inner_text(self, *, timeout: int) -> str: + assert timeout == 10_000 + return collector.TASK_READY_MARKER + + class _Page: + url = "https://qianchuan.jinritemai.com/tools/task-center/download-task" + + def expect_download(self, *, timeout: int) -> _DownloadInfo: + assert timeout == 60_000 + return _DownloadInfo() + + def locator(self, selector: str) -> _Body: + assert selector == "body" + return _Body() + + monkeypatch.setattr(collector, "_matching_task_rows", lambda *args: _Rows()) + monkeypatch.setattr( + collector, "_wait_task_center_ready", lambda page, **kwargs: None + ) + + actual = collector._download_task_from_page( + _Page(), + date(2026, 9, 29), + date(2026, 9, 29), + wait_timeout_ms=0, + ) + + assert actual is expected_download + assert state["count_calls"] == 2