first commit

inventory_sync/__init__.py

@@ -0,0 +1,2 @@
+"""Inventory sync package."""
+

inventory_sync/cli.py

@@ -0,0 +1,49 @@
+from __future__ import annotations
+
+import argparse
+import json
+import sys
+
+from .feishu import FeishuClient
+from .postgres import query_postgres
+from .service import run_feishu_summary, run_sync_brand
+from .settings import build_app_config
+
+
+def parse_args() -> argparse.Namespace:
+    parser = argparse.ArgumentParser(description="Feishu Base and PostgreSQL helper")
+    parser.add_argument(
+        "--action",
+        choices=["feishu", "postgres", "sync-brand", "all"],
+        default="feishu",
+        help="要执行的动作，默认 feishu",
+    )
+    return parser.parse_args()
+
+
+def main() -> int:
+    try:
+        args = parse_args()
+        config = build_app_config()
+        if config.env_path:
+            print(f"已加载环境变量文件: {config.env_path}", file=sys.stderr)
+        else:
+            print("WARNING: 未找到 .env 文件，将仅使用当前 shell 环境变量", file=sys.stderr)
+
+        client = FeishuClient(config)
+        result: dict[str, object] = {}
+
+        if args.action in {"feishu", "all"}:
+            result["feishu"] = run_feishu_summary(client)
+        if args.action in {"postgres", "all"}:
+            result["postgres"] = query_postgres(config)
+        if args.action in {"sync-brand", "all"}:
+            postgres_result = result.get("postgres")
+            result["sync_brand"] = run_sync_brand(config, client, postgres_result if isinstance(postgres_result, dict) else None)
+
+        print(json.dumps(result, ensure_ascii=False, indent=2))
+        return 0
+    except Exception as exc:  # noqa: BLE001
+        print(f"执行失败: {exc}", file=sys.stderr)
+        return 1
+

inventory_sync/feishu.py

@@ -0,0 +1,256 @@
+from __future__ import annotations
+
+import json
+import time
+import uuid
+from typing import Any
+from urllib import error, parse, request
+
+from .settings import AppConfig
+from .specs import FIELD_SPECS, LEGACY_NAME_MAP, PRIMARY_FIELD_NAME
+
+
+FEISHU_API_BASE = "https://open.feishu.cn/open-apis"
+
+
+class FeishuApiError(RuntimeError):
+    """Raised when the Feishu API returns a non-zero code."""
+
+
+class FeishuClient:
+    def __init__(self, config: AppConfig) -> None:
+        self.config = config
+
+    def request_json(
+        self,
+        url: str,
+        *,
+        method: str = "GET",
+        headers: dict[str, str] | None = None,
+        payload: dict[str, Any] | None = None,
+    ) -> dict[str, Any]:
+        body = None
+        request_headers = {"Content-Type": "application/json; charset=utf-8"}
+        if headers:
+            request_headers.update(headers)
+        if payload is not None:
+            body = json.dumps(payload).encode("utf-8")
+
+        req = request.Request(url, data=body, headers=request_headers, method=method)
+        try:
+            with request.urlopen(req, timeout=30) as response:
+                return json.loads(response.read().decode("utf-8"))
+        except error.HTTPError as exc:
+            detail = exc.read().decode("utf-8", errors="replace")
+            raise RuntimeError(f"请求失败: {exc.code} {detail}") from exc
+        except error.URLError as exc:
+            raise RuntimeError(f"无法连接飞书接口: {exc.reason}") from exc
+
+    def get_tenant_access_token(self) -> str:
+        data = self.request_json(
+            f"{FEISHU_API_BASE}/auth/v3/tenant_access_token/internal",
+            method="POST",
+            payload={"app_id": self.config.feishu_app_id, "app_secret": self.config.feishu_app_secret},
+        )
+        if data.get("code") != 0:
+            raise FeishuApiError(f"获取 tenant_access_token 失败: {data}")
+        token = data.get("tenant_access_token")
+        if not token:
+            raise FeishuApiError("接口未返回 tenant_access_token")
+        return token
+
+    def build_base_url(self) -> str:
+        base_url = self.config.feishu_base_url
+        app_token = self.config.feishu_app_token
+        if app_token and "/base/" not in base_url and "/wiki/" not in base_url:
+            return f"{base_url}/{app_token}"
+        return base_url
+
+    def get_wiki_node_info(self, access_token: str, node_token: str) -> dict[str, Any]:
+        data = self.request_json(
+            f"{FEISHU_API_BASE}/wiki/v2/spaces/get_node?token={parse.quote(node_token)}",
+            headers={"Authorization": f"Bearer {access_token}"},
+        )
+        if data.get("code") != 0:
+            raise FeishuApiError(f"获取知识空间节点信息失败: {data}")
+        node_info = data.get("data", {}).get("node")
+        if not node_info:
+            raise FeishuApiError("未获取到知识空间节点信息")
+        return node_info
+
+    def parse_base_info(self, base_url: str, access_token: str) -> dict[str, str | None]:
+        parsed_url = parse.urlparse(base_url)
+        pathname = parsed_url.path
+        app_token = pathname.rstrip("/").split("/")[-1]
+        if "/wiki/" in pathname:
+            node_info = self.get_wiki_node_info(access_token, app_token)
+            app_token = node_info.get("obj_token", app_token)
+        query_params = parse.parse_qs(parsed_url.query)
+        return {
+            "base_url": base_url,
+            "app_token": app_token,
+            "selected_table_id": query_params.get("table", [None])[0],
+            "selected_view_id": query_params.get("view", [None])[0],
+        }
+
+    def enable_advanced_permission(self, access_token: str, app_token: str) -> bool:
+        try:
+            data = self.request_json(
+                f"{FEISHU_API_BASE}/bitable/v1/apps/{app_token}",
+                method="PUT",
+                headers={"Authorization": f"Bearer {access_token}"},
+                payload={"is_advanced": True},
+            )
+        except Exception:
+            return False
+        return data.get("code") == 0
+
+    def _paginate(self, access_token: str, url_prefix: str, page_size: int) -> list[dict[str, Any]]:
+        items: list[dict[str, Any]] = []
+        page_token = ""
+        while True:
+            query = parse.urlencode({"page_size": page_size, **({"page_token": page_token} if page_token else {})})
+            data = self.request_json(
+                f"{url_prefix}?{query}",
+                headers={"Authorization": f"Bearer {access_token}"},
+            )
+            if data.get("code") != 0:
+                raise FeishuApiError(f"分页接口失败: {data}")
+            items.extend(data.get("data", {}).get("items") or [])
+            page_token = data.get("data", {}).get("page_token") or ""
+            if not data.get("data", {}).get("has_more", False):
+                break
+        return items
+
+    def list_tables(self, access_token: str, app_token: str) -> list[dict[str, Any]]:
+        return self._paginate(access_token, f"{FEISHU_API_BASE}/bitable/v1/apps/{app_token}/tables", 100)
+
+    def list_fields(self, access_token: str, app_token: str, table_id: str) -> list[dict[str, Any]]:
+        return self._paginate(access_token, f"{FEISHU_API_BASE}/bitable/v1/apps/{app_token}/tables/{table_id}/fields", 100)
+
+    def list_records(self, access_token: str, app_token: str, table_id: str) -> list[dict[str, Any]]:
+        return self._paginate(access_token, f"{FEISHU_API_BASE}/bitable/v1/apps/{app_token}/tables/{table_id}/records", 500)
+
+    def delete_field(self, access_token: str, app_token: str, table_id: str, field_id: str) -> None:
+        data = self.request_json(
+            f"{FEISHU_API_BASE}/bitable/v1/apps/{app_token}/tables/{table_id}/fields/{field_id}",
+            method="DELETE",
+            headers={"Authorization": f"Bearer {access_token}"},
+        )
+        if data.get("code") != 0:
+            raise FeishuApiError(f"删除字段失败({field_id}): {data}")
+
+    def update_field(self, access_token: str, app_token: str, table_id: str, field_id: str, payload: dict[str, Any]) -> None:
+        data = self.request_json(
+            f"{FEISHU_API_BASE}/bitable/v1/apps/{app_token}/tables/{table_id}/fields/{field_id}",
+            method="PUT",
+            headers={"Authorization": f"Bearer {access_token}"},
+            payload=payload,
+        )
+        if data.get("code") != 0:
+            raise FeishuApiError(f"更新字段失败({payload.get('field_name')}): {data}")
+
+    def create_field(self, access_token: str, app_token: str, table_id: str, payload: dict[str, Any]) -> None:
+        data = self.request_json(
+            f"{FEISHU_API_BASE}/bitable/v1/apps/{app_token}/tables/{table_id}/fields?client_token={uuid.uuid4()}",
+            method="POST",
+            headers={"Authorization": f"Bearer {access_token}"},
+            payload=payload,
+        )
+        if data.get("code") != 0:
+            raise FeishuApiError(f"新增字段失败({payload.get('field_name')}): {data}")
+
+    def batch_delete_records(self, access_token: str, app_token: str, table_id: str, record_ids: list[str]) -> int:
+        deleted = 0
+        for start in range(0, len(record_ids), 500):
+            chunk = record_ids[start : start + 500]
+            data = self.request_json(
+                f"{FEISHU_API_BASE}/bitable/v1/apps/{app_token}/tables/{table_id}/records/batch_delete",
+                method="POST",
+                headers={"Authorization": f"Bearer {access_token}"},
+                payload={"records": chunk},
+            )
+            if data.get("code") != 0:
+                raise FeishuApiError(f"删除记录失败: {data}")
+            deleted += len(chunk)
+            time.sleep(0.2)
+        return deleted
+
+    def batch_create_records(self, access_token: str, app_token: str, table_id: str, records: list[dict[str, Any]]) -> int:
+        created = 0
+        for start in range(0, len(records), 500):
+            chunk = records[start : start + 500]
+            data = self.request_json(
+                f"{FEISHU_API_BASE}/bitable/v1/apps/{app_token}/tables/{table_id}/records/batch_create?client_token={uuid.uuid4()}",
+                method="POST",
+                headers={"Authorization": f"Bearer {access_token}"},
+                payload={"records": chunk},
+            )
+            if data.get("code") != 0:
+                raise FeishuApiError(f"新增记录失败: {data}")
+            created += len(chunk)
+            time.sleep(0.2)
+        return created
+
+    def ensure_target_fields(self, access_token: str, app_token: str, table_id: str) -> 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("未找到主字段")
+
+        duplicate_primary = next(
+            (field for field in fields if not field.get("is_primary") and field.get("field_name") == PRIMARY_FIELD_NAME),
+            None,
+        )
+        if duplicate_primary:
+            self.delete_field(access_token, app_token, table_id, duplicate_primary["field_id"])
+            time.sleep(0.2)
+            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("删除重复字段后未找到主字段")
+
+        if primary_field.get("field_name") != PRIMARY_FIELD_NAME:
+            payload = {"field_name": PRIMARY_FIELD_NAME, "type": primary_field.get("type")}
+            if primary_field.get("property") is not None:
+                payload["property"] = primary_field.get("property")
+            self.update_field(access_token, app_token, table_id, primary_field["field_id"], payload)
+            time.sleep(0.2)
+
+        for old_name, new_name in LEGACY_NAME_MAP.items():
+            fields = self.list_fields(access_token, app_token, table_id)
+            field = next((item for item in fields if item.get("field_name") == old_name), None)
+            if field and old_name != new_name:
+                payload = {"field_name": new_name, "type": field.get("type")}
+                if field.get("property") is not None:
+                    payload["property"] = field.get("property")
+                self.update_field(access_token, app_token, table_id, field["field_id"], payload)
+                time.sleep(0.2)
+
+        fields = self.list_fields(access_token, app_token, table_id)
+        fields_by_name = {field.get("field_name"): field for field in fields if field.get("field_name")}
+        for spec in FIELD_SPECS:
+            field = fields_by_name.get(spec.feishu_name)
+            if not field:
+                payload = {"field_name": spec.feishu_name, "type": spec.field_type}
+                if spec.formatter is not None:
+                    payload["property"] = {"formatter": spec.formatter}
+                self.create_field(access_token, app_token, table_id, payload)
+                time.sleep(0.2)
+                fields = self.list_fields(access_token, app_token, table_id)
+                fields_by_name = {item.get("field_name"): item for item in fields if item.get("field_name")}
+                field = fields_by_name.get(spec.feishu_name)
+            if not field:
+                continue
+            current_formatter = (field.get("property") or {}).get("formatter")
+            if field.get("type") != spec.field_type or current_formatter != spec.formatter:
+                payload = {"field_name": spec.feishu_name, "type": spec.field_type}
+                if spec.formatter is not None:
+                    payload["property"] = {"formatter": spec.formatter}
+                elif field.get("property") is not None:
+                    payload["property"] = field.get("property")
+                self.update_field(access_token, app_token, table_id, field["field_id"], payload)
+                time.sleep(0.2)
+
+        return [field.get("field_name") for field in self.list_fields(access_token, app_token, table_id) if field.get("field_name")]
+

inventory_sync/postgres.py

@@ -0,0 +1,51 @@
+from __future__ import annotations
+
+import csv
+import io
+import os
+import subprocess
+from typing import Any
+
+from .settings import AppConfig
+from .sql import POSTGRES_QUERY
+
+
+def query_postgres(config: AppConfig) -> dict[str, Any]:
+    env = os.environ.copy()
+    env["PGPASSWORD"] = config.analytics_db_password
+    command = [
+        "/usr/local/bin/psql",
+        "-h",
+        config.analytics_db_host,
+        "-p",
+        config.analytics_db_port,
+        "-U",
+        config.analytics_db_user,
+        "-d",
+        config.analytics_db_name,
+        "-v",
+        "ON_ERROR_STOP=1",
+        "-c",
+        "SET TIME ZONE 'Asia/Shanghai';",
+        "-c",
+        f"COPY ({POSTGRES_QUERY}) TO STDOUT WITH CSV HEADER",
+    ]
+
+    try:
+        completed = subprocess.run(
+            command,
+            check=True,
+            capture_output=True,
+            text=True,
+            env=env,
+        )
+    except subprocess.CalledProcessError as exc:
+        detail = exc.stderr.strip() or exc.stdout.strip()
+        raise RuntimeError(f"PostgreSQL 查询失败: {detail}") from exc
+
+    cleaned_output = "\n".join(
+        line for line in completed.stdout.splitlines() if line.strip() and line.strip() != "SET"
+    )
+    rows = list(csv.DictReader(io.StringIO(cleaned_output)))
+    return {"row_count": len(rows), "rows": rows}
+

inventory_sync/service.py

@@ -0,0 +1,117 @@
+from __future__ import annotations
+
+from typing import Any
+
+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
+
+
+def build_feishu_record(row: dict[str, Any]) -> 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:
+        if spec.source_name not in row:
+            raise KeyError(f"映射缺少 SQL 列: {spec.source_name} -> {spec.feishu_name}")
+        value = row.get(spec.source_name)
+        if value is None or value == "":
+            fields[spec.feishu_name] = None
+        elif spec.field_type == 2:
+            number = float(value)
+            fields[spec.feishu_name] = int(round(number)) if spec.scale == 0 else round(number, spec.scale or 0)
+        else:
+            fields[spec.feishu_name] = str(value)
+    return {"fields": fields}
+
+
+def run_feishu_summary(client: FeishuClient) -> dict[str, Any]:
+    access_token = client.get_tenant_access_token()
+    base_info = client.parse_base_info(client.build_base_url(), access_token)
+    advanced_permission_enabled = client.enable_advanced_permission(access_token, base_info["app_token"] or "")
+    tables = client.list_tables(access_token, base_info["app_token"] or "")
+    return {
+        "base_url": base_info["base_url"],
+        "app_token": base_info["app_token"],
+        "auth_mode": "tenant_access_token",
+        "selected_table_id": base_info["selected_table_id"],
+        "selected_view_id": base_info["selected_view_id"],
+        "advanced_permission_enabled": advanced_permission_enabled,
+        "folder": {
+            "name": base_info["app_token"],
+            "type": "base",
+            "tables": [
+                {"table_id": table.get("table_id"), "name": table.get("name"), "revision": table.get("revision")}
+                for table in tables
+            ],
+        },
+        "table_count": len(tables),
+    }
+
+
+def sync_brand_target(
+    client: FeishuClient,
+    access_token: str,
+    app_token: str,
+    tables: list[dict[str, Any]],
+    postgres_rows: list[dict[str, Any]],
+    target: BrandSyncTarget,
+) -> dict[str, Any]:
+    target_table = next(
+        (table for table in tables if table.get("name") == target.table_name),
+        None,
+    )
+    if not target_table:
+        raise FeishuApiError(f"未找到目标飞书表: {target.table_name}")
+    table_id = target_table.get("table_id")
+    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]
+    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")]
+    deleted_count = client.batch_delete_records(access_token, app_token, table_id, existing_record_ids) if existing_record_ids else 0
+    created_count = client.batch_create_records(
+        access_token,
+        app_token,
+        table_id,
+        [build_feishu_record(row) for row in filtered_rows],
+    ) if filtered_rows else 0
+
+    return {
+        "brand_name": target.brand_name,
+        "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),
+        "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
+    ]
+
+    return {
+        "target_count": len(BRAND_SYNC_TARGETS),
+        "source_row_count": len(postgres_rows or []),
+        "targets": results,
+    }

inventory_sync/settings.py

@@ -0,0 +1,85 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+from pathlib import Path
+import os
+
+
+@dataclass(frozen=True)
+class AppConfig:
+    env_path: Path | None
+    feishu_app_id: str
+    feishu_app_secret: str
+    feishu_base_url: str
+    feishu_app_token: str | None
+    analytics_db_host: str
+    analytics_db_port: str
+    analytics_db_user: str
+    analytics_db_password: str
+    analytics_db_name: str
+
+
+def load_env_file(env_path: Path) -> dict[str, str]:
+    values: dict[str, str] = {}
+    if not env_path.exists():
+        return values
+
+    for raw_line in env_path.read_text(encoding="utf-8").splitlines():
+        line = raw_line.strip()
+        if not line or line.startswith("#") or "=" not in line:
+            continue
+        key, value = line.split("=", 1)
+        values[key.strip()] = value.strip().strip("'").strip('"')
+    return values
+
+
+def load_env_candidates() -> tuple[dict[str, str], Path | None]:
+    script_path = Path(__file__).resolve()
+    candidates = [
+        script_path.parent.parent / ".env",
+        script_path.parent.parent / "project (1)" / ".env",
+        script_path.parent.parent.parent / ".env",
+        Path.cwd() / ".env",
+    ]
+    seen: set[Path] = set()
+    for candidate in candidates:
+        candidate = candidate.resolve()
+        if candidate in seen:
+            continue
+        seen.add(candidate)
+        values = load_env_file(candidate)
+        if values:
+            return values, candidate
+    return {}, None
+
+
+def get_config(config: dict[str, str], *keys: str) -> str | None:
+    for key in keys:
+        value = config.get(key) or os.getenv(key)
+        if value:
+            return value
+    return None
+
+
+def require_config(config: dict[str, str], *keys: str) -> str:
+    value = get_config(config, *keys)
+    if value:
+        return value
+    raise KeyError(f"缺少必填配置: {' / '.join(keys)}")
+
+
+def build_app_config() -> AppConfig:
+    config, env_path = load_env_candidates()
+    return AppConfig(
+        env_path=env_path,
+        feishu_app_id=require_config(config, "FEISHU_APP_ID", "APP_ID"),
+        feishu_app_secret=require_config(config, "FEISHU_APP_SECRET", "FEISHU_APP_SECRE", "APP_SECRET"),
+        feishu_base_url=require_config(config, "FEISHU_BASE_URL", "BASE_URL").rstrip("/"),
+        feishu_app_token=get_config(config, "FEISHU_APP_TOKEN"),
+        analytics_db_host=require_config(config, "ANALYTICS_DB_HOST"),
+        analytics_db_port=require_config(config, "ANALYTICS_DB_PORT"),
+        analytics_db_user=require_config(config, "ANALYTICS_DB_USER"),
+        analytics_db_password=require_config(config, "ANALYTICS_DB_PASSWORD"),
+        analytics_db_name=require_config(config, "ANALYTICS_DB_NAME"),
+    )
+

inventory_sync/specs.py

@@ -0,0 +1,51 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+
+PRIMARY_FIELD_NAME = "货品编码"
+PRIMARY_FIELD_SOURCE = "统一规格编码"
+
+
+@dataclass(frozen=True)
+class BrandSyncTarget:
+    brand_name: str
+    table_name: str
+
+
+@dataclass(frozen=True)
+class FieldSpec:
+    feishu_name: str
+    source_name: str
+    field_type: int
+    formatter: str | None = None
+    scale: int | None = None
+
+
+BRAND_SYNC_TARGETS = [
+    BrandSyncTarget(brand_name="羽心堂", table_name="羽心堂"),
+]
+
+FIELD_SPECS = [
+    FieldSpec("品牌", "品牌名称", 1),
+    FieldSpec("分类", "一级分类", 1),
+    FieldSpec("产品类别", "货品简称", 1),
+    FieldSpec("货品名称", "商品名称", 1),
+    FieldSpec("条形码", "条形码", 1),
+    FieldSpec("零售价", "零售价", 2, "0.00", 2),
+    FieldSpec("店铺名称", "店铺名称", 1),
+    FieldSpec("店铺可用库存", "店铺挂单量", 2, "0", 0),
+    FieldSpec("总库存", "总库存", 2, "0", 0),
+    FieldSpec("本月销售数量", "本月销售数量", 2, "0", 0),
+    FieldSpec("上月销售数量", "上月销售数量", 2, "0", 0),
+    FieldSpec("上上月销售数量", "上上月销售数量", 2, "0", 0),
+]
+
+LEGACY_NAME_MAP = {
+    "品牌名称": "品牌",
+    "一级分类": "分类",
+    "货品简称": "产品类别",
+    "商品名称": "货品名称",
+    "店铺挂单量": "店铺可用库存",
+    "统一规格编码": PRIMARY_FIELD_NAME,
+}

inventory_sync/sql.py

@@ -0,0 +1,127 @@
+POSTGRES_QUERY = """
+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
+    GROUP BY "统一规格编码"
+),
+
+shop_listing_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_shop_listing_snapshot f
+        WHERE f."数据来源系统" = '旺店通店铺'
+          AND NULLIF(TRIM(f."统一规格编码"), '') IS NOT NULL
+    ) t
+    WHERE t.rn = 1
+),
+
+sales_summary AS (
+    SELECT
+        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."统一规格编码"
+),
+
+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_listing_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."统一规格编码"
+LEFT JOIN product_dim pd
+    ON sl."统一规格编码" = pd."统一规格编码"
+ORDER BY sl."统一规格编码", sl."店铺主键"
+"""
+