fix(product): 千川任务中心账户选择页兜底与 jd_spus jsonb 查询修复

千川任务中心深链缺账户上下文时会被重定向到账户选择页,按 .account-list-item
卡片自动选中目标广告主后继续下载;jd_spus 列改 jsonb 后查询改用
jsonb_typeof/jsonb_array_length 防类型报错。
This commit is contained in:
wangyunlong committed 2026-10-09 13:45:30 +08:00
1 parent df11e195b7
commit 90282ae766
4 files changed
+248 -14

No files matched your search

@@ -47,6 +47,10 @@ TASK_CENTER_URL = (
"https://qianchuan.jinritemai.com/tools/task-center/download-task" "https://qianchuan.jinritemai.com/tools/task-center/download-task"
"?aavid={advertiser_id}" "?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_SCENE_MARKER = "全域推广数据-推商品"
TASK_READY_MARKER = "下载任务列表" TASK_READY_MARKER = "下载任务列表"
TASK_COMPLETE_MARKER = "生成完成" TASK_COMPLETE_MARKER = "生成完成"
@@ -188,6 +192,70 @@ def _validate_task_center_page(page) -> None:
raise DouyinQianchuanDownloadError("巨量千川下载任务中心未达到可用状态") 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( def _download_task_from_page(
page, page,
start_date: date, start_date: date,
@@ -206,15 +274,11 @@ def _download_task_from_page(
except Exception as exc: except Exception as exc:
if ( if (
"execution context was destroyed" not in str(exc).casefold() "execution context was destroyed" not in str(exc).casefold()
or navigation_retries >= 1 or navigation_retries >= 2
): ):
raise raise
navigation_retries += 1 navigation_retries += 1
page.wait_for_function( _wait_task_center_ready(page)
"(marker) => (document.body?.innerText || '').includes(marker)",
arg=TASK_READY_MARKER,
timeout=15_000,
)
_validate_task_center_page(page) _validate_task_center_page(page)
continue continue
if has_rows: if has_rows:
@@ -248,15 +312,11 @@ def _download_from_task_center(
task_page = context.new_page() task_page = context.new_page()
try: try:
task_page.goto( task_page.goto(
TASK_CENTER_URL.format(advertiser_id=quote(ADVERTISER_ID)), TASK_CENTER_PAGE_URL,
wait_until="domcontentloaded", wait_until="domcontentloaded",
timeout=90_000, timeout=90_000,
) )
task_page.wait_for_function( _wait_task_center_ready(task_page)
"(marker) => (document.body?.innerText || '').includes(marker)",
arg=TASK_READY_MARKER,
timeout=90_000,
)
_validate_task_center_page(task_page) _validate_task_center_page(task_page)
return _download_task_from_page( return _download_task_from_page(
task_page, task_page,
@@ -1437,7 +1437,8 @@ def load_jd_product_references(
""" """
SELECT style_name, jd_spus SELECT style_name, jd_spus
FROM dim_style 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() rows = cursor.fetchall()
@@ -503,7 +503,8 @@ def load_tm_item_ids_for_styles(style_names: List[str]) -> Dict[str, List[str]]:
""" """
SELECT style_name, tm_spus SELECT style_name, tm_spus
FROM dim_style 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() rows = cur.fetchall()
@@ -4,6 +4,7 @@ from datetime import date
from pathlib import Path from pathlib import Path
import collect_douyin_qianchuan_ads as collector import collect_douyin_qianchuan_ads as collector
import pytest
def test_qianchuan_browser_uses_scrapling_stealth_and_persistent_state( 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 collector._matching_task_rows = original
assert actual is expected_download 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