diff --git a/.gitlab-ci.yml b/.gitlab-ci.yml index 4aa6364..c5f0b42 100644 --- a/.gitlab-ci.yml +++ b/.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 diff --git a/config/commands.json b/config/commands.json index 15fa71d..c4ba062 100644 --- a/config/commands.json +++ b/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", diff --git a/config/runtime-bindings.json b/config/runtime-bindings.json index 50bcbdc..0845234 100644 --- a/config/runtime-bindings.json +++ b/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", diff --git a/src/gyxx_flow/modules/product_commerce/collect_erp_sales_theme_daily.py b/src/gyxx_flow/modules/product_commerce/collect_erp_sales_theme_daily.py new file mode 100644 index 0000000..4f1557e --- /dev/null +++ b/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()) diff --git a/src/gyxx_flow/modules/product_commerce/db/__init__.py b/src/gyxx_flow/modules/product_commerce/db/__init__.py index 91410c9..30b29fe 100644 --- a/src/gyxx_flow/modules/product_commerce/db/__init__.py +++ b/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) diff --git a/src/gyxx_flow/modules/product_commerce/db/schema.sql b/src/gyxx_flow/modules/product_commerce/db/schema.sql index ba231a1..ab62349 100644 --- a/src/gyxx_flow/modules/product_commerce/db/schema.sql +++ b/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); diff --git a/tests/modules/product_commerce/test_erp_sales_theme_daily.py b/tests/modules/product_commerce/test_erp_sales_theme_daily.py new file mode 100644 index 0000000..65b5dde --- /dev/null +++ b/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 diff --git a/tests/test_runtime_integrations.py b/tests/test_runtime_integrations.py index c7afdfd..6aff8c4 100644 --- a/tests/test_runtime_integrations.py +++ b/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(first.command_ids) == 157 + assert len(scripts.command_ids) == 61 + assert len(first.command_ids) == 158 bindings = [first.binding_for(command_id) for command_id in first.command_ids] - assert len({binding.cdp_port for binding in bindings}) == 157 + assert len({binding.cdp_port for binding in bindings}) == 158 assert all(22000 <= binding.cdp_port <= 22999 for binding in bindings) assert all(binding.cdp_url.startswith("http://127.0.0.1:") for binding in bindings) assert bindings == [second.binding_for(item.script_id) for item in bindings] diff --git a/tests/test_script_catalog.py b/tests/test_script_catalog.py index 8b3605e..7e6c77d 100644 --- a/tests/test_script_catalog.py +++ b/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",