feat(product): 聚水潭销售主题分析按日采集工作流

collect_erp_sales_theme_daily.py 采集 bi.erp321.com「报表→销售主题分析→按日统计」，
异步导出 + OSS 直连兜底，固定品牌+店铺口径，写 erp_sales_theme_daily_metrics
（db 层建表与 upsert 同步，每日一行幂等覆盖）。注册 product.erp_sales_theme_daily.collect
手动命令，复用 erp-account 登录绑定新增 CDP 22167；目录/集成计数 60→61，
新脚本纳入 CI 聚焦 lint 清单。

.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

@@ -168,6 +168,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",
@@ -1346,6 +1347,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",

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

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