diff --git a/src/gyxx_flow/adapters/workflow_config.py b/src/gyxx_flow/adapters/workflow_config.py index 9fa3c04..c81b3ff 100644 --- a/src/gyxx_flow/adapters/workflow_config.py +++ b/src/gyxx_flow/adapters/workflow_config.py @@ -33,6 +33,25 @@ DEFAULT_SEED_PATH = ( / "workflow-dynamic-config-seed.json" ) +DIM_STYLE_RUNTIME_SQL = """ + SELECT s.id, + s.style_name AS name, + COALESCE(s.brand, '') AS brand, + COALESCE(s.erp_style_codes, '[]'::jsonb) AS erp_style_codes, + jsonb_build_object( + 'tm', COALESCE(s.tm_spus, '[]'::jsonb), + 'jd', COALESCE(s.jd_spus, '[]'::jsonb), + 'dy', COALESCE(s.dy_spus, '[]'::jsonb), + 'jd_self', COALESCE(s.jd_self_skus, '[]'::jsonb) + ) AS platform_product_ids, + COALESCE(s.bitable_weekly_url, '') AS sales_bitable_url, + COALESCE(s.series_name, '') AS series_name, + COALESCE(s.is_series_anchor, FALSE) AS is_series_anchor + FROM dim_style s + WHERE COALESCE(s.style_name, '') <> '' + ORDER BY s.style_name, s.id +""" + CONFIG_FIELDS = ( "platform", "erp_codes", @@ -298,8 +317,9 @@ def merge_lifecycle_style_records( """Overlay lifecycle product identity on legacy target configuration rows. ``workflow_dynamic_*`` still owns Feishu target URLs and persona targets, while - ``cmt_styles`` owns ERP codes and product IDs. The result intentionally keeps - the old flat runtime contract so existing collectors need no per-workflow fork. + ``dim_style`` (产品生命进程建档) owns ERP codes, product IDs and the sales sheet. + The result intentionally keeps the old flat runtime contract so existing + collectors need no per-workflow fork. """ dynamic_by_style: dict[str, list[Mapping[str, Any]]] = {} @@ -311,7 +331,7 @@ def merge_lifecycle_style_records( lifecycle_by_style: dict[str, Mapping[str, Any]] = {} lifecycle_order: list[str] = [] for row in lifecycle_rows: - style_name = " ".join(str(row.get("name") or "").split()) + style_name = " ".join(str(row.get("name") or row.get("style_name") or "").split()) if not style_name or style_name in lifecycle_by_style: continue lifecycle_by_style[style_name] = row @@ -379,6 +399,19 @@ def merge_lifecycle_style_records( if item_id not in dynamic_unknown_ids.setdefault(label, []): dynamic_unknown_ids[label].append(item_id) + # 销量飞书表：生命进程建档的地址优先，动态配置只在建档为空时兜底。 + sales_bitable_url = str(lifecycle.get("sales_bitable_url") or "").strip() + if not sales_bitable_url and legacy_rows: + sales_bitable_url = str(legacy_rows[0].get("sales_bitable_url") or "").strip() + series_name = str(lifecycle.get("series_name") or "").strip() + is_series_anchor = bool(lifecycle.get("is_series_anchor")) + + # 生命进程建档是唯一权威：销量表与系列归属直接写进运行时记录， + # 子款即使没填平台编码/销量表，也能被下游按系列主款补齐。 + template["sales_bitable_url"] = sales_bitable_url + template["series_name"] = series_name + template["is_series_anchor"] = is_series_anchor + if managed: labels = [] for label in LIFECYCLE_PLATFORM_LABELS.values(): @@ -413,7 +446,7 @@ def merge_lifecycle_style_records( else list(dynamic_ids.get(key, dynamic_unknown_ids.get(label, []))) ) output["source_record_id"] = ( - f"cmt_style:{lifecycle.get('id')}" if key and managed else output.get("source_record_id") + f"lifecycle_style:{lifecycle.get('id')}" if key and managed else output.get("source_record_id") ) records.append(output) @@ -1158,16 +1191,24 @@ class PostgresWorkflowConfigStore: "WHERE s.enabled ORDER BY s.style_name, p.id" ) rows = cursor.fetchall() - cursor.execute("SELECT to_regclass('public.cmt_styles') AS table_name") - cmt_table = cursor.fetchone() or {} - if cmt_table.get("table_name"): - cursor.execute( - "SELECT id, name, brand, erp_style_codes, platform_product_ids " - "FROM cmt_styles ORDER BY name, id" - ) + cursor.execute("SELECT to_regclass('public.dim_style') AS table_name") + dim_table = cursor.fetchone() or {} + if dim_table.get("table_name"): + # 产品生命进程（dim_style）是款式身份的唯一权威来源： + # ERP 款式编码逐款填写，平台商品编码与销量表允许整系列只填主款。 + cursor.execute(DIM_STYLE_RUNTIME_SQL) lifecycle_rows = cursor.fetchall() else: - lifecycle_rows = [] + cursor.execute("SELECT to_regclass('public.cmt_styles') AS table_name") + cmt_table = cursor.fetchone() or {} + if cmt_table.get("table_name"): + cursor.execute( + "SELECT id, name, brand, erp_style_codes, platform_product_ids " + "FROM cmt_styles ORDER BY name, id" + ) + lifecycle_rows = cursor.fetchall() + else: + lifecycle_rows = [] personas: dict[int, dict[str, str]] = {} for row in rows: if row.get("platform_enabled") is False: diff --git a/src/gyxx_flow/modules/product_commerce/backfill_erp_all_shop_daily.py b/src/gyxx_flow/modules/product_commerce/backfill_erp_all_shop_daily.py index fcffc7d..326db2c 100644 --- a/src/gyxx_flow/modules/product_commerce/backfill_erp_all_shop_daily.py +++ b/src/gyxx_flow/modules/product_commerce/backfill_erp_all_shop_daily.py @@ -18,7 +18,7 @@ from scrapling.fetchers import DynamicFetcher try: from . import collect_erp_yesterday_metrics as erp from .collect_retry_utils import retry_step - from .config.style_config_loader import StyleConfigLoader, has_complete_style_identity + from .config.style_config_loader import StyleConfigLoader from .db import ( get_conn, init_schema, @@ -35,7 +35,7 @@ try: except ImportError: # Direct script execution from the module root. import collect_erp_yesterday_metrics as erp from collect_retry_utils import retry_step - from config.style_config_loader import StyleConfigLoader, has_complete_style_identity + from config.style_config_loader import StyleConfigLoader from db import get_conn, init_schema, upsert_erp_all_shop_style_daily_metrics from erp_login_product_analysis import ( DEFAULT_CONFIG, @@ -88,7 +88,7 @@ def build_style_plans(payload: Mapping[str, Any]) -> tuple[StylePlan, ...]: if str(code).strip() ) ) - if not style_name or not codes or not has_complete_style_identity(config): + if not style_name or not codes: continue for code in codes: previous = code_owners.setdefault(code, style_name) diff --git a/src/gyxx_flow/modules/product_commerce/config/style_config_loader.py b/src/gyxx_flow/modules/product_commerce/config/style_config_loader.py index 9a5d0a2..ab2aa13 100644 --- a/src/gyxx_flow/modules/product_commerce/config/style_config_loader.py +++ b/src/gyxx_flow/modules/product_commerce/config/style_config_loader.py @@ -32,7 +32,7 @@ except ImportError: # Direct script execution from the module root. from runtime_paths import NORMALIZED_DATA_ROOT, STATE_ROOT, managed_data_path -CACHE_SCHEMA_VERSION = 5 +CACHE_SCHEMA_VERSION = 6 # 表格字段名（按实际表格设置，支持别名） FIELD_ALIASES = { @@ -266,6 +266,8 @@ class StyleConfigLoader: "style_analysis_bitable_url", "" ), "destinations": row.get("destinations", []), + "series_name": row.get("series_name", ""), + "is_series_anchor": bool(row.get("is_series_anchor", False)), } ) return records @@ -315,6 +317,9 @@ class StyleConfigLoader: "jd_persona_bitable": {}, "style_analysis_bitable": {}, "destinations": [], + "series_name": "", + "is_series_anchor": False, + "series": {}, } # ── 第一遍:款式级字段 ────────────────────────────────────── @@ -332,6 +337,13 @@ class StyleConfigLoader: if brand and not style["brand"]: style["brand"] = brand + # 系列归属（产品生命进程建档维护）：同系列可共享平台编码与销量表 + series_name = str(record.get("series_name") or "").strip() + if series_name and not style["series_name"]: + style["series_name"] = series_name + if record.get("is_series_anchor"): + style["is_series_anchor"] = True + destinations = record.get("destinations") if isinstance(destinations, list) and destinations and not style["destinations"]: style["destinations"] = [ @@ -427,6 +439,9 @@ class StyleConfigLoader: elif platform == "jd_self" and ids: style["jd_self"]["skus"] = ids + # ── 第三遍: 系列内共享（平台编码 / 销量表只要求主款填写）────────── + self._apply_series_sharing(styles) + return { "styles": styles, "source": "database", @@ -434,6 +449,46 @@ class StyleConfigLoader: "schema_version": CACHE_SCHEMA_VERSION, } + @staticmethod + def _apply_series_sharing(styles: dict[str, dict]) -> None: + """把同系列的平台编码归属与销量表摊到每个子款。 + + 产品生命进程的录入约定：ERP 款式编码逐款必填，平台商品编码和 + 销量飞书表整系列只需要主款（锚点）填一次。这里只做「归属」标注 + 和销量表继承，不复制平台商品编码——同一个商品 ID 若被同系列多款 + 分别采集，会把同一份访客/加购重复写进每日指标。 + """ + + by_series: dict[str, list[str]] = {} + for name, style in styles.items(): + series = str(style.get("series_name") or "").strip() + if series: + by_series.setdefault(series, []).append(name) + + for series, members in by_series.items(): + # 锚点优先，其余按名称稳定排序，保证每次运行归属一致 + members.sort(key=lambda n: (not styles[n].get("is_series_anchor"), n)) + platform_owner: dict[str, str] = {} + for platform, ids_key in PRODUCT_ID_FIELDS.items(): + for member in members: + if styles[member].get(platform, {}).get(ids_key): + platform_owner[platform] = member + break + sales_owner = next( + (m for m in members if (styles[m].get("sales_bitable") or {}).get("base_token")), + "", + ) + for member in members: + style = styles[member] + style["series"] = { + "name": series, + "is_anchor": bool(style.get("is_series_anchor")), + "platform_owner": dict(platform_owner), + "sales_owner": sales_owner, + } + if not style["sales_bitable"] and sales_owner: + style["sales_bitable"] = dict(styles[sales_owner]["sales_bitable"]) + # ------------------------------------------------------------------ # 缓存与回退 # ------------------------------------------------------------------ @@ -656,10 +711,11 @@ class StyleConfigLoader: ] def get_erp_styles_input(self) -> list[dict]: + """ERP 编码清单：只要求款式编码，平台编码/销量表由系列主款承担。""" return [ {"style_name": name, "erp_style_codes": cfg["erp_codes"]} for name, cfg in self.load()["styles"].items() - if self._has_complete_identity(cfg) + if cfg.get("erp_codes") ] def get_bitable_style_map(self) -> dict: @@ -745,14 +801,17 @@ class StyleConfigLoader: payload = self.load() result = {p: {"collectable": [], "skipped": []} for p in ("tm", "jd", "dy")} for name, cfg in payload["styles"].items(): - identity_missing = self._identity_missing(cfg) + # 平台编码在下面按平台逐项判定，这里不重复报「平台商品编码」 + identity_missing = [ + item for item in self._identity_missing(cfg) if item != "平台商品编码" + ] for plat in ("tm", "jd", "dy"): ids_key = "spus" if plat == "jd" else "item_ids" missing = list(identity_missing) if not self._cfg_has_sales_bitable(cfg): missing.append("销量目标地址") if not cfg.get(plat, {}).get(ids_key): - missing.append(f"{plat}商品编码") + missing.append(self._platform_missing_label(cfg, plat)) if missing: result[plat]["skipped"].append( {"style": name, "reason": "缺 " + "/".join(dict.fromkeys(missing))} @@ -761,11 +820,25 @@ class StyleConfigLoader: result[plat]["collectable"].append(name) return result + def _platform_missing_label(self, cfg: dict, platform: str) -> str: + """平台商品编码缺失的提示文案。 + + 同系列其它款已填编码时，本款不采平台流量是预期行为（平台数据是 + 商品/系列级，重复采集会把同一份访客数写进多个款），因此文案指向 + 承担款而不是报了「缺配置」。 + """ + + owner = (cfg.get("series") or {}).get("platform_owner", {}).get(platform, "") + if owner and owner != cfg.get("name"): + return f"{platform}商品编码（同系列「{owner}」已在采，本款只采 ERP 销量）" + return f"{platform}商品编码" + def get_erp_daily_styles(self) -> dict: - """ERP daily 门：ERP 编码、至少一个平台商品编码和销量目标都要有。 + """ERP daily 门：只要求 ERP 款式编码（有销量表写入需求时再校验表地址）。 返回 {collectable, skipped, brand_map: {style: brand}}。 - ERP 一次跑覆盖 jd/dy/tm 三个平台，平台商品编码只用于确认款式已完成商品身份配置。 + ERP 销量按款式编码逐款查，平台商品编码与销量表都是系列级配置， + 由同系列主款承担，子款不需要重复填写。 brand_map 给 collect_erp_yesterday_metrics.py 按品牌选店铺 (光影行星旗下店铺 vs ozko旗舰店)。 """ @@ -774,7 +847,9 @@ class StyleConfigLoader: skipped: list[dict] = [] brand_map: dict[str, str] = {} for name, cfg in payload["styles"].items(): - missing = self._identity_missing(cfg) + missing: list[str] = [] + if not cfg.get("erp_codes"): + missing.append("ERP款式编码") if not self._cfg_has_sales_bitable(cfg): missing.append("销量目标地址") if not missing: diff --git a/src/gyxx_flow/modules/product_commerce/sync_monthly_sales_sheet.py b/src/gyxx_flow/modules/product_commerce/sync_monthly_sales_sheet.py index c794fb7..13f5cfa 100644 --- a/src/gyxx_flow/modules/product_commerce/sync_monthly_sales_sheet.py +++ b/src/gyxx_flow/modules/product_commerce/sync_monthly_sales_sheet.py @@ -23,10 +23,7 @@ try: collect_monthly_erp_metrics, monthly_report_interval, ) - from .config.style_config_loader import ( - StyleConfigLoader, - has_complete_style_identity, - ) + from .config.style_config_loader import StyleConfigLoader from .lark_cli_runtime import run_lark_cli except ImportError: # Direct script execution from the module root. from collect_erp_monthly_metrics import ( @@ -35,10 +32,7 @@ except ImportError: # Direct script execution from the module root. collect_monthly_erp_metrics, monthly_report_interval, ) - from config.style_config_loader import ( - StyleConfigLoader, - has_complete_style_identity, - ) + from config.style_config_loader import StyleConfigLoader from lark_cli_runtime import run_lark_cli _MARKDOWN_LINK = re.compile(r"^\s*\[([^]]+)]\([^)]*\)\s*$") @@ -330,7 +324,7 @@ def plan_erp_styles( str(code).strip() for code in raw_codes if str(code).strip() ) ) - if not codes or not has_complete_style_identity(style_config): + if not codes: missing.append(target.product_name) continue key = normalized_name(style_name)