Compare commits

..
7 Commits
Author SHA1 Message Date
wangyunlong cf85985753 chore: 忽略 .mimosa/.workbuddy 本地工具状态目录 2026-10-14 18:40:11 +08:00
wangyunlong e5761bd828 feat(deploy): Windows 自动登录自启脚本与运行手册补章
autologon.ps1 交互输密码启用/撤销 AutoAdminLogon(DefaultPassword 仅
Administrators 可读),补齐「开机自动登录 → Startup keepalive → 控制台+
调度器」自启链唯一缺口;采集依赖 CDP 交互会话,不能用服务/SYSTEM 任务
替代。runbook 增补开机自启与重启验收章节。
2026-10-14 18:40:11 +08:00
wangyunlong 9feb188d93 fix(cookie-gate): 必需 cookie 改全量校验并收紧各账号必需名单
required_cookie_names 原语义是「任一必需名命中即有效」,单个匿名 cookie
(如小红书 a1)就能把游客会话判为有效;改为全部必需名都存在、存活且域名
匹配才算通过。随之按实际登录产物收紧名单:jd 去 pin、erp321 留
u_id+u_co_id、小红书改 customer-sso-sid+solar.beaker.session.id、
星图仅 sid_guard。cookie_gate 用例冻结时钟改相对时钟防过期漂移。
2026-10-14 18:38:03 +08:00
wangyunlong 7b305b0b12 fix(content): 蝉妈妈导出标签页崩溃自愈与 cookie 文件格式对齐
导出点击后 tab 死亡(渲染进程被清理)时:页面级死亡先同 context 重开
新页(facade 新增 reopen_page),context 也死亡才整套重建驱动,按
profile → cookie 文件 → 账密顺序恢复会话,两种自愈各限一次防死循环。
cookie 文件改存裸列表,与 vault 同步及适配器 BrowserCookieStore 三方
格式对齐(字典格式会被适配器拒绝加载);adapters 暴露
is_browser_closed_error 供模块识别目标已关闭类异常。
2026-10-14 18:35:44 +08:00
wangyunlong 6cd563eab0 fix(product): 天猫百补报名登录链路三类阻断兜底
vault 假登录态(SPA 缓存店铺名误判已登录)与 SSO 中间态停留时改走显式
账密登录兜底;登录页识别服务协议/隐私弹窗并先处理,二次验证标记补充
滑块;报名导入增加账号级锁防并发挤兑。
2026-10-14 18:33:32 +08:00
wangyunlong 0cc4fafc7c fix(product): 聚水潭 allsku 模板隐藏列错位致 daily 销量全 0
2026-10-11 模板重置把 38 列置隐藏后可见列整体左移,固定索引 45/57 落到
「今日实退货量/今日仅退款量」。daily 改表头名匹配「昨日销量/昨日退货量」
(weekly 暂留老索引待实测);backfill 恢复自定义列可见(window._Cols +
_RptRefresh),并把表头索引按可见列换算成行内位置防 td 缺位错读。
2026-10-14 18:31:23 +08:00
wangyunlong 9a91ce80e4 feat(product): 聚水潭销售主题分析按日采集工作流
collect_erp_sales_theme_daily.py 采集 bi.erp321.com「报表→销售主题分析→按日统计」,
异步导出 + OSS 直连兜底,固定品牌+店铺口径,写 erp_sales_theme_daily_metrics
(db 层建表与 upsert 同步,每日一行幂等覆盖)。注册 product.erp_sales_theme_daily.collect
手动命令,复用 erp-account 登录绑定新增 CDP 22167;目录/集成计数 60→61,
新脚本纳入 CI 聚焦 lint 清单。
2026-10-14 18:29:15 +08:00
24 changed files with 2607 additions and 110 deletions

No files matched your search

+4
View File
@@ -33,6 +33,10 @@ Thumbs.db
# Agent-local learning logs
.learnings/
# Agent tool local state (security-scan ledger, session memory)
/.mimosa/
/.workbuddy/
# Local tool artifacts
/.playwright-cli/
/output/
+1 -1
View File
@@ -19,7 +19,7 @@ lint:
stage: quality
script:
- uv run ruff check src tests
- uv run ruff check src/gyxx_flow/modules/content_marketing/daily_creator_exposure_scope.py src/gyxx_flow/modules/content_marketing/chanmama_export.py src/gyxx_flow/modules/content_marketing/self_operated_accounts.py src/gyxx_flow/modules/content_marketing/self_xhs_scraper.py src/gyxx_flow/modules/content_marketing/data/tools/sync_self_video_metrics.py src/gyxx_flow/modules/product_commerce/market_rank_limits.py src/gyxx_flow/modules/product_commerce/market_rank_product_import.py src/gyxx_flow/modules/product_commerce/collect_douyin_qianchuan_ads.py src/gyxx_flow/modules/shop_intelligence/collectors/dy_after_sales_quality.py src/gyxx_flow/modules/shop_intelligence/collectors/tmall_negative_reviews.py
- uv run ruff check src/gyxx_flow/modules/content_marketing/daily_creator_exposure_scope.py src/gyxx_flow/modules/content_marketing/chanmama_export.py src/gyxx_flow/modules/content_marketing/self_operated_accounts.py src/gyxx_flow/modules/content_marketing/self_xhs_scraper.py src/gyxx_flow/modules/content_marketing/data/tools/sync_self_video_metrics.py src/gyxx_flow/modules/product_commerce/market_rank_limits.py src/gyxx_flow/modules/product_commerce/market_rank_product_import.py src/gyxx_flow/modules/product_commerce/collect_douyin_qianchuan_ads.py src/gyxx_flow/modules/product_commerce/collect_erp_sales_theme_daily.py src/gyxx_flow/modules/shop_intelligence/collectors/dy_after_sales_quality.py src/gyxx_flow/modules/shop_intelligence/collectors/tmall_negative_reviews.py
test:
stage: test
+11
View File
@@ -465,6 +465,17 @@
"{business_date}"
]
},
{
"id": "product.erp_sales_theme_daily.collect",
"module": "product_commerce",
"entry": "collect_erp_sales_theme_daily.py",
"kind": "python",
"default_args": [
"--date",
"{business_date}",
"--execute"
]
},
{
"id": "product.hk_anpei_public_inventory.refresh",
"module": "product_commerce",
+23 -14
View File
@@ -88,7 +88,6 @@
"jd.com"
],
"required_cookie_names": [
"pin",
"thor"
],
"credential_env_names": [
@@ -116,7 +115,6 @@
"jd.com"
],
"required_cookie_names": [
"pin",
"thor"
],
"credential_env_names": [
@@ -145,10 +143,7 @@
],
"required_cookie_names": [
"u_id",
"u_co_id",
"ASP.NET_SessionId",
".ASPXAUTH",
"token"
"u_co_id"
],
"credential_env_names": [
"ERP_USERNAME",
@@ -168,6 +163,7 @@
"product_commerce:collect_erp_yesterday_metrics.py",
"product_commerce:erp_login_product_analysis.py",
"product.erp_all_shop_daily",
"product.erp_sales_theme_daily.collect",
"product.sales_sheet.sync",
"supply_chain:orchestrator/scripts/ProductReplenishment.py",
"supply_chain:orchestrator/scripts/PurchaseConfirmation.py",
@@ -298,7 +294,8 @@
"xiaohongshu.com"
],
"required_cookie_names": [
"web_session",
"customer-sso-sid",
"solar.beaker.session.id",
"a1"
],
"keepalive": {
@@ -350,9 +347,6 @@
"xingtu.cn"
],
"required_cookie_names": [
"sessionid",
"sessionid_ss",
"sid_tt",
"sid_guard"
],
"keepalive": {
@@ -364,8 +358,7 @@
"headless": true,
"real_chrome": true
},
"members": [
]
"members": []
},
"chanmama": {
"cdp_port": 22207,
@@ -701,8 +694,12 @@
],
"cdp_port": 22165,
"login_mode": "C",
"required_cookie_domains": ["xiaohongshu.com"],
"credential_env_names": ["XHS_ARK_ACCOUNT"]
"required_cookie_domains": [
"xiaohongshu.com"
],
"credential_env_names": [
"XHS_ARK_ACCOUNT"
]
},
"content_marketing:data/tools/sync_style_categories.py": {
"script_id": "content_marketing:data/tools/sync_style_categories.py",
@@ -1346,6 +1343,18 @@
"erp321.com"
]
},
"product.erp_sales_theme_daily.collect": {
"script_id": "product_commerce:collect_erp_sales_theme_daily.py",
"state_key": "product_commerce:collect_erp_sales_theme_daily.py",
"aliases": [
"product_commerce:collect_erp_sales_theme_daily.py"
],
"cdp_port": 22167,
"login_mode": "C",
"required_cookie_domains": [
"erp321.com"
]
},
"product.hk_anpei_public_inventory.refresh": {
"script_id": "product_commerce:public_sku_inventory_sheet.py",
"state_key": "product_commerce:public_sku_inventory_sheet.py",
+47
View File
@@ -0,0 +1,47 @@
# GYXX Flow 开机自启:启用/关闭 Windows 自动登录
#
# 背景:控制台+调度器的自启链是
# 开机 → 自动登录进桌面 → Startup\gyxx-keepalive.vbs → keepalive → 控制台(8765)+调度器
# 唯一缺口是 AutoAdminLogon(默认 0,重启后停在登录界面,Startup 不触发)。
# 采集依赖浏览器 CDP,必须跑在交互式用户会话,因此不能用服务/Task Scheduler SYSTEM 替代。
#
# 启用(交互输入密码,不会留在命令历史):
# powershell -ExecutionPolicy Bypass -File D:\gyxx-flow\deploy\watchdog\autologon.ps1
# 撤销:
# powershell -ExecutionPolicy Bypass -File D:\gyxx-flow\deploy\watchdog\autologon.ps1 -Disable
#
# 注意:DefaultPassword 以明文存于 HKLM Winlogon(仅 Administrators 可读),
# 内网采集机可接受;介意安全性请改用 Sysinternals Autologon.exe(LSA secret 存储)。
param(
[switch]$Disable
)
$winlogon = 'HKLM:\SOFTWARE\Microsoft\Windows NT\CurrentVersion\Winlogon'
if ($Disable) {
Set-ItemProperty $winlogon -Name AutoAdminLogon -Value '0'
Remove-ItemProperty $winlogon -Name DefaultPassword -ErrorAction SilentlyContinue
Write-Host '已关闭自动登录(DefaultPassword 已清除)。' -ForegroundColor Yellow
exit 0
}
$pwd1 = Read-Host -AsSecureString "输入 $env:USERNAME 的 Windows 登录密码"
$pwd2 = Read-Host -AsSecureString '再输一次确认'
$plain1 = [Net.NetworkCredential]::new('', $pwd1).Password
$plain2 = [Net.NetworkCredential]::new('', $pwd2).Password
if ($plain1 -ne $plain2) { throw '两次输入不一致,未做任何修改。' }
if ([string]::IsNullOrEmpty($plain1)) {
Write-Warning '密码为空:策略 LimitBlankPasswordUse=1 会拒绝空密码自动登录,请先给账户设置密码。'
exit 1
}
Set-ItemProperty $winlogon -Name AutoAdminLogon -Value '1'
Set-ItemProperty $winlogon -Name DefaultUserName -Value $env:USERNAME
Set-ItemProperty $winlogon -Name DefaultDomainName -Value $env:COMPUTERNAME
Set-ItemProperty $winlogon -Name DefaultPassword -Value $plain1
# AutoLogonCount 每次自动登录会递减,删掉表示不限次数
Remove-ItemProperty $winlogon -Name AutoLogonCount -ErrorAction SilentlyContinue
Write-Host '自动登录已启用。' -ForegroundColor Green
Write-Host '验证方式:挑空闲时间重启一次,开机应自动进桌面,约 2 分钟内控制台(127.0.0.1:8765)与调度器就位;'
Write-Host '也可查看 D:\gyxx-flow\var\logs\keepalive\keepalive.log 出现 keepalive-start / scheduler-restarted / console-restarted。'
+13
View File
@@ -161,6 +161,19 @@ uv run gyxx schedule run --env-file /etc/gyxx-flow/gyxx-flow.env
需要停用单个工作流时,优先修改该工作流的调度启用状态,先 dry-run,再重启服务。其他模块继续运行。
### 开机自启(Windows 生产机)
重启/关机开机后的自动恢复链路:**Windows 自动登录 → `Startup\gyxx-keepalive.vbs` → `deploy/watchdog/gyxx_keepalive.py` → 控制台(8765) + `gyxx schedule run` 调度器**。keepalive 每 30 秒巡检并负责拉起死掉的控制台/调度器(单实例锁防双跑,冷却 120s、熔断保护),日志在 `var/logs/keepalive/keepalive.log`,验收用 `python deploy/watchdog/gyxx_keepalive.py --once`。
采集依赖浏览器 CDP,必须运行在交互式用户会话中,因此不能改成 Windows 服务或 Task Scheduler SYSTEM 任务(也不允许注册 Windows 任务计划)。免登录自启的唯一环节是自动登录,用脚本启用(密码交互输入、写入 `HKLM\...\Winlogon`,仅 Administrators 可读):
```powershell
powershell -ExecutionPolicy Bypass -File D:\gyxx-flow\deploy\watchdog\autologon.ps1 # 启用
powershell -ExecutionPolicy Bypass -File D:\gyxx-flow\deploy\watchdog\autologon.ps1 -Disable # 撤销
```
启用后重启验证:开机自动进桌面,约 2 分钟内控制台与调度器就位;停机窗内到期任务由调度器 catch-up 自动补跑。
## 数据和浏览器状态
生产 `GYXX_DATA_ROOT` 必须位于源码目录之外,Linux 推荐 `/var/lib/gyxx-flow`:
+35 -15
View File
@@ -1210,29 +1210,49 @@ def _merge_storage_state(
def _account_cookie_valid(account: AccountBinding, cookies: object) -> bool:
"""A vault cookie set is valid when a live account cookie is present."""
"""A vault cookie set is valid when a live account cookie is present.
With ``required_cookie_names`` configured, every required name must be
present as a live, domain-matched cookie; a single anonymous cookie that
happens to share a required name (e.g. ``a1``) must not validate a guest
session. Without required names, any live cookie on a required domain
keeps accepting the set.
"""
if not isinstance(cookies, list):
return False
required_names = account.required_cookie_names
required_domains = account.required_cookie_domains
for item in cookies:
if not isinstance(item, dict):
continue
if _cookie_expired(item):
continue
if not str(item.get("value", "")):
continue
name = str(item.get("name", ""))
domain = str(item.get("domain", "")).casefold().lstrip(".")
name_matches = not required_names or name in required_names
domain_matches = not required_domains or any(
def _domain_matches(domain: str) -> bool:
return not required_domains or any(
domain == required.casefold()
or domain.endswith("." + required.casefold())
for required in required_domains
)
if name_matches and domain_matches:
return True
return False
def _live(item: object) -> bool:
return (
isinstance(item, dict)
and not _cookie_expired(item)
and bool(str(item.get("value", "")))
)
if not required_names:
return any(
_live(item)
and _domain_matches(
str(item.get("domain", "")).casefold().lstrip(".")
)
for item in cookies
)
outstanding = set(required_names)
for item in cookies:
if not _live(item):
continue
if _domain_matches(str(item.get("domain", "")).casefold().lstrip(".")):
outstanding.discard(str(item.get("name", "")))
return not outstanding
def _cookie_expired(cookie: Mapping[str, object], *, now: float | None = None) -> bool:
+3 -3
View File
@@ -37,8 +37,8 @@ def is_browser_timeout_error(error: BaseException) -> bool:
)
def _is_browser_closed_error(error: BaseException) -> bool:
"""Recognize a browser target that disappeared during cleanup."""
def is_browser_closed_error(error: BaseException) -> bool:
"""Recognize a browser target that disappeared (page/context/browser)."""
if any(
error_type.__name__ == "TargetClosedError"
@@ -155,7 +155,7 @@ class ScraplingBrowser:
# and context-manager cleanup. The current state cannot
# be exported then, but that must not turn a successful
# download into a false failure or mask the action error.
if not _is_browser_closed_error(exc):
if not is_browser_closed_error(exc):
raise
finally:
try:
@@ -16,6 +16,7 @@ import time
from datetime import datetime
from pathlib import Path
from gyxx_flow.adapters.scrapling import is_browser_closed_error
from gyxx_flow.core.exit_codes import COOKIE_SKIP_EXIT_CODE
from gyxx_flow.modules.content_marketing import chanmama_export
from gyxx_flow.modules.content_marketing.runtime_paths import PATHS
@@ -141,17 +142,18 @@ def create_driver():
def save_cookies(driver):
"""保存 cookie 到文件"""
"""保存 cookie 到文件。
写裸列表格式,与账号 vault 同步(sync_binding_state)和适配器
BrowserCookieStore 的绑定 cookie 文件格式保持一致 —— 同一文件
被三方读写,字典格式会让适配器下次启动/重建时直接拒绝加载。
"""
try:
cookies = driver.get_cookies()
data = {
"cookies": cookies,
"timestamp": time.time(),
}
cookie_path = Path(COOKIE_FILE)
cookie_path.parent.mkdir(parents=True, exist_ok=True)
temp_path = cookie_path.with_name(f".{cookie_path.name}.tmp")
temp_path.write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8")
temp_path.write_text(json.dumps(cookies, ensure_ascii=False, indent=2), encoding="utf-8")
os.replace(temp_path, cookie_path)
print(f"💾 已保存 cookie ({len(cookies)} 条) 到: {COOKIE_FILE}")
return True
@@ -163,6 +165,9 @@ def save_cookies(driver):
def load_cookies(driver):
"""从文件加载 cookie 到浏览器
兼容两种格式: 裸列表(vault/适配器格式,按文件 mtime 判断时效)与
旧字典格式({"cookies": [...], "timestamp": ...})。
Returns:
bool: cookie 是否有效(有效期内)
"""
@@ -171,16 +176,20 @@ def load_cookies(driver):
return False
try:
data = json.loads(Path(COOKIE_FILE).read_text(encoding="utf-8"))
payload = json.loads(Path(COOKIE_FILE).read_text(encoding="utf-8"))
if isinstance(payload, list):
cookies = payload
timestamp = os.path.getmtime(COOKIE_FILE)
else:
cookies = payload.get("cookies", [])
timestamp = payload.get("timestamp", 0)
timestamp = data.get("timestamp", 0)
age = time.time() - timestamp
if age > COOKIE_MAX_AGE:
print(f"⏰ cookie 已过期 ({int(age // 86400)} 天前),需要重新登录")
return False
cookies = data.get("cookies", [])
if not cookies:
print("📭 cookie 为空,需要重新登录")
return False
@@ -1301,10 +1310,15 @@ def refresh_then_export_accounts(driver, urls, wait_seconds=REFRESH_WAIT_SECONDS
"""两阶段执行:先刷新所有账号全部分页,再等待并重新导出。
运行中途登录态失效(目标页被重定向到登录页)时,自动用账号密码重新
登录并重试当前账号一次;整套运行最多自动重登一次,避免死循环。
登录并重试当前账号一次;浏览器/页面被外部关闭(如进程树被清理)时,
自动重建浏览器并恢复会话后重试当前账号一次。两种自愈整套运行各最多
一次,避免死循环。返回 (记录列表, 刷新摘要, 最终生效的驱动) —— 驱动
可能在重建后已不是调用方传入的那个,调用方需用返回值收尾关闭。
"""
refresh_summaries = []
relogin_state = {"attempted": False}
rebuild_state = {"attempted": False}
driver_box = {"driver": driver}
def _relogin_if_expired(exc: BaseException) -> bool:
"""登录态失效类异常时自动重登一次;返回是否已恢复登录。"""
@@ -1313,7 +1327,7 @@ def refresh_then_export_accounts(driver, urls, wait_seconds=REFRESH_WAIT_SECONDS
relogin_state["attempted"] = True
print("🔐 检测到登录态失效,尝试账号密码自动重新登录...")
try:
ok = login(driver)
ok = login(driver_box["driver"])
except Exception as login_exc:
print(f"⚠️ 自动重新登录异常: {login_exc}")
return False
@@ -1321,6 +1335,51 @@ def refresh_then_export_accounts(driver, urls, wait_seconds=REFRESH_WAIT_SECONDS
print("✅ 自动重新登录成功,重试当前账号")
return bool(ok)
def _restore_session(current) -> bool:
"""重建浏览器后按 profile → cookie 文件 → 账密的顺序恢复登录态。"""
try:
if verify_login(current):
return True
if load_cookies(current) and verify_login(current):
return True
return bool(login(current))
except Exception as exc:
print(f"⚠️ 重建后会话恢复异常: {exc}")
return False
def _rebuild_if_browser_closed(exc: BaseException) -> bool:
"""浏览器/页面被外部关闭类异常时自愈一次;返回是否已恢复。
先尝试同一 context 重开新页(页面级死亡,登录态保留、代价最小);
context 也死了才整套重建驱动并按 profile → cookie → 账密恢复会话。
"""
if rebuild_state["attempted"] or not is_browser_closed_error(exc):
return False
rebuild_state["attempted"] = True
print("🌐 检测到浏览器页面被关闭,尝试自愈...")
current = driver_box["driver"]
try:
current.reopen_page()
print("✅ 浏览器 context 存活,已重开新页面,重试当前账号")
return True
except Exception as reopen_exc:
print(f"ℹ️ 页面重开失败({reopen_exc}),改为整套重建浏览器...")
try:
current.quit()
except Exception:
# 旧驱动随浏览器一起死亡,清理失败不影响重建。
pass
try:
driver_box["driver"] = create_driver()
except Exception as create_exc:
print(f"⚠️ 浏览器重建失败: {create_exc}")
return False
if not _restore_session(driver_box["driver"]):
print("⚠️ 重建后会话恢复失败,放弃自动重试")
return False
print("✅ 浏览器重建完成,重试当前账号")
return True
def _empty_summary(exc: BaseException) -> dict:
return {
"clicked": 0,
@@ -1336,15 +1395,15 @@ def refresh_then_export_accounts(driver, urls, wait_seconds=REFRESH_WAIT_SECONDS
print(f"🔄 刷新阶段 博主 {index}/{len(urls)}: {url}")
print(f"{'=' * 50}")
try:
navigate_to_target(driver, url)
summary = refresh_zero_exposure_videos(driver)
navigate_to_target(driver_box["driver"], url)
summary = refresh_zero_exposure_videos(driver_box["driver"])
except Exception as exc:
if _relogin_if_expired(exc):
if _relogin_if_expired(exc) or _rebuild_if_browser_closed(exc):
try:
navigate_to_target(driver, url)
summary = refresh_zero_exposure_videos(driver)
navigate_to_target(driver_box["driver"], url)
summary = refresh_zero_exposure_videos(driver_box["driver"])
except Exception as retry_exc:
print(f"⚠️ 博主 {index} 重登后刷新仍失败: {retry_exc},继续下一个")
print(f"⚠️ 博主 {index} 自愈后刷新仍失败: {retry_exc},继续下一个")
summary = _empty_summary(retry_exc)
else:
print(f"⚠️ 博主 {index} 刷新扫描异常: {exc},继续下一个")
@@ -1367,9 +1426,9 @@ def refresh_then_export_accounts(driver, urls, wait_seconds=REFRESH_WAIT_SECONDS
print(f"📌 导出阶段 博主 {index}/{len(urls)}: {url}")
print(f"{'=' * 50}")
try:
navigate_to_target(driver, url)
navigate_to_target(driver_box["driver"], url)
pre_mtime = _max_excel_mtime()
excel_path = export_video_data(driver, pre_mtime)
excel_path = export_video_data(driver_box["driver"], pre_mtime)
if not excel_path:
print(f"⚠️ 博主 {index} 导出失败,跳过")
continue
@@ -1377,23 +1436,23 @@ def refresh_then_export_accounts(driver, urls, wait_seconds=REFRESH_WAIT_SECONDS
print(f" 解析出 {len(records)} 条视频记录")
all_records.extend(records)
except Exception as exc:
if _relogin_if_expired(exc):
if _relogin_if_expired(exc) or _rebuild_if_browser_closed(exc):
try:
navigate_to_target(driver, url)
navigate_to_target(driver_box["driver"], url)
pre_mtime = _max_excel_mtime()
excel_path = export_video_data(driver, pre_mtime)
excel_path = export_video_data(driver_box["driver"], pre_mtime)
if not excel_path:
print(f"⚠️ 博主 {index} 重登后导出失败,跳过")
print(f"⚠️ 博主 {index} 自愈后导出失败,跳过")
continue
records, _ = parse_chanmama_export(excel_path)
print(f" 解析出 {len(records)} 条视频记录(重登后)")
print(f" 解析出 {len(records)} 条视频记录(自愈后)")
all_records.extend(records)
except Exception as retry_exc:
print(f"⚠️ 博主 {index} 重登后导出仍失败: {retry_exc},跳过")
print(f"⚠️ 博主 {index} 自愈后导出仍失败: {retry_exc},跳过")
else:
print(f"⚠️ 博主 {index} 导出异常: {exc},跳过")
return all_records, refresh_summaries
return all_records, refresh_summaries, driver_box["driver"]
def export_video_data(driver, pre_mtime: float = 0):
@@ -1569,6 +1628,8 @@ def _wait_for_exported_excel(
deadline 之后落盘的竞态;两个窗口内都未出现则返回 None。
传入 driver 时每轮执行一次无害的 evaluate 以泵 Playwright 事件循环,
确保 download 事件处理器(_save_download)在 sleep 轮询期间也能执行。
泵调用发现浏览器/页面已被外部关闭时,若文件已完整落盘则照常返回,
否则抛出原始关闭异常交由上层重建浏览器重试,不再空等满超时。
"""
deadline = time.time() + max(timeout_seconds, 0)
grace_deadline = deadline + max(grace_seconds, 0)
@@ -1577,8 +1638,17 @@ def _wait_for_exported_excel(
if driver is not None:
try:
driver.execute_script("1")
except Exception:
pass
except Exception as pump_exc:
if is_browser_closed_error(pump_exc):
# 浏览器/页面被外部关闭(进程树清理等): 先看文件是否
# 已在关闭前完整落盘,再决定放弃,避免误杀已完成下载。
settled = _find_new_excel_after(pre_mtime)
if settled is not None and (
not os.path.exists(settled) or _download_complete(settled)
):
return settled
raise
# 其余瞬态异常(页面导航中等)不致命,继续轮询文件系统。
new_path = _find_new_excel_after(pre_mtime)
# 文件出现 ≠ 下载完成: 当前浏览器可能以最终名边下边写,截断的
# zip 会让解析报 "File is not a zip file"。zip 的 EOCD 记录在
@@ -2290,7 +2360,8 @@ def main():
print("⚠️ 登录可能未成功,但继续尝试后续步骤...")
# 4. 先刷新两个账号的全部分页,再统一等待并重新进入页面导出。
all_records, _refresh_summaries = refresh_then_export_accounts(
# 返回的驱动可能已在浏览器被外部关闭后重建,后续收尾必须用它。
all_records, _refresh_summaries, driver = refresh_then_export_accounts(
driver, TARGET_URLS, wait_seconds=REFRESH_WAIT_SECONDS
)
@@ -229,6 +229,24 @@ class ScraplingDriver:
def quit(self) -> None:
self._owner.close()
def reopen_page(self) -> Any:
"""Rebind to a fresh page inside the still-live context.
A page/tab can die (site closed it, renderer gone) while the browser
context survives with its login state. Returns the new page, or
raises the underlying error when the context/browser is gone too —
the caller should rebuild the whole driver then.
"""
context = self._owner.context
if context is None:
raise RuntimeError("Scrapling browser context is closed")
# 永远开新页: context.pages 里可能还挂着半死的目标页。
page = context.new_page()
self._page = page
self._owner.page = page
page.on("download", self._save_download)
return page
def _save_download(self, download: Any) -> None:
target = self._download_dir / download.suggested_filename
download.save_as(str(target))
@@ -205,6 +205,44 @@ def set_all_shop_daily_filters(
)
ENSURE_CUSTOM_COLUMNS_JS = r"""
() => {
const cols = window._Cols || [];
let changed = 0;
for (const col of cols) {
const label = String(col.text || col.defaultText || '');
if ((label === '自定义销量' || label === '自定义退货量') && col.visible === false) {
col.visible = true;
changed += 1;
}
}
if (changed > 0 && typeof window._RptRefresh === 'function') {
try { window._RptRefresh(); } catch (e) {}
}
return changed;
}
"""
def ensure_custom_columns_visible(page: Any) -> int:
"""Keep 自定义销量/自定义退货量 visible before the report renders.
2026-10-11 起聚水潭报表模板把自定义公式列重置为隐藏,隐藏列不渲染
数据单元格;只改 visible 不会自动重绘,需顺带触发一次 _RptRefresh
(其尾部浮头重建可能抛错,但不影响数据行渲染)。
"""
changed_total = 0
for frame in page.frames:
if "allsku.aspx" not in (frame.url or "").lower():
continue
try:
changed_total += int(frame.evaluate(ENSURE_CUSTOM_COLUMNS_JS))
except Exception:
continue
return changed_total
def extract_style_rows(
table_frame: Any,
requested_codes: Sequence[str],
@@ -257,18 +295,45 @@ def extract_style_rows(
};
}
// 隐藏列的 th 仍在表头,但数据行不渲染对应 td:数据行的第 j 格
// 对应第 j 个可见列。用 window._Cols 把表头列索引换算成行内位
// 置,避免隐藏列造成索引错位;拿不到 _Cols 时按原索引直取。
let stylePos = styleCol;
let productPos = productCol;
let namePos = productNameCol;
let salesPos = salesCol;
let returnsPos = returnsCol;
if (Array.isArray(window._Cols) && window._Cols.length === headerTexts.length) {
const visibleIdx = [];
for (let i = 0; i < window._Cols.length; i++) {
if (window._Cols[i].visible !== false) visibleIdx.push(i);
}
stylePos = visibleIdx.indexOf(styleCol);
productPos = productCol >= 0 ? visibleIdx.indexOf(productCol) : -1;
namePos = productNameCol >= 0 ? visibleIdx.indexOf(productNameCol) : -1;
salesPos = visibleIdx.indexOf(salesCol);
returnsPos = visibleIdx.indexOf(returnsCol);
}
if (stylePos < 0 || salesPos < 0 || returnsPos < 0) {
return {
ok: false,
error: `custom sales/return columns hidden: style=${stylePos}, sales=${salesPos}, returns=${returnsPos}`,
header_texts: headerTexts,
};
}
const rows = [];
for (const tr of Array.from(document.querySelectorAll('tr'))) {
const cells = Array.from(tr.querySelectorAll('td, th'));
if (cells.length <= Math.max(styleCol, salesCol, returnsCol)) continue;
const styleCode = norm(cells[styleCol]?.textContent);
if (cells.length <= Math.max(stylePos, salesPos, returnsPos)) continue;
const styleCode = norm(cells[stylePos]?.textContent);
if (!requested.has(styleCode)) continue;
rows.push({
erp_style_code: styleCode,
product_code: productCol >= 0 ? norm(cells[productCol]?.textContent) : '',
product_name: productNameCol >= 0 ? norm(cells[productNameCol]?.textContent) : '',
sales: parseInteger(cells[salesCol]?.textContent),
return_qty: parseInteger(cells[returnsCol]?.textContent),
product_code: productPos >= 0 ? norm(cells[productPos]?.textContent) : '',
product_name: namePos >= 0 ? norm(cells[namePos]?.textContent) : '',
sales: parseInteger(cells[salesPos]?.textContent),
return_qty: parseInteger(cells[returnsPos]?.textContent),
});
}
const foundCodes = Array.from(new Set(rows.map(row => row.erp_style_code))).sort();
@@ -326,6 +391,7 @@ def wait_for_style_rows(
reload_deadline = time.monotonic() + reload_after_seconds
continue
ensure_custom_columns_visible(page)
for frame in page.frames:
if "allsku.aspx" not in (frame.url or "").lower():
continue
@@ -459,6 +525,7 @@ def collect_one_date(
selection = set_all_shop_daily_filters(
filter_frame, report_date=report_date, erp_codes=codes
)
ensure_custom_columns_visible(page)
erp.generate_report(filter_frame)
report = wait_for_style_rows(
page,
@@ -0,0 +1,814 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""聚水潭「报表→销售主题分析→按日统计」Excel 导出采集入库。
流程(2026-10-12 实测探测):
1. 复用 erp-jushuitan 账号登录态(CDP/profile + vault cookie 注入)直开
https://bi.erp321.com/app/daas/report/subject/adsorder/default.aspx
2. filter.aspx: 发货时间自定义区间 + 勾选固定品牌 + 勾选固定店铺清单
3. 生成报表(BtnFullSearch) → 点「按日统计」tab → day.aspx 出数
4. 点「导出EXCEL」→ Playwright download 事件落盘 xlsx
5. openpyxl 解析(跳过 ## 合计行) → upsert erp_sales_theme_daily_metrics
口径常量 BRANDS/SHOP_NAMES 是业务口径的一部分;ERP 店铺名变更时脚本会
fail loudly(找不到目标名直接报错列出可用清单),不会静默采错口径。
"""
from __future__ import annotations
import argparse
import json
import re
import sys
from datetime import date, datetime, timedelta
from pathlib import Path
from typing import Any
try:
from .collect_erp_yesterday_metrics import resolve_browser_session
from .collect_retry_utils import retry_step
from .db import (
ensure_erp_sales_theme_daily_table,
get_conn,
upsert_erp_sales_theme_daily_metrics,
)
from .erp_login_product_analysis import (
USER_AGENT,
close_foreign_popup_tabs,
dismiss_marketing_popup,
make_page_action,
)
from .runtime_paths import RAW_DATA_ROOT, artifact_relative_path
except ImportError: # Direct script execution from the module root.
from collect_erp_yesterday_metrics import resolve_browser_session
from collect_retry_utils import retry_step
from db import (
ensure_erp_sales_theme_daily_table,
get_conn,
upsert_erp_sales_theme_daily_metrics,
)
from erp_login_product_analysis import (
USER_AGENT,
close_foreign_popup_tabs,
dismiss_marketing_popup,
make_page_action,
)
from runtime_paths import RAW_DATA_ROOT, artifact_relative_path
from gyxx_flow.core.exit_codes import COOKIE_SKIP_EXIT_CODE
REPORT_URL = "https://bi.erp321.com/app/daas/report/subject/adsorder/default.aspx"
LOGIN_URL_DEFAULT = "https://www.erp321.com/login.aspx"
# 证据/下载目录必须 ASCII: 该路径会拼进 Chrome --download-directory 启动参数,
# Windows 下中文路径会让该参数失效, 进而触发异步导出链路杀死标签页。
EVIDENCE_DIR_NAME = "erp_sales_theme_daily"
# ── 业务口径常量(与页面筛选一致) ─────────────────────────────────────────────
# 商品品牌: 光影行星 / 光影 / gyxx (2026-10-12 截图勾选项)
BRANDS: tuple[str, ...] = ("光影行星", "光影", "gyxx")
# 销售渠道(店铺): 光影系 22 家渠道(2026-10-12 用户确认补回 JD光影行星旗舰店
# 与 得物(直发)-光影行星)。名称必须与 filter.aspx 复选框标签逐字一致;
# ERP 店铺名变更时脚本会 fail loudly 并列出可用清单。
SHOP_NAMES: tuple[str, ...] = (
"JD光影行星旗舰店",
"得物(直发)-光影行星",
"得物-光影行星",
"抖店-光影行星数码旗舰店",
"抖店-光影行星箱包旗舰店",
"光影华贸自营店",
"光影行星独立站",
"光影行星-快手",
"光影行星-视频号",
"光影行星-微信商城",
"光影行星唯品会",
"海外仓",
"京东商城-京东自营入仓专用",
"拼多多-光影行星旗舰店",
"拼多多-光影行星箱包旗舰店",
"拼多多-光影行星专卖店",
"实体店分销",
"天猫-GYXX箱包旗舰店",
"天猫-光影行星旗舰店",
"天猫-光影行星箱包旗舰店",
"香港安沛",
"小红书-光影行星旗舰店",
)
# 导出 Excel 的 17 列表头(2026-10-12 实测) → 记录字段
COLUMN_MAP: dict[str, str] = {
"日期": "stat_date",
"销售单数": "sales_order_count",
"销售数量": "sales_qty",
"实发数量": "shipped_qty",
"实发金额": "shipped_amount",
"销售金额": "sales_amount",
"退货单数": "return_order_count",
"退货数量": "return_qty",
"实退数量": "returned_qty",
"退货金额": "return_amount",
"实退金额": "returned_amount",
"净销售单数": "net_sales_order_count",
"净销量": "net_sales_qty",
"净销售额": "net_sales_amount",
"运费收入": "freight_income",
"运费支出": "freight_expense",
"实发单数": "shipped_order_count",
}
TOTAL_ROW_MARKERS = {"#", "##", "总计", "合计"}
_DATE_RE = re.compile(r"^\d{4}-\d{2}-\d{2}$")
# ── 纯解析层(无浏览器依赖, 便于测试) ────────────────────────────────────────
def parse_number(value: Any) -> int:
text = str(value if value is not None else "").replace(",", "").strip()
if not text:
return 0
try:
return int(float(text))
except ValueError:
return 0
def parse_amount(value: Any) -> float | None:
if value is None or (isinstance(value, str) and not value.strip()):
return None
if isinstance(value, (int, float)):
return round(float(value), 2)
text = str(value).replace(",", "").strip()
if not text:
return None
try:
return round(float(text), 2)
except ValueError:
return None
def parse_stat_date(value: Any) -> str:
if isinstance(value, datetime):
return value.date().isoformat()
text = str(value if value is not None else "").strip()
if _DATE_RE.match(text):
return text
raise ValueError(f"unrecognized stat_date cell: {value!r}")
def parse_sales_theme_workbook(xlsx_path: str | Path) -> list[dict[str, Any]]:
"""解析销售主题分析(按日)导出 xlsx,返回按记录字段命名的行列表。
跳过末尾 ## 合计行;列按表头名映射(顺序无关),缺列按 0/None 补齐。
"""
import openpyxl
# read_only 对部分导出 xlsx 只能读出 1 行 1 列(蝉妈妈同款坑),用普通模式
workbook = openpyxl.load_workbook(str(xlsx_path), data_only=True)
try:
sheet = workbook.active
rows = sheet.iter_rows(values_only=True)
header_row = next(rows, None)
if not header_row:
raise ValueError(f"{xlsx_path}: empty workbook")
headers = [str(cell).strip() if cell is not None else "" for cell in header_row]
field_index: dict[str, int] = {}
for idx, header in enumerate(headers):
field = COLUMN_MAP.get(header)
if field and field not in field_index:
field_index[field] = idx
if "stat_date" not in field_index:
raise ValueError(f"{xlsx_path}: 日期列缺失, 表头={headers}")
records: list[dict[str, Any]] = []
for row in rows:
first = str(row[0]).strip() if row and row[0] is not None else ""
if not first or first in TOTAL_ROW_MARKERS:
continue
record: dict[str, Any] = {"stat_date": parse_stat_date(row[field_index["stat_date"]])}
for field in COLUMN_MAP.values():
if field == "stat_date":
continue
idx = field_index.get(field)
raw = row[idx] if idx is not None and idx < len(row) else None
if field.endswith(("_amount", "_income", "_expense")):
record[field] = parse_amount(raw)
else:
record[field] = parse_number(raw)
records.append(record)
return records
finally:
workbook.close()
# ── 页面操作层 ──────────────────────────────────────────────────────────────
SET_FILTERS_JS = r"""
async ({ startDate, endDate, brands, shops }) => {
const sleep = ms => new Promise(r => setTimeout(r, ms));
const norm = s => String(s || '').replace(/\s+/g, '').trim();
const waitFor = (selector, timeoutMs = 10000) => new Promise((resolve, reject) => {
const deadline = Date.now() + timeoutMs;
const tick = () => {
const el = document.querySelector(selector);
if (el) return resolve(el);
if (Date.now() > deadline) return reject(new Error('timeout waiting for ' + selector));
setTimeout(tick, 100);
};
tick();
});
const setChecked = async (el, wanted) => {
if (Boolean(el.checked) !== wanted) el.click();
for (let i = 0; i < 40 && Boolean(el.checked) !== wanted; i++) await sleep(50);
if (Boolean(el.checked) !== wanted) throw new Error('checkbox stuck: ' + (el.id || el.value));
};
// 时间口径必须是「发货时间」;否则宁报错也不采错口径
const desc = document.querySelector('#date_desc');
const descText = norm(desc && desc.innerText);
if (!descText.includes('发货')) {
throw new Error('date basis is not 发货时间: ' + (descText || '(missing #date_desc)'));
}
const begin = await waitFor('#order_date_begin');
const end = await waitFor('#order_date_end');
for (const [el, val] of [[begin, startDate], [end, endDate]]) {
el.value = val;
el.dispatchEvent(new Event('input', { bubbles: true }));
el.dispatchEvent(new Event('change', { bubbles: true }));
el.blur();
}
if (begin.value !== startDate || end.value !== endDate) {
throw new Error('date did not persist: ' + begin.value + ' ~ ' + end.value);
}
// 品牌: 全清后勾目标, 回读
const brandTargets = new Set(brands);
const brandInputs = Array.from(document.querySelectorAll('input[name="brand_name"]'));
for (const el of brandInputs) await setChecked(el, brandTargets.has(el.value));
const checkedBrands = brandInputs.filter(el => el.checked).map(el => el.value);
const missingBrands = brands.filter(b => !checkedBrands.includes(b));
if (missingBrands.length) {
throw new Error('brands not applied: ' + missingBrands.join(',') +
' available=' + brandInputs.map(el => el.value).join(','));
}
// 店铺: 按标签精确匹配; 目标名必须全部存在, 否则报错并列出可用清单
const labelOf = input => norm(
(input.closest('label') || input.parentElement || input).innerText
);
const shopInputs = Array.from(document.querySelectorAll('input[name="shop_id"]'));
const byLabel = new Map();
for (const el of shopInputs) {
const label = labelOf(el);
if (!label) continue;
if (byLabel.has(label)) throw new Error('duplicate shop label: ' + label);
byLabel.set(label, el);
}
const shopTargets = shops.map(norm);
const missingShops = shopTargets.filter(name => !byLabel.has(name));
if (missingShops.length) {
throw new Error('shops not found in filter: ' + missingShops.join(',') +
' | available=' + Array.from(byLabel.keys()).join(','));
}
for (const el of shopInputs) await setChecked(el, shopTargets.includes(labelOf(el)));
const checkedShops = shopInputs.filter(el => el.checked).map(labelOf);
return {
dateBegin: begin.value,
dateEnd: end.value,
dateBasis: descText,
brands: checkedBrands,
shops: checkedShops,
};
}
"""
CLICK_DAY_TAB_JS = r"""
() => {
const ownText = el => Array.from(el.childNodes)
.filter(n => n.nodeType === 3).map(n => n.textContent).join('').replace(/\s+/g, ' ').trim();
const tab = Array.from(document.querySelectorAll('div,span,li,a'))
.find(el => ownText(el) === '按日统计');
if (!tab) throw new Error('按日统计 tab not found on report shell');
(tab.closest('div,li,a') || tab).click();
return true;
}
"""
DAY_BODY_SNAPSHOT_JS = r"""
() => {
const text = (document.body.innerText || '').replace(/\s+/g, ' ');
const noData = text.includes('没有任何数据') || text.includes('无数据');
const pending = text.includes('点击左侧');
// 报表按日期倒序且分页, 首页只含最近若干天; 边界日期可能不存在
// (如今日尚未出数), 因此抽出页面全部日期供区间判断。
const dates = Array.from(text.matchAll(/\d{4}-\d{2}-\d{2}/g)).map(m => m[0]);
return { text: text.slice(0, 2000), noData, pending, dates };
}
"""
CLICK_EXPORT_JS = r"""
() => {
const ownText = el => Array.from(el.childNodes)
.filter(n => n.nodeType === 3).map(n => n.textContent).join('').replace(/\s+/g, ' ').trim();
const span = Array.from(document.querySelectorAll('span'))
.find(el => ownText(el) === '导出EXCEL');
if (!span) throw new Error('导出EXCEL button not found in day report');
const target = span.closest('.btn_line') || span.parentElement;
target.dispatchEvent(new MouseEvent('mousedown', { bubbles: true }));
target.dispatchEvent(new MouseEvent('mouseup', { bubbles: true }));
target.click();
return String(target.className);
}
"""
def _is_page_closed_error(exc: BaseException) -> bool:
"""沿着异常链判断是否为 Playwright 页面/上下文/浏览器被关闭。"""
seen: set[int] = set()
node: BaseException | None = exc
while node is not None and id(node) not in seen:
seen.add(id(node))
if type(node).__name__ == "TargetClosedError":
return True
if "Target page, context or browser has been closed" in str(node):
return True
node = node.__cause__ or node.__context__
return False
def find_frame(page: Any, needle: str, wait_seconds: int = 20) -> Any:
import time
deadline = time.monotonic() + wait_seconds
while True:
for frame in page.frames:
if needle in (frame.url or ""):
return frame
if time.monotonic() >= deadline:
raise RuntimeError(
f"frame {needle} not found; frames={[f.url for f in page.frames]}"
)
page.wait_for_timeout(400)
def wait_day_report(
page: Any,
*,
start_date: str,
end_date: str,
timeout_seconds: int,
) -> dict[str, Any]:
"""等待 day.aspx 出现本区间数据(含起始日字符串)并连续两次稳定。"""
import time
deadline = time.monotonic() + timeout_seconds
stable_key = None
stable_count = 0
last_error = "day report did not return data"
while time.monotonic() < deadline:
frame = next(
(f for f in page.frames if "adsorder/day.aspx" in (f.url or "")), None
)
if frame is None:
last_error = "day.aspx frame not present"
page.evaluate(CLICK_DAY_TAB_JS)
page.wait_for_timeout(2000)
continue
try:
snapshot = frame.evaluate(DAY_BODY_SNAPSHOT_JS)
except Exception as exc:
last_error = f"{type(exc).__name__}: {exc}"
page.wait_for_timeout(1000)
continue
if snapshot.get("pending"):
last_error = "day report still waiting for 生成报表"
regenerate_report(page)
page.wait_for_timeout(3000)
stable_key = None
stable_count = 0
continue
fresh = any(
start_date <= day <= end_date for day in snapshot.get("dates") or ()
)
if snapshot.get("no_data"):
# 空结果也要求稳定出现两次,防止复用上一次查询的空页
key = ("no_data", hash(snapshot["text"]))
if key == stable_key:
stable_count += 1
else:
stable_key, stable_count = key, 1
if stable_count >= 2:
return {"ok": True, "no_data": True, "head": snapshot["text"][:200]}
page.wait_for_timeout(1000)
continue
if not fresh:
last_error = "day report body does not cover requested range"
stable_key = None
stable_count = 0
page.wait_for_timeout(1000)
continue
key = hash(snapshot["text"])
if key == stable_key:
stable_count += 1
else:
stable_key, stable_count = key, 1
if stable_count >= 2:
return {"ok": True, "no_data": False, "head": snapshot["text"][:200]}
page.wait_for_timeout(1000)
raise TimeoutError(f"timeout waiting for day report: {last_error}")
def regenerate_report(page: Any) -> None:
filter_frame = next(
(f for f in page.frames if "adsorder/filter.aspx" in (f.url or "")), None
)
if filter_frame is None:
return
filter_frame.evaluate(
"() => { if (typeof window.BtnFullSearch === 'function') window.BtnFullSearch(); }"
)
def _download_oss_export(url: str, evidence_dir: Path) -> Path:
"""带签名 OSS 导出 URL 的直连兜底下载(页面死亡后仍可获取文件)。
ERP 异步导出会先上传 OSS 再返回带签名的 jstzbstatic URL;签名有有效期
(约 20 分钟),命中后立即下载。直连不走系统代理(国内站)。
"""
import re as _re
import urllib.request
if not _re.match(r"^https://jstzbstatic\.erp321\.com/export/[A-Za-z0-9]+\.xlsx\?", url):
raise RuntimeError(f"refusing non-export download url: {url[:80]}")
opener = urllib.request.build_opener(urllib.request.ProxyHandler({}))
request = urllib.request.Request(url, headers={"User-Agent": USER_AGENT})
with opener.open(request, timeout=120) as response:
payload = response.read()
safe_root = evidence_dir.resolve()
target = (safe_root / f"销售主题分析_按日_fallback_{datetime.now():%Y%m%d%H%M%S}.xlsx").resolve()
if target.parent != safe_root:
raise RuntimeError("fallback download path escaped evidence dir")
target.write_bytes(payload)
return target
def export_day_excel(
page: Any,
*,
evidence_dir: Path,
export_timeout_seconds: int,
) -> Path:
"""点击导出EXCEL并接住浏览器 download 事件, 落盘到证据目录。
双通道取文件: (1) Playwright download 事件 + save_as; (2) ERP 异步导出
产物走 jstzbstatic OSS 签名 URL —— BI 页在异步导出链路上偶发杀死标签页
(2026-10-12 实测, 蝉妈妈同款病), 页面死了 download 事件作废, 但签名 URL
已经出现在网络请求里, 直连下载不受页面生命周期影响。
"""
import time
day_frame = find_frame(page, "adsorder/day.aspx", wait_seconds=15)
received: list[Path] = []
oss_urls: list[str] = []
def on_download(download: Any) -> None:
try:
if download.url and download.url not in oss_urls:
oss_urls.append(download.url)
name = download.suggested_filename or "销售主题分析_按日.xlsx"
if not name.lower().endswith(".xlsx"):
name += ".xlsx"
target = evidence_dir / name
download.save_as(str(target))
received.append(target)
except Exception as exc:
print(f"[WARN] download save failed: {type(exc).__name__}: {exc}", flush=True)
def on_request(request: Any) -> None:
url = request.url or ""
if "jstzbstatic.erp321.com/export/" in url and ".xlsx" in url and url not in oss_urls:
oss_urls.append(url)
page.on("download", on_download)
page.context.on("request", on_request)
try:
day_frame.evaluate(CLICK_EXPORT_JS)
deadline = time.monotonic() + export_timeout_seconds
while time.monotonic() < deadline and not received:
try:
page.wait_for_timeout(1000)
except Exception:
# 页面已死, save_as 通道作废; 直接转入 OSS 直连兜底
break
finally:
try:
page.remove_listener("download", on_download)
except Exception:
pass
try:
page.context.remove_listener("request", on_request)
except Exception:
pass
if received:
return received[0]
if oss_urls:
path = _download_oss_export(oss_urls[-1], evidence_dir)
print(f"[OK] 导出走 OSS 直连兜底下载: {path.name}", flush=True)
return path
raise TimeoutError(
f"export download did not arrive within {export_timeout_seconds}s"
)
def collect_export(
page: Any,
*,
start_date: str,
end_date: str,
brands: tuple[str, ...],
shops: tuple[str, ...],
evidence_dir: Path,
data_wait_seconds: int,
export_timeout_seconds: int,
) -> dict[str, Any]:
"""一轮完整采集: 打开报表 → 设筛选 → 生成 → 按日 → 导出。"""
if "adsorder/default.aspx" not in (page.url or ""):
page.goto(REPORT_URL, wait_until="domcontentloaded", timeout=60000)
page.wait_for_timeout(3000)
dismiss_marketing_popup(page)
close_foreign_popup_tabs(getattr(page, "context", None), page)
filter_frame = find_frame(page, "adsorder/filter.aspx", wait_seconds=30)
selection = filter_frame.evaluate(
SET_FILTERS_JS,
{"startDate": start_date, "endDate": end_date, "brands": list(brands), "shops": list(shops)},
)
regenerate_report(page)
page.wait_for_timeout(1500)
page.evaluate(CLICK_DAY_TAB_JS)
report_state = wait_day_report(
page,
start_date=start_date,
end_date=end_date,
timeout_seconds=data_wait_seconds,
)
export_path = export_day_excel(
page,
evidence_dir=evidence_dir,
export_timeout_seconds=export_timeout_seconds,
)
return {"selection": selection, "report": report_state, "export_path": export_path}
# ── CLI ─────────────────────────────────────────────────────────────────────
def parse_date_arg(value: str) -> date:
try:
return date.fromisoformat(value)
except ValueError as exc:
raise argparse.ArgumentTypeError("date must be YYYY-MM-DD") from exc
def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
parser = argparse.ArgumentParser(
description="聚水潭销售主题分析(按日)导出采集入库。"
)
parser.add_argument("--date", type=parse_date_arg, default=None,
help="业务日期(区间结束日); 缺省取昨天")
parser.add_argument("--days", type=int, default=1,
help="发货时间区间天数(含结束日); --date 默认 1 天")
parser.add_argument("--from", dest="from_date", type=parse_date_arg, default=None,
help="显式区间起始日(与 --to 一起用, 覆盖 --date/--days)")
parser.add_argument("--to", dest="to_date", type=parse_date_arg, default=None,
help="显式区间结束日")
parser.add_argument("--execute", action="store_true",
help="正式执行采集并写 PostgreSQL")
parser.add_argument("--headless", action="store_true")
parser.add_argument("--timeout-ms", type=int, default=120_000)
parser.add_argument("--data-wait-timeout", type=int, default=180,
help="等待按日报表出数的秒数(默认 180)")
parser.add_argument("--export-timeout", type=int, default=120,
help="等待导出下载的秒数(默认 120)")
parser.add_argument("--report-attempts", type=int, default=2,
help="整轮采集重试次数(含首次)")
return parser.parse_args(argv)
def resolve_window(args: argparse.Namespace) -> tuple[date, date]:
if args.from_date or args.to_date:
if not (args.from_date and args.to_date):
raise SystemExit("--from 与 --to 必须成对使用")
start, end = args.from_date, args.to_date
else:
end = args.date or (date.today() - timedelta(days=1))
start = end - timedelta(days=args.days - 1)
if end < start:
raise SystemExit(f"区间无效: {start} ~ {end}")
if (end - start).days + 1 > 366:
raise SystemExit("发货时间区间不能超过 366 天")
return start, end
def main(argv: list[str] | None = None) -> int:
from scrapling.fetchers import DynamicFetcher
args = parse_args(argv)
start, end = resolve_window(args)
evidence_dir = RAW_DATA_ROOT / EVIDENCE_DIR_NAME / f"{start.isoformat()}_{end.isoformat()}"
plan = {
"range": [start.isoformat(), end.isoformat()],
"days": (end - start).days + 1,
"brands": list(BRANDS),
"shops": list(SHOP_NAMES),
"write_table": "erp_sales_theme_daily_metrics",
"execute": bool(args.execute),
}
print(json.dumps(plan, ensure_ascii=False), flush=True)
if not args.execute:
print("[DRY-RUN] Add --execute to collect and write PostgreSQL.")
return 0
import os
username = os.environ.get("ERP_USERNAME", "").strip()
login_secret = (
os.environ.get("ERP_PASSWORD", "").strip()
or os.environ.get("GYXX_SUPPLY_ERP_PASSWORD", "").strip()
)
login_url = os.environ.get("GYXX_ERP_LOGIN_URL", "").strip() or LOGIN_URL_DEFAULT
# 成员绑定的 CDP 常驻浏览器并不一定在跑(无 keepalive);只有真正在听才用,
# 否则回退隔离 profile + vault cookie 注入(restore_erp_runtime_state)。
cdp_url, user_data_dir = resolve_browser_session({})
evidence_dir.mkdir(parents=True, exist_ok=True)
state: dict[str, Any] = {}
def page_action(page: Any) -> None:
open_action = make_page_action(
login_url=login_url,
username=username,
**{"pass" + "word": login_secret},
home_url=REPORT_URL,
timeout_ms=args.timeout_ms,
captcha_checks=24,
captcha_interval_seconds=5,
hold_seconds=0,
state=state,
login_only=True,
)
open_action(page)
if not state.get("ok"):
return
try:
state["collected"] = retry_step(
lambda: collect_export(
page,
start_date=start.isoformat(),
end_date=end.isoformat(),
brands=BRANDS,
shops=SHOP_NAMES,
evidence_dir=evidence_dir,
data_wait_seconds=max(args.data_wait_timeout, min(plan["days"] * 3, 600)),
export_timeout_seconds=args.export_timeout,
),
max_attempts=max(args.report_attempts, 1),
sleep_seconds=5,
label="ERP销售主题分析导出",
)
except BaseException as exc:
# ERP BI 页偶发「点导出后 tab 崩溃」(蝉妈妈同款病)。页面已死时原地
# 重试毫无意义:记录后正常返回,由外层重开一个全新浏览器会话自愈。
if _is_page_closed_error(exc):
print(
f"[WARN] 报表页在导出阶段崩溃/关闭({type(exc).__name__}),"
"本轮作废,由外层重开会话自愈",
flush=True,
)
state["page_crashed"] = True
else:
state["collect_error"] = f"{type(exc).__name__}: {exc}"
fetch_kwargs: dict[str, Any] = {
"headless": args.headless,
"load_dom": True,
"network_idle": True,
"timeout": args.timeout_ms,
"wait": 1000,
# scrapling 要求 retries>=1;页面级异常已在 page_action 内转入
# state 标志,重试语义由下方外层循环持有(每轮 = 全新浏览器会话)
"retries": 1,
"useragent": USER_AGENT,
"additional_args": {
"viewport": {"width": 1600, "height": 950},
"screen": {"width": 1600, "height": 950},
"accept_downloads": True,
},
# ERP 是国内站, 代理出口会被 WAF 拉黑(Clash 教训), 一律直连;
# 必须显式给下载目录: 覆盖>90天数据的导出走异步导出中心
# (CreateAsyncExport → OSS 签名 URL), headless Chrome 没有下载目录时
# 这条链路会直接弄死标签页(2026-10-12 实测)。
"extra_flags": [
"--no-proxy-server",
"--disable-dev-shm-usage",
f"--download-directory={evidence_dir}",
],
"page_action": page_action,
}
if cdp_url:
fetch_kwargs["cdp_url"] = cdp_url
if user_data_dir:
fetch_kwargs["user_data_dir"] = user_data_dir
fetch_kwargs["real_chrome"] = True
passes = max(args.report_attempts, 1)
last_error = "collect did not complete"
for attempt in range(1, passes + 1):
for flag in ("collected", "page_crashed", "collect_error"):
state.pop(flag, None)
if attempt > 1:
print(f"[RETRY] 第 {attempt - 1}/{passes - 1} 轮重开浏览器会话重试采集", flush=True)
try:
DynamicFetcher.fetch(REPORT_URL, **fetch_kwargs)
except Exception as exc:
message = str(exc)
if state.get("collected"):
# 文件已到手, 采集后任何浏览器/响应转换错误都无关紧要
print(
f"[WARN] 报表已导出,忽略 Scrapling 收尾错误: {type(exc).__name__}",
flush=True,
)
break
last_error = f"fetch: {type(exc).__name__}: {message}"
print(f"[FAIL] {last_error}", flush=True)
continue
if state.get("collected"):
break
if state.get("page_crashed"):
last_error = "report page closed during export"
continue
if state.get("collect_error"):
last_error = state["collect_error"]
print(f"[FAIL] {last_error}", flush=True)
continue
break
if not state.get("ok"):
print(
"[SKIPPED_COOKIE] ERP 登录态失效且无法自动恢复: "
f"{state.get('error', 'unknown error')}",
flush=True,
)
return COOKIE_SKIP_EXIT_CODE
collected = state.get("collected")
if not collected:
print(f"[FAIL] 采集未完成: {last_error}", flush=True)
return 1
export_path: Path = collected["export_path"]
records = parse_sales_theme_workbook(export_path)
if not records:
print(f"[FAIL] 导出文件未解析出数据行: {export_path}", flush=True)
return 1
expected_days = {(start + timedelta(days=offset)).isoformat() for offset in range(plan["days"])}
got_days = {record["stat_date"] for record in records}
outside = sorted(got_days - expected_days)
if outside:
print(f"[FAIL] 导出包含区间外日期 {outside[:5]}", flush=True)
return 1
missing = sorted(expected_days - got_days)
if missing:
print(f"[WARN] 区间内 {len(missing)} 天无数据行(ERP 未出行): {missing[:10]}", flush=True)
source = artifact_relative_path(export_path)
for record in records:
record["date_basis"] = "发货时间"
record["brands"] = list(BRANDS)
record["shops"] = list(collected["selection"].get("shops") or ())
record["source_file"] = source
manifest = {
"range": plan["range"],
"collected_at": datetime.now().isoformat(timespec="seconds"),
"selection": collected["selection"],
"report": collected["report"],
"export_file": str(export_path),
"row_count": len(records),
"missing_days": missing,
}
manifest_path = evidence_dir / "manifest.json"
manifest_path.write_text(json.dumps(manifest, ensure_ascii=False, indent=2), encoding="utf-8")
with get_conn() as conn:
ensure_erp_sales_theme_daily_table(conn)
written = upsert_erp_sales_theme_daily_metrics(conn, records)
total_sales_amount = sum(float(r.get("sales_amount") or 0) for r in records)
print(
f"[DB] {start}~{end} upserted {written} days; "
f"shops={len(collected['selection'].get('shops') or ())}; "
f"sales_amount_total={total_sales_amount:.2f}; evidence={source}",
flush=True,
)
return 0
if __name__ == "__main__":
sys.exit(main())
@@ -84,8 +84,14 @@ MODE = "daily"
# 2026-06-11: daily 走 cell 45/57 老路径 (保护 run_daily_collect.bat 定时任务, data.json 口径不变);
# weekly 同样 cell 45/57 (2026-06-27 run_weekly_collect.bat 定时任务已删除, --mode weekly 留作手动回采);
# backfill 单日窗口 + 表头匹配 (自定义销量 / 自定义退货量 + SPU 行汇总), 跟 daily/weekly 区分开
# 2026-10-11: 聚水潭重置 allsku 报表模板, 38 列被置隐藏后可见列整体左移 5 列, 固定可见索引
# 45/57 落到了「今日实退货量/今日仅退款量」(查询日清晨≈0) 上, daily 销量全 0。
# daily 改为表头名匹配「昨日销量/昨日退货量」+ x 坐标最近格; weekly 暂留老索引
# (语义未实测验证, 周采集复用时需先确认 7 天窗口下各列含义)。
WEEKLY_SALES_HEADER = "__idx__45"
WEEKLY_RETURNS_HEADER = "__idx__57"
DAILY_SALES_HEADER = "昨日销量"
DAILY_RETURNS_HEADER = "昨日退货量"
BACKFILL_SALES_HEADER = "自定义销量"
BACKFILL_RETURNS_HEADER = "自定义退货量"
@@ -292,8 +298,11 @@ def report_date_range() -> tuple[str, str]:
def _sales_returns_idx() -> tuple[str, str]:
if MODE in ("weekly", "daily"):
# daily / weekly 都走 cell 45/57 老路径, 跟之前 data.json 口径一致
if MODE == "daily":
# 表头匹配 (昨日销量 / 昨日退货量): 模板列可见性变化会让固定索引错位
return DAILY_SALES_HEADER, DAILY_RETURNS_HEADER
if MODE == "weekly":
# weekly 仍走 cell 45/57 老路径, 跟之前 data.json 口径一致
return WEEKLY_SALES_HEADER, WEEKLY_RETURNS_HEADER
# backfill 走表头匹配 (自定义销量 / 自定义退货量 + SPU 行求和)
return BACKFILL_SALES_HEADER, BACKFILL_RETURNS_HEADER
@@ -395,6 +404,44 @@ def _extract_by_idx(table_frame: Any, sales_idx: int, returns_idx: int) -> dict[
)
ENSURE_CUSTOM_COLUMNS_JS = r"""
() => {
const cols = window._Cols || [];
let changed = 0;
for (const col of cols) {
const label = String(col.text || col.defaultText || '');
if ((label === '自定义销量' || label === '自定义退货量') && col.visible === false) {
col.visible = true;
changed += 1;
}
}
if (changed > 0 && typeof window._RptRefresh === 'function') {
try { window._RptRefresh(); } catch (e) {}
}
return changed;
}
"""
def ensure_custom_columns_visible(page: Any) -> int:
"""backfill 模式: 把模板重置隐藏的自定义销量/退货量列恢复可见。
2026-10-11 起聚水潭报表模板把自定义公式列重置为隐藏,隐藏列不渲染
数据单元格;只改 visible 不会自动重绘,需顺带触发一次 _RptRefresh。
与 backfill_erp_all_shop_daily.py 同源(该脚本 10-13 实测有效)。
"""
changed_total = 0
for frame in page.frames:
if "allsku.aspx" not in (frame.url or "").lower():
continue
try:
changed_total += int(frame.evaluate(ENSURE_CUSTOM_COLUMNS_JS))
except Exception:
continue
return changed_total
def _extract_by_header(table_frame: Any, sales_header: str, returns_header: str) -> dict[str, Any]:
"""backfill 模式专用: 按表头列名匹配 sales / returns 列, 直接读每 SPU 行对应 cell 的 innerText 求和.
@@ -487,13 +534,35 @@ def _extract_by_header(table_frame: Any, sales_header: str, returns_header: str)
};
}
// 4) 累加: 每行 cells[salesCol] / cells[returnsCol] 的 innerText
// 4) 累加: 每行 cells[salesPos] / cells[returnsPos]
// 2026-10-11 起模板把自定义公式列重置为隐藏: 隐藏列的 th 仍在表头,
// 但数据行不渲染对应 td, 表头索引直接取行内格子会错位。用
// window._Cols 把表头列索引换算成行内可见位置; 拿不到 _Cols 时按原索引直取。
let salesPos = salesCol;
let returnsPos = returnsCol;
if (Array.isArray(window._Cols) && window._Cols.length === headerCells.length) {
const visibleIdx = [];
for (let i = 0; i < window._Cols.length; i++) {
if (window._Cols[i].visible !== false) visibleIdx.push(i);
}
salesPos = visibleIdx.indexOf(salesCol);
returnsPos = visibleIdx.indexOf(returnsCol);
}
if (salesPos < 0 || returnsPos < 0) {
return {
ok: false,
yesterday_sales: 0,
yesterday_returns: 0,
error: `custom sales/return columns hidden: salesPos=${salesPos} returnsPos=${returnsPos}`,
header_texts: headerTexts,
};
}
let totalSales = 0, totalReturns = 0;
const rowBreakdown = [];
for (const tr of dataRows) {
const cells = Array.from(tr.querySelectorAll('td, th'));
const salesCell = cells[salesCol];
const returnsCell = cells[returnsCol];
const salesCell = cells[salesPos];
const returnsCell = cells[returnsPos];
const head = (tr.innerText || tr.textContent || '').slice(0, 60);
if (!salesCell || !returnsCell) {
rowBreakdown.push({
@@ -662,7 +731,7 @@ def _extract_with_text(table_frame: Any) -> tuple[dict[str, Any], str]:
ERP DOM often contains NBSP (\xa0) etc; Scrapling/Playwright on Windows gbk
rejects those on return, so we use encodeURIComponent + tryUTF8 to make it ASCII-safe.
按 MODE 选 idx 模式 (daily) 或 header 模式 (weekly).
按 MODE 选 idx 模式 (weekly) 或 header 模式 (daily / backfill)。
"""
sales_key, returns_key = _sales_returns_idx()
@@ -673,17 +742,17 @@ def _extract_with_text(table_frame: Any) -> tuple[dict[str, Any], str]:
else:
js_args = {"salesHeader": sales_key, "returnsHeader": returns_key, "mode": "header"}
# 走 SPU 行 + 自定义列名匹配 + 求和 (因为 sumRow 在 ERP 系统里不含自定义段)
if sales_key != "__idx__45": # backfill 模式
# backfill 走 SPU 行 + 自定义列名匹配 + 求和 (因为 sumRow 在 ERP 系统里不含自定义段);
# daily 的 昨日销量/昨日退货量 是原生列, sumRow 里就有, 走下面 payload 的
# mode='header' 分支 (表头名 -> x 坐标 -> sumRow 最近格), 不经过 _extract_by_header。
if MODE == "backfill":
header_result = _extract_by_header(table_frame, sales_key, returns_key)
# _extract_by_header 返回 dict, 没有 body_text; wait_for_metrics 需要 erp_code in body_text 校验
# 用 erp_code placeholder 骗校验 (它会查 erp_code in summary_text, 而 summary_text 不一定含 erp_code)
# 但新逻辑在 row_breakdown 里已有 erp_code 关联的 SPU 编码 (10384xxx), 校验一定会过
return header_result, header_result.get("summary_text", "") or "PLACEHOLDER"
# 老 idx 模式
sales_idx = int(sales_key.removeprefix("__idx__"))
returns_idx = int(returns_key.removeprefix("__idx__"))
# weekly 老 idx 模式 + daily header 模式共用这段 payload (js_args 已按模式构造)
payload = table_frame.evaluate(
"""
(args) => {
@@ -712,6 +781,9 @@ def _extract_with_text(table_frame: Any) -> tuple[dict[str, Any], str]:
}
let salesCell = null, returnsCell = null;
// salesCol/returnsCol 供末尾 return 的 *_col_idx 引用, 必须在 if/else 外声明
// (2026-10-13 daily 切到 header 模式后首测即 ReferenceError 的教训)
let salesCol = -1, returnsCol = -1;
if (args.mode === 'idx') {
const cells = Array.from(sumRow.children).map((cell) => {
const box = cell.getBoundingClientRect();
@@ -745,8 +817,8 @@ def _extract_with_text(table_frame: Any) -> tuple[dict[str, Any], str]:
}
return -1;
};
const salesCol = findCol(args.salesHeader);
const returnsCol = findCol(args.returnsHeader);
salesCol = findCol(args.salesHeader);
returnsCol = findCol(args.returnsHeader);
if (salesCol < 0 || returnsCol < 0) {
return { ok: false, yesterday_sales: 0, yesterday_returns: 0,
error: `header not found: salesCol=${salesCol} returnsCol=${returnsCol}`,
@@ -977,6 +1049,9 @@ def wait_for_metrics(page: Any, erp_code: str, timeout_seconds: int = 45,
# previous_url 可能在 _reload_allsku_iframe 后被重置为 None
while True:
# backfill 模式: 报表刷新会把自定义列的可见性重置回隐藏, 每轮恢复一次
if MODE == "backfill":
ensure_custom_columns_visible(page)
# 先检测"操作频繁"弹窗 (在主页 DOM, 不在 allsku iframe)
if filter_frame is not None and freq_dialog_handled < MAX_FREQ_DIALOG:
if _detect_and_close_freq_dialog(page):
@@ -1176,6 +1251,9 @@ def _run_one_code(page: Any, filter_frame: Any, platform: dict[str, Any], erp_co
"""
clear_code(filter_frame)
set_platform_and_code(filter_frame, platform["shops"], erp_code)
if MODE == "backfill":
# 自定义销量/退货量列被模板重置隐藏时先恢复可见, 否则数据行不渲染对应格
ensure_custom_columns_visible(page)
prev_state = _get_allsku_frame_state(page)
generate_report(filter_frame)
return wait_for_metrics(
@@ -68,6 +68,9 @@
fact_market_rank_notifications
写: market_rank_hermes_notification.py (Hermes 机器人通知结果;文件名兼容保留)
读: 同一收件人同日通知去重 / 发送审计
erp_sales_theme_daily_metrics
写: collect_erp_sales_theme_daily.py (聚水潭「报表→销售主题分析→按日统计」Excel 导出入库)
读: 固定品牌+店铺口径的全渠道按日销售汇总查询
所有 upsert 都用 ON CONFLICT ... DO UPDATE, 重复跑覆盖不双写。
"""
@@ -1908,3 +1911,160 @@ def upsert_market_rank_notification(conn, record: dict) -> int:
if not row:
raise RuntimeError("市场排行飞书通知归档未返回记录 id")
return int(row[0])
# ── erp_sales_theme_daily_metrics ────────────────────────────────────────────
def ensure_erp_sales_theme_daily_table(conn) -> None:
"""幂等创建聚水潭销售主题分析(按日)汇总表,不执行其他 schema 迁移。"""
sql = """
CREATE TABLE IF NOT EXISTS erp_sales_theme_daily_metrics (
id BIGSERIAL PRIMARY KEY,
stat_date DATE NOT NULL,
sales_order_count INTEGER NOT NULL DEFAULT 0,
sales_qty INTEGER NOT NULL DEFAULT 0,
shipped_qty INTEGER NOT NULL DEFAULT 0,
shipped_amount NUMERIC(18, 2) NOT NULL DEFAULT 0,
sales_amount NUMERIC(18, 2) NOT NULL DEFAULT 0,
return_order_count INTEGER NOT NULL DEFAULT 0,
return_qty INTEGER NOT NULL DEFAULT 0,
returned_qty INTEGER NOT NULL DEFAULT 0,
return_amount NUMERIC(18, 2) NOT NULL DEFAULT 0,
returned_amount NUMERIC(18, 2) NOT NULL DEFAULT 0,
net_sales_order_count INTEGER NOT NULL DEFAULT 0,
net_sales_qty INTEGER NOT NULL DEFAULT 0,
net_sales_amount NUMERIC(18, 2) NOT NULL DEFAULT 0,
freight_income NUMERIC(18, 2),
freight_expense NUMERIC(18, 2),
shipped_order_count INTEGER NOT NULL DEFAULT 0,
date_basis TEXT NOT NULL DEFAULT '发货时间',
brands TEXT[] NOT NULL DEFAULT '{}',
shops TEXT[] NOT NULL DEFAULT '{}',
source_file TEXT,
imported_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
UNIQUE (stat_date)
);
CREATE INDEX IF NOT EXISTS erp_sales_theme_daily_metrics_date_idx
ON erp_sales_theme_daily_metrics (stat_date DESC);
COMMENT ON TABLE erp_sales_theme_daily_metrics IS
'聚水潭「报表→销售主题分析→按日统计」Excel 导出入库;固定品牌+店铺口径,每日一行幂等覆盖。';
COMMENT ON COLUMN erp_sales_theme_daily_metrics.date_basis IS
'报表时间口径,当前固定为发货时间(send_date)。';
COMMENT ON COLUMN erp_sales_theme_daily_metrics.brands IS
'本次查询勾选的商品品牌清单(筛选页 input[name=brand_name] 值)。';
COMMENT ON COLUMN erp_sales_theme_daily_metrics.shops IS
'本次查询勾选的销售渠道(店铺)名称清单。';
"""
with conn.cursor() as cur:
cur.execute(sql)
_SALES_THEME_INT_FIELDS = frozenset({
"sales_order_count",
"sales_qty",
"shipped_qty",
"return_order_count",
"return_qty",
"returned_qty",
"net_sales_order_count",
"net_sales_qty",
"shipped_order_count",
})
_SALES_THEME_COLUMNS = (
"sales_order_count",
"sales_qty",
"shipped_qty",
"shipped_amount",
"sales_amount",
"return_order_count",
"return_qty",
"returned_qty",
"return_amount",
"returned_amount",
"net_sales_amount",
"net_sales_order_count",
"net_sales_qty",
"shipped_order_count",
"freight_income",
"freight_expense",
)
# 表内 NOT NULL 的金额列;导出里无退货/无数据的日期这些单元格为空, 统一落 0
# (运费收入/支出两列允许 NULL, 与导出口径一致)。
_SALES_THEME_REQUIRED_AMOUNTS = frozenset({
"shipped_amount",
"sales_amount",
"return_amount",
"returned_amount",
"net_sales_amount",
})
def upsert_erp_sales_theme_daily_metrics(conn, records: Sequence[dict]) -> int:
"""幂等写入销售主题分析按日汇总,主键为统计日期。
records: [{'stat_date', 'sales_order_count', ..., 'brands', 'shops', 'source_file'}, ...]
数量字段缺省按 0 处理;brands/shops 写入前去重排序,保证重跑结果稳定。
"""
if not records:
return 0
dedup: dict[Any, dict] = {}
for record in records:
key = record["stat_date"]
if str(key).strip() == "":
raise ValueError("stat_date must not be empty")
dedup[key] = dict(record)
rows = []
for record in dedup.values():
row = [record["stat_date"]]
for field in _SALES_THEME_COLUMNS:
value = record.get(field)
if field in _SALES_THEME_INT_FIELDS:
row.append(int(value or 0))
elif field in _SALES_THEME_REQUIRED_AMOUNTS:
row.append(round(float(value), 2) if value is not None else 0)
else:
row.append(round(float(value), 2) if value is not None else None)
row.append(str(record.get("date_basis") or "发货时间"))
row.append(sorted({str(item).strip() for item in record.get("brands") or () if str(item).strip()}))
row.append(sorted({str(item).strip() for item in record.get("shops") or () if str(item).strip()}))
row.append(record.get("source_file"))
rows.append(tuple(row))
sql = """
INSERT INTO erp_sales_theme_daily_metrics
(stat_date, sales_order_count, sales_qty, shipped_qty,
shipped_amount, sales_amount,
return_order_count, return_qty, returned_qty,
return_amount, returned_amount, net_sales_amount,
net_sales_order_count, net_sales_qty,
shipped_order_count, freight_income, freight_expense,
date_basis, brands, shops, source_file)
VALUES %s
ON CONFLICT (stat_date) DO UPDATE SET
sales_order_count = EXCLUDED.sales_order_count,
sales_qty = EXCLUDED.sales_qty,
shipped_qty = EXCLUDED.shipped_qty,
shipped_amount = EXCLUDED.shipped_amount,
sales_amount = EXCLUDED.sales_amount,
return_order_count = EXCLUDED.return_order_count,
return_qty = EXCLUDED.return_qty,
returned_qty = EXCLUDED.returned_qty,
return_amount = EXCLUDED.return_amount,
returned_amount = EXCLUDED.returned_amount,
net_sales_order_count = EXCLUDED.net_sales_order_count,
net_sales_qty = EXCLUDED.net_sales_qty,
net_sales_amount = EXCLUDED.net_sales_amount,
freight_income = EXCLUDED.freight_income,
freight_expense = EXCLUDED.freight_expense,
shipped_order_count = EXCLUDED.shipped_order_count,
date_basis = EXCLUDED.date_basis,
brands = EXCLUDED.brands,
shops = EXCLUDED.shops,
source_file = EXCLUDED.source_file,
updated_at = NOW()
"""
with conn.cursor() as cur:
execute_values(cur, sql, rows)
return len(rows)
@@ -1273,3 +1273,37 @@ COMMENT ON TABLE fact_video_publications IS
'附件级视频发布幂等账本;点击发布后无明确回执必须进入 ambiguous,禁止自动重试。';
COMMENT ON TABLE fact_video_publication_attempts IS
'视频发布每次尝试的标题、商品、话题、标签、提交时点与确认凭据。';
-- 聚水潭「报表→销售主题分析→按日统计」Excel 导出入库。
-- 口径:发货时间区间 + 固定品牌(光影行星/光影/gyxx) + 固定勾选店铺清单,
-- 采集脚本 collect_erp_sales_theme_daily.py;与 erp_all_shop_style_daily_metrics
-- (全店铺按款式) 分表,本表是全渠道合并口径的按日汇总。
CREATE TABLE IF NOT EXISTS erp_sales_theme_daily_metrics (
id BIGSERIAL PRIMARY KEY,
stat_date DATE NOT NULL,
sales_order_count INTEGER NOT NULL DEFAULT 0, -- 销售单数
sales_qty INTEGER NOT NULL DEFAULT 0, -- 销售数量
shipped_qty INTEGER NOT NULL DEFAULT 0, -- 实发数量
shipped_amount NUMERIC(18, 2) NOT NULL DEFAULT 0, -- 实发金额
sales_amount NUMERIC(18, 2) NOT NULL DEFAULT 0, -- 销售金额
return_order_count INTEGER NOT NULL DEFAULT 0, -- 退货单数
return_qty INTEGER NOT NULL DEFAULT 0, -- 退货数量
returned_qty INTEGER NOT NULL DEFAULT 0, -- 实退数量
return_amount NUMERIC(18, 2) NOT NULL DEFAULT 0, -- 退货金额
returned_amount NUMERIC(18, 2) NOT NULL DEFAULT 0, -- 实退金额
net_sales_order_count INTEGER NOT NULL DEFAULT 0, -- 净销售单数
net_sales_qty INTEGER NOT NULL DEFAULT 0, -- 净销量
net_sales_amount NUMERIC(18, 2) NOT NULL DEFAULT 0, -- 净销售额
freight_income NUMERIC(18, 2), -- 运费收入(可空)
freight_expense NUMERIC(18, 2), -- 运费支出(可空)
shipped_order_count INTEGER NOT NULL DEFAULT 0, -- 实发单数
date_basis TEXT NOT NULL DEFAULT '发货时间',
brands TEXT[] NOT NULL DEFAULT '{}',
shops TEXT[] NOT NULL DEFAULT '{}',
source_file TEXT,
imported_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
UNIQUE (stat_date)
);
CREATE INDEX IF NOT EXISTS erp_sales_theme_daily_metrics_date_idx
ON erp_sales_theme_daily_metrics (stat_date DESC);
@@ -29,15 +29,19 @@ from runtime_paths import TMP_ROOT
from gyxx_flow.adapters.browser import BrowserCookieStore
from gyxx_flow.adapters.scrapling import ScraplingBrowser, is_browser_timeout_error
from gyxx_flow.core.config import Settings
from gyxx_flow.core.locks import LockManager, ResourceBusyError
_CONFIG_NAME = "tmall-baibu.json"
_LOGIN_PAGE_URL = "https://login.taobao.com/member/login.jhtml"
_EXPORT_TIMEOUT_SECONDS = 240
_EXPORT_MAX_ATTEMPTS = 3
_EXPORT_RETRY_DELAYS_SECONDS = (5, 15)
_PAGE_TIMEOUT_MS = 60_000
_IMPORT_TIMEOUT_SECONDS = 300
_IMPORT_ACCEPTED_AFTER_PROCESSING_SECONDS = 120
_STATUS_SETTLE_TIMEOUT_SECONDS = 5
_ACCOUNT_LOCK_TIMEOUT_SECONDS = 120
_RESULT_REFRESH_INTERVAL_SECONDS = 8
_DRAFT_ACTION_TIMEOUT_SECONDS = 30
_DRAFT_NAVIGATION_TIMEOUT_SECONDS = 15
@@ -103,6 +107,7 @@ _LOGIN_SECOND_FACTOR_MARKERS = (
"安全验证",
"二次验证",
"验证码",
"滑块",
)
@@ -805,6 +810,41 @@ def _find_password_login_form(
return locate()
def _is_login_agreement_visible(page: Any) -> bool:
"""Return whether Tmall's 服务协议及隐私保护 popup is blocking the login."""
for context in _page_contexts(page):
get_by_role = getattr(context, "get_by_role", None)
if not callable(get_by_role):
continue
try:
dialogs = get_by_role("dialog")
for index in range(min(dialogs.count(), 8)):
dialog = dialogs.nth(index)
try:
if not dialog.is_visible():
continue
text = _normalized_page_text(dialog.inner_text(timeout=2_000))
except Exception:
continue
if "服务协议" in text or "隐私" in text:
return True
except Exception:
continue
return False
def _click_login_agreement_accept(page: Any) -> bool:
for context in _page_contexts(page):
for candidate in _visible_text_candidates(context, "同意"):
try:
candidate.click(timeout=8_000)
return True
except Exception:
continue
return False
def _login_if_needed(
entry: BaibuEntry,
page: Any,
@@ -817,6 +857,11 @@ def _login_if_needed(
return False
login_url = _is_login_url(str(getattr(page, "url", "")))
form = _find_password_login_form(page, use_password_form=False)
if form is None and not login_url:
# A redirect can land on a scanner-tab seller login page (qianniu)
# whose URL carries no login marker; probe the password tab before
# deciding there is nothing to log in to.
form = _find_password_login_form(page, use_password_form=True)
if not login_url and form is None:
return False
@@ -852,10 +897,92 @@ def _login_if_needed(
raise BaibuApplyError(
f"{entry.name} 自动填写账号密码并提交登录失败({type(exc).__name__})"
) from exc
# The first password submit can pop the 服务协议及隐私保护 confirmation
# which swallows the submit; accept it and submit once more so the login
# actually runs. Both popups and form re-renders are tolerated.
for _attempt in range(2):
wait_for_timeout = getattr(page, "wait_for_timeout", None)
if callable(wait_for_timeout):
wait_for_timeout(2_000)
else:
time.sleep(2)
if not _is_login_agreement_visible(page):
break
if not _click_login_agreement_accept(page):
break
print(f"[百补] {entry.name} 已确认登录服务协议弹窗,重新提交登录")
refreshed = _find_password_login_form(page, use_password_form=True)
if refreshed is None:
break
_context, username, secret_input, submit = refreshed
try:
username.fill(credentials.username)
secret_input.fill(credentials.password)
submit.click(timeout=15_000)
except Exception as exc:
raise BaibuApplyError(
f"{entry.name} 协议确认后重新提交登录失败({type(exc).__name__})"
) from exc
print(f"[百补] {entry.name} 已提交密码登录,等待店铺身份确认")
return True
def _goto_tolerant(page: Any, url: str) -> None:
"""Navigate, tolerating an aborted load when a redirect takes over.
An unauthenticated apply URL is answered with a redirect chain that can
abort the original navigation from inside the browser; treat that as the
navigation landing wherever the chain pointed and let it settle.
"""
try:
page.goto(url, wait_until="domcontentloaded", timeout=_PAGE_TIMEOUT_MS)
except Exception as exc:
if "ERR_ABORTED" not in str(exc):
raise
wait_for_timeout = getattr(page, "wait_for_timeout", None)
if callable(wait_for_timeout):
wait_for_timeout(_LOGIN_FORM_SETTLE_MILLISECONDS)
else:
time.sleep(_LOGIN_FORM_SETTLE_MILLISECONDS / 1000)
def _explicit_password_login(
entry: BaibuEntry,
page: Any,
*,
manual_login: bool,
) -> None:
"""Drive the Tmall password form when the apply page hid its login state.
The myseller/qn apply SPA can sit on an SSO interstitial that never shows
a login URL, or render the cached shop name on an expired session. In
both cases ``_login_if_needed`` sees nothing to do. Re-entering the apply
URL redirects to the seller login page (qianniu) which carries the right
merchant context; only when that shows no form fall back to the plain
login page. Login there, return to the apply URL, and confirm identity.
"""
print(f"[百补] {entry.name} 显式重新进入报名页并走密码登录")
_goto_tolerant(page, entry.apply_url)
wait_for_timeout = getattr(page, "wait_for_timeout", None)
if callable(wait_for_timeout):
wait_for_timeout(_LOGIN_FORM_SETTLE_MILLISECONDS * 3)
else:
time.sleep(_LOGIN_FORM_SETTLE_MILLISECONDS / 1000 * 3)
login_attempted = _login_if_needed(entry, page, manual_login=manual_login)
if not login_attempted:
_goto_tolerant(page, _LOGIN_PAGE_URL)
login_attempted = _login_if_needed(entry, page, manual_login=manual_login)
_goto_tolerant(page, entry.apply_url)
_wait_for_authenticated(
entry,
page,
login_attempted=login_attempted,
manual_login=manual_login,
)
def _click_text(page: Any, labels: Iterable[str]) -> bool:
contexts = _page_contexts(page)
for label in labels:
@@ -1040,7 +1167,10 @@ def _open_batch_import(
f"{entry.name} 页面已有批量导入操作正在处理,已停止,避免重复提交"
)
time.sleep(0.5)
raise BaibuApplyError(f"{entry.name} 页面没有找到可用的「商品批量导入」入口")
raise BaibuApplyError(
f"{entry.name} 页面没有找到可用的「商品批量导入」入口"
"(活动可能已报名/已结束或会话无效,请人工打开报名页确认)"
)
def _wait_import_result(
@@ -1062,6 +1192,7 @@ def _wait_import_result(
last_processing_log = 0.0
last_refresh = 0.0
observed_new_operation = False
first_new_operation_at: float | None = None
while time.monotonic() < deadline:
text = _page_text(page)
operation = _latest_batch_import_status(text)
@@ -1070,6 +1201,8 @@ def _wait_import_result(
or operation.signature != previous_operation.signature
):
observed_new_operation = True
if first_new_operation_at is None:
first_new_operation_at = time.monotonic()
if operation.state == "success":
print(
f"[百补] {entry.name} 页面已完成导入处理,"
@@ -1082,6 +1215,21 @@ def _wait_import_result(
f"忽略行结果并按导入提交成功:{operation.detail}"
)
return
if (
observed_new_operation
and first_new_operation_at is not None
and time.monotonic() - first_new_operation_at
>= _IMPORT_ACCEPTED_AFTER_PROCESSING_SECONDS
):
# Tmall has already acknowledged the batch ("执行了商品批量导入
# 操作,正在处理中"); the submission itself is the business
# effect, so a slow per-row summary must not fail the run.
print(
f"[百补] {entry.name} 天猫已受理批量导入,"
f"{_IMPORT_ACCEPTED_AFTER_PROCESSING_SECONDS} 秒内未返回行结果,"
"按导入提交成功,行结果以报名页面为准"
)
return
if observed_new_operation and any(marker in text for marker in _SUCCESS_MARKERS):
print(f"[百补] {entry.name} 页面已确认导入提交,忽略行结果")
return
@@ -1284,9 +1432,20 @@ def _complete_draft_products(page: Any, entry: BaibuEntry) -> None:
if not entry.complete_drafts_after_import:
return
if not _click_draft_tab(page):
count = _draft_count(_page_text(page))
if count is None and not _has_visible_text(page, ("完善商品",)):
# Right after the import acknowledgement the SPA may still be hydrating;
# the draft tab, the draft count and the per-row action rarely all appear
# on the first probe, so retry briefly before declaring the page broken.
deadline = time.monotonic() + _DRAFT_NAVIGATION_TIMEOUT_SECONDS
while time.monotonic() < deadline:
if _click_draft_tab(page):
break
text = _page_text(page)
if _draft_count(text) is not None or _has_visible_text(page, ("完善商品",)):
break
if any(marker in text for marker in _DRAFT_EMPTY_MARKERS):
break
time.sleep(0.5)
else:
raise BaibuApplyError(f"{entry.name} 未找到草稿页入口")
draft_count, action_count = _wait_for_draft_state(page, entry)
@@ -1439,6 +1598,8 @@ def _has_account_cookie(
def _persist_account_state(
browser: ScraplingBrowser,
entry: BaibuEntry,
*,
session_lock_held: bool = False,
) -> bool:
"""Publish a verified login to the account vault without leaking cookies."""
@@ -1463,8 +1624,12 @@ def _persist_account_state(
)
cookies = _account_cookies(cookies, required_domains)
if not _has_account_cookie(cookies, required_names):
names = sorted({str(item.get("name", "")) for item in cookies})
preview = "、".join(names[:12]) if names else "无"
raise BaibuApplyError(
f"{entry.name} 登录后未获得可用的账号 cookie,未更新账号登录态"
f"{entry.name} 登录后未获得可用的账号 cookie"
f"(域内 cookie:{preview};要求含:{'、'.join(required_names)}),"
"未更新账号登录态"
)
raw_state = context.storage_state()
@@ -1472,23 +1637,33 @@ def _persist_account_state(
filtered_state = _account_storage_state(state, cookies, required_domains)
cookie_file, storage_state_file = paths
store = BrowserCookieStore(cookie_file, storage_state_file)
lock_root = cookie_file.parent.parent / "locks"
def _save() -> None:
try:
store.save_cookies(cookies)
store.save_storage_state(filtered_state)
except (OSError, ValueError, TypeError) as exc:
raise BaibuApplyError(
f"{entry.name} 保存账号登录态失败({type(exc).__name__})"
) from exc
if session_lock_held:
# The whole browser session already holds the scheduler-side
# browser-account lock; re-acquiring here would self-deadlock.
_save()
else:
lock_root = Settings.from_env().data_root / "state" / "locks"
try:
with LockManager(lock_root).acquire(
f"browser-account:{entry.account_id}",
owner=f"tmall-baibu:{entry.entry_id}",
timeout_seconds=0,
):
store.save_cookies(cookies)
store.save_storage_state(filtered_state)
_save()
except ResourceBusyError as exc:
raise BaibuApplyError(
f"{entry.name} 账号登录态正在被其他进程更新,未覆盖账号登录态"
) from exc
except (OSError, ValueError, TypeError) as exc:
raise BaibuApplyError(
f"{entry.name} 保存账号登录态失败({type(exc).__name__})"
) from exc
print(f"[百补] 已将 {entry.account_id} 的已验证登录态同步到账号 vault")
return True
@@ -1498,6 +1673,33 @@ def _apply_entry(
template: Path,
*,
manual_login: bool = False,
) -> None:
# Hold the scheduler-side account browser lock for the whole session.
# The resident keepalive heartbeats the same profile on an interval and
# its timeout-0 acquire skips busy accounts, so this lock is what stops
# the heartbeat from killing the import mid-run (TargetClosedError /
# silent page death). Members must use the same lock directory as the
# scheduler (Settings.data_root/state/locks), not one derived from the
# vault paths.
locks = LockManager(Settings.from_env().data_root / "state" / "locks")
try:
with locks.acquire(
f"browser-account:{entry.account_id}",
owner=f"tmall-baibu:{entry.entry_id}",
timeout_seconds=_ACCOUNT_LOCK_TIMEOUT_SECONDS,
):
_apply_entry_locked(entry, template, manual_login=manual_login)
except ResourceBusyError as exc:
raise BaibuApplyError(
f"{entry.name} 账号浏览器正被保活/登录占用,请稍后重试"
) from exc
def _apply_entry_locked(
entry: BaibuEntry,
template: Path,
*,
manual_login: bool = False,
) -> None:
browser = _start_browser()
try:
@@ -1516,14 +1718,36 @@ def _apply_entry(
page,
manual_login=manual_login,
)
try:
_wait_for_authenticated(
entry,
page,
login_attempted=login_attempted,
manual_login=manual_login,
)
except BaibuApplyError:
if login_attempted or manual_login:
raise
# The apply page may sit on an SSO interstitial without a login
# URL, so the passive check above never triggers the credentials;
# fall back to the explicit password login once.
_explicit_password_login(entry, page, manual_login=manual_login)
login_attempted = True
browser.persist_bound_state()
_persist_account_state(browser, entry)
try:
_persist_account_state(browser, entry, session_lock_held=True)
except BaibuApplyError as exc:
# A failed vault refresh can also mean the shop name was rendered
# from cache on top of a guest session; retry once through the
# explicit login before giving up on the vault refresh. The
# refresh only caches this session for later runs and must never
# block the import submission itself.
print(f"[WARN] {exc};尝试显式重新登录后重试", file=sys.stderr)
try:
_explicit_password_login(entry, page, manual_login=manual_login)
_persist_account_state(browser, entry, session_lock_held=True)
except BaibuApplyError as retry_exc:
print(f"[WARN] {retry_exc};本次继续执行报名导入", file=sys.stderr)
print(f"[百补] 已确认账号 {entry.account_id},打开 {entry.name} 报名入口")
previous_operation = _open_batch_import(page, entry)
time.sleep(1)
@@ -1,3 +1,5 @@
import json
import time
from datetime import datetime
from pathlib import Path
from types import SimpleNamespace
@@ -33,7 +35,7 @@ def test_refresh_then_export_waits_once_after_all_accounts(monkeypatch):
monkeypatch.setattr(chanmama, "export_video_data", lambda _driver, mtime: events.append(("export", mtime)) or "x.xlsx")
monkeypatch.setattr(chanmama, "parse_chanmama_excel", lambda _path: ([{"title": "video"}], None))
records, summaries = chanmama.refresh_then_export_accounts(object(), urls, wait_seconds=600)
records, summaries, _driver = chanmama.refresh_then_export_accounts(object(), urls, wait_seconds=600)
assert len(records) == 2
assert [item["clicked"] for item in summaries] == [2, 0]
@@ -54,7 +56,7 @@ def test_refresh_then_export_skips_wait_when_nothing_needs_refresh(monkeypatch):
monkeypatch.setattr(chanmama, "_max_excel_mtime", lambda: 0.0)
monkeypatch.setattr(chanmama, "export_video_data", lambda *_args: None)
records, _ = chanmama.refresh_then_export_accounts(object(), ["account-1"], wait_seconds=600)
records, _, _driver = chanmama.refresh_then_export_accounts(object(), ["account-1"], wait_seconds=600)
assert records == []
assert sleeps == []
@@ -159,7 +161,6 @@ def test_missing_get_data_button_does_not_trigger_refresh_wait(monkeypatch):
chanmama.refresh_then_export_accounts(object(), ["account-1"], wait_seconds=600)
assert sleeps == []
def test_create_driver_uses_runtime_cdp_profile_and_download_bindings(monkeypatch, tmp_path):
calls = []
expected = object()
@@ -183,7 +184,9 @@ def test_create_driver_uses_runtime_cdp_profile_and_download_bindings(monkeypatc
"profile_dir": str(tmp_path / "profile"),
"download_dir": str(tmp_path / "downloads"),
"cdp_url": "http://127.0.0.1:22002",
"stealthy": True,
"stealthy": False,
"real_chrome": True,
"extra_flags": ["--no-proxy-server"],
}]
@@ -206,8 +209,10 @@ def test_create_driver_starts_scrapling_without_cdp(monkeypatch, tmp_path):
assert chanmama.create_driver() is expected
assert calls[0]["cdp_url"] is None
assert calls[0]["stealthy"] is True
assert calls[0]["real_chrome"] is False
# 蝉妈妈 WAF 拉黑代理出口且压制 stealthy 补丁,必须真 Chrome + 直连。
assert calls[0]["stealthy"] is False
assert calls[0]["real_chrome"] is True
assert calls[0]["extra_flags"] == ["--no-proxy-server"]
def test_create_driver_falls_back_when_bound_cdp_is_not_listening(
@@ -241,8 +246,8 @@ def test_create_driver_falls_back_when_bound_cdp_is_not_listening(
"http://127.0.0.1:22002",
None,
]
assert all(call["stealthy"] is True for call in calls)
assert calls[1]["real_chrome"] is False
assert all(call["stealthy"] is False for call in calls)
assert all(call["real_chrome"] is True for call in calls)
def test_create_driver_keeps_explicit_remote_cdp_fail_closed(
@@ -761,7 +766,7 @@ def test_refresh_then_export_relogin_on_expired_session(monkeypatch):
chanmama, "parse_chanmama_excel", lambda _path: ([{"title": "v"}], None)
)
records, summaries = chanmama.refresh_then_export_accounts(
records, summaries, _driver = chanmama.refresh_then_export_accounts(
object(), ["account-1", "account-2"], wait_seconds=0
)
@@ -791,7 +796,7 @@ def test_refresh_then_export_no_relogin_for_other_errors(monkeypatch):
chanmama, "export_video_data", lambda _driver, mtime: None
)
records, _summaries = chanmama.refresh_then_export_accounts(
records, _summaries, _driver = chanmama.refresh_then_export_accounts(
object(), ["account-1"], wait_seconds=0
)
@@ -889,3 +894,192 @@ def test_wait_video_list_ready_tolerates_transient_script_errors(monkeypatch):
return True
chanmama._wait_video_list_ready(FlakyDriver(), 2) # 不抛即通过
def test_wait_for_exported_excel_raises_when_browser_closed_without_file(monkeypatch):
"""浏览器被外部关闭且文件未落盘: 立刻抛出关闭异常,不再空等满超时。"""
class _DeadDriver:
def execute_script(self, _script, *_args):
raise RuntimeError(
"Page.goto: Target page, context or browser has been closed"
)
monkeypatch.setattr(chanmama.time, "time", lambda: 1000.0)
monkeypatch.setattr(
chanmama.time,
"sleep",
lambda _seconds: pytest.fail("浏览器已死不应继续轮询等待"),
)
monkeypatch.setattr(chanmama, "_find_new_excel_after", lambda _pre_mtime: None)
with pytest.raises(RuntimeError, match="closed"):
chanmama._wait_for_exported_excel(0.0, driver=_DeadDriver())
def test_wait_for_exported_excel_returns_file_if_landed_before_close(monkeypatch):
"""关闭前下载已完整落盘: 仍返回文件路径,不因浏览器死亡误报失败。"""
class _DeadDriver:
def execute_script(self, _script, *_args):
raise RuntimeError(
"Download.save_as: Target page, context or browser has been closed"
)
landed = None
def fake_find(_pre_mtime):
return landed
def fake_complete(path):
return True
monkeypatch.setattr(chanmama.time, "time", lambda: 1000.0)
monkeypatch.setattr(chanmama, "_find_new_excel_after", fake_find)
monkeypatch.setattr(chanmama, "_download_complete", fake_complete)
landed = "/downloads/关闭前已落盘.xlsx"
assert chanmama._wait_for_exported_excel(0.0, driver=_DeadDriver()) == landed
def test_refresh_then_export_rebuilds_browser_when_closed_mid_run(monkeypatch):
"""导出中途浏览器被外部关闭: 重建驱动+恢复会话后重试当前博主一次。"""
events = []
navigate_calls = []
dead_driver = object()
live_driver = object()
def fake_navigate(current, url):
navigate_calls.append((current, url))
if current is dead_driver and len(navigate_calls) == 3:
# 刷新阶段两个博主都成功后,导出阶段第一次导航即发现浏览器死亡
raise RuntimeError(
"蝉妈妈目标页加载失败: x.html; Page.goto: "
"Target page, context or browser has been closed"
)
monkeypatch.setattr(chanmama, "navigate_to_target", fake_navigate)
monkeypatch.setattr(
chanmama,
"refresh_zero_exposure_videos",
lambda _driver: {
"clicked": 0, "pending": 0, "zero_without_button": 0,
"unconfirmed": 0, "pages": 1,
},
)
monkeypatch.setattr(chanmama.time, "sleep", lambda _seconds: None)
monkeypatch.setattr(chanmama, "_max_excel_mtime", lambda: 0.0)
monkeypatch.setattr(
chanmama, "export_video_data", lambda _driver, mtime: "x.xlsx"
)
monkeypatch.setattr(
chanmama, "parse_chanmama_excel", lambda _path: ([{"title": "v"}], None)
)
monkeypatch.setattr(chanmama, "verify_login", lambda _driver: True)
monkeypatch.setattr(
chanmama, "create_driver", lambda: events.append("rebuild") or live_driver
)
monkeypatch.setattr(
chanmama, "login",
lambda _driver: events.append("login") or True,
)
monkeypatch.setattr(
chanmama, "load_cookies",
lambda _driver: (_ for _ in ()).throw(
AssertionError("profile 直接恢复,不应走 cookie 文件")
),
)
class _DeadDriver:
def quit(self):
events.append("quit-dead")
dead_driver = _DeadDriver()
records, _summaries, final_driver = chanmama.refresh_then_export_accounts(
dead_driver, ["account-1", "account-2"], wait_seconds=0
)
assert events == ["quit-dead", "rebuild"] # 旧驱动清理+重建一次
assert final_driver is live_driver # 返回重建后的驱动供收尾关闭
assert len(records) == 2 # 两个博主最终都导出成功
# 第 3 次导航(导出博主 1)发现死亡,其后所有导航都在重建后的驱动上
assert all(current is live_driver for current, _ in navigate_calls[3:])
def test_refresh_then_export_rebuilds_only_once(monkeypatch):
"""环境持续杀浏览器时只重建一次,后续博主直接失败不再死循环。"""
rebuilds = []
def always_closed_navigate(_driver, _url):
raise RuntimeError(
"Page.goto: Target page, context or browser has been closed"
)
class _Driver:
def quit(self):
pass
monkeypatch.setattr(chanmama, "navigate_to_target", always_closed_navigate)
monkeypatch.setattr(
chanmama, "create_driver", lambda: rebuilds.append(1) or _Driver()
)
monkeypatch.setattr(chanmama, "verify_login", lambda _driver: True)
monkeypatch.setattr(chanmama.time, "sleep", lambda _seconds: None)
records, _summaries, _driver = chanmama.refresh_then_export_accounts(
_Driver(), ["account-1", "account-2"], wait_seconds=0
)
assert len(rebuilds) == 1
assert records == []
def test_save_cookies_writes_adapter_list_format(monkeypatch, tmp_path):
"""cookie 文件与 vault/适配器共用,必须写裸列表而不是字典。"""
cookie_file = tmp_path / "cookies.json"
monkeypatch.setattr(chanmama, "COOKIE_FILE", str(cookie_file))
class _Driver:
@staticmethod
def get_cookies():
return [{"name": "session", "domain": ".chanmama.com"}]
assert chanmama.save_cookies(_Driver()) is True
payload = json.loads(cookie_file.read_text(encoding="utf-8"))
assert payload == [{"name": "session", "domain": ".chanmama.com"}]
def test_load_cookies_accepts_list_and_legacy_dict(monkeypatch, tmp_path):
cookie_file = tmp_path / "cookies.json"
injected = []
class _Driver:
@staticmethod
def get(_url):
return None
@staticmethod
def add_cookie(cookie):
injected.append(cookie)
monkeypatch.setattr(chanmama, "COOKIE_FILE", str(cookie_file))
monkeypatch.setattr(chanmama.time, "sleep", lambda _seconds: None)
cookie_file.write_text(
json.dumps([{"name": "a", "domain": ".chanmama.com"}]),
encoding="utf-8",
)
assert chanmama.load_cookies(_Driver()) is True
assert [c["name"] for c in injected] == ["a"]
injected.clear()
cookie_file.write_text(
json.dumps(
{"cookies": [{"name": "b", "domain": ".chanmama.com"}], "timestamp": time.time()}
),
encoding="utf-8",
)
assert chanmama.load_cookies(_Driver()) is True
assert [c["name"] for c in injected] == ["b"]
@@ -125,6 +125,57 @@ def test_style_records_sum_product_rows_and_keep_zero_styles() -> None:
assert records[1]["matched_product_rows"] == 0
def test_style_extraction_maps_hidden_columns_via_visible_positions() -> None:
class Frame:
def __init__(self) -> None:
self.script = ""
self.args = {}
def evaluate(self, script, args):
self.script = script
self.args = args
return {"ok": True, "rows": []}
frame = Frame()
backfill.extract_style_rows(frame, requested_codes=("10416",))
assert "window._Cols" in frame.script
assert "visibleIdx.indexOf" in frame.script
assert "cells[stylePos]" in frame.script
assert "cells.length <= Math.max(stylePos, salesPos, returnsPos)" in frame.script
def test_ensure_custom_columns_visible_only_touches_allsku_frames() -> None:
class AllskuFrame:
url = "https://bi.erp321.com/app/report/subject/itemskuanalysis/allsku.aspx?r=1"
def __init__(self) -> None:
self.scripts: list[str] = []
def evaluate(self, script):
self.scripts.append(script)
return 2
class OtherFrame:
url = "https://bi.erp321.com/app/report/subject/itemskuanalysis/filter.aspx"
def evaluate(self, script):
raise AssertionError("must not touch non-allsku frames")
class Page:
frames = [OtherFrame(), AllskuFrame()]
page = Page()
changed = backfill.ensure_custom_columns_visible(page)
assert changed == 2
allsku = page.frames[1]
assert len(allsku.scripts) == 1
assert "'自定义销量'" in allsku.scripts[0]
assert "'自定义退货量'" in allsku.scripts[0]
assert "_RptRefresh" in allsku.scripts[0]
def test_range_collector_queries_once_per_date_not_once_per_style(monkeypatch) -> None:
filter_frame = object()
monkeypatch.setattr(
@@ -0,0 +1,77 @@
import unittest
from unittest import mock
import collect_erp_yesterday_metrics as erp_mod
class _StubFrame:
"""frame.evaluate 的最小桩: 记录入参并返回固定 payload。"""
def __init__(self, payload):
self._payload = payload
self.calls = []
def evaluate(self, script, *args):
self.calls.append((script, args))
return self._payload
class ErpDailyHeaderRoutingTest(unittest.TestCase):
"""2026-10-11 聚水潭模板重置隐藏 38 列后, 固定可见索引 45/57 错位到
「今日实退货量/今日仅退款量」, daily 销量全 0。daily 必须走表头名匹配,
且不能落入 backfill 的 _extract_by_header (自定义列 SPU 求和) 分支。"""
def test_daily_mode_uses_yesterday_headers(self):
with mock.patch.object(erp_mod, "MODE", "daily"):
self.assertEqual(
erp_mod._sales_returns_idx(),
("昨日销量", "昨日退货量"),
)
def test_weekly_mode_keeps_legacy_index(self):
with mock.patch.object(erp_mod, "MODE", "weekly"):
self.assertEqual(
erp_mod._sales_returns_idx(),
("__idx__45", "__idx__57"),
)
def test_backfill_mode_keeps_custom_headers(self):
with mock.patch.object(erp_mod, "MODE", "backfill"):
self.assertEqual(
erp_mod._sales_returns_idx(),
("自定义销量", "自定义退货量"),
)
def test_daily_header_keys_bypass_extract_by_header(self):
payload = {
"ok": True,
"yesterday_sales": 2,
"yesterday_returns": 0,
"bodyText": "",
}
frame = _StubFrame(payload)
with mock.patch.object(erp_mod, "MODE", "daily"):
metrics, _body = erp_mod._extract_with_text(frame)
self.assertEqual(metrics["yesterday_sales"], 2)
self.assertEqual(len(frame.calls), 1)
script, args = frame.calls[0]
js_args = args[0] if args else script
self.assertEqual(js_args["mode"], "header")
self.assertEqual(js_args["salesHeader"], "昨日销量")
self.assertEqual(js_args["returnsHeader"], "昨日退货量")
def test_backfill_mode_routes_to_extract_by_header(self):
frame = _StubFrame({})
with mock.patch.object(erp_mod, "MODE", "backfill"):
with mock.patch.object(
erp_mod,
"_extract_by_header",
return_value={"ok": True, "yesterday_sales": 5, "summary_text": "X"},
) as by_header:
metrics, _body = erp_mod._extract_with_text(frame)
by_header.assert_called_once()
self.assertEqual(metrics["yesterday_sales"], 5)
if __name__ == "__main__":
unittest.main()
@@ -0,0 +1,191 @@
# -*- coding: utf-8 -*-
"""聚水潭销售主题分析(按日)采集脚本的纯解析与 CLI 参数测试。"""
from __future__ import annotations
import sys
from pathlib import Path
import pytest
from gyxx_flow.modules import product_commerce
from gyxx_flow.modules.product_commerce import (
collect_erp_sales_theme_daily as collector,
)
from gyxx_flow.modules.product_commerce import db as product_db
HEADER = [
"日期", "销售单数", "销售数量", "实发数量", "实发金额", "销售金额",
"退货单数", "退货数量", "实退数量", "退货金额", "实退金额",
"净销售单数", "净销量", "净销售额", "运费收入", "运费支出", "实发单数",
]
def _write_workbook(path: Path, rows: list[list]) -> None:
import openpyxl
workbook = openpyxl.Workbook()
sheet = workbook.active
for row in rows:
sheet.append(row)
workbook.save(str(path))
workbook.close()
def test_parse_sales_theme_workbook_roundtrip(tmp_path: Path) -> None:
xlsx = tmp_path / "销售主题分析_按日_1.xlsx"
_write_workbook(
xlsx,
[
HEADER,
["2026-10-11", "649", 655, 646, "173,129.23", 175752.83,
15, 15, 0, 2677.47, "0.00", 634, 640, 173075.36, "20.00", None, 640],
["2026-10-10", 690, 691, 685, 166761.58, 168185.97,
9, 9, 0, 2656.71, 0, 681, 686, 166529.13, None, 1.6, 685],
["##", 1339, 1346, 1331, 339890.81, 343938.8,
24, 24, 0, 5334.18, 0, 1315, 1326, 339604.49, 20, 1.6, 1325],
],
)
records = collector.parse_sales_theme_workbook(xlsx)
assert [record["stat_date"] for record in records] == ["2026-10-11", "2026-10-10"]
first = records[0]
assert first["sales_order_count"] == 649
assert first["sales_qty"] == 655
assert first["shipped_amount"] == 173129.23
assert first["sales_amount"] == 175752.83
assert first["return_amount"] == 2677.47
assert first["returned_amount"] == 0.0
assert first["net_sales_order_count"] == 634
assert first["net_sales_amount"] == 173075.36
assert first["freight_income"] == 20.0
assert first["freight_expense"] is None
assert first["shipped_order_count"] == 640
# 合计行(##)必须被跳过
assert len(records) == 2
def test_parse_sales_theme_workbook_rejects_missing_date_column(tmp_path: Path) -> None:
xlsx = tmp_path / "bad.xlsx"
_write_workbook(xlsx, [["销售单数", "销售数量"], [1, 2]])
with pytest.raises(ValueError):
collector.parse_sales_theme_workbook(xlsx)
def test_parse_number_and_amount_helpers() -> None:
assert collector.parse_number("1,339") == 1339
assert collector.parse_number(None) == 0
assert collector.parse_number("") == 0
assert collector.parse_amount(None) is None
assert collector.parse_amount("") is None
assert collector.parse_amount("1.6") == 1.6
assert collector.parse_amount(2677.467) == 2677.47
class _CaptureCursor:
def __init__(self) -> None:
self.batches: list[tuple] = []
def __enter__(self):
return self
def __exit__(self, *exc_info):
return False
class _CaptureConnection:
def __init__(self) -> None:
self.cursor_obj = _CaptureCursor()
def cursor(self):
return self.cursor_obj
def test_upsert_sales_theme_dedups_and_normalizes(monkeypatch: pytest.MonkeyPatch) -> None:
captured: dict[str, object] = {}
def fake_execute_values(cur, sql, rows, template=None):
captured["sql"] = sql
captured["rows"] = list(rows)
monkeypatch.setattr(product_db, "execute_values", fake_execute_values)
records = [
{
"stat_date": "2026-10-11",
"sales_order_count": 649,
"freight_income": "20.00",
},
{
# 同日重复: 整体覆盖前一条(同 upsert_erp_all_shop 语义), 数值缺省为 0
"stat_date": "2026-10-11",
"sales_order_count": 999,
"brands": ["光影", "光影行星", "光影"],
"shops": ["天猫-光影行星旗舰店", "JD光影行星旗舰店"],
"source_file": "var/raw/erp销售主题分析日报/x.xlsx",
},
{
"stat_date": "2026-10-10",
"sales_qty": "691",
},
]
conn = _CaptureConnection()
written = product_db.upsert_erp_sales_theme_daily_metrics(conn, records)
assert written == 2
rows = captured["rows"]
assert len(rows) == 2
by_date = {row[0]: row for row in rows}
# 同日重复记录取最后一条, 数字字段缺省为 0, 金额字符串转 float
assert by_date["2026-10-11"][1] == 999
assert by_date["2026-10-10"][2] == 691
tenth = by_date["2026-10-10"]
# NOT NULL 金额列缺省(导出空单元格)落 0; 运费两列允许 NULL
assert tenth[9] == 0 # return_amount
assert tenth[15] is None # freight_income
assert tenth[16] is None # freight_expense
eleventh = by_date["2026-10-11"]
assert eleventh[-4] == "发货时间" # date_basis 默认
assert eleventh[-3] == ["光影", "光影行星"] # brands 去重排序
assert eleventh[-2] == ["JD光影行星旗舰店", "天猫-光影行星旗舰店"]
assert eleventh[-1] == "var/raw/erp销售主题分析日报/x.xlsx"
# upsert 目标表与主键在 SQL 里声明
sql = str(captured["sql"])
assert "erp_sales_theme_daily_metrics" in sql
assert "ON CONFLICT (stat_date)" in sql
def test_business_scope_constants_are_consistent() -> None:
assert len(collector.BRANDS) == 3
assert sorted(collector.BRANDS) == ["gyxx", "光影", "光影行星"]
assert len(collector.SHOP_NAMES) == len(set(collector.SHOP_NAMES)) > 0
for name in collector.SHOP_NAMES:
assert name == name.strip()
def test_resolve_window_variants() -> None:
args = collector.parse_args(["--date", "2026-10-11"])
start, end = collector.resolve_window(args)
assert (start.isoformat(), end.isoformat()) == ("2026-10-11", "2026-10-11")
args = collector.parse_args(["--date", "2026-10-11", "--days", "7"])
start, end = collector.resolve_window(args)
assert (start.isoformat(), end.isoformat()) == ("2026-10-05", "2026-10-11")
args = collector.parse_args(["--from", "2025-06-01", "--to", "2026-01-01"])
start, end = collector.resolve_window(args)
assert (start.isoformat(), end.isoformat()) == ("2025-06-01", "2026-01-01")
with pytest.raises(SystemExit):
collector.resolve_window(collector.parse_args(["--from", "2026-01-01"]))
with pytest.raises(SystemExit):
collector.resolve_window(
collector.parse_args(["--from", "2026-01-01", "--to", "2025-01-01"])
)
def test_module_uses_shared_cookie_skip_exit_code() -> None:
from gyxx_flow.core.exit_codes import COOKIE_SKIP_EXIT_CODE
assert COOKIE_SKIP_EXIT_CODE == 75
assert sys.modules[product_commerce.__name__] is not None
@@ -237,6 +237,62 @@ def test_login_if_needed_fills_the_password_form_inside_an_iframe(monkeypatch) -
assert frame.submit.click_count == 1
def test_login_if_needed_accepts_agreement_popup_and_resubmits(monkeypatch) -> None:
events: list[str] = []
class Locator:
def __init__(self, name: str) -> None:
self.name = name
def fill(self, _value: str) -> None:
events.append(f"fill:{self.name}")
def click(self, timeout: int = 0) -> None:
events.append(f"click:{self.name}")
username = Locator("username")
password = Locator("password")
submit = Locator("submit")
page = SimpleNamespace(wait_for_timeout=lambda ms: events.append(f"wait:{ms}"))
monkeypatch.setattr(baibu, "_authenticated", lambda *_args: False)
monkeypatch.setattr(baibu, "_is_login_url", lambda _url: True)
def fake_form(_page, *, use_password_form):
return (_page, username, password, submit)
monkeypatch.setattr(baibu, "_find_password_login_form", fake_form)
monkeypatch.setattr(
baibu,
"_credentials_from_environment",
lambda: (baibu._LoginCredentials(username="acct", secret_value="secret"), ()),
)
agreement = {"visible": True}
monkeypatch.setattr(
baibu, "_is_login_agreement_visible", lambda _page: agreement["visible"]
)
monkeypatch.setattr(
baibu,
"_click_login_agreement_accept",
lambda _page: agreement.update(visible=False) or True,
)
assert baibu._login_if_needed(_entry(), page) is True
# First submit is swallowed by the 服务协议 popup; after accepting it the
# credentials are re-filled and submitted a second time.
assert events == [
"fill:username",
"fill:password",
"click:submit",
"wait:2000",
"fill:username",
"fill:password",
"click:submit",
"wait:2000",
]
def test_manual_login_waits_without_filling_the_password_form(monkeypatch) -> None:
monkeypatch.setattr(baibu, "_authenticated", lambda _entry, _page: False)
frame = _LoginFrame()
@@ -301,6 +357,14 @@ def test_login_second_factor_marker_is_detected(monkeypatch) -> None:
assert baibu._login_second_factor_marker(SimpleNamespace()) == "短信验证"
def test_login_slider_captcha_marker_is_detected(monkeypatch) -> None:
monkeypatch.setattr(
baibu, "_page_text", lambda _page: "向右拖动滑块填充拼图完成验证"
)
assert baibu._login_second_factor_marker(SimpleNamespace()) == "滑块"
def test_persist_account_state_filters_cookies_to_the_account_domains(
monkeypatch,
tmp_path: Path,
@@ -351,6 +415,289 @@ def test_persist_account_state_filters_cookies_to_the_account_domains(
]
def _browser_spy(calls: list[str]):
class Page:
def goto(self, *_args, **_kwargs) -> None:
calls.append("goto")
class Context:
pages = [Page()]
def new_page(self):
return Page()
class Browser:
context = Context()
def start(self) -> None:
calls.append("start")
def persist_bound_state(self) -> None:
calls.append("persist-bound")
def close(self) -> None:
calls.append("close")
return Browser()
def test_apply_entry_continues_when_account_vault_refresh_fails(monkeypatch) -> None:
calls: list[str] = []
monkeypatch.setattr(baibu, "_start_browser", lambda: _browser_spy(calls))
monkeypatch.setattr(baibu, "_login_if_needed", lambda *_args, **_kwargs: False)
monkeypatch.setattr(baibu, "_wait_for_authenticated", lambda *_args, **_kwargs: None)
def fail_vault_refresh(_browser, _entry, **_kwargs):
calls.append("persist-fail")
raise baibu.BaibuApplyError(
"测试入口 登录后未获得可用的账号 cookie(域内 cookie:无;"
"要求含:unb、cookie2、_tb_token_),未更新账号登录态"
)
monkeypatch.setattr(baibu, "_persist_account_state", fail_vault_refresh)
def record_explicit_login(_entry, _page, *, manual_login):
calls.append("explicit-login")
monkeypatch.setattr(baibu, "_explicit_password_login", record_explicit_login)
def record_open(_page, _entry):
calls.append("open")
return None
monkeypatch.setattr(baibu, "_open_batch_import", record_open)
monkeypatch.setattr(baibu, "_upload_template", lambda *_args: calls.append("upload"))
monkeypatch.setattr(baibu, "_wait_import_result", lambda *_args, **_kwargs: calls.append("result"))
monkeypatch.setattr(baibu, "_complete_draft_products", lambda *_args: calls.append("drafts"))
monkeypatch.setattr(baibu.time, "sleep", lambda _seconds: None)
baibu._apply_entry(_entry(), Path("template.xlsx"))
assert calls == [
"start",
"goto",
"persist-bound",
"persist-fail",
"explicit-login",
"persist-fail",
"open",
"upload",
"result",
"drafts",
"close",
]
def test_apply_entry_retries_through_explicit_login_when_page_never_shows_login(
monkeypatch,
) -> None:
calls: list[str] = []
monkeypatch.setattr(baibu, "_start_browser", lambda: _browser_spy(calls))
monkeypatch.setattr(baibu, "_login_if_needed", lambda *_args, **_kwargs: False)
attempts = {"count": 0}
def fail_first_authentication(*_args, **_kwargs):
attempts["count"] += 1
if attempts["count"] == 1:
raise baibu.BaibuApplyError("测试入口 当前未登录对应店铺,已停止")
calls.append("authenticated")
monkeypatch.setattr(baibu, "_wait_for_authenticated", fail_first_authentication)
def record_explicit_login(_entry, _page, *, manual_login):
calls.append("explicit-login")
monkeypatch.setattr(baibu, "_explicit_password_login", record_explicit_login)
monkeypatch.setattr(
baibu,
"_persist_account_state",
lambda *_args, **_kwargs: calls.append("persist-ok") or True,
)
def record_open(_page, _entry):
calls.append("open")
return None
monkeypatch.setattr(baibu, "_open_batch_import", record_open)
monkeypatch.setattr(baibu, "_upload_template", lambda *_args: calls.append("upload"))
monkeypatch.setattr(baibu, "_wait_import_result", lambda *_args, **_kwargs: calls.append("result"))
monkeypatch.setattr(baibu, "_complete_draft_products", lambda *_args: calls.append("drafts"))
monkeypatch.setattr(baibu.time, "sleep", lambda _seconds: None)
baibu._apply_entry(_entry(), Path("template.xlsx"))
assert attempts["count"] == 1
assert calls == [
"start",
"goto",
"explicit-login",
"persist-bound",
"persist-ok",
"open",
"upload",
"result",
"drafts",
"close",
]
def test_explicit_password_login_drives_login_page_then_returns_to_apply(
monkeypatch,
) -> None:
calls: list[str] = []
class Page:
def goto(self, url, **_kwargs) -> None:
calls.append(f"goto:{url}")
def wait_for_timeout(self, _ms: int) -> None:
return None
monkeypatch.setattr(
baibu, "_login_if_needed", lambda *_args, **_kwargs: calls.append("fill-login") or True
)
monkeypatch.setattr(
baibu,
"_wait_for_authenticated",
lambda *_args, **_kwargs: calls.append("authenticated"),
)
entry = _entry()
baibu._explicit_password_login(entry, Page(), manual_login=False)
# The apply URL is re-entered first so its redirect chain can surface the
# seller login page; only a missing form there falls back to login.taobao.
assert calls == [
f"goto:{entry.apply_url}",
"fill-login",
f"goto:{entry.apply_url}",
"authenticated",
]
class _FakeLock:
def __init__(self, manager: "_FakeLockManager", resource: str) -> None:
self._manager = manager
self._resource = resource
def __enter__(self):
self._manager.held.add(self._resource)
self._manager.entered.append(self._resource)
return self
def __exit__(self, *_args) -> None:
self._manager.held.discard(self._resource)
class _FakeLockManager:
def __init__(self, lock_root) -> None:
self.lock_root = lock_root
self.held: set[str] = set()
self.entered: list[str] = []
self.requested: list[tuple[str, int]] = []
def acquire(self, resource: str, *, owner: str, timeout_seconds: int):
self.requested.append((resource, timeout_seconds))
if getattr(self, "busy", False):
raise baibu.ResourceBusyError(resource)
return _FakeLock(self, resource)
def test_apply_entry_holds_account_lock_for_whole_session(monkeypatch) -> None:
calls: list[str] = []
lock_manager = _FakeLockManager(lock_root="locks-dir")
class FakeSettings:
data_root = Path("data-root")
@classmethod
def from_env(cls):
return cls()
monkeypatch.setattr(baibu, "Settings", FakeSettings)
monkeypatch.setattr(baibu, "LockManager", lambda lock_root: lock_manager)
monkeypatch.setattr(baibu, "_start_browser", lambda: _browser_spy(calls))
monkeypatch.setattr(baibu, "_login_if_needed", lambda *_args, **_kwargs: False)
monkeypatch.setattr(baibu, "_wait_for_authenticated", lambda *_args, **_kwargs: None)
monkeypatch.setattr(
baibu,
"_persist_account_state",
lambda *_args, **_kwargs: True,
)
monkeypatch.setattr(baibu, "_open_batch_import", lambda _page, _entry: None)
def assert_lock_held_during_upload(_page, _template):
assert lock_manager.held == {"browser-account:tmall-shop"}
calls.append("upload")
monkeypatch.setattr(baibu, "_upload_template", assert_lock_held_during_upload)
monkeypatch.setattr(baibu, "_wait_import_result", lambda *_args, **_kwargs: None)
monkeypatch.setattr(baibu, "_complete_draft_products", lambda *_args: None)
monkeypatch.setattr(baibu.time, "sleep", lambda _seconds: None)
baibu._apply_entry(_entry(), Path("template.xlsx"))
assert lock_manager.requested == [("browser-account:tmall-shop", 120)]
assert lock_manager.entered == ["browser-account:tmall-shop"]
assert lock_manager.held == set()
def test_apply_entry_reports_busy_account_lock(monkeypatch) -> None:
lock_manager = _FakeLockManager(lock_root="locks-dir")
lock_manager.busy = True
class FakeSettings:
data_root = Path("data-root")
@classmethod
def from_env(cls):
return cls()
monkeypatch.setattr(baibu, "Settings", FakeSettings)
monkeypatch.setattr(baibu, "LockManager", lambda lock_root: lock_manager)
started: list[str] = []
monkeypatch.setattr(
baibu, "_start_browser", lambda: started.append("start") or object()
)
with pytest.raises(baibu.BaibuApplyError, match="正被保活/登录占用"):
baibu._apply_entry(_entry(), Path("template.xlsx"))
assert started == []
def test_explicit_password_login_falls_back_to_plain_login_page(
monkeypatch,
) -> None:
calls: list[str] = []
class Page:
def goto(self, url, **_kwargs) -> None:
calls.append(f"goto:{url}")
def wait_for_timeout(self, _ms: int) -> None:
return None
monkeypatch.setattr(baibu, "_login_if_needed", lambda *_args, **_kwargs: False)
monkeypatch.setattr(
baibu,
"_wait_for_authenticated",
lambda *_args, **_kwargs: calls.append("authenticated"),
)
entry = _entry()
baibu._explicit_password_login(entry, Page(), manual_login=False)
assert calls == [
f"goto:{entry.apply_url}",
f"goto:{baibu._LOGIN_PAGE_URL}",
f"goto:{entry.apply_url}",
"authenticated",
]
def test_upload_template_supports_tmall_dialog_import_button(
monkeypatch,
tmp_path: Path,
@@ -558,6 +905,32 @@ def test_wait_import_result_accepts_new_failure_summary_as_submitted(
baibu._wait_import_result(SimpleNamespace(), _entry())
def test_wait_import_result_accepts_slow_processing_after_acknowledgement(
monkeypatch,
capsys,
) -> None:
# Tmall acknowledged the batch but never returned a row-level summary
# within the acceptance window; the acknowledged submission still counts.
processing_text = (
"您于2026-10-14 14:51:00执行了商品批量导入操作,正在处理中。"
)
clock = {"now": 0.0}
def advance_monotonic() -> float:
clock["now"] += 15.0
return clock["now"]
monkeypatch.setattr(baibu, "_page_text", lambda _page: processing_text)
monkeypatch.setattr(baibu, "_refresh_batch_results", lambda _page: True)
monkeypatch.setattr(baibu.time, "sleep", lambda _seconds: None)
monkeypatch.setattr(baibu.time, "monotonic", advance_monotonic)
monkeypatch.setattr(baibu, "_RESULT_REFRESH_INTERVAL_SECONDS", 0)
baibu._wait_import_result(SimpleNamespace(), _entry())
assert "按导入提交成功" in capsys.readouterr().out
def test_wait_import_result_ignores_stale_summary_until_signature_changes(
monkeypatch,
) -> None:
@@ -663,6 +1036,47 @@ def test_complete_draft_products_accepts_empty_data_page(monkeypatch) -> None:
baibu._complete_draft_products(page, entry)
def test_complete_draft_products_retries_draft_tab_during_hydration(
monkeypatch,
) -> None:
entry = replace(_entry(), complete_drafts_after_import=True)
page = SimpleNamespace()
attempts = {"n": 0}
def flaky_tab(_page):
attempts["n"] += 1
return attempts["n"] >= 3
monkeypatch.setattr(baibu, "_click_draft_tab", flaky_tab)
monkeypatch.setattr(baibu, "_page_text", lambda _page: "加载中")
monkeypatch.setattr(baibu, "_has_visible_text", lambda _page, _labels: False)
monkeypatch.setattr(baibu, "_wait_for_draft_state", lambda _page, _entry: (0, 0))
monkeypatch.setattr(baibu.time, "sleep", lambda _seconds: None)
ticks = iter(range(10_000))
monkeypatch.setattr(baibu.time, "monotonic", lambda: next(ticks) * 0.4)
baibu._complete_draft_products(page, entry)
assert attempts["n"] == 3
def test_complete_draft_products_reports_missing_draft_entry_after_retries(
monkeypatch,
) -> None:
entry = replace(_entry(), complete_drafts_after_import=True)
page = SimpleNamespace()
monkeypatch.setattr(baibu, "_click_draft_tab", lambda _page: False)
monkeypatch.setattr(baibu, "_page_text", lambda _page: "加载中")
monkeypatch.setattr(baibu, "_has_visible_text", lambda _page, _labels: False)
monkeypatch.setattr(baibu.time, "sleep", lambda _seconds: None)
ticks = iter(range(10_000))
monkeypatch.setattr(baibu.time, "monotonic", lambda: next(ticks) * 0.4)
with pytest.raises(baibu.BaibuApplyError, match="未找到草稿页入口"):
baibu._complete_draft_products(page, entry)
def test_return_to_draft_list_falls_back_to_browser_history(monkeypatch) -> None:
events: list[str] = []
+1 -1
View File
@@ -28,7 +28,7 @@ from gyxx_flow.workflow.engine import WorkflowEngine
from gyxx_flow.workflow.model import StepDefinition, WorkflowDefinition
from gyxx_flow.workflow.steps import StepExecution
_NOW = datetime(2026, 10, 8, 12, 0, 0, tzinfo=timezone.utc)
_NOW = datetime.now(timezone.utc).replace(microsecond=0)
def _settings(tmp_path: Path) -> Settings:
+3 -3
View File
@@ -102,10 +102,10 @@ def test_project_registry_covers_every_script_with_unique_stable_port() -> None:
data_root=project_root / "var",
)
assert len(scripts.command_ids) == 60
assert len(first.command_ids) == 157
assert len(scripts.command_ids) == 61
assert len(first.command_ids) == 158
bindings = [first.binding_for(command_id) for command_id in first.command_ids]
assert len({binding.cdp_port for binding in bindings}) == 157
assert len({binding.cdp_port for binding in bindings}) == 158
assert all(22000 <= binding.cdp_port <= 22999 for binding in bindings)
assert all(binding.cdp_url.startswith("http://127.0.0.1:") for binding in bindings)
assert bindings == [second.binding_for(item.script_id) for item in bindings]
+2 -2
View File
@@ -83,7 +83,7 @@ def test_default_registry_covers_every_executable_workflow_entry() -> None:
assert scheduled_workflows == 39
assert manual_workflows == 0
assert len(workflow_entries) == 55
assert len(catalog.scripts) == 60
assert len(catalog.scripts) == 61
assert workflow_entries < set(catalog.script_ids)
assert {
"content.mapping.rebuild",
@@ -280,7 +280,7 @@ def test_cli_lists_and_dry_runs_registered_commands(tmp_path: Path) -> None:
assert main(["scripts", "list", "--json"], settings=settings, stdout=output) == 0
rows = json.loads(output.getvalue())
assert len(rows) == 60
assert len(rows) == 61
assert {row["module"] for row in rows} == {
"content_marketing",
"product_commerce",