chore(product): 废弃并删除 db/sync_dim_style.py
采集端运行时身份已改为直接读 dim_style，这个脚本的投影方向变成"自己写自己"： 既补不了旧 cmt_styles 台账数据，upsert 又是 EXCLUDED 全量覆盖，跑一次有可能把 空值写回 dim_style。款式新增与改款统一走 data-hub「产品生命进程」建档页。 - 删除 src/gyxx_flow/modules/product_commerce/db/sync_dim_style.py（390 行） - 移除 config/runtime-bindings.json 绑定（cdp_port 22070）与 source-manifests 条目 - 测试计数同步：PRODUCTION_PYTHON 65 -> 64，集成绑定 158 -> 157 - docs/dynamic-workflow-config.md 补废弃说明；运行时路径图与表说明由 cmt_styles 更正为 dim_style - db/__init__.py 表来源说明同步 验证：tests 根级套件 11 failed / 820 passed，与改动前完全一致（失败均为清单 sha256 与既有环境问题，非本次引入）。
This commit is contained in:
wangyunlong committed
1 parent
ee6fd4170e
commit
423f8d1f75
7 files changed
+22
-421
No files matched your search
@@ -1119,14 +1119,6 @@
|
||||
],
|
||||
"cdp_port": 22163
|
||||
},
|
||||
"product_commerce:db/sync_dim_style.py": {
|
||||
"script_id": "product_commerce:db/sync_dim_style.py",
|
||||
"state_key": "product_commerce:db/sync_dim_style.py",
|
||||
"aliases": [
|
||||
"product_commerce:db/sync_dim_style.py"
|
||||
],
|
||||
"cdp_port": 22070
|
||||
},
|
||||
"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",
|
||||
|
||||
@@ -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": "fc19a57bcbc023ab20156820a708d64c5526fad33ef5eb5fdea56c0bb04b84e3",
|
||||
"target_sha256": "2cbd79e4d984d556bac35a0aeb5b8f803309fd3fac2399da459cd61a3aa2484c",
|
||||
"transformed": true,
|
||||
"source_relative_path": "db/sync_dim_style.py",
|
||||
"target_relative_path": "src/gyxx_flow/modules/product_commerce/db/sync_dim_style.py"
|
||||
},
|
||||
{
|
||||
"category": "production_source",
|
||||
"source_sha256": "2dc00b88fb5f28c2db619074e97c4a7c446aec29926af9a6890065409c8fac0e",
|
||||
|
||||
@@ -56,14 +56,15 @@ PostgreSQL 的动态配置表。生命进程负责款式名、品牌、ERP 款
|
||||
|
||||
数据库使用父子表：
|
||||
|
||||
- `cmt_styles`：产品生命进程主表，一个款式一行，保存 ERP 编码、四个平台商品编码和 NFC 数量。
|
||||
- `dim_style`：产品生命进程主表，一个款式一行，保存 ERP 款式编码、四个平台商品编码、系列归属和销量飞书表；采集端运行时身份的唯一权威来源。
|
||||
- `cmt_styles`：笔记分析用的款式挂载表（`cmt_notes_master.style_id` 引用其 id），已退出 product 采集身份，仅在 `dim_style` 不可用时作兜底。
|
||||
- `workflow_dynamic_styles`：一个款式一行，保存共享飞书目标及内容工作流参数。
|
||||
- `workflow_dynamic_style_platforms`：一个款式可有多个平台子行，保留平台画像表、启停状态和旧商品 ID 兼容快照。
|
||||
- 旧 `workflow_dynamic_configs` 保留为原始迁移审计，不再参与运行时读取。
|
||||
- `erp_codes`、`item_ids` 使用 PostgreSQL 数组，页面接受中英文逗号或换行并自动去重。
|
||||
- `destinations` 使用通用 JSONB 目标列表，每项包含 `key`、`label`、`url`、`description`、`enabled`；六个旧目标字段仍同步保留，保证已有工作流兼容。后续新增分析逻辑只需增加一个稳定的 `key` 和对应飞书表地址。
|
||||
- 款式带递增 `revision`；整页保存和删除必须提交 `If-Match`，避免并发覆盖。
|
||||
- `cmt_styles.platform_product_ids` 的四个键固定为 `tm`（天猫）、`jd`（京东旗舰店）、`jd_self`（京东自营）、`dy`（抖音）；空数组表示该平台尚未补齐。
|
||||
- `dim_style` 的四个平台列固定为 `tm_spus`（天猫）、`jd_spus`（京东旗舰店）、`jd_self_skus`（京东自营）、`dy_spus`（抖音）；空数组表示该平台尚未补齐。同系列只需主款填写，子款靠系列归属继承。
|
||||
|
||||
## 首次迁移
|
||||
|
||||
@@ -98,31 +99,36 @@ uv run --env-file 'D:/product-collector-analyze-flow/.env' `
|
||||
uv run --env-file 'D:/product-collector-analyze-flow/.env' `
|
||||
--project 'D:/gyxx-flow' `
|
||||
python -m gyxx_flow.migration.sync_feishu_style_identity --apply
|
||||
|
||||
uv run --env-file 'D:/product-collector-analyze-flow/.env' `
|
||||
--project 'D:/gyxx-flow' `
|
||||
python 'src/gyxx_flow/modules/product_commerce/db/sync_dim_style.py'
|
||||
```
|
||||
|
||||
脚本按款式合并旧表的多平台行，只写入 `cmt_styles` 的品牌、ERP 编码和四个平台商品编码，
|
||||
不会覆盖 NFC、图片、类目或市场标签；空源字段不会清空生命进程已有身份。拼多多等当前四键
|
||||
不会覆盖 NFC、图片、类目或市场标签；空源字段不会清空已有身份。拼多多等当前四键
|
||||
模型之外的平台只保留在动态平台配置/审计中，不会伪装成天猫、京东或抖音编码。默认是预览，
|
||||
必须显式传 `--apply` 才会写数据库。
|
||||
|
||||
> **`db/sync_dim_style.py` 已废弃并删除（2026-10-28）。**
|
||||
> 采集端运行时身份改为直接读 `dim_style`，该脚本再跑就是把 `dim_style` 投影回它自己，
|
||||
> 既补不了旧台账数据，也可能把空值写回。款式新增与改款统一走 data-hub「产品生命进程」
|
||||
> 建档页，下一轮调度自动生效，不需要再手工同步。
|
||||
|
||||
服务器必须注入 `GYXX_POSTGRES_DSN`，或完整的 `PG_HOST/PG_PORT/PG_DB/PG_USER/PG_PASSWORD`。
|
||||
没有数据库且没有数据库生成的运行缓存时，相关工作流会明确失败，不会静默回读旧飞书主表。
|
||||
|
||||
## 运行时路径
|
||||
|
||||
```text
|
||||
产品生命进程 cmt_styles (商品身份)
|
||||
产品生命进程 dim_style (商品身份：ERP 编码 / 四平台编码 / 系列归属 / 销量表)
|
||||
+
|
||||
workflow_dynamic_styles + workflow_dynamic_style_platforms (目标/画像/启停)
|
||||
↓
|
||||
统一运行时聚合
|
||||
统一运行时聚合（dim_style 优先，cmt_styles 仅在 dim_style 缺失时兜底）
|
||||
├─ 完整身份门 → StyleConfigLoader → 商品经营工作流
|
||||
└─ 目标兼容输出 → feishu_mapping → 内容营销工作流 → 各业务飞书目标表
|
||||
```
|
||||
|
||||
系列共享规则：ERP 款式编码逐款必填；平台商品编码与销量飞书表整系列只需主款填写，
|
||||
子款通过系列归属继承销量表，但**不复制平台编码**——同一个商品 ID 被多款重复采集会把
|
||||
访客/加购重复入账。
|
||||
|
||||
`StyleConfigLoader` 只在数据库短暂不可用时使用最近一次数据库成功读取后生成的本地缓存；
|
||||
`feishu_mapping` 的目标表字段缓存仍保留，用来减少对各业务飞书表的字段查询。
|
||||
@@ -27,8 +27,10 @@
|
||||
写: check_nine_day_decline.py (去重)
|
||||
读: check_nine_day_decline.py._already_notified
|
||||
dim_style
|
||||
写: db/sync_dim_style.py (人工触发，投影产品生命进程身份)
|
||||
读: bitable 同步 / 业务查询
|
||||
写: data-hub 建档页 (产品生命进程，采集端身份的唯一权威来源)
|
||||
读: 采集端运行时取款 / bitable 同步 / 业务查询
|
||||
注: 原 db/sync_dim_style.py 已废弃并删除（改造后它读的就是 dim_style，
|
||||
投影等于自己写自己）
|
||||
fact_platform_xlsx
|
||||
写: 暂无 caller (db/__init__.py.upsert_platform_xlsx_rows 已写好，import 脚本待补)
|
||||
读: 平台 XLSX 明细查询
|
||||
|
||||
@@ -1,390 +0,0 @@
|
||||
"""同步 dim_style：4 个数据源自动 upsert。
|
||||
|
||||
数据源:
|
||||
产品生命进程 + 动态配置运行时聚合 -> 商品编码、ERP 编码和飞书目标地址
|
||||
|
||||
历史文件、采集产物和硬编码仅保留在本文件的兼容函数中，不再作为同步入口，
|
||||
避免人工修改生命进程后被旧脚本反向覆盖。
|
||||
|
||||
SKU 主表的飞书地址 (bitable_sku_master_*) 由人工维护，不在自动同步范围。
|
||||
|
||||
用法:
|
||||
.venv/Scripts/python.exe db/sync_dim_style.py # 全量
|
||||
.venv/Scripts/python.exe db/sync_dim_style.py --dry-run # 扫描不入库
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import re
|
||||
import sys
|
||||
from datetime import datetime
|
||||
from pathlib import Path
|
||||
from typing import Any, Iterable
|
||||
|
||||
PROJECT_ROOT = Path(__file__).resolve().parent.parent
|
||||
sys.path.insert(0, str(PROJECT_ROOT))
|
||||
sys.path.insert(0, str(PROJECT_ROOT.parents[2]))
|
||||
from runtime_paths import RAW_DATA_ROOT, runtime_script
|
||||
from gyxx_flow.adapters.workflow_config import runtime_workflow_configs # noqa: E402
|
||||
|
||||
from db import get_conn # noqa: E402
|
||||
|
||||
|
||||
PLATFORM_DIRS = {
|
||||
"jd": "jd单品数据",
|
||||
"dy": "dy单品数据",
|
||||
"tm": "tm单品数据",
|
||||
}
|
||||
|
||||
|
||||
# ── 数据源 1: bitable_style_map.json ────────────────────────────────────────
|
||||
|
||||
def load_bitable_weekly() -> dict[str, dict[str, str]]:
|
||||
"""从 bitable_style_map.json 抓每个款式的飞书周报表 base/table。"""
|
||||
path = PROJECT_ROOT / "bitable_style_map.json"
|
||||
if not path.exists():
|
||||
return {}
|
||||
raw = json.loads(path.read_text(encoding="utf-8-sig"))
|
||||
out: dict[str, dict[str, str]] = {}
|
||||
for style_name, info in raw.items():
|
||||
if not isinstance(info, dict) or not info.get("enabled", True):
|
||||
continue
|
||||
base = info.get("base_token", "")
|
||||
table = info.get("table_id", "")
|
||||
if not base or not table:
|
||||
continue
|
||||
url = f"https://feishu.cn/base/{base}?table={table}"
|
||||
out[style_name] = {
|
||||
"bitable_weekly_base": base,
|
||||
"bitable_weekly_table": table,
|
||||
"bitable_weekly_url": url,
|
||||
}
|
||||
return out
|
||||
|
||||
|
||||
# ── 数据源 2: jd_self SKU 集合 ──────────────────────────────────────────────
|
||||
|
||||
def load_jd_self_skus() -> dict[str, list[str]]:
|
||||
"""从 jd_self_inventory_sales_collector.STYLE_SKU_GROUPS 抓款→SKU 集合。
|
||||
|
||||
直接解析源文件里 STYLE_SKU_GROUPS = [...] 的字面量，避免 import 拉起 playwright。
|
||||
"""
|
||||
path = runtime_script("jd_self_inventory_sales_collector.py")
|
||||
text = path.read_text(encoding="utf-8")
|
||||
needle = "STYLE_SKU_GROUPS = "
|
||||
i = text.find(needle)
|
||||
if i < 0:
|
||||
return {}
|
||||
i += len(needle)
|
||||
# 平衡括号找匹配的 ]
|
||||
depth = 0
|
||||
in_str: str | None = None
|
||||
escape = False
|
||||
end = None
|
||||
for j in range(i, len(text)):
|
||||
ch = text[j]
|
||||
if in_str:
|
||||
if escape:
|
||||
escape = False
|
||||
elif ch == "\\":
|
||||
escape = True
|
||||
elif ch == in_str:
|
||||
in_str = None
|
||||
continue
|
||||
if ch in ("'", '"'):
|
||||
in_str = ch
|
||||
elif ch == "[":
|
||||
depth += 1
|
||||
elif ch == "]":
|
||||
depth -= 1
|
||||
if depth == 0:
|
||||
end = j + 1
|
||||
break
|
||||
if end is None:
|
||||
return {}
|
||||
chunk = text[i:end]
|
||||
try:
|
||||
# 用 ast.literal_eval 解析 Python 字面量（兼容末尾逗号 / 单引号）
|
||||
import ast
|
||||
groups = ast.literal_eval(chunk)
|
||||
except Exception as exc:
|
||||
print(f" [WARN] 解析 STYLE_SKU_GROUPS 失败: {exc}")
|
||||
return {}
|
||||
return {g["style_name"]: list(g.get("skus") or []) for g in groups if "style_name" in g}
|
||||
|
||||
|
||||
# ── 数据源 3: 各平台 SPU 集合 ───────────────────────────────────────────────
|
||||
|
||||
def _latest_style_json(plat_dir: Path, style_name: str) -> Path | None:
|
||||
"""取某款式在 plat_dir 下的最新日期 JSON。
|
||||
|
||||
目录名约定: <款名>_<YYYY-MM-DD> (jd/dy/tm 主流) 或 业务款式_<YYYY-MM-DD> (tm 部分)
|
||||
|
||||
优先级:
|
||||
1) <safe>_<date>.json (jd / dy / tm 主流)
|
||||
2) 业务款式_<date>.json (tm 习惯命名)
|
||||
3) data.json (兜底，但 data.json 通常是 ERP 抓数，可能没有 SPU 字段)
|
||||
"""
|
||||
safe = "".join("_" if ch in '\\/:*?"<>|' else ch for ch in style_name).strip(" ._")
|
||||
# 只匹配以 "<safe>_" 开头的目录，避免拿到别的款式的目录
|
||||
cands: list[tuple[str, Path]] = []
|
||||
for d in plat_dir.iterdir():
|
||||
if not d.is_dir() or not d.name.startswith(safe + "_"):
|
||||
continue
|
||||
date_part = d.name[len(safe) + 1:]
|
||||
try:
|
||||
datetime.strptime(date_part, "%Y-%m-%d")
|
||||
except ValueError:
|
||||
continue
|
||||
cands.append((date_part, d))
|
||||
if not cands:
|
||||
return None
|
||||
cands.sort(key=lambda x: x[0], reverse=True)
|
||||
latest_date, latest_dir = cands[0]
|
||||
|
||||
# 优先级 1: <safe>_<date>.json
|
||||
p1 = latest_dir / f"{safe}_{latest_date}.json"
|
||||
if p1.exists():
|
||||
return p1
|
||||
# 优先级 2: 业务款式_<date>.json (tm)
|
||||
p2 = latest_dir / f"业务款式_{latest_date}.json"
|
||||
if p2.exists():
|
||||
return p2
|
||||
# 优先级 3: data.json
|
||||
p3 = latest_dir / "data.json"
|
||||
if p3.exists():
|
||||
return p3
|
||||
return None
|
||||
|
||||
|
||||
def _extract_spus(payload: dict, platform: str) -> list[str]:
|
||||
"""从款式 JSON payload 提取 SPU/商品 ID 列表。
|
||||
|
||||
各平台字段名:
|
||||
jd: payload['匹配SPU'] 或 payload['spu列表']
|
||||
dy: payload['matched_product_ids'] 或 payload['product_ids']
|
||||
tm: payload['style']['匹配商品ID'] 或 payload['style']['SPU']
|
||||
"""
|
||||
if platform == "jd":
|
||||
return list(payload.get("匹配SPU") or payload.get("spu列表") or [])
|
||||
if platform == "dy":
|
||||
return list(payload.get("matched_product_ids") or payload.get("product_ids") or [])
|
||||
if platform == "tm":
|
||||
style = payload.get("style") or {}
|
||||
return list(style.get("匹配商品ID") or style.get("SPU") or [])
|
||||
return []
|
||||
|
||||
|
||||
def load_platform_spus() -> dict[str, dict[str, list[str]]]:
|
||||
"""返回 {style_name: {platform: [spu, ...]}}，跨日期取最近一次抓数。"""
|
||||
out: dict[str, dict[str, list[str]]] = {}
|
||||
# 先枚举所有款式名（从 weekly_style_metrics 拿权威款式列表）
|
||||
data_root = RAW_DATA_ROOT
|
||||
# 用 daily_style_metrics 拿所有款式名
|
||||
try:
|
||||
with get_conn() as conn:
|
||||
with conn.cursor() as cur:
|
||||
cur.execute("SELECT DISTINCT style_name FROM daily_style_metrics")
|
||||
style_names = [r[0] for r in cur.fetchall()]
|
||||
except Exception:
|
||||
style_names = []
|
||||
|
||||
for style_name in style_names:
|
||||
out.setdefault(style_name, {})
|
||||
for platform, dir_name in PLATFORM_DIRS.items():
|
||||
plat_dir = data_root / dir_name
|
||||
if not plat_dir.exists():
|
||||
continue
|
||||
json_path = _latest_style_json(plat_dir, style_name)
|
||||
if not json_path:
|
||||
continue
|
||||
try:
|
||||
payload = json.loads(json_path.read_text(encoding="utf-8-sig"))
|
||||
except Exception:
|
||||
continue
|
||||
spus = _extract_spus(payload, platform)
|
||||
if spus:
|
||||
out[style_name][f"{platform}_spus"] = spus
|
||||
return out
|
||||
|
||||
|
||||
# ── 数据源 4: ERP 编码（PG 聚合）──────────────────────────────────────────
|
||||
|
||||
def load_erp_codes() -> dict[str, list[str]]:
|
||||
"""从 daily_style_metrics 聚每个款式的 ERP 编码。"""
|
||||
out: dict[str, list[str]] = {}
|
||||
try:
|
||||
with get_conn() as conn:
|
||||
with conn.cursor() as cur:
|
||||
cur.execute("""
|
||||
SELECT style_name, ARRAY(
|
||||
SELECT DISTINCT unnest(erp_style_codes)
|
||||
FROM daily_style_metrics s
|
||||
WHERE s.style_name = dm.style_name
|
||||
AND s.erp_style_codes IS NOT NULL
|
||||
) AS codes
|
||||
FROM (SELECT DISTINCT style_name FROM daily_style_metrics) dm
|
||||
""")
|
||||
for style_name, codes in cur.fetchall():
|
||||
out[style_name] = [c for c in (codes or []) if c]
|
||||
except Exception as exc:
|
||||
print(f" [WARN] 读 daily_style_metrics 失败: {exc}")
|
||||
return out
|
||||
|
||||
|
||||
# ── 合并 + upsert ──────────────────────────────────────────────────────────
|
||||
|
||||
def _merge_records(
|
||||
style_names: Iterable[str],
|
||||
weekly: dict[str, dict],
|
||||
self_skus: dict[str, list[str]],
|
||||
spus_map: dict[str, dict],
|
||||
erp_map: dict[str, list[str]],
|
||||
) -> list[dict]:
|
||||
rows: list[dict] = []
|
||||
for name in style_names:
|
||||
rows.append({
|
||||
"style_name": name,
|
||||
"jd_spus": spus_map.get(name, {}).get("jd_spus", []) or [],
|
||||
"dy_spus": spus_map.get(name, {}).get("dy_spus", []) or [],
|
||||
"tm_spus": spus_map.get(name, {}).get("tm_spus", []) or [],
|
||||
"jd_self_skus": self_skus.get(name, []) or [],
|
||||
"erp_codes": erp_map.get(name, []) or [],
|
||||
"bitable_weekly_url": weekly.get(name, {}).get("bitable_weekly_url", ""),
|
||||
"bitable_weekly_base": weekly.get(name, {}).get("bitable_weekly_base", ""),
|
||||
"bitable_weekly_table": weekly.get(name, {}).get("bitable_weekly_table", ""),
|
||||
# bitable_sku_master_* 由人工维护，不动
|
||||
})
|
||||
return rows
|
||||
|
||||
|
||||
def load_lifecycle_records() -> list[dict[str, Any]]:
|
||||
"""读取生命进程身份与动态目标的统一运行时视图。"""
|
||||
|
||||
return runtime_workflow_configs()
|
||||
|
||||
|
||||
def _lifecycle_to_dim_rows(records: list[dict[str, Any]]) -> list[dict[str, Any]]:
|
||||
grouped: dict[str, dict[str, Any]] = {}
|
||||
platform_fields = {
|
||||
"天猫": "tm_spus",
|
||||
"京东旗舰店": "jd_spus",
|
||||
"京东": "jd_spus",
|
||||
"京东自营": "jd_self_skus",
|
||||
"抖音": "dy_spus",
|
||||
}
|
||||
for record in records:
|
||||
name = str(record.get("style_name") or "").strip()
|
||||
if not name:
|
||||
continue
|
||||
row = grouped.setdefault(
|
||||
name,
|
||||
{
|
||||
"style_name": name,
|
||||
"jd_spus": [],
|
||||
"dy_spus": [],
|
||||
"tm_spus": [],
|
||||
"jd_self_skus": [],
|
||||
"erp_codes": [],
|
||||
"bitable_weekly_url": "",
|
||||
"bitable_weekly_base": "",
|
||||
"bitable_weekly_table": "",
|
||||
},
|
||||
)
|
||||
for code in record.get("erp_codes") or []:
|
||||
code = str(code).strip()
|
||||
if code and code not in row["erp_codes"]:
|
||||
row["erp_codes"].append(code)
|
||||
target_url = str(record.get("sales_bitable_url") or "").strip()
|
||||
if target_url and not row["bitable_weekly_url"]:
|
||||
row["bitable_weekly_url"] = target_url
|
||||
match = re.search(
|
||||
r"/base/([A-Za-z0-9]+)\?[^\s]*table=([A-Za-z0-9]+)",
|
||||
target_url,
|
||||
)
|
||||
if match:
|
||||
row["bitable_weekly_base"] = match.group(1)
|
||||
row["bitable_weekly_table"] = match.group(2)
|
||||
field = platform_fields.get(str(record.get("platform") or "").strip())
|
||||
if not field:
|
||||
continue
|
||||
for product_id in record.get("item_ids") or []:
|
||||
product_id = str(product_id).strip()
|
||||
if product_id and product_id not in row[field]:
|
||||
row[field].append(product_id)
|
||||
return list(grouped.values())
|
||||
|
||||
|
||||
def upsert_dim_style(conn, records: list[dict]) -> int:
|
||||
if not records:
|
||||
return 0
|
||||
def arr(v):
|
||||
# 空 list / None 都写空数组 {}（PG 数组维度不空），非空 list 原样写
|
||||
if v is None:
|
||||
return []
|
||||
return list(v)
|
||||
payload = [
|
||||
(
|
||||
r["style_name"],
|
||||
arr(r.get("jd_spus")),
|
||||
arr(r.get("dy_spus")),
|
||||
arr(r.get("tm_spus")),
|
||||
arr(r.get("jd_self_skus")),
|
||||
arr(r.get("erp_codes")),
|
||||
r.get("bitable_weekly_url") or None,
|
||||
r.get("bitable_weekly_base") or None,
|
||||
r.get("bitable_weekly_table") or None,
|
||||
)
|
||||
for r in records
|
||||
]
|
||||
sql = """
|
||||
INSERT INTO dim_style
|
||||
(style_name, jd_spus, dy_spus, tm_spus, jd_self_skus, erp_codes,
|
||||
bitable_weekly_url, bitable_weekly_base, bitable_weekly_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_weekly_url = EXCLUDED.bitable_weekly_url,
|
||||
bitable_weekly_base = EXCLUDED.bitable_weekly_base,
|
||||
bitable_weekly_table = EXCLUDED.bitable_weekly_table,
|
||||
-- bitable_sku_master_* 字段：保留人工已填值
|
||||
bitable_sku_master_url = COALESCE(dim_style.bitable_sku_master_url, EXCLUDED.bitable_sku_master_url),
|
||||
bitable_sku_master_base = COALESCE(dim_style.bitable_sku_master_base, EXCLUDED.bitable_sku_master_base),
|
||||
bitable_sku_master_table = COALESCE(dim_style.bitable_sku_master_table, EXCLUDED.bitable_sku_master_table),
|
||||
updated_at = NOW()
|
||||
"""
|
||||
from psycopg2.extras import execute_values
|
||||
with conn.cursor() as cur:
|
||||
execute_values(cur, sql, payload)
|
||||
return len(payload)
|
||||
|
||||
|
||||
def main():
|
||||
ap = argparse.ArgumentParser(description="同步 dim_style (款式 SPU/SKU + 飞书表地址)")
|
||||
ap.add_argument("--dry-run", action="store_true", help="扫描不入库")
|
||||
args = ap.parse_args()
|
||||
|
||||
runtime_records = load_lifecycle_records()
|
||||
records = _lifecycle_to_dim_rows(runtime_records)
|
||||
|
||||
print(f"[INFO] 数据源: product_lifecycle_runtime={len(runtime_records)} 条")
|
||||
print(f"[INFO] 合并款式: {len(records)}")
|
||||
if args.dry_run:
|
||||
for r in records[:5]:
|
||||
print(f" - {r['style_name']}: jd_spus={r['jd_spus']} jd_self_skus={r['jd_self_skus']} weekly_url={r['bitable_weekly_url'][:60]}")
|
||||
return 0
|
||||
|
||||
with get_conn() as conn:
|
||||
n = upsert_dim_style(conn, records)
|
||||
print(f"[OK] dim_style upsert: {n} 行")
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
@@ -41,7 +41,6 @@ PRODUCTION_PYTHON = {
|
||||
"config/__init__.py",
|
||||
"config/style_config_loader.py",
|
||||
"db/__init__.py",
|
||||
"db/sync_dim_style.py",
|
||||
"db/sync_sku_master.py",
|
||||
"dy_audience_profile_collect.py",
|
||||
"dy_product_scraping.py",
|
||||
@@ -197,7 +196,7 @@ def _load_runtime_paths():
|
||||
|
||||
|
||||
def test_product_runtime_contains_the_complete_reviewed_source_snapshot() -> None:
|
||||
assert len(PRODUCTION_PYTHON) == 65
|
||||
assert len(PRODUCTION_PYTHON) == 64
|
||||
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)
|
||||
|
||||
@@ -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) == 158
|
||||
assert len(first.command_ids) == 157
|
||||
bindings = [first.binding_for(command_id) for command_id in first.command_ids]
|
||||
assert len({binding.cdp_port for binding in bindings}) == 158
|
||||
assert len({binding.cdp_port for binding in bindings}) == 157
|
||||
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]
|
||||
|
||||
Reference in new issue
Block a user