commit 86f120bc181e7e5f39181da6bb5295781757c35a Author: BXY0414 <1079823339@qq.com> Date: Mon Apr 20 15:02:17 2026 +0800 first commit diff --git a/.DS_Store b/.DS_Store new file mode 100644 index 0000000..c1709d2 Binary files /dev/null and b/.DS_Store differ diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..ee203ff --- /dev/null +++ b/.gitignore @@ -0,0 +1,2 @@ +.env +.feishu_oauth_tokens.json diff --git a/REFACTOR_PLAN.md b/REFACTOR_PLAN.md new file mode 100644 index 0000000..40af555 --- /dev/null +++ b/REFACTOR_PLAN.md @@ -0,0 +1,352 @@ +# 飞书库存同步项目重构方案 + +## 1. 重构目标 + +当前项目的核心能力是从 PostgreSQL 查询库存、挂单量、销售和商品信息，按品牌筛选数据，并同步写入对应的飞书多维表格。 + +本次重构的目标是将原先集中在单个脚本中的逻辑拆分为更清晰、可维护、可扩展的工程结构： + +- 将配置读取、数据库查询、飞书 API、业务同步、命令行入口拆分到独立模块。 +- 将飞书字段名称、SQL 字段来源、字段类型、数字格式统一维护在映射表中。 +- 保留原有脚本入口，降低使用成本，避免改变已有执行习惯。 +- 提高后续新增品牌、新增字段、新增同步动作时的可读性和安全性。 + +## 2. 当前目录结构 + +```text +inventory_analytics/ +├── .env +├── .gitignore +├── list_feishu_base_tables.py +├── REFACTOR_PLAN.md +└── inventory_sync/ + ├── __init__.py + ├── cli.py + ├── feishu.py + ├── postgres.py + ├── service.py + ├── settings.py + ├── specs.py + └── sql.py +``` + +## 3. 模块职责 + +### 3.1 入口脚本 + +文件：`list_feishu_base_tables.py` + +职责： + +- 作为项目的兼容入口。 +- 将执行逻辑转交给 `inventory_sync.cli.main()`。 +- 保持原命令可继续使用。 + +示例： + +```bash +python3 list_feishu_base_tables.py --action sync-brand +``` + +### 3.2 命令行模块 + +文件：`inventory_sync/cli.py` + +职责： + +- 解析 `--action` 参数。 +- 初始化配置和飞书客户端。 +- 根据 action 调用对应服务。 +- 统一输出 JSON 格式结果。 + +当前支持的 action： + +- `feishu`：读取飞书 Base 下的表信息。 +- `postgres`：执行 PostgreSQL 查询并输出结果。 +- `sync-brand`：按照 `BRAND_SYNC_TARGETS` 配置筛选品牌数据并同步到对应飞书表。 +- `all`：依次执行飞书读取、数据库查询和品牌同步。 + +### 3.3 配置模块 + +文件：`inventory_sync/settings.py` + +职责： + +- 自动查找并加载 `.env` 文件。 +- 从 `.env` 或当前 shell 环境变量中读取配置。 +- 校验必填配置。 +- 构建统一的 `AppConfig` 配置对象。 + +当前必填配置： + +```text +FEISHU_APP_ID +FEISHU_APP_SECRET 或 FEISHU_APP_SECRE +FEISHU_BASE_URL +ANALYTICS_DB_HOST +ANALYTICS_DB_PORT +ANALYTICS_DB_USER +ANALYTICS_DB_PASSWORD +ANALYTICS_DB_NAME +``` + +可选配置： + +```text +FEISHU_APP_TOKEN +``` + +### 3.4 字段映射模块 + +文件：`inventory_sync/specs.py` + +职责： + +- 定义品牌到目标飞书表的同步配置。 +- 定义主字段名称和来源字段。 +- 定义 SQL 查询结果字段到飞书表字段的映射关系。 +- 定义飞书字段类型和数字格式。 +- 定义历史字段名到新字段名的重命名关系。 + +当前同步目标： + +```python +BRAND_SYNC_TARGETS = [ + BrandSyncTarget(brand_name="羽心堂", table_name="羽心堂"), +] +PRIMARY_FIELD_NAME = "货品编码" +PRIMARY_FIELD_SOURCE = "统一规格编码" +``` + +当前字段映射： + +| 飞书字段名 | SQL 来源字段 | 飞书字段类型 | 显示格式 | +| --- | --- | --- | --- | +| 货品编码 | 统一规格编码 | 文本主字段 | 文本 | +| 品牌 | 品牌名称 | 文本 | 文本 | +| 分类 | 一级分类 | 文本 | 文本 | +| 产品类别 | 货品简称 | 文本 | 文本 | +| 货品名称 | 商品名称 | 文本 | 文本 | +| 条形码 | 条形码 | 文本 | 文本 | +| 零售价 | 零售价 | 数字 | 保留 2 位小数 | +| 店铺名称 | 店铺名称 | 文本 | 文本 | +| 店铺可用库存 | 店铺挂单量 | 数字 | 不保留小数 | +| 总库存 | 总库存 | 数字 | 不保留小数 | +| 本月销售数量 | 本月销售数量 | 数字 | 不保留小数 | +| 上月销售数量 | 上月销售数量 | 数字 | 不保留小数 | +| 上上月销售数量 | 上上月销售数量 | 数字 | 不保留小数 | + +### 3.5 SQL 模块 + +文件：`inventory_sync/sql.py` + +职责： + +- 集中维护 PostgreSQL 查询 SQL。 +- 避免业务代码中混入大段 SQL。 +- 后续如需调整库存、销量、商品维度逻辑，只需要修改该文件。 + +当前 SQL 逻辑包括： + +- 取每个统一规格编码、仓库的最新库存快照。 +- 汇总统一规格编码维度的总库存。 +- 取每个统一规格编码、店铺的最新挂单量快照。 +- 汇总本月、上月、上上月销售数量。 +- 关联商品维表和店铺维表。 +- 输出用于飞书同步的明细结果。 + +### 3.6 PostgreSQL 模块 + +文件：`inventory_sync/postgres.py` + +职责： + +- 使用 `.env` 中的数据库配置连接 PostgreSQL。 +- 执行 `inventory_sync/sql.py` 中定义的 SQL。 +- 将查询结果解析为结构化字典列表。 +- 对外返回统一结构，供 CLI 或同步服务使用。 + +### 3.7 飞书 API 模块 + +文件：`inventory_sync/feishu.py` + +职责： + +- 获取飞书 `tenant_access_token`。 +- 解析飞书 Base 地址和 `app_token`。 +- 查询飞书 Base 下的表列表。 +- 查询、创建、更新飞书字段。 +- 根据字段映射创建或修正目标表字段。 +- 分页读取记录。 +- 批量删除记录。 +- 批量创建记录。 + +该模块只关心飞书接口本身，不关心具体业务品牌和 SQL 字段。 + +### 3.8 业务服务模块 + +文件：`inventory_sync/service.py` + +职责： + +- 编排完整业务流程。 +- 根据配置查询 PostgreSQL。 +- 按 `BRAND_SYNC_TARGETS` 逐个筛选品牌数据。 +- 根据 `specs.py` 中的映射表构造飞书记录。 +- 确认目标飞书表字段存在且类型正确。 +- 删除目标表已有记录。 +- 批量写入最新记录。 + +当前同步策略是全量覆盖： + +```text +查询 PostgreSQL -> 遍历品牌同步配置 -> 筛选目标品牌 -> 确认飞书字段 -> 删除目标表已有记录 -> 批量插入新记录 +``` + +## 4. 执行方式 + +### 4.1 查看飞书表 + +```bash +python3 list_feishu_base_tables.py --action feishu +``` + +### 4.2 查询 PostgreSQL 数据 + +```bash +python3 list_feishu_base_tables.py --action postgres +``` + +### 4.3 同步“羽心堂”数据到飞书表 + +```bash +python3 list_feishu_base_tables.py --action sync-brand +``` + +### 4.4 执行完整流程 + +```bash +python3 list_feishu_base_tables.py --action all +``` + +## 5. 字段调整方式 + +如果只需要调整飞书表头名称、字段来源或字段类型，优先修改 `inventory_sync/specs.py`。 + +例如，将飞书表中的“店铺可用库存”改名为“可售库存”，只需要调整： + +```python +FieldSpec("店铺可用库存", "店铺挂单量", 2, "0", 0) +``` + +改为： + +```python +FieldSpec("可售库存", "店铺挂单量", 2, "0", 0) +``` + +如果 SQL 输出字段名发生变化，需要同步调整 `source_name`。 + +例如 SQL 输出从 `店铺挂单量` 改为 `店铺库存`，则需要调整： + +```python +FieldSpec("店铺可用库存", "店铺库存", 2, "0", 0) +``` + +## 6. 新增字段流程 + +新增字段建议按以下顺序处理： + +1. 在 `inventory_sync/sql.py` 中确认 SQL 已输出该字段。 +2. 在 `inventory_sync/specs.py` 的 `FIELD_SPECS` 中增加映射。 +3. 如果是数字字段，设置 `field_type=2`，并配置 `formatter` 和 `scale`。 +4. 执行 `python3 list_feishu_base_tables.py --action sync-brand`。 +5. 检查飞书表头和数据是否符合预期。 + +示例： + +```python +FieldSpec("新字段飞书名称", "SQL查询结果字段名", 1) +FieldSpec("新数字字段", "SQL查询结果数字字段名", 2, "0", 0) +FieldSpec("新金额字段", "SQL查询结果金额字段名", 2, "0.00", 2) +``` + +## 7. 新增品牌同步方式 + +当前版本已经使用 `BRAND_SYNC_TARGETS` 管理品牌和目标飞书表的对应关系，配置位置在 `inventory_sync/specs.py`： + +```python +BRAND_SYNC_TARGETS = [ + BrandSyncTarget(brand_name="羽心堂", table_name="羽心堂"), +] +``` + +后续新增品牌时，只需要在列表中增加一行，并确保飞书 Base 中存在同名或指定名称的目标表： + +```python +BRAND_SYNC_TARGETS = [ + BrandSyncTarget(brand_name="羽心堂", table_name="羽心堂"), + BrandSyncTarget(brand_name="其他品牌", table_name="其他品牌"), +] +``` + +执行 `python3 list_feishu_base_tables.py --action sync-brand` 时，程序会先查询一次 PostgreSQL，再按该列表逐个品牌同步。 + +## 8. 当前设计取舍 + +### 8.1 保留原脚本入口 + +保留 `list_feishu_base_tables.py` 是为了不改变已有使用方式。后续如果项目继续扩大，可以新增更标准的入口，例如： + +```bash +python3 -m inventory_sync.cli --action sync-brand +``` + +### 8.2 使用全量覆盖同步 + +当前采用删除旧记录后重新插入的方式，优点是逻辑简单、结果确定，不需要处理复杂的增量比对。 + +需要注意的是，如果目标飞书表中存在人工维护的额外行，全量覆盖会删除这些记录。后续如需保留人工数据，应改为基于主键“货品编码 + 店铺名称”做 upsert。 + +### 8.3 字段映射集中维护 + +所有“SQL 字段名 -> 飞书表头名”的关系集中放在 `specs.py`，这样可以避免同步逻辑中到处散落字段名，降低后续调整表头时漏改的风险。 + +## 9. 后续优化建议 + +- 增加日志模块，将当前 print 输出改为结构化日志。 +- 增加 dry-run 模式，只预览将要删除和写入的数据，不真正修改飞书。 +- 增加 upsert 模式，避免每次全量删除重建。 +- 增加单元测试，重点覆盖字段映射、数字转换、配置校验。 +- 增加同步前数据质量检查，例如主字段为空、重复主键、数字无法转换。 +- 支持为不同品牌配置不同字段映射，适配不同品牌表结构差异。 +- 将 SQL 拆分为独立 `.sql` 文件，便于数据同学维护。 + +## 10. 验证命令 + +重构后建议至少执行以下命令验证： + +```bash +PYTHONPYCACHEPREFIX=/tmp/pycache python3 -m py_compile list_feishu_base_tables.py inventory_sync/*.py +python3 list_feishu_base_tables.py --action feishu +python3 list_feishu_base_tables.py --action postgres +python3 list_feishu_base_tables.py --action sync-brand +``` + +如果 `sync-brand` 返回类似以下字段，说明同步流程执行成功： + +```json +{ + "target_count": 1, + "source_row_count": 305, + "targets": [ + { + "brand_name": "羽心堂", + "target_table_name": "羽心堂", + "filtered_row_count": 200, + "deleted_record_count": 198, + "created_record_count": 200 + } + ] +} +``` diff --git a/inventory_sync/__init__.py b/inventory_sync/__init__.py new file mode 100644 index 0000000..295f5c9 --- /dev/null +++ b/inventory_sync/__init__.py @@ -0,0 +1,2 @@ +"""Inventory sync package.""" + diff --git a/inventory_sync/cli.py b/inventory_sync/cli.py new file mode 100644 index 0000000..5aea783 --- /dev/null +++ b/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 + diff --git a/inventory_sync/feishu.py b/inventory_sync/feishu.py new file mode 100644 index 0000000..618d5ff --- /dev/null +++ b/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")] + diff --git a/inventory_sync/postgres.py b/inventory_sync/postgres.py new file mode 100644 index 0000000..56479ec --- /dev/null +++ b/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} + diff --git a/inventory_sync/service.py b/inventory_sync/service.py new file mode 100644 index 0000000..4a00471 --- /dev/null +++ b/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, + } diff --git a/inventory_sync/settings.py b/inventory_sync/settings.py new file mode 100644 index 0000000..66522f3 --- /dev/null +++ b/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"), + ) + diff --git a/inventory_sync/specs.py b/inventory_sync/specs.py new file mode 100644 index 0000000..98fa691 --- /dev/null +++ b/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, +} diff --git a/inventory_sync/sql.py b/inventory_sync/sql.py new file mode 100644 index 0000000..bb1c0d0 --- /dev/null +++ b/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."店铺主键" +""" + diff --git a/list_feishu_base_tables.py b/list_feishu_base_tables.py new file mode 100644 index 0000000..23d4b41 --- /dev/null +++ b/list_feishu_base_tables.py @@ -0,0 +1,8 @@ +#!/usr/bin/env python3 +"""CLI entrypoint for inventory sync.""" + +from inventory_sync.cli import main + + +if __name__ == "__main__": + raise SystemExit(main())