调整 格式

inventory_sync/feishu.py

@@ -4,6 +4,7 @@ import json
 import re
 import time
 import uuid
+from collections.abc import Sequence
 from typing import Any
 from urllib import error, parse, request
 
@@ -201,19 +202,44 @@ class FeishuClient:
             time.sleep(0.2)
         return created
 
-    def ensure_target_fields(self, access_token: str, app_token: str, table_id: str) -> list[str]:
+    def ensure_target_fields(
+        self,
+        access_token: str,
+        app_token: str,
+        table_id: str,
+        field_specs: Sequence[Any] = FIELD_SPECS,
+    ) -> list[str]:
         fields = self.list_fields(access_token, app_token, table_id)
         primary_field = next((field for field in fields if field.get("is_primary")), None)
         if not primary_field:
             raise FeishuApiError("未找到主字段")
-        expected_field_names = {spec.feishu_name for spec in FIELD_SPECS}
+        expected_field_names = {spec.feishu_name for spec in field_specs}
+        expected_field_order = [spec.feishu_name for spec in field_specs]
         actual_managed_field_names = {
             field.get("field_name")
             for field in fields
             if not field.get("is_primary") and field.get("field_name") in expected_field_names
         }
+        actual_expected_field_order = [
+            field.get("field_name")
+            for field in fields
+            if not field.get("is_primary") and field.get("field_name") in expected_field_names
+        ]
+        obsolete_managed_field_names = {
+            field.get("field_name")
+            for field in fields
+            if (
+                not field.get("is_primary")
+                and field.get("field_name") not in expected_field_names
+                and self._is_managed_field(field.get("field_name") or "")
+            )
+        }
         primary_matches = primary_field.get("field_name") == PRIMARY_FIELD_NAME
-        fields_match = actual_managed_field_names == expected_field_names
+        fields_match = (
+            actual_managed_field_names == expected_field_names
+            and actual_expected_field_order == expected_field_order
+            and not obsolete_managed_field_names
+        )
         if primary_matches and fields_match:
             return [field.get("field_name") for field in fields if field.get("field_name")]
 
@@ -242,11 +268,11 @@ class FeishuClient:
             field_id = field.get("field_id")
             if field.get("is_primary") or not field_name or not field_id:
                 continue
-            if self._is_managed_field(field_name):
+            if field_name in expected_field_names or self._is_managed_field(field_name):
                 self.delete_field(access_token, app_token, table_id, field_id)
                 time.sleep(0.2)
 
-        for spec in FIELD_SPECS:
+        for spec in field_specs:
             payload = {"field_name": spec.feishu_name, "type": spec.field_type}
             if spec.formatter is not None:
                 payload["property"] = {"formatter": spec.formatter}

inventory_sync/service.py

@@ -1,6 +1,8 @@
 from __future__ import annotations
 
 import logging
+from collections import OrderedDict
+from collections.abc import Sequence
 from typing import Any
 
 from inventory_analytics.logging_utils import log_event as emit_log_event
@@ -8,7 +10,7 @@ from inventory_analytics.logging_utils import log_event as emit_log_event
 from .feishu import FeishuApiError, FeishuClient
 from .postgres import query_postgres
 from .settings import AppConfig
-from .specs import BRAND_SYNC_TARGETS, FIELD_SPECS, PRIMARY_FIELD_NAME, PRIMARY_FIELD_SOURCE, BrandSyncTarget
+from .specs import BRAND_SYNC_TARGETS, FIELD_SPECS, PRIMARY_FIELD_NAME, PRIMARY_FIELD_SOURCE, BrandSyncTarget, FieldSpec
 
 
 logger = logging.getLogger("inventory_sync.service")
@@ -22,6 +24,7 @@ LOG_FIELD_LABELS = {
     "rows": "行数",
     "source_rows": "源数据行数",
     "stage": "阶段",
+    "shop_count": "店铺数量",
     "table_count": "表数量",
     "table_name": "表名",
     "target_count": "目标数量",
@@ -32,17 +35,20 @@ def log_event(level: int, event: str, **fields: Any) -> None:
     emit_log_event(logger, LOG_FIELD_LABELS, level, event, **fields)
 
 
-def build_feishu_record(row: dict[str, Any]) -> dict[str, Any]:
+def build_feishu_record(row: dict[str, Any], field_specs: Sequence[FieldSpec] = FIELD_SPECS) -> dict[str, Any]:
     if PRIMARY_FIELD_SOURCE not in row:
         raise KeyError(f"映射缺少主字段来源列: {PRIMARY_FIELD_SOURCE}")
 
     fields: dict[str, Any] = {
         PRIMARY_FIELD_NAME: "" if row.get(PRIMARY_FIELD_SOURCE) is None else str(row.get(PRIMARY_FIELD_SOURCE)),
     }
-    for spec in FIELD_SPECS:
+    for spec in field_specs:
         if spec.source_name not in row:
-            raise KeyError(f"映射缺少 SQL 列: {spec.source_name} -> {spec.feishu_name}")
+            if spec.source_name in {base_spec.source_name for base_spec in FIELD_SPECS}:
-        value = row.get(spec.source_name)
+                raise KeyError(f"映射缺少 SQL 列: {spec.source_name} -> {spec.feishu_name}")
+            value = 0
+        else:
+            value = row.get(spec.source_name)
         if value is None or value == "":
             fields[spec.feishu_name] = None
         elif spec.field_type == 2:
@@ -53,6 +59,54 @@ def build_feishu_record(row: dict[str, Any]) -> dict[str, Any]:
     return {"fields": fields}
 
 
+def _to_number(value: Any) -> float:
+    if value is None or value == "":
+        return 0.0
+    return float(value)
+
+
+def build_shop_stock_field_specs(shop_names: Sequence[str]) -> list[FieldSpec]:
+    return [FieldSpec(shop_name, shop_name, 2, "0", 0) for shop_name in shop_names]
+
+
+def build_channel_inventory_field_specs(shop_names: Sequence[str]) -> list[FieldSpec]:
+    shop_stock_specs = build_shop_stock_field_specs(shop_names)
+    insert_after_field = "零售价"
+    field_specs: list[FieldSpec] = []
+    for spec in FIELD_SPECS:
+        field_specs.append(spec)
+        if spec.feishu_name == insert_after_field:
+            field_specs.extend(shop_stock_specs)
+    return field_specs
+
+
+def pivot_shop_stock_rows(rows: list[dict[str, Any]]) -> tuple[list[dict[str, Any]], list[str]]:
+    shop_names = list(OrderedDict.fromkeys(str(row.get("店铺名称") or "").strip() for row in rows if row.get("店铺名称")))
+    rows_by_code: OrderedDict[str, dict[str, Any]] = OrderedDict()
+    for row in rows:
+        code = "" if row.get(PRIMARY_FIELD_SOURCE) is None else str(row.get(PRIMARY_FIELD_SOURCE))
+        if code not in rows_by_code:
+            rows_by_code[code] = {
+                PRIMARY_FIELD_SOURCE: row.get(PRIMARY_FIELD_SOURCE),
+                "品牌名称": row.get("品牌名称"),
+                "货品简称": row.get("货品简称"),
+                "一级分类": row.get("一级分类"),
+                "商品名称": row.get("商品名称"),
+                "条形码": row.get("条形码"),
+                "零售价": row.get("零售价"),
+                "总库存": row.get("总库存"),
+                "可用总库存": 0,
+            }
+        shop_name = str(row.get("店铺名称") or "").strip()
+        if not shop_name:
+            continue
+        stock = _to_number(row.get("店铺挂单量"))
+        current_stock = _to_number(rows_by_code[code].get(shop_name))
+        rows_by_code[code][shop_name] = current_stock + stock
+        rows_by_code[code]["可用总库存"] = _to_number(rows_by_code[code].get("可用总库存")) + stock
+    return list(rows_by_code.values()), shop_names
+
+
 def run_feishu_summary(client: FeishuClient) -> dict[str, Any]:
     log_event(logging.INFO, "开始获取飞书访问令牌", stage="feishu_auth")
     access_token = client.get_tenant_access_token()
@@ -102,8 +156,10 @@ def sync_brand_target(
     if not table_id:
         raise FeishuApiError(f"目标飞书表缺少 table_id: {target.table_name}")
 
-    target_rows = postgres_rows or []
+    source_rows = postgres_rows or []
-    available_fields = client.ensure_target_fields(access_token, app_token, table_id)
+    target_rows, shop_names = pivot_shop_stock_rows(source_rows)
+    field_specs = build_channel_inventory_field_specs(shop_names)
+    available_fields = client.ensure_target_fields(access_token, app_token, table_id, field_specs)
     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")]
     log_event(
@@ -119,7 +175,7 @@ def sync_brand_target(
         access_token,
         app_token,
         table_id,
-        [build_feishu_record(row) for row in target_rows],
+        [build_feishu_record(row, field_specs) for row in target_rows],
     ) if target_rows else 0
     log_event(
         logging.INFO,
@@ -129,6 +185,7 @@ def sync_brand_target(
         table_name=target.table_name,
         deleted_records=deleted_count,
         created_records=created_count,
+        shop_count=len(shop_names),
     )
 
     return {
@@ -137,7 +194,7 @@ def sync_brand_target(
         "target_table_name": target.table_name,
         "target_table_id": table_id,
         "available_fields": available_fields,
-        "source_row_count": len(target_rows),
+        "source_row_count": len(source_rows),
         "filtered_row_count": len(target_rows),
         "deleted_record_count": deleted_count,
         "created_record_count": created_count,

inventory_sync/specs.py

@@ -42,12 +42,8 @@ FIELD_SPECS = [
     FieldSpec("分类", "一级分类", 1),
     FieldSpec("条形码", "条形码", 1),
     FieldSpec("零售价", "零售价", 2, "0.00", 2),
-    FieldSpec("店铺名称", "店铺名称", 1),
+    FieldSpec("可用总库存", "可用总库存", 2, "0", 0),
-    FieldSpec("店铺可用库存", "店铺挂单量", 2, "0", 0),
     FieldSpec("全渠道总库存", "总库存", 2, "0", 0),
-    FieldSpec(month_sales_field_name(0), "本月销售数量", 2, "0", 0),
-    FieldSpec(month_sales_field_name(1), "上月销售数量", 2, "0", 0),
-    FieldSpec(month_sales_field_name(2), "上上月销售数量", 2, "0", 0),
     FieldSpec("品牌", "品牌名称", 1),
     FieldSpec("产品类别", "货品简称", 1),
 ]
@@ -57,6 +53,7 @@ LEGACY_NAME_MAP = {
     "一级分类": "分类",
     "货品简称": "产品类别",
     "商品名称": "货品名称",
+    "店铺名称": "店铺名称",
     "店铺挂单量": "店铺可用库存",
     "本月销售数量": month_sales_field_name(0),
     "上月销售数量": month_sales_field_name(1),

inventory_sync/sql.py

@@ -89,41 +89,6 @@ shop_stock_latest AS (
         t."店铺主键"
 ),
 
-sales_summary AS (
-    SELECT
-        fs."统一规格编码",
-        fs."店铺主键",
-        SUM(
-            CASE
-                WHEN fs."发货时间" >= date_trunc('month', now())
-                 AND fs."发货时间" < date_trunc('month', now()) + INTERVAL '1 month'
-                THEN fs."销售数量"
-                ELSE 0
-            END
-        ) AS "本月销售数量",
-        SUM(
-            CASE
-                WHEN fs."发货时间" >= date_trunc('month', now()) - INTERVAL '1 month'
-                 AND fs."发货时间" < date_trunc('month', now())
-                THEN fs."销售数量"
-                ELSE 0
-            END
-        ) AS "上月销售数量",
-        SUM(
-            CASE
-                WHEN fs."发货时间" >= date_trunc('month', now()) - INTERVAL '2 month'
-                 AND fs."发货时间" < date_trunc('month', now()) - INTERVAL '1 month'
-                THEN fs."销售数量"
-                ELSE 0
-            END
-        ) AS "上上月销售数量"
-    FROM analytics_layer.fact_sales fs
-    WHERE NULLIF(TRIM(fs."统一规格编码"), '') IS NOT NULL
-      AND fs."发货时间" >= date_trunc('month', now()) - INTERVAL '2 month'
-      AND fs."发货时间" < date_trunc('month', now()) + INTERVAL '1 month'
-    GROUP BY fs."统一规格编码", fs."店铺主键"
-),
-
 product_dim AS (
     SELECT DISTINCT ON ("统一规格编码")
         "统一规格编码",
@@ -148,18 +113,12 @@ SELECT
     pd."零售价",
     ds."店铺名称",
     COALESCE(sl."可发库存", 0) AS "店铺挂单量",
-    COALESCE(inv."总库存", 0) AS "总库存",
+    COALESCE(inv."总库存", 0) AS "总库存"
-    COALESCE(ss."本月销售数量", 0) AS "本月销售数量",
-    COALESCE(ss."上月销售数量", 0) AS "上月销售数量",
-    COALESCE(ss."上上月销售数量", 0) AS "上上月销售数量"
 FROM shop_stock_latest sl
 LEFT JOIN inventory_summary inv
     ON sl."统一规格编码" = inv."统一规格编码"
 LEFT JOIN analytics_layer.dim_shop ds
     ON sl."店铺主键" = ds."店铺主键"
-LEFT JOIN sales_summary ss
-    ON sl."统一规格编码" = ss."统一规格编码"
-    AND sl."店铺主键" = ss."店铺主键"
 LEFT JOIN product_dim pd
     ON sl."统一规格编码" = pd."统一规格编码"
 WHERE pd."品牌名称" = {brand_filter}