From 2f31d5872fa3d63125ac38d3fe3124408846cad8 Mon Sep 17 00:00:00 2001 From: wangyunlong <3080433063@qq.com> Date: Wed, 28 Oct 2026 14:51:44 +0800 Subject: [PATCH] =?UTF-8?q?chore(product):=20=E5=BA=9F=E5=BC=83=E5=B9=B6?= =?UTF-8?q?=E5=88=A0=E9=99=A4=20db/sync=5Fsku=5Fmaster.py?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 与上一并废弃的 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 | 8 - config/source-manifests/product_commerce.json | 8 - docs/dynamic-workflow-config.md | 6 +- .../product_commerce/db/sync_sku_master.py | 170 ------------------ tests/test_product_source_migration.py | 3 +- tests/test_runtime_integrations.py | 4 +- 6 files changed, 7 insertions(+), 192 deletions(-) delete mode 100644 src/gyxx_flow/modules/product_commerce/db/sync_sku_master.py diff --git a/config/runtime-bindings.json b/config/runtime-bindings.json index 883ee25..b6b33e0 100644 --- a/config/runtime-bindings.json +++ b/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", diff --git a/config/source-manifests/product_commerce.json b/config/source-manifests/product_commerce.json index d0eeebb..0f5c0e8 100644 --- a/config/source-manifests/product_commerce.json +++ b/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", diff --git a/docs/dynamic-workflow-config.md b/docs/dynamic-workflow-config.md index cc0aee1..a4e75b0 100644 --- a/docs/dynamic-workflow-config.md +++ b/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` 和 -`db/sync_sku_master.py` 也已改为读取同一项目数据库，避免从非定时入口绕回旧主表。 +共 12 个定时工作流依赖这套动态配置。维护脚本 `feishu_comment_batch.py` 也已改为读取同一项目数据库， +避免从非定时入口绕回旧主表。原先承担「动态配置 → 款式身份」投影的 `db/sync_dim_style.py` 与 +`db/sync_sku_master.py` 均已废弃：运行时身份已直读 `dim_style`，这两个脚本跑一次等于把自己写回自己， +款式新增与改款统一走 data-hub「产品生命进程」建档页。 `product.alert.daily`、商品导入、万相台广告、视频发布和市场排行使用各自数据库或专用配置， 不依赖旧配置主表，因此不在本次切换范围内。 diff --git a/src/gyxx_flow/modules/product_commerce/db/sync_sku_master.py b/src/gyxx_flow/modules/product_commerce/db/sync_sku_master.py deleted file mode 100644 index 5b67886..0000000 --- a/src/gyxx_flow/modules/product_commerce/db/sync_sku_master.py +++ /dev/null @@ -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()) diff --git a/tests/test_product_source_migration.py b/tests/test_product_source_migration.py index 1538ac2..b2b23fc 100644 --- a/tests/test_product_source_migration.py +++ b/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) diff --git a/tests/test_runtime_integrations.py b/tests/test_runtime_integrations.py index 27f5a6d..32d4c76 100644 --- a/tests/test_runtime_integrations.py +++ b/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]