chore(product): 废弃并删除 db/sync_sku_master.py

与上一并废弃的 sync_dim_style.py 同类问题，且更彻底：

- 身份列自投影：脚本读 runtime_workflow_configs()（现已直读 dim_style），
  再把 jd_spus/dy_spus/tm_spus/jd_self_skus/erp_codes 全量 EXCLUDED 覆盖写回
  dim_style。对「同系列只填主款平台编码」的子款，它会走动态配置兜底把旧商品 ID
  反写，绕过系列共享规则
- 唯一还写的 bitable_sku_master_url/base/table 是 sales_bitable_url 的副本：
  生产库 24 行与 bitable_weekly_* 完全相同、无一行独有，且采集端、data-hub
  后端、运维脚本均无任何读取点（data-hub 审计文档标注「flow 只写，全生态不读」）

删除后款式身份的唯一写入口是 data-hub「产品生命进程」建档页。
bitable_sku_master_* 三列保留在库中不删（存量值不动，需要时可低成本恢复）。

- 删除 src/gyxx_flow/modules/product_commerce/db/sync_sku_master.py（170 行）
- 移除 config/runtime-bindings.json 绑定（cdp_port 22071）与 source-manifests 条目
- 测试计数同步：PRODUCTION_PYTHON 64 -> 63，集成绑定 157 -> 156
- docs/dynamic-workflow-config.md 合并两个废弃脚本的说明

验证：tests 根级套件 11 failed / 820 passed / 1 skipped，与改动前一致。

config/runtime-bindings.json

@@ -1119,14 +1119,6 @@
       ],
       "cdp_port": 22163
     },
-    "product_commerce:db/sync_sku_master.py": {
-      "script_id": "product_commerce:db/sync_sku_master.py",
-      "state_key": "product_commerce:db/sync_sku_master.py",
-      "aliases": [
-        "product_commerce:db/sync_sku_master.py"
-      ],
-      "cdp_port": 22071
-    },
     "product_commerce:dy_audience_profile_collect.py": {
       "script_id": "product_commerce:dy_audience_profile_collect.py",
       "state_key": "product_commerce:dy_audience_profile_collect.py",

config/source-manifests/product_commerce.json

@@ -139,14 +139,6 @@
       "source_relative_path": "db/__init__.py",
       "target_relative_path": "src/gyxx_flow/modules/product_commerce/db/__init__.py"
     },
-    {
-      "category": "production_source",
-      "source_sha256": "2dc00b88fb5f28c2db619074e97c4a7c446aec29926af9a6890065409c8fac0e",
-      "target_sha256": "840f98d9fb79557016a59a2e3a1194a1e328a1574b4670f419c99ac85731d623",
-      "transformed": true,
-      "source_relative_path": "db/sync_sku_master.py",
-      "target_relative_path": "src/gyxx_flow/modules/product_commerce/db/sync_sku_master.py"
-    },
     {
       "category": "production_source",
       "source_sha256": "78c74c0e52e3651b3b042a2095e596ab250e60309c7ae2de905850822aad4922",

docs/dynamic-workflow-config.md

@@ -33,8 +33,10 @@ PostgreSQL 的动态配置表。生命进程负责款式名、品牌、ERP 款
 | `product.sales_sheet.daily` | 款式与 ERP 编码 | `sync_monthly_sales_sheet.py` |
 | `product.erp_all_shop_daily` | 全部款式与 ERP 编码 | `backfill_erp_all_shop_daily.py` |
 
-共 12 个定时工作流依赖这套动态配置。维护脚本 `feishu_comment_batch.py` 和
+共 12 个定时工作流依赖这套动态配置。维护脚本 `feishu_comment_batch.py` 也已改为读取同一项目数据库，
-`db/sync_sku_master.py` 也已改为读取同一项目数据库，避免从非定时入口绕回旧主表。
+避免从非定时入口绕回旧主表。原先承担「动态配置 → 款式身份」投影的 `db/sync_dim_style.py` 与
+`db/sync_sku_master.py` 均已废弃：运行时身份已直读 `dim_style`，这两个脚本跑一次等于把自己写回自己，
+款式新增与改款统一走 data-hub「产品生命进程」建档页。
 
 `product.alert.daily`、商品导入、万相台广告、视频发布和市场排行使用各自数据库或专用配置，
 不依赖旧配置主表，因此不在本次切换范围内。

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

@@ -1,170 +0,0 @@
-"""同步项目动态配置 → dim_style。
-
-从项目数据库读取配置，按款式聚合 4 个平台 ID 集合 + ERP 编码 + 飞书链接，
-upsert 到 dim_style 的 jd_spus / dy_spus / tm_spus / jd_self_skus / erp_codes / bitable_sku_master_* 字段。
-
-用法:
-    .venv/Scripts/python.exe db/sync_sku_master.py            # 全量拉 + 入库
-    .venv/Scripts/python.exe db/sync_sku_master.py --dry-run  # 拉数据 + 显示不入库
-
-"""
-
-from __future__ import annotations
-
-import argparse
-import re
-import sys
-from collections import defaultdict
-from pathlib import Path
-
-from gyxx_flow.adapters.workflow_config import runtime_workflow_configs
-
-PROJECT_ROOT = Path(__file__).resolve().parent.parent
-sys.path.insert(0, str(PROJECT_ROOT))
-
-from db import get_conn  # noqa: E402
-
-
-def fetch_sku_master_records() -> list[dict]:
-    """从项目数据库读取全部启用的动态配置。"""
-    return runtime_workflow_configs()
-
-
-def _split_ids(value) -> list[str]:
-    """兼容数据库数组和旧的逗号、换行分隔值。"""
-    if value is None:
-        return []
-    if isinstance(value, (list, tuple, set)):
-        return [str(item).strip() for item in value if str(item).strip()]
-    s = str(value)
-    # 先按换行 split（京东自营风格），再按逗号 split（其它平台风格）
-    parts: list[str] = []
-    for line in s.splitlines():
-        for piece in line.split(","):
-            piece = piece.strip()
-            if piece:
-                parts.append(piece)
-    return parts
-
-
-def aggregate(records: list[dict]) -> dict[str, dict]:
-    """按款式聚合: jd_spus / dy_spus / tm_spus / jd_self_skus / erp_codes / sku_master_link。"""
-    by_style: dict[str, dict] = defaultdict(lambda: {
-        "jd_spus": [], "dy_spus": [], "tm_spus": [], "jd_self_skus": [],
-        "erp_codes": [], "sku_master_url": "", "sku_master_base": "",
-        "sku_master_table": "",
-    })
-    for r in records:
-        plat_field = r.get("platform")
-        plat_list = plat_field if isinstance(plat_field, list) else ([plat_field] if plat_field else [])
-        plat_s = ",".join(str(p) for p in plat_list)
-        name = str(r.get("style_name") or "").strip()
-        if not name:
-            continue
-        ids_field = r.get("item_ids")
-        id_list = _split_ids(ids_field)
-        erp_field = r.get("erp_codes")
-        erp_list = _split_ids(erp_field)
-        link_field = str(r.get("sales_bitable_url") or "")
-
-        if "京东自营" in plat_s:
-            by_style[name]["jd_self_skus"].extend(id_list)
-        elif "京东" in plat_s:
-            by_style[name]["jd_spus"].extend(id_list)
-        if "抖音" in plat_s:
-            by_style[name]["dy_spus"].extend(id_list)
-        if "天猫" in plat_s:
-            by_style[name]["tm_spus"].extend(id_list)
-        for ec in erp_list:
-            if ec not in by_style[name]["erp_codes"]:
-                by_style[name]["erp_codes"].append(ec)
-
-        if link_field and not by_style[name]["sku_master_url"]:
-            # 解析 https://.../base/XXXXX?table=YYYYY[&view=ZZZZ]
-            m = re.search(
-                r"(https?://[^/]+/base/([A-Za-z0-9]+)\?[^\s]*table=([A-Za-z0-9]+)[^\s]*)",
-                link_field,
-            )
-            if m:
-                by_style[name]["sku_master_url"] = m.group(1)
-                by_style[name]["sku_master_base"] = m.group(2)
-                by_style[name]["sku_master_table"] = m.group(3)
-            else:
-                # fallback: 整段存
-                by_style[name]["sku_master_url"] = link_field
-    return by_style
-
-
-def upsert_from_sku_master(conn, by_style: dict[str, dict]) -> int:
-    """upsert SKU 主表数据到 dim_style (只更新 SPU/SKU/ERP/sku_master_* 字段，其它保留)。"""
-    if not by_style:
-        return 0
-    from psycopg2.extras import execute_values
-
-    def arr(v):
-        if v is None:
-            return []
-        return list(v)
-
-    payload = [
-        (
-            name,
-            arr(s.get("jd_spus")),
-            arr(s.get("dy_spus")),
-            arr(s.get("tm_spus")),
-            arr(s.get("jd_self_skus")),
-            arr(s.get("erp_codes")),
-            s.get("sku_master_url") or None,
-            s.get("sku_master_base") or None,
-            s.get("sku_master_table") or None,
-        )
-        for name, s in by_style.items()
-    ]
-    sql = """
-        INSERT INTO dim_style
-            (style_name, jd_spus, dy_spus, tm_spus, jd_self_skus, erp_codes,
-             bitable_sku_master_url, bitable_sku_master_base, bitable_sku_master_table)
-        VALUES %s
-        ON CONFLICT (style_name) DO UPDATE SET
-            jd_spus              = EXCLUDED.jd_spus,
-            dy_spus              = EXCLUDED.dy_spus,
-            tm_spus              = EXCLUDED.tm_spus,
-            jd_self_skus         = EXCLUDED.jd_self_skus,
-            erp_codes            = EXCLUDED.erp_codes,
-            bitable_sku_master_url   = EXCLUDED.bitable_sku_master_url,
-            bitable_sku_master_base  = EXCLUDED.bitable_sku_master_base,
-            bitable_sku_master_table = EXCLUDED.bitable_sku_master_table,
-            updated_at = NOW()
-    """
-    with conn.cursor() as cur:
-        execute_values(cur, sql, payload)
-    return len(payload)
-
-
-def main():
-    ap = argparse.ArgumentParser(description="同步项目动态配置 → dim_style")
-    ap.add_argument("--dry-run", action="store_true")
-    args = ap.parse_args()
-
-    print("[STEP] 读取项目动态配置 ...")
-    records = fetch_sku_master_records()
-    print(f"[INFO] 拉取 {len(records)} 条记录")
-    by_style = aggregate(records)
-    print(f"[INFO] 聚合 {len(by_style)} 个款式")
-
-    for name in sorted(by_style.keys())[:5]:
-        s = by_style[name]
-        print(f"  - {name}: jd={s['jd_spus']} dy={s['dy_spus']} tm={s['tm_spus']} jd_self={s['jd_self_skus']} erp={s['erp_codes']}")
-
-    if args.dry_run:
-        print("[DRY] 不入库")
-        return 0
-
-    with get_conn() as conn:
-        n = upsert_from_sku_master(conn, by_style)
-    print(f"[OK] dim_style upsert: {n} 行")
-    return 0
-
-
-if __name__ == "__main__":
-    sys.exit(main())

tests/test_product_source_migration.py

@@ -41,7 +41,6 @@ PRODUCTION_PYTHON = {
     "config/__init__.py",
     "config/style_config_loader.py",
     "db/__init__.py",
-    "db/sync_sku_master.py",
     "dy_audience_profile_collect.py",
     "dy_product_scraping.py",
     "erp_login_product_analysis.py",
@@ -196,7 +195,7 @@ def _load_runtime_paths():
 
 
 def test_product_runtime_contains_the_complete_reviewed_source_snapshot() -> None:
-    assert len(PRODUCTION_PYTHON) == 64
+    assert len(PRODUCTION_PYTHON) == 63
     assert all((MODULE_ROOT / relative).is_file() for relative in PRODUCTION_PYTHON)
     assert all((MODULE_ROOT / relative).is_file() for relative in RESOURCES)
     assert all((MIGRATED_TEST_ROOT / name).is_file() for name in LEGACY_TESTS)

tests/test_runtime_integrations.py

@@ -103,9 +103,9 @@ def test_project_registry_covers_every_script_with_unique_stable_port() -> None:
     )
 
     assert len(scripts.command_ids) == 61
-    assert len(first.command_ids) == 157
+    assert len(first.command_ids) == 156
     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}) == 156
     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]