chore: 忽略 .mimosa/.workbuddy 本地工具状态目录

autologon.ps1 交互输密码启用/撤销 AutoAdminLogon（DefaultPassword 仅
Administrators 可读），补齐「开机自动登录 → Startup keepalive → 控制台+
调度器」自启链唯一缺口；采集依赖 CDP 交互会话，不能用服务/SYSTEM 任务
替代。runbook 增补开机自启与重启验收章节。

.gitignore

@@ -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/

.gitlab-ci.yml

@@ -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

config/commands.json

@@ -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",

config/runtime-bindings.json

@@ -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",
+        "u_co_id"
-        "ASP.NET_SessionId",
-        ".ASPXAUTH",
-        "token"
       ],
       "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"],
+      "required_cookie_domains": [
-      "credential_env_names": ["XHS_ARK_ACCOUNT"]
+        "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",

deploy/watchdog/autologon.ps1

@@ -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。'

docs/runbook.md

@@ -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`：

src/gyxx_flow/adapters/integration.py

@@ -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):
+    def _domain_matches(domain: str) -> bool:
-            continue
+        return not required_domains or any(
-        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(
             domain == required.casefold()
             or domain.endswith("." + required.casefold())
             for required in required_domains
         )
-        if name_matches and domain_matches:
+
-            return True
+    def _live(item: object) -> bool:
-    return False
+        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:

src/gyxx_flow/adapters/scrapling.py

@@ -37,8 +37,8 @@ def is_browser_timeout_error(error: BaseException) -> bool:
     )
 
 
-def _is_browser_closed_error(error: BaseException) -> bool:
+def is_browser_closed_error(error: BaseException) -> bool:
-    """Recognize a browser target that disappeared during cleanup."""
+    """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:

src/gyxx_flow/modules/content_marketing/chanmama_scraper.py

@@ -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)
+            navigate_to_target(driver_box["driver"], url)
-            summary = refresh_zero_exposure_videos(driver)
+            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)
+                    navigate_to_target(driver_box["driver"], url)
-                    summary = refresh_zero_exposure_videos(driver)
+                    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:
+            except Exception as pump_exc:
-                pass
+                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
         )
 

src/gyxx_flow/modules/content_marketing/scrapling_browser_facade.py

@@ -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))

src/gyxx_flow/modules/product_commerce/backfill_erp_all_shop_daily.py

@@ -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;
+            if (cells.length <= Math.max(stylePos, salesPos, returnsPos)) continue;
-            const styleCode = norm(cells[styleCol]?.textContent);
+            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_code: productPos >= 0 ? norm(cells[productPos]?.textContent) : '',
-              product_name: productNameCol >= 0 ? norm(cells[productNameCol]?.textContent) : '',
+              product_name: namePos >= 0 ? norm(cells[namePos]?.textContent) : '',
-              sales: parseInteger(cells[salesCol]?.textContent),
+              sales: parseInteger(cells[salesPos]?.textContent),
-              return_qty: parseInteger(cells[returnsCol]?.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,

src/gyxx_flow/modules/product_commerce/collect_erp_sales_theme_daily.py

@@ -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())

src/gyxx_flow/modules/product_commerce/collect_erp_yesterday_metrics.py

@@ -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"):
+    if MODE == "daily":
-        # daily / weekly 都走 cell 45/57 老路径, 跟之前 data.json 口径一致
+        # 表头匹配 (昨日销量 / 昨日退货量): 模板列可见性变化会让固定索引错位
+        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 salesCell = cells[salesPos];
-            const returnsCell = cells[returnsCol];
+            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 系统里不含自定义段)
+    # backfill 走 SPU 行 + 自定义列名匹配 + 求和 (因为 sumRow 在 ERP 系统里不含自定义段);
-    if sales_key != "__idx__45":  # backfill 模式
+    # 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 模式
+    # weekly 老 idx 模式 + daily header 模式共用这段 payload (js_args 已按模式构造)
-    sales_idx = int(sales_key.removeprefix("__idx__"))
-    returns_idx = int(returns_key.removeprefix("__idx__"))
     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);
+            salesCol = findCol(args.salesHeader);
-            const returnsCol = findCol(args.returnsHeader);
+            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(

src/gyxx_flow/modules/product_commerce/db/__init__.py

@@ -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)

src/gyxx_flow/modules/product_commerce/db/schema.sql

@@ -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);

src/gyxx_flow/modules/product_commerce/tmall_baibu_apply_common.py

@@ -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,10 +1432,21 @@ def _complete_draft_products(page: Any, entry: BaibuEntry) -> None:
     if not entry.complete_drafts_after_import:
         return
 
-    if not _click_draft_tab(page):
+    # Right after the import acknowledgement the SPA may still be hydrating;
-        count = _draft_count(_page_text(page))
+    # the draft tab, the draft count and the per-row action rarely all appear
-        if count is None and not _has_visible_text(page, ("完善商品",)):
+    # on the first probe, so retry briefly before declaring the page broken.
-            raise BaibuApplyError(f"{entry.name} 未找到草稿页入口")
+    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)
     if draft_count == 0:
@@ -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"
+
-    try:
+    def _save() -> None:
-        with LockManager(lock_root).acquire(
+        try:
-            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)
-    except ResourceBusyError as exc:
+        except (OSError, ValueError, TypeError) as exc:
-        raise BaibuApplyError(
+            raise BaibuApplyError(
-            f"{entry.name} 账号登录态正在被其他进程更新，未覆盖账号登录态"
+                f"{entry.name} 保存账号登录态失败（{type(exc).__name__}）"
-        ) from exc
+            ) from exc
-    except (OSError, ValueError, TypeError) as exc:
+
-        raise BaibuApplyError(
+    if session_lock_held:
-            f"{entry.name} 保存账号登录态失败（{type(exc).__name__}）"
+        # The whole browser session already holds the scheduler-side
-        ) from exc
+        # 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,
+            ):
+                _save()
+        except ResourceBusyError as exc:
+            raise BaibuApplyError(
+                f"{entry.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,
         )
-        _wait_for_authenticated(
+        try:
-            entry,
+            _wait_for_authenticated(
-            page,
+                entry,
-            login_attempted=login_attempted,
+                page,
-            manual_login=manual_login,
+                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)

tests/modules/content_marketing/test_chanmama_refresh_before_export.py

@@ -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
+    # 蝉妈妈 WAF 拉黑代理出口且压制 stealthy 补丁,必须真 Chrome + 直连。
-    assert calls[0]["real_chrome"] is False
+    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 all(call["stealthy"] is False for call in calls)
-    assert calls[1]["real_chrome"] is False
+    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"]

tests/modules/product_commerce/test_erp_all_shop_daily_backfill.py

@@ -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(

tests/modules/product_commerce/test_erp_daily_header_routing.py

@@ -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()

tests/modules/product_commerce/test_erp_sales_theme_daily.py

@@ -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

tests/modules/product_commerce/test_tmall_baibu_apply.py

@@ -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] = []
 

tests/test_cookie_gate.py

@@ -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:

tests/test_runtime_integrations.py

@@ -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(scripts.command_ids) == 61
-    assert len(first.command_ids) == 157
+    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]

tests/test_script_catalog.py

@@ -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",