inventory_analytics/inventory_sync/sql.py

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."统一规格编码",
        t."仓库主键",
        t."物理总库存"
    FROM (
        SELECT
            f."统一规格编码",
            f."仓库主键",
            f."快照时间",
            f."物理总库存",
            ROW_NUMBER() OVER (
                PARTITION BY f."统一规格编码", f."仓库主键"
                ORDER BY f."快照时间" DESC
            ) AS rn
        FROM analytics_layer.fact_inventory_snapshot f
        WHERE f."数据来源系统" = '旺店通'
          AND NULLIF(TRIM(f."统一规格编码"), '') IS NOT NULL
    ) t
    WHERE t.rn = 1
),

inventory_summary AS (
    SELECT
        "统一规格编码",
        SUM("物理总库存") AS "总库存"
    FROM inventory_latest
    WHERE "仓库主键" IN ({warehouse_key_filter})
    GROUP BY "统一规格编码"
),

shop_stock_latest AS (
    SELECT
        t."统一规格编码",
        t."店铺主键",
        SUM(t."可发库存") AS "可发库存"
    FROM (
        SELECT
            f."统一规格编码",
            f."店铺主键",
            f."仓库主键",
            f."快照时间",
            f."可发库存",
            f."状态",
            ROW_NUMBER() OVER (
                PARTITION BY 
                    f."统一规格编码",
                    f."店铺主键",
                    f."仓库主键"
                ORDER BY f."快照时间" DESC
            ) AS rn
        FROM analytics_layer.fact_shop_stock_snapshot f
        WHERE f."数据来源系统" = '旺店通店铺'
          AND NULLIF(TRIM(f."统一规格编码"), '') IS NOT NULL
    ) t
    WHERE t.rn = 1
      AND t."状态" <> '已失效'
      AND t."可发库存" <> 0
    GROUP BY
        t."统一规格编码",
        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 ("统一规格编码")
        "统一规格编码",
        "品牌名称",
        "货品简称",
        "一级分类",
        "商品名称",
        "条形码",
        "零售价"
    FROM analytics_layer.dim_product
    WHERE NULLIF(TRIM("统一规格编码"), '') IS NOT NULL
    ORDER BY "统一规格编码"
)

SELECT
    sl."统一规格编码",
    pd."品牌名称",
    pd."货品简称",
    pd."一级分类",
    pd."商品名称",
    pd."条形码",
    pd."零售价",
    ds."店铺名称",
    COALESCE(sl."可发库存", 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}
ORDER BY sl."统一规格编码", sl."店铺主键"
"""


POSTGRES_QUERY = build_postgres_query()