feat(product): 取款主源切到 dim_style,平台编码与销量表按系列共享

产品生命进程建档(dim_style)成为款式身份唯一权威,采集端不再依赖
workflow_dynamic_styles + cmt_styles 双份维护:

- workflow_config 运行时改读 dim_style(ERP 编码、四平台商品 ID、销量表、
  系列归属),cmt_styles 降级为 dim_style 不存在时的兜底;source_record_id
  前缀由 cmt_style 改为 lifecycle_style
- merge_lifecycle_style_records 把 series_name / is_series_anchor /
  sales_bitable_url 写进运行时记录,建档地址优先于动态配置
- StyleConfigLoader 新增 _apply_series_sharing:按系列标注各平台编码的
  承担款,子款继承主款销量表,但不复制平台商品编码,避免同一份访客/加购
  被同系列多款重复入账
- ERP 三道门(ERP 昨日采集、全店铺日报、月度销量表)只校验款式编码;
  平台编码缺失时提示「同系列某款已在采,本款只采 ERP 销量」
- CACHE_SCHEMA_VERSION 5 -> 6,旧缓存自动失效

系(系列)级配置结果:ERP 门可采款数 33 -> 53,全店铺日报取款计划 42 -> 198。
This commit is contained in:
wangyunlong committed 2026-10-28 14:20:59 +08:00
1 parent 3cf7048140
commit ee6fd4170e
4 files changed
+133 -23

No files matched your search

+45 -4
View File
@@ -33,6 +33,25 @@ DEFAULT_SEED_PATH = (
/ "workflow-dynamic-config-seed.json" / "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 = ( CONFIG_FIELDS = (
"platform", "platform",
"erp_codes", "erp_codes",
@@ -298,8 +317,9 @@ def merge_lifecycle_style_records(
"""Overlay lifecycle product identity on legacy target configuration rows. """Overlay lifecycle product identity on legacy target configuration rows.
``workflow_dynamic_*`` still owns Feishu target URLs and persona targets, while ``workflow_dynamic_*`` still owns Feishu target URLs and persona targets, while
``cmt_styles`` owns ERP codes and product IDs. The result intentionally keeps ``dim_style`` (产品生命进程建档) owns ERP codes, product IDs and the sales sheet.
the old flat runtime contract so existing collectors need no per-workflow fork. 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]]] = {} 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_by_style: dict[str, Mapping[str, Any]] = {}
lifecycle_order: list[str] = [] lifecycle_order: list[str] = []
for row in lifecycle_rows: 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: if not style_name or style_name in lifecycle_by_style:
continue continue
lifecycle_by_style[style_name] = row 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, []): if item_id not in dynamic_unknown_ids.setdefault(label, []):
dynamic_unknown_ids[label].append(item_id) 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: if managed:
labels = [] labels = []
for label in LIFECYCLE_PLATFORM_LABELS.values(): 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, []))) else list(dynamic_ids.get(key, dynamic_unknown_ids.get(label, [])))
) )
output["source_record_id"] = ( 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) records.append(output)
@@ -1158,6 +1191,14 @@ class PostgresWorkflowConfigStore:
"WHERE s.enabled ORDER BY s.style_name, p.id" "WHERE s.enabled ORDER BY s.style_name, p.id"
) )
rows = cursor.fetchall() rows = cursor.fetchall()
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:
cursor.execute("SELECT to_regclass('public.cmt_styles') AS table_name") cursor.execute("SELECT to_regclass('public.cmt_styles') AS table_name")
cmt_table = cursor.fetchone() or {} cmt_table = cursor.fetchone() or {}
if cmt_table.get("table_name"): if cmt_table.get("table_name"):
@@ -18,7 +18,7 @@ from scrapling.fetchers import DynamicFetcher
try: try:
from . import collect_erp_yesterday_metrics as erp from . import collect_erp_yesterday_metrics as erp
from .collect_retry_utils import retry_step 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 ( from .db import (
get_conn, get_conn,
init_schema, init_schema,
@@ -35,7 +35,7 @@ try:
except ImportError: # Direct script execution from the module root. except ImportError: # Direct script execution from the module root.
import collect_erp_yesterday_metrics as erp import collect_erp_yesterday_metrics as erp
from collect_retry_utils import retry_step 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 db import get_conn, init_schema, upsert_erp_all_shop_style_daily_metrics
from erp_login_product_analysis import ( from erp_login_product_analysis import (
DEFAULT_CONFIG, DEFAULT_CONFIG,
@@ -88,7 +88,7 @@ def build_style_plans(payload: Mapping[str, Any]) -> tuple[StylePlan, ...]:
if str(code).strip() 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 continue
for code in codes: for code in codes:
previous = code_owners.setdefault(code, style_name) previous = code_owners.setdefault(code, style_name)
@@ -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 from runtime_paths import NORMALIZED_DATA_ROOT, STATE_ROOT, managed_data_path
CACHE_SCHEMA_VERSION = 5 CACHE_SCHEMA_VERSION = 6
# 表格字段名(按实际表格设置,支持别名) # 表格字段名(按实际表格设置,支持别名)
FIELD_ALIASES = { FIELD_ALIASES = {
@@ -266,6 +266,8 @@ class StyleConfigLoader:
"style_analysis_bitable_url", "" "style_analysis_bitable_url", ""
), ),
"destinations": row.get("destinations", []), "destinations": row.get("destinations", []),
"series_name": row.get("series_name", ""),
"is_series_anchor": bool(row.get("is_series_anchor", False)),
} }
) )
return records return records
@@ -315,6 +317,9 @@ class StyleConfigLoader:
"jd_persona_bitable": {}, "jd_persona_bitable": {},
"style_analysis_bitable": {}, "style_analysis_bitable": {},
"destinations": [], "destinations": [],
"series_name": "",
"is_series_anchor": False,
"series": {},
} }
# ── 第一遍:款式级字段 ────────────────────────────────────── # ── 第一遍:款式级字段 ──────────────────────────────────────
@@ -332,6 +337,13 @@ class StyleConfigLoader:
if brand and not style["brand"]: if brand and not style["brand"]:
style["brand"] = 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") destinations = record.get("destinations")
if isinstance(destinations, list) and destinations and not style["destinations"]: if isinstance(destinations, list) and destinations and not style["destinations"]:
style["destinations"] = [ style["destinations"] = [
@@ -427,6 +439,9 @@ class StyleConfigLoader:
elif platform == "jd_self" and ids: elif platform == "jd_self" and ids:
style["jd_self"]["skus"] = ids style["jd_self"]["skus"] = ids
# ── 第三遍: 系列内共享(平台编码 / 销量表只要求主款填写)──────────
self._apply_series_sharing(styles)
return { return {
"styles": styles, "styles": styles,
"source": "database", "source": "database",
@@ -434,6 +449,46 @@ class StyleConfigLoader:
"schema_version": CACHE_SCHEMA_VERSION, "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]: def get_erp_styles_input(self) -> list[dict]:
"""ERP 编码清单:只要求款式编码,平台编码/销量表由系列主款承担。"""
return [ return [
{"style_name": name, "erp_style_codes": cfg["erp_codes"]} {"style_name": name, "erp_style_codes": cfg["erp_codes"]}
for name, cfg in self.load()["styles"].items() 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: def get_bitable_style_map(self) -> dict:
@@ -745,14 +801,17 @@ class StyleConfigLoader:
payload = self.load() payload = self.load()
result = {p: {"collectable": [], "skipped": []} for p in ("tm", "jd", "dy")} result = {p: {"collectable": [], "skipped": []} for p in ("tm", "jd", "dy")}
for name, cfg in payload["styles"].items(): 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"): for plat in ("tm", "jd", "dy"):
ids_key = "spus" if plat == "jd" else "item_ids" ids_key = "spus" if plat == "jd" else "item_ids"
missing = list(identity_missing) missing = list(identity_missing)
if not self._cfg_has_sales_bitable(cfg): if not self._cfg_has_sales_bitable(cfg):
missing.append("销量目标地址") missing.append("销量目标地址")
if not cfg.get(plat, {}).get(ids_key): if not cfg.get(plat, {}).get(ids_key):
missing.append(f"{plat}商品编码") missing.append(self._platform_missing_label(cfg, plat))
if missing: if missing:
result[plat]["skipped"].append( result[plat]["skipped"].append(
{"style": name, "reason": "缺 " + "/".join(dict.fromkeys(missing))} {"style": name, "reason": "缺 " + "/".join(dict.fromkeys(missing))}
@@ -761,11 +820,25 @@ class StyleConfigLoader:
result[plat]["collectable"].append(name) result[plat]["collectable"].append(name)
return result 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: def get_erp_daily_styles(self) -> dict:
"""ERP daily 门:ERP 编码、至少一个平台商品编码和销量目标都要有。 """ERP daily 门:只要求 ERP 款式编码(有销量表写入需求时再校验表地址)。
返回 {collectable, skipped, brand_map: {style: brand}}。 返回 {collectable, skipped, brand_map: {style: brand}}。
ERP 一次跑覆盖 jd/dy/tm 三个平台,平台商品编码只用于确认款式已完成商品身份配置。 ERP 销量按款式编码逐款查,平台商品编码与销量表都是系列级配置,
由同系列主款承担,子款不需要重复填写。
brand_map 给 collect_erp_yesterday_metrics.py 按品牌选店铺 brand_map 给 collect_erp_yesterday_metrics.py 按品牌选店铺
(光影行星旗下店铺 vs ozko旗舰店)。 (光影行星旗下店铺 vs ozko旗舰店)。
""" """
@@ -774,7 +847,9 @@ class StyleConfigLoader:
skipped: list[dict] = [] skipped: list[dict] = []
brand_map: dict[str, str] = {} brand_map: dict[str, str] = {}
for name, cfg in payload["styles"].items(): 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): if not self._cfg_has_sales_bitable(cfg):
missing.append("销量目标地址") missing.append("销量目标地址")
if not missing: if not missing:
@@ -23,10 +23,7 @@ try:
collect_monthly_erp_metrics, collect_monthly_erp_metrics,
monthly_report_interval, monthly_report_interval,
) )
from .config.style_config_loader import ( from .config.style_config_loader import StyleConfigLoader
StyleConfigLoader,
has_complete_style_identity,
)
from .lark_cli_runtime import run_lark_cli from .lark_cli_runtime import run_lark_cli
except ImportError: # Direct script execution from the module root. except ImportError: # Direct script execution from the module root.
from collect_erp_monthly_metrics import ( from collect_erp_monthly_metrics import (
@@ -35,10 +32,7 @@ except ImportError: # Direct script execution from the module root.
collect_monthly_erp_metrics, collect_monthly_erp_metrics,
monthly_report_interval, monthly_report_interval,
) )
from config.style_config_loader import ( from config.style_config_loader import StyleConfigLoader
StyleConfigLoader,
has_complete_style_identity,
)
from lark_cli_runtime import run_lark_cli from lark_cli_runtime import run_lark_cli
_MARKDOWN_LINK = re.compile(r"^\s*\[([^]]+)]\([^)]*\)\s*$") _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() 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) missing.append(target.product_name)
continue continue
key = normalized_name(style_name) key = normalized_name(style_name)