diff --git a/inventory_sync/feishu.py b/inventory_sync/feishu.py index a8e786b..41ffbff 100644 --- a/inventory_sync/feishu.py +++ b/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} diff --git a/inventory_sync/service.py b/inventory_sync/service.py index bd121db..ac11acc 100644 --- a/inventory_sync/service.py +++ b/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}") - value = row.get(spec.source_name) + if spec.source_name in {base_spec.source_name for base_spec in FIELD_SPECS}: + 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 [] - available_fields = client.ensure_target_fields(access_token, app_token, table_id) + source_rows = postgres_rows or [] + 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, diff --git a/inventory_sync/specs.py b/inventory_sync/specs.py index 83eb077..e829205 100644 --- a/inventory_sync/specs.py +++ b/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), diff --git a/inventory_sync/sql.py b/inventory_sync/sql.py index 6bec581..b186e91 100644 --- a/inventory_sync/sql.py +++ b/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(ss."本月销售数量", 0) AS "本月销售数量", - COALESCE(ss."上月销售数量", 0) AS "上月销售数量", - COALESCE(ss."上上月销售数量", 0) AS "上上月销售数量" + COALESCE(inv."总库存", 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}