diff --git a/REFACTOR_PLAN.md b/REFACTOR_PLAN.md index 0ee969c..883c6f2 100644 --- a/REFACTOR_PLAN.md +++ b/REFACTOR_PLAN.md @@ -63,7 +63,7 @@ python3 list_feishu_base_tables.py --action sync-brand - `feishu`：读取飞书 Base 下的表信息。 - `postgres`：执行 PostgreSQL 查询并输出结果。 -- `sync-brand`：按照 `BRAND_SYNC_TARGETS` 配置筛选品牌数据并同步到对应飞书表。 +- `sync-brand`：按照 `BRAND_SYNC_TARGETS` 中的品牌和仓库主键组合配置查询数据，并同步到对应飞书表。 - `all`：依次执行飞书读取、数据库查询和品牌同步。 ### 3.3 配置模块 @@ -102,7 +102,7 @@ FEISHU_APP_TOKEN 职责： -- 定义品牌到目标飞书表的同步配置。 +- 定义品牌、目标飞书表和仓库主键的组合同步配置。 - 定义主字段名称和来源字段。 - 定义 SQL 查询结果字段到飞书表字段的映射关系。 - 定义飞书字段类型和数字格式。 @@ -112,12 +112,18 @@ FEISHU_APP_TOKEN ```python BRAND_SYNC_TARGETS = [ - BrandSyncTarget(brand_name="羽心堂", table_name="羽心堂"), + BrandSyncTarget(brand_name="羽心堂", table_name="羽心堂", warehouse_keys=(2, 3, 7, 13)), ] PRIMARY_FIELD_NAME = "货品编码" PRIMARY_FIELD_SOURCE = "统一规格编码" ``` +`BrandSyncTarget` 中三个字段含义： + +- `brand_name`：SQL 中用于筛选 `pd."品牌名称"` 的品牌名称。 +- `table_name`：该品牌同步写入的飞书目标表名。 +- `warehouse_keys`：该品牌参与总库存汇总的仓库主键列表。 + 当前字段映射和展示顺序： | 飞书字段名 | SQL 来源字段 | 飞书字段类型 | 显示格式 | @@ -149,10 +155,11 @@ PRIMARY_FIELD_SOURCE = "统一规格编码" 当前 SQL 逻辑包括： - 取每个统一规格编码、仓库的最新库存快照。 -- 汇总统一规格编码维度的总库存。 -- 取每个统一规格编码、店铺的最新挂单量快照。 -- 按店铺和统一规格编码汇总本月、上月、上上月销售数量。 +- 按当前品牌配置的 `warehouse_keys` 筛选仓库，并汇总统一规格编码维度的总库存。 +- 取每个统一规格编码、店铺的最新可发库存快照，并排除已失效状态。 +- 按统一规格编码汇总本月、上月、上上月销售数量。 - 关联商品维表和店铺维表。 +- 按当前品牌配置的 `brand_name` 筛选品牌数据。 - 输出用于飞书同步的明细结果。 ### 3.6 PostgreSQL 模块 @@ -162,7 +169,7 @@ PRIMARY_FIELD_SOURCE = "统一规格编码" 职责： - 使用 `.env` 中的数据库配置连接 PostgreSQL。 -- 执行 `inventory_sync/sql.py` 中定义的 SQL。 +- 根据 `BrandSyncTarget` 中的品牌和仓库主键配置生成并执行 SQL。 - 将查询结果解析为结构化字典列表。 - 对外返回统一结构，供 CLI 或同步服务使用。 @@ -190,8 +197,8 @@ PRIMARY_FIELD_SOURCE = "统一规格编码" 职责： - 编排完整业务流程。 -- 根据配置查询 PostgreSQL。 -- 按 `BRAND_SYNC_TARGETS` 逐个筛选品牌数据。 +- 按 `BRAND_SYNC_TARGETS` 逐个读取品牌和仓库主键配置。 +- 根据每个品牌目标的配置查询 PostgreSQL。 - 根据 `specs.py` 中的映射表构造飞书记录。 - 确认目标飞书表字段存在且类型正确。 - 删除目标表已有记录。 @@ -200,7 +207,7 @@ PRIMARY_FIELD_SOURCE = "统一规格编码" 当前同步策略是全量覆盖： ```text -查询 PostgreSQL -> 遍历品牌同步配置 -> 筛选目标品牌 -> 确认飞书字段 -> 删除目标表已有记录 -> 批量插入新记录 +遍历品牌同步配置 -> 按品牌和仓库主键查询 PostgreSQL -> 确认飞书字段 -> 删除目标表已有记录 -> 批量插入新记录 ``` ## 4. 执行方式 @@ -273,24 +280,24 @@ FieldSpec("新金额字段", "SQL查询结果金额字段名", 2, "0.00", 2) ## 7. 新增品牌同步方式 -当前版本已经使用 `BRAND_SYNC_TARGETS` 管理品牌和目标飞书表的对应关系，配置位置在 `inventory_sync/specs.py`： +当前版本使用 `BRAND_SYNC_TARGETS` 管理品牌、目标飞书表和仓库主键的组合关系，配置位置在 `inventory_sync/specs.py`： ```python BRAND_SYNC_TARGETS = [ - BrandSyncTarget(brand_name="羽心堂", table_name="羽心堂"), + BrandSyncTarget(brand_name="羽心堂", table_name="羽心堂", warehouse_keys=(2, 3, 7, 13)), ] ``` -后续新增品牌时，只需要在列表中增加一行，并确保飞书 Base 中存在同名或指定名称的目标表： +后续新增品牌时，只需要在列表中增加一行，并确保飞书 Base 中存在同名或指定名称的目标表。每个品牌可以配置自己的仓库主键列表： ```python BRAND_SYNC_TARGETS = [ - BrandSyncTarget(brand_name="羽心堂", table_name="羽心堂"), - BrandSyncTarget(brand_name="其他品牌", table_name="其他品牌"), + BrandSyncTarget(brand_name="羽心堂", table_name="羽心堂", warehouse_keys=(2, 3, 7, 13)), + BrandSyncTarget(brand_name="其他品牌", table_name="其他品牌", warehouse_keys=(9, 10)), ] ``` -执行 `python3 list_feishu_base_tables.py --action sync-brand` 时，程序会先查询一次 PostgreSQL，再按该列表逐个品牌同步。 +执行 `python3 list_feishu_base_tables.py --action sync-brand` 时，程序会遍历该列表。每个目标都会使用自己的 `brand_name` 和 `warehouse_keys` 生成 SQL，再同步到对应的飞书表。 ## 8. 当前设计取舍 diff --git a/inventory_sync/postgres.py b/inventory_sync/postgres.py index 56479ec..ad52cfd 100644 --- a/inventory_sync/postgres.py +++ b/inventory_sync/postgres.py @@ -7,10 +7,26 @@ import subprocess from typing import Any from .settings import AppConfig -from .sql import POSTGRES_QUERY +from .specs import BrandSyncTarget +from .sql import DEFAULT_SYNC_TARGET, POSTGRES_QUERY, build_postgres_query -def query_postgres(config: AppConfig) -> dict[str, Any]: +def query_postgres( + config: AppConfig, + brand_name: str | None = None, + warehouse_keys: tuple[int | str, ...] | None = None, + target: BrandSyncTarget | None = None, +) -> dict[str, Any]: + active_brand_name = brand_name or (target.brand_name if target else DEFAULT_SYNC_TARGET.brand_name) + active_warehouse_keys = warehouse_keys or (target.warehouse_keys if target else DEFAULT_SYNC_TARGET.warehouse_keys) + query = ( + POSTGRES_QUERY + if brand_name is None and warehouse_keys is None and target is None + else build_postgres_query( + brand_name=active_brand_name, + warehouse_keys=active_warehouse_keys, + ) + ) env = os.environ.copy() env["PGPASSWORD"] = config.analytics_db_password command = [ @@ -28,7 +44,7 @@ def query_postgres(config: AppConfig) -> dict[str, Any]: "-c", "SET TIME ZONE 'Asia/Shanghai';", "-c", - f"COPY ({POSTGRES_QUERY}) TO STDOUT WITH CSV HEADER", + f"COPY ({query}) TO STDOUT WITH CSV HEADER", ] try: @@ -48,4 +64,3 @@ def query_postgres(config: AppConfig) -> dict[str, Any]: ) rows = list(csv.DictReader(io.StringIO(cleaned_output))) return {"row_count": len(rows), "rows": rows} - diff --git a/inventory_sync/service.py b/inventory_sync/service.py index 4a00471..e51d2b6 100644 --- a/inventory_sync/service.py +++ b/inventory_sync/service.py @@ -58,8 +58,8 @@ def sync_brand_target( access_token: str, app_token: str, tables: list[dict[str, Any]], - postgres_rows: list[dict[str, Any]], target: BrandSyncTarget, + postgres_rows: list[dict[str, Any]] | None = None, ) -> dict[str, Any]: target_table = next( (table for table in tables if table.get("name") == target.table_name), @@ -71,7 +71,7 @@ def sync_brand_target( if not table_id: raise FeishuApiError(f"目标飞书表缺少 table_id: {target.table_name}") - filtered_rows = [row for row in (postgres_rows or []) if row.get("品牌名称") == target.brand_name] + target_rows = postgres_rows or [] available_fields = client.ensure_target_fields(access_token, app_token, table_id) existing_records = client.list_records(access_token, app_token, table_id) existing_record_ids = [record.get("record_id") for record in existing_records if record.get("record_id")] @@ -80,38 +80,42 @@ def sync_brand_target( access_token, app_token, table_id, - [build_feishu_record(row) for row in filtered_rows], - ) if filtered_rows else 0 + [build_feishu_record(row) for row in target_rows], + ) if target_rows else 0 return { "brand_name": target.brand_name, + "warehouse_keys": target.warehouse_keys, "target_table_name": target.table_name, "target_table_id": table_id, "available_fields": available_fields, - "source_row_count": len(postgres_rows or []), - "filtered_row_count": len(filtered_rows), + "source_row_count": len(target_rows), + "filtered_row_count": len(target_rows), "deleted_record_count": deleted_count, "created_record_count": created_count, } def run_sync_brand(config: AppConfig, client: FeishuClient, postgres_result: dict[str, Any] | None = None) -> dict[str, Any]: - postgres_payload = postgres_result or query_postgres(config) - postgres_rows = postgres_payload["rows"] - access_token = client.get_tenant_access_token() base_info = client.parse_base_info(client.build_base_url(), access_token) app_token = base_info["app_token"] or "" client.enable_advanced_permission(access_token, app_token) tables = client.list_tables(access_token, app_token) - results = [ - sync_brand_target(client, access_token, app_token, tables, postgres_rows, target) - for target in BRAND_SYNC_TARGETS - ] + results = [] + source_row_count = 0 + for index, target in enumerate(BRAND_SYNC_TARGETS): + if postgres_result is not None and index == 0: + postgres_payload = postgres_result + else: + postgres_payload = query_postgres(config, target=target) + postgres_rows = postgres_payload["rows"] + source_row_count += len(postgres_rows or []) + results.append(sync_brand_target(client, access_token, app_token, tables, target, postgres_rows)) return { "target_count": len(BRAND_SYNC_TARGETS), - "source_row_count": len(postgres_rows or []), + "source_row_count": source_row_count, "targets": results, } diff --git a/inventory_sync/specs.py b/inventory_sync/specs.py index 8145344..cc52e92 100644 --- a/inventory_sync/specs.py +++ b/inventory_sync/specs.py @@ -12,6 +12,7 @@ PRIMARY_FIELD_SOURCE = "统一规格编码" class BrandSyncTarget: brand_name: str table_name: str + warehouse_keys: tuple[int, ...] @dataclass(frozen=True) @@ -24,7 +25,7 @@ class FieldSpec: BRAND_SYNC_TARGETS = [ - BrandSyncTarget(brand_name="羽心堂", table_name="羽心堂"), + BrandSyncTarget(brand_name="羽心堂", table_name="羽心堂", warehouse_keys=(2, 3, 7, 13)), ] diff --git a/inventory_sync/sql.py b/inventory_sync/sql.py index 909057f..ff1494f 100644 --- a/inventory_sync/sql.py +++ b/inventory_sync/sql.py @@ -1,4 +1,31 @@ -POSTGRES_QUERY = """ +from __future__ import annotations + +from collections.abc import Sequence + +from .specs import BRAND_SYNC_TARGETS + + +DEFAULT_SYNC_TARGET = BRAND_SYNC_TARGETS[0] + + +def _quote_sql_literal(value: str) -> str: + return "'" + value.replace("'", "''") + "'" + + +def _format_warehouse_keys(warehouse_keys: Sequence[int | str]) -> str: + cleaned_keys = tuple(int(key) for key in warehouse_keys) + if not cleaned_keys: + raise ValueError("warehouse_keys 不能为空") + return ", ".join(str(key) for key in cleaned_keys) + + +def build_postgres_query( + brand_name: str = DEFAULT_SYNC_TARGET.brand_name, + warehouse_keys: Sequence[int | str] = DEFAULT_SYNC_TARGET.warehouse_keys, +) -> str: + brand_filter = _quote_sql_literal(brand_name) + warehouse_key_filter = _format_warehouse_keys(warehouse_keys) + return f""" WITH inventory_latest AS ( SELECT t."统一规格编码", @@ -26,29 +53,32 @@ inventory_summary AS ( "统一规格编码", SUM("物理总库存") AS "总库存" FROM inventory_latest + WHERE "仓库主键" IN ({warehouse_key_filter}) GROUP BY "统一规格编码" ), -shop_listing_latest AS ( +shop_stock_latest AS ( SELECT t."统一规格编码", t."店铺主键", - t."商品挂单量" + t."可发库存" FROM ( SELECT f."统一规格编码", f."店铺主键", f."快照时间", - f."商品挂单量", + f."可发库存", + f."状态", ROW_NUMBER() OVER ( PARTITION BY f."统一规格编码", f."店铺主键" ORDER BY f."快照时间" DESC ) AS rn - FROM analytics_layer.fact_shop_listing_snapshot f + FROM analytics_layer.fact_shop_stock_snapshot f WHERE f."数据来源系统" = '旺店通店铺' AND NULLIF(TRIM(f."统一规格编码"), '') IS NOT NULL ) t WHERE t.rn = 1 + AND t."状态" <> '已失效' ), sales_summary AS ( @@ -109,12 +139,12 @@ SELECT pd."条形码", pd."零售价", ds."店铺名称", - COALESCE(sl."商品挂单量", 0) AS "店铺挂单量", + COALESCE(sl."可发库存", 0) AS "店铺挂单量", COALESCE(inv."总库存", 0) AS "总库存", COALESCE(ss."本月销售数量", 0) AS "本月销售数量", COALESCE(ss."上月销售数量", 0) AS "上月销售数量", COALESCE(ss."上上月销售数量", 0) AS "上上月销售数量" -FROM shop_listing_latest sl +FROM shop_stock_latest sl LEFT JOIN inventory_summary inv ON sl."统一规格编码" = inv."统一规格编码" LEFT JOIN analytics_layer.dim_shop ds @@ -124,5 +154,9 @@ LEFT JOIN sales_summary ss AND sl."店铺主键" = ss."店铺主键" LEFT JOIN product_dim pd ON sl."统一规格编码" = pd."统一规格编码" +WHERE pd."品牌名称" = {brand_filter} ORDER BY sl."统一规格编码", sl."店铺主键" """ + + +POSTGRES_QUERY = build_postgres_query()