BXY0414/inventory_analytics
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

重构

Browse Source
This commit is contained in:
BXY0414
2026-04-29 14:06:41 +08:00
parent 65216cd4cf
commit 8e4c2eef98
37 changed files with 2686 additions and 1985 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.DS_Store
Vendored
BIN
View File
Binary file not shown.
.gitignore
+1
View File
@@ -5,3 +5,4 @@ inventory_sync/__pycache__
batch_inventory_data/*
logs
reports
batch_inventory_data
Dockerfile
+48
View File
@@ -0,0 +1,48 @@
FROM ghcr.io/astral-sh/uv:latest AS uv
FROM python:3.12-slim-bookworm
COPY --from=uv /uv /bin/uv
ARG UV_INDEX_URL=https://pypi.tuna.tsinghua.edu.cn/simple
ENV PYTHONUNBUFFERED=1 \
PYTHONDONTWRITEBYTECODE=1 \
UV_SYSTEM_PYTHON=1 \
UV_INDEX_URL=${UV_INDEX_URL} \
TZ=Asia/Shanghai \
PYTHONPATH="/app:${PYTHONPATH}"
RUN echo "# 清华大学 Debian 镜像源配置\n\
deb https://mirrors.tuna.tsinghua.edu.cn/debian/ bookworm main contrib non-free non-free-firmware\n\
deb-src https://mirrors.tuna.tsinghua.edu.cn/debian/ bookworm main contrib non-free non-free-firmware\n\
\n\
deb https://mirrors.tuna.tsinghua.edu.cn/debian-security/ bookworm-security main contrib non-free non-free-firmware\n\
deb-src https://mirrors.tuna.tsinghua.edu.cn/debian-security/ bookworm-security main contrib non-free non-free-firmware\n\
\n\
deb https://mirrors.tuna.tsinghua.edu.cn/debian/ bookworm-updates main contrib non-free non-free-firmware\n\
deb-src https://mirrors.tuna.tsinghua.edu.cn/debian/ bookworm-updates main contrib non-free non-free-firmware" > /etc/apt/sources.list
RUN apt-get update && apt-get install -y --no-install-recommends \
supervisor \
tzdata \
&& ln -snf /usr/share/zoneinfo/Asia/Shanghai /etc/localtime \
&& echo "Asia/Shanghai" > /etc/timezone \
&& rm -rf /var/lib/apt/lists/*
WORKDIR /app
COPY pyproject.toml /app/pyproject.toml
RUN uv pip install -r /app/pyproject.toml
COPY . /app
RUN chmod +x /app/docker-entrypoint.sh \
&& mkdir -p /app/logs \
&& mkdir -p /var/log/supervisor
COPY supervisord.conf /etc/supervisor/conf.d/supervisord.conf
ENTRYPOINT ["/bin/sh", "/app/docker-entrypoint.sh"]
CMD ["/usr/bin/supervisord", "-c", "/etc/supervisor/conf.d/supervisord.conf"]
PROJECT_GUIDE.md
+350 -154
View File
@@ -1,45 +1,137 @@
# Inventory Analytics 工程说明
# Inventory Analytics 项目说明
## 项目概览
## 1. 项目概览
当前工程包含两条独立业务链路：
当前项目围绕 3 条业务链路运行：
- 委外批次库存上传：通过本地网页上传 Excel/CSV 文件，保存原始文件，并逐行写入 Raw PostgreSQL 表。
- 飞书库存同步：从 Analytics PostgreSQL 查询库存、挂单量、销售和商品信息，并同步到飞书多维表格。
- 委外批次库存上传：通过本地网页上传 Excel/CSV 文件，归档原始文件，并逐行写入 Raw PostgreSQL。
- 渠道库存统计同步飞书：从 Analytics PostgreSQL 查询渠道库存、挂单量、销售与商品信息，并同步到飞书多维表格。
- 批次库存统计报表：读取最新批次库存、关联 Analytics 店铺库存与近 30 天销售，生成本地报表并可同步到飞书。
## 目录结构
命名说明：
- “渠道库存统计”是框架层命名，当前默认业务目标仍是羽心堂渠道库存。
- “批次库存统计”是框架层命名，当前默认业务目标仍是羽心堂批次库存。
- 后续扩展到其他品牌时，优先复用这两条通用链路，而不是继续新增带品牌名的脚本和包名。
项目已经完成一轮框架调整：
- 新实现逐步迁入 `inventory_analytics/` 包。
- 根目录脚本保留兼容入口。
- `scripts/` 目录提供新的标准执行入口。
- 旧的 `inventory_sync/` 仍保留，当前被渠道库存统计链路复用，属于过渡层。
## 2. 当前目录结构
```text
inventory_analytics/
├── .env
├── pyproject.toml
├── PROJECT_GUIDE.md
├── REFACTOR_PLAN.md
├── upload_inventory_page.py
├── Dockerfile
├── docker-compose.yaml
├── docker-entrypoint.sh
├── supervisord.conf
├── prefect.yaml
├── prefect_inventory_scheduler.py
├── generate_batch_inventory_report.py
├── generate_yxt_batch_inventory_report.py
├── list_feishu_base_tables.py
└── inventory_sync/
├── __init__.py
├── cli.py
├── feishu.py
├── postgres.py
├── service.py
├── settings.py
├── specs.py
└── sql.py
├── sync_channel_inventory.py
├── upload_inventory_page.py
├── scripts/
│ ├── generate_batch_inventory_report.py
│ ├── sync_channel_inventory.py
│ └── upload_batch_inventory.py
├── inventory_analytics/
│ ├── __init__.py
│ ├── config.py
│ ├── logging_utils.py
│ ├── paths.py
│ ├── db/
│ │ └── __init__.py
│ ├── feishu/
│ │ ├── __init__.py
│ │ └── client.py
│ ├── batch_upload/
│ │ ├── __init__.py
│ │ └── web.py
│ ├── channel_inventory_sync/
│ │ ├── __init__.py
│ │ ├── cli.py
│ │ ├── entrypoint.py
│ │ ├── queries.py
│ │ ├── service.py
│ │ └── specs.py
│ └── batch_inventory_report/
│ ├── __init__.py
│ ├── cli.py
│ └── service.py
├── inventory_sync/
│ ├── __init__.py
│ ├── cli.py
│ ├── feishu.py
│ ├── postgres.py
│ ├── service.py
│ ├── settings.py
│ ├── specs.py
│ └── sql.py
├── batch_inventory_data/
├── logs/
└── reports/
```
运行后会自动创建或使用：
说明：
```text
batch_inventory_data/
```
- `inventory_analytics/` 是当前主包，后续新增实现优先放这里。
- `scripts/` 是推荐执行入口。
- 根目录兼容入口仍保留，避免旧命令立即失效。
- `inventory_sync/` 当前仍在使用，但属于待进一步收敛的历史模块。
该目录保存上传的委外批次库存原始文件，不纳入业务代码结构。
## 3. 模块职责
## 环境变量
### 3.1 通用基础层
### 委外批次库存上传
| 文件 | 说明 |
| --- | --- |
| `inventory_analytics/config.py` | 统一加载 `.env`、读取数据库和飞书配置 |
| `inventory_analytics/logging_utils.py` | 统一日志初始化与敏感信息脱敏 |
| `inventory_analytics/paths.py` | 统一维护项目根目录、日志目录、报表目录、上传目录 |
上传页使用 `.env` 中的 Raw 数据库配置：
### 3.2 上传链路
| 文件 | 说明 |
| --- | --- |
| `scripts/upload_batch_inventory.py` | 推荐入口 |
| `upload_inventory_page.py` | 兼容入口 |
| `inventory_analytics/batch_upload/web.py` | 上传页、文件解析、Raw 写库实现 |
### 3.3 渠道库存统计链路
| 文件 | 说明 |
| --- | --- |
| `scripts/sync_channel_inventory.py` | 推荐入口，默认等价于 `--action all` |
| `sync_channel_inventory.py` | 通用兼容入口 |
| `list_feishu_base_tables.py` | 历史兼容入口 |
| `inventory_analytics/channel_inventory_sync/entrypoint.py` | 新入口日志封装 |
| `inventory_analytics/channel_inventory_sync/cli.py` | 当前复用旧 `inventory_sync.cli` |
| `inventory_sync/` | 当前渠道库存统计实际实现所在模块 |
### 3.4 批次库存统计链路
| 文件 | 说明 |
| --- | --- |
| `scripts/generate_batch_inventory_report.py` | 推荐入口 |
| `generate_batch_inventory_report.py` | 通用兼容入口 |
| `generate_yxt_batch_inventory_report.py` | 历史兼容入口 |
| `inventory_analytics/batch_inventory_report/service.py` | 报表查询、汇总、导出、飞书同步实现 |
## 4. 环境变量
项目主要使用以下配置：
### 4.1 Raw 数据库
```text
RAW_DB_HOST
@@ -49,32 +141,9 @@ RAW_DB_PASSWORD
RAW_DB_NAME
```
写入目标：
### 4.2 Analytics 数据库
```text
raw_data.wangdian_batch_inventory
```
表结构由 `upload_inventory_page.py` 内的 `get_create_table_sql()` 创建：
```sql
payload JSONB NOT NULL,
start_time TIMESTAMP WITH TIME ZONE,
end_time TIMESTAMP WITH TIME ZONE,
created_at TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT NOW(),
reserved_1 TEXT,
reserved_2 TEXT
```
### 飞书库存同步
飞书同步使用 `.env` 中的飞书和 Analytics 数据库配置：
```text
FEISHU_APP_ID
FEISHU_APP_SECRET
FEISHU_BASE_URL
FEISHU_APP_TOKEN
ANALYTICS_DB_HOST
ANALYTICS_DB_PORT
ANALYTICS_DB_USER
@@ -82,22 +151,92 @@ ANALYTICS_DB_PASSWORD
ANALYTICS_DB_NAME
```
`FEISHU_APP_TOKEN` 可选；当 `FEISHU_BASE_URL` 已包含 `/base/` 或 `/wiki/` 时，可以不单独配置。
### 4.3 飞书配置
## 委外批次库存上传
```text
FEISHU_APP_ID
FEISHU_APP_SECRET
FEISHU_BASE_URL
FEISHU_APP_TOKEN
```
说明：
- `FEISHU_APP_TOKEN` 可选。
- 当 `FEISHU_BASE_URL` 已包含 `/base/` 或 `/wiki/` 时，可以不额外配置 `FEISHU_APP_TOKEN`。
- 配置优先从项目根目录 `.env` 读取，若缺失则回退到当前 shell 环境变量。
## 5. 执行方式
### 5.1 推荐入口
启动上传页：
```bash
python3 upload_inventory_page.py
python3 scripts/upload_batch_inventory.py
```
打开页面：
启动后访问：
```text
http://127.0.0.1:8080
```
执行渠道库存统计完整流程：
```bash
python3 scripts/sync_channel_inventory.py
```
这条命令默认等价于：
```bash
python3 scripts/sync_channel_inventory.py --action all
```
按动作拆分执行：
```bash
python3 scripts/sync_channel_inventory.py --action feishu
python3 scripts/sync_channel_inventory.py --action postgres
python3 scripts/sync_channel_inventory.py --action sync-brand
python3 scripts/sync_channel_inventory.py --action all
```
生成批次库存统计报表并同步飞书：
```bash
python3 scripts/generate_batch_inventory_report.py
```
只生成本地报表，不同步飞书：
```bash
python3 scripts/generate_batch_inventory_report.py --skip-feishu-sync
```
自定义输出路径：
```bash
python3 scripts/generate_batch_inventory_report.py --output reports/custom.xlsx
```
### 5.2 兼容入口
以下旧命令或兼容命令仍可继续使用：
```bash
python3 upload_inventory_page.py
python3 sync_channel_inventory.py --action all
python3 list_feishu_base_tables.py --action all
python3 generate_batch_inventory_report.py
python3 generate_yxt_batch_inventory_report.py
```
## 6. 各链路行为说明
### 6.1 委外批次库存上传
支持文件：
- `.xlsx`
@@ -113,144 +252,201 @@ http://127.0.0.1:8080
上传成功后会执行：
1. 保存文件到 `batch_inventory_data/`。
2. 解析 Excel/CSV 为行数据。
3. 创建 Raw schema/table 和 `created_at` 索引。
4. 每一行插入一条 JSONB payload。
1. 保存原始文件到 `batch_inventory_data/`
2. 解析 Excel 或 CSV
3. 自动确认 `raw_data.wangdian_batch_inventory` 表存在
4. 每一行写入一条 JSONB payload
payload 示例：
```json
{
"source_file": "委外批次库存.xlsx",
"stored_file": "20260423_172410_委外批次库存.xlsx",
"uploaded_at": "2026-04-23T18:20:00",
"total_count": 120,
"inventory_data": {
"货品编码": "SKU001",
"批次": "BATCH001",
"数量": 10
}
}
```
## 飞书库存同步
查看飞书 Base 表：
```bash
python3 list_feishu_base_tables.py --action feishu
```
只查询 PostgreSQL：
```bash
python3 list_feishu_base_tables.py --action postgres
```
同步品牌库存到飞书：
```bash
python3 list_feishu_base_tables.py --action sync-brand
```
执行完整流程：
```bash
python3 list_feishu_base_tables.py --action all
```
同步目标配置在 `inventory_sync/specs.py`：
```python
BRAND_SYNC_TARGETS = [
BrandSyncTarget(brand_name="羽心堂", table_name="羽心堂", warehouse_keys=(2, 3, 7, 13)),
]
```
字段映射也集中维护在 `inventory_sync/specs.py` 的 `FIELD_SPECS`。
## 日志说明
项目使用结构化事件日志，格式：
写入表：
```text
YYYY-MM-DD HH:MM:SS LEVEL logger_name 事件=中文事件描述 中文字段=字段值 ...
raw_data.wangdian_batch_inventory
```
上传页日志输出到 stdout，典型事件包括：
### 6.2 渠道库存统计同步飞书
当前支持动作：
```text
事件=上传服务启动中
事件=收到上传请求
事件=已解析上传文件
事件=上传文件已保存
事件=开始解析库存文件
事件=已解析Excel工作表
事件=已加载Raw数据库配置
事件=Raw数据表确认完成
事件=Raw数据写入完成
事件=上传请求处理完成
feishu
postgres
sync-brand
all
```
飞书同步日志输出到 stderr，JSON 结果仍输出到 stdout，便于命令行管道消费。典型事件包括：
说明：
- 当前脚本名已经泛化为“渠道库存统计”，但内部动作参数 `sync-brand` 仍来自历史实现，后续可再统一。
- 当前默认同步目标仍是羽心堂渠道库存。
渠道库存统计目标和字段映射当前仍维护在：
- `inventory_sync/specs.py`
- `inventory_sync/sql.py`
### 6.3 批次库存统计报表
当前报表链路会：
1. 读取 `raw_data.wangdian_batch_inventory` 中最新批次数据
2. 提取商家编码
3. 查询 Analytics 中对应店铺可发库存、商品信息、近 30 天销售数量
4. 生成合并报表
5. 输出到 `reports/`
6. 可选同步到飞书 `批次库存` 表
说明：
- 当前命名已泛化为“批次库存统计”
- 当前默认业务规则仍针对羽心堂
默认输出文件名格式：
```text
事件=飞书同步入口启动
事件=飞书同步动作开始
事件=已加载环境变量文件
事件=开始读取飞书Base信息
事件=开始查询PostgreSQL
事件=开始同步品牌数据
事件=开始同步目标飞书表
事件=目标飞书表同步完成
事件=飞书同步入口完成
reports/批次库存统计_YYYYMMDD_HHMMSS.xlsx
```
## 查看和关闭上传服务
## 7. 字段同步策略
查看 8080 端口：
渠道库存统计和批次库存统计在同步飞书时，都采用相同策略：
1. 先读取当前飞书表字段
2. 判断当前字段结构是否与期望配置一致
3. 如果一致：只删除旧记录，再写入新记录
4. 如果不一致：删除受管字段并按配置重新创建，再写入新记录
## 8. 日志说明
日志统一采用结构化事件格式：
```text
YYYY-MM-DD HH:MM:SS LEVEL logger_name 事件=中文事件描述 中文字段=字段值
```
主要日志文件：
```text
logs/upload_inventory_page.log
logs/sync_channel_inventory.log
logs/generate_batch_inventory_report.log
```
说明：
- 上传页日志主要输出到 stdout，并写入 `upload_inventory_page.log`
- 渠道库存统计日志主要输出到 stderr，并写入 `sync_channel_inventory.log`
- 批次库存统计日志输出到 stdout，并写入 `generate_batch_inventory_report.log`
## 9. 推荐验证方式
验证渠道库存统计入口：
```bash
lsof -nP -iTCP:8080 -sTCP:LISTEN
python3 scripts/sync_channel_inventory.py --action feishu
```
关闭服务：
验证上传页：
```bash
kill <PID>
python3 scripts/upload_batch_inventory.py
```
如果普通关闭失败：
验证批次库存统计本地导出：
```bash
kill -9 <PID>
python3 scripts/generate_batch_inventory_report.py --skip-feishu-sync
```
## 验证命令
语法检查：
编译校验：
```bash
python3 -m py_compile upload_inventory_page.py list_feishu_base_tables.py inventory_sync/*.py
PYTHONPYCACHEPREFIX=/tmp/inventory_analytics_pycache python3 -m compileall inventory_analytics scripts generate_batch_inventory_report.py generate_yxt_batch_inventory_report.py sync_channel_inventory.py list_feishu_base_tables.py upload_inventory_page.py prefect_inventory_scheduler.py
```
查看飞书同步帮助：
## 10. Docker 与 Prefect 部署
项目已补齐以下部署文件：
```text
Dockerfile
docker-compose.yaml
docker-entrypoint.sh
supervisord.conf
prefect.yaml
prefect_inventory_scheduler.py
```
### 10.1 调度目标
当前通过 Prefect 定义了 2 个 deployment，都会在每天 `09:15`（`Asia/Shanghai`）执行：
- 渠道库存统计并写入飞书
- 批次库存统计报表并写入飞书
### 10.2 依赖前提
需要先准备：
- 可访问的 Prefect API
- 已创建的 Prefect work pool：`inventory-analytics-pool`
- 项目根目录 `.env`
如果 Prefect API 跑在宿主机本地，容器默认使用：
```text
http://host.docker.internal:4200/api
```
### 10.3 本地注册 deployment
```bash
python3 list_feishu_base_tables.py --help
prefect deploy -p prefect.yaml --all
```
本地启动上传页：
### 10.4 启动 Docker worker
```bash
python3 upload_inventory_page.py
docker compose up -d --build
```
## 维护原则
这会启动一个容器，容器内通过 `supervisord` 常驻运行：
- 上传 Raw 数据保持原始粒度，每行一条 payload，不在上传阶段做复杂业务清洗。
- 飞书同步使用 `inventory_sync/specs.py` 管理品牌目标和字段映射。
- SQL 调整优先修改 `inventory_sync/sql.py`。
- 入口脚本保持稳定，便于人工执行和后续定时任务接入。
- 日志新增时优先使用 `事件=<中文描述>`，并输出足够定位问题的关键字段。
```text
prefect worker start --pool "inventory-analytics-pool" --type process
```
### 10.5 手工触发 Prefect flow
立即执行渠道库存统计：
```bash
python3 prefect_inventory_scheduler.py --run-channel-inventory-sync
```
立即执行批次库存统计：
```bash
python3 prefect_inventory_scheduler.py --run-batch-inventory-report
```
### 10.6 本地 serve 模式
```bash
python3 prefect_inventory_scheduler.py --serve-channel-inventory-sync
python3 prefect_inventory_scheduler.py --serve-batch-inventory-report
```
### 10.7 调度入口说明
Prefect 调度脚本不会走 shell 子进程，而是直接调用当前项目入口逻辑：
- 渠道库存统计等价于 `python3 scripts/sync_channel_inventory.py --action all`
- 批次库存统计等价于 `python3 scripts/generate_batch_inventory_report.py`
## 11. 维护建议
- 新增功能优先落在 `inventory_analytics/` 包内，不再继续向根目录增加重脚本。
- 新命令优先使用 `scripts/` 目录中的入口。
- 渠道库存统计后续应继续把 `inventory_sync/` 中的实现迁入 `inventory_analytics/channel_inventory_sync/` 与 `inventory_analytics/feishu/`。
- 公共配置、日志、路径处理统一走 `config.py`、`logging_utils.py`、`paths.py`。
- 涉及渠道库存统计 SQL 和字段映射时，当前仍优先修改 `inventory_sync/sql.py` 与 `inventory_sync/specs.py`。
docker-compose.yaml
+33
View File
@@ -0,0 +1,33 @@
services:
prefect-inventory-worker:
build:
context: .
dockerfile: Dockerfile
args:
- UV_INDEX_URL=https://pypi.tuna.tsinghua.edu.cn/simple
image: inventory-analytics-prefect:0.1.0
container_name: prefect-inventory-worker
restart: unless-stopped
env_file:
- .env
environment:
- PYTHONUNBUFFERED=1
- PREFECT_API_URL=${PREFECT_API_URL:-http://host.docker.internal:4200/api}
extra_hosts:
- "host.docker.internal:host-gateway"
volumes:
- .:/app
- ./logs:/app/logs
- ./reports:/app/reports
- uv_cache:/root/.cache/uv
deploy:
resources:
limits:
cpus: "2.0"
memory: 4G
reservations:
cpus: "0.5"
memory: 512M
volumes:
uv_cache:
docker-entrypoint.sh
+14
View File
@@ -0,0 +1,14 @@
#!/bin/sh
set -eu
APP_DIR="/app"
if [ -f "$APP_DIR/.env" ]; then
set -a
. "$APP_DIR/.env"
set +a
fi
mkdir -p "$APP_DIR/logs" "$APP_DIR/reports"
exec "$@"
generate_batch_inventory_report.py
+8
View File
@@ -0,0 +1,8 @@
#!/usr/bin/env python3
"""Compatibility entrypoint for batch inventory report generation."""
from inventory_analytics.batch_inventory_report.cli import main
if __name__ == "__main__":
raise SystemExit(main())
generate_yxt_batch_inventory_report.py
+2 -679
View File
@@ -1,684 +1,7 @@
#!/usr/bin/env python3
"""Generate YXT batch inventory report by joining Raw batch stock with Analytics shop stock."""
"""Compatibility entrypoint for batch inventory report generation."""
from __future__ import annotations
import argparse
from dataclasses import dataclass
from datetime import datetime
import logging
from logging.handlers import RotatingFileHandler
from pathlib import Path
import re
import sys
import time
from typing import Any
import pandas as pd
import psycopg2
from inventory_sync.feishu import FeishuApiError, FeishuClient
from inventory_sync.settings import build_app_config
from inventory_sync.settings import load_env_candidates, require_config
PROJECT_ROOT = Path(__file__).resolve().parent
LOG_DIR = PROJECT_ROOT / "logs"
REPORT_DIR = PROJECT_ROOT / "reports"
LOG_FILE = LOG_DIR / "generate_yxt_batch_inventory_report.log"
LOGGER_NAME = "yxt_batch_inventory_report"
BATCH_FEISHU_TABLE_NAME = "批次库存"
SENSITIVE_PATTERN = re.compile(r"(?i)(password|passwd|secret|token|authorization)\s*[:=]\s*([^\s,;]+)")
RAW_BATCH_SQL = """
WITH latest_batch AS (
SELECT MAX(created_at) AS latest_created_at
FROM raw_data.wangdian_batch_inventory
),
latest_raw AS (
SELECT r.*
FROM raw_data.wangdian_batch_inventory r
JOIN latest_batch lb
ON r.created_at = lb.latest_created_at
)
SELECT
r.payload -> 'inventory_data' ->> '仓库' AS "仓库",
NULLIF(r.payload -> 'inventory_data' ->> '保质期', '')::integer AS "保质期",
r.payload -> 'inventory_data' ->> '批次号' AS "批次号",
r.payload -> 'inventory_data' ->> '规格码' AS "规格码",
NULLIF(r.payload -> 'inventory_data' ->> '冻结库存', '')::numeric(18,4) AS "冻结库存",
NULLIF(
r.payload -> 'inventory_data' ->> '创建时间',
'0000-00-00 00:00:00'
)::timestamp AS "创建时间",
r.payload -> 'inventory_data' ->> '商家编码' AS "商家编码",
NULLIF(r.payload -> 'inventory_data' ->> '库存日期', '0000-00-00')::date AS "库存日期",
NULLIF(r.payload -> 'inventory_data' ->> '次品标记', '')::integer AS "次品标记",
NULLIF(r.payload -> 'inventory_data' ->> '正常库存', '')::numeric(18,4) AS "正常库存",
NULLIF(
r.payload -> 'inventory_data' ->> '生产日期',
'0000-00-00 00:00:00'
)::timestamp AS "生产日期",
r.payload -> 'inventory_data' ->> '规格名称' AS "规格名称",
r.payload -> 'inventory_data' ->> '货品名称' AS "货品名称",
NULLIF(
r.payload -> 'inventory_data' ->> '过期日期',
'0000-00-00 00:00:00'
)::timestamp AS "过期日期",
NULLIF(r.payload -> 'inventory_data' ->> '距离到期天数', '')::integer AS "距离到期天数",
r.payload -> 'inventory_data' ->> '保质期计算方式' AS "保质期计算方式"
FROM latest_raw r
WHERE COALESCE(
r.payload -> 'inventory_data' ->> '货品名称',
''
) NOT LIKE '%测试%'
ORDER BY
r.payload -> 'inventory_data' ->> '商家编码',
r.payload -> 'inventory_data' ->> '仓库',
r.payload -> 'inventory_data' ->> '批次号';
"""
ANALYTICS_LOCK_SQL_TEMPLATE = """
WITH raw_codes AS (
SELECT code AS "统一规格编码"
FROM (
VALUES {code_values}
) AS t(code)
),
latest_shop_stock 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
AND TRIM(f."统一规格编码") IN (
SELECT "统一规格编码" FROM raw_codes
)
) t
WHERE t.rn = 1
AND t."状态" <> '已失效'
GROUP BY
t."统一规格编码",
t."店铺主键"
),
product_dedup AS (
SELECT DISTINCT ON (TRIM("统一规格编码"))
TRIM("统一规格编码") AS "统一规格编码",
"品牌名称",
"商品名称",
"一级分类"
FROM analytics_layer.dim_product
WHERE NULLIF(TRIM("统一规格编码"), '') IS NOT NULL
AND TRIM("统一规格编码") IN (
SELECT "统一规格编码" FROM raw_codes
)
ORDER BY TRIM("统一规格编码"), "商品主键" DESC
),
sales_30d AS (
SELECT
TRIM(fs."统一规格编码") AS "统一规格编码",
SUM(fs."销售数量") AS "近30天销售数量"
FROM analytics_layer.fact_sales fs
WHERE NULLIF(TRIM(fs."统一规格编码"), '') IS NOT NULL
AND TRIM(fs."统一规格编码") IN (
SELECT "统一规格编码" FROM raw_codes
)
AND fs."发货时间" >= now() - INTERVAL '30 days'
AND fs."发货时间" < now()
GROUP BY TRIM(fs."统一规格编码")
),
shop_pivot AS (
SELECT
l."统一规格编码",
SUM(CASE WHEN s."店铺名称" = '有赞_大眼睛买买买商店' THEN l."可发库存" ELSE 0 END) AS "有赞_大眼睛买买买商店",
SUM(CASE WHEN s."店铺名称" = '店铺-有赞-麦漾' THEN l."可发库存" ELSE 0 END) AS "店铺-有赞-麦漾",
SUM(CASE WHEN s."店铺名称" = '店铺-天猫-羽心堂旗舰店' THEN l."可发库存" ELSE 0 END) AS "店铺-天猫-羽心堂旗舰店",
SUM(CASE WHEN s."店铺名称" = '店铺-抖音-羽心堂旗舰店' THEN l."可发库存" ELSE 0 END) AS "店铺-抖音-羽心堂旗舰店"
FROM latest_shop_stock l
LEFT JOIN analytics_layer.dim_shop s
ON l."店铺主键" = s."店铺主键"
GROUP BY l."统一规格编码"
)
SELECT
rc."统一规格编码",
pd."品牌名称",
pd."商品名称",
pd."一级分类",
COALESCE(sp."有赞_大眼睛买买买商店", 0) AS "有赞_大眼睛买买买商店",
COALESCE(sp."店铺-有赞-麦漾", 0) AS "店铺-有赞-麦漾",
COALESCE(sp."店铺-天猫-羽心堂旗舰店", 0) AS "店铺-天猫-羽心堂旗舰店",
COALESCE(sp."店铺-抖音-羽心堂旗舰店", 0) AS "店铺-抖音-羽心堂旗舰店",
COALESCE(sa."近30天销售数量", 0) AS "近30天销售数量"
FROM raw_codes rc
LEFT JOIN product_dedup pd
ON rc."统一规格编码" = pd."统一规格编码"
LEFT JOIN shop_pivot sp
ON rc."统一规格编码" = sp."统一规格编码"
LEFT JOIN sales_30d sa
ON rc."统一规格编码" = sa."统一规格编码"
ORDER BY rc."统一规格编码";
"""
WAREHOUSE_FILTER = {
"杭州飞蚁仓",
"飞蚁-大眼睛仓",
"北京配思云仓",
"上海顺丰松江仓",
}
logger = logging.getLogger(LOGGER_NAME)
LOG_FIELD_LABELS = {
"available_fields": "可用字段",
"created_records": "新增记录数",
"database": "数据库",
"deleted_records": "删除记录数",
"elapsed_ms": "耗时毫秒",
"env_path": "环境文件",
"existing_records": "已有记录数",
"filtered_rows": "筛选后行数",
"join_key": "关联字段",
"log_file": "日志文件",
"matched_rows": "匹配行数",
"output": "输出文件",
"raw_rows": "Raw行数",
"report_rows": "报表行数",
"rows": "行数",
"table": "表",
"table_id": "表ID",
"table_name": "表名",
"unmatched_rows": "未匹配行数",
"warehouse_count": "仓库数量",
}
def sanitize_value(value: Any) -> Any:
if isinstance(value, str):
return SENSITIVE_PATTERN.sub(r"\1=***", value)
return value
def configure_logging() -> None:
LOG_DIR.mkdir(parents=True, exist_ok=True)
formatter = logging.Formatter(
fmt="%(asctime)s %(levelname)s %(name)s %(message)s",
datefmt="%Y-%m-%d %H:%M:%S",
)
stream_handler = logging.StreamHandler(sys.stdout)
stream_handler.setFormatter(formatter)
file_handler = RotatingFileHandler(LOG_FILE, maxBytes=5 * 1024 * 1024, backupCount=5, encoding="utf-8")
file_handler.setFormatter(formatter)
root_logger = logging.getLogger()
root_logger.handlers.clear()
root_logger.setLevel(logging.INFO)
root_logger.addHandler(stream_handler)
root_logger.addHandler(file_handler)
def log_event(level: int, event: str, **fields: Any) -> None:
safe_event = sanitize_value(event)
details = " ".join(
f"{LOG_FIELD_LABELS.get(key, key)}={sanitize_value(value)!r}" for key, value in fields.items()
)
logger.log(level, "事件=%s%s", safe_event, f" {details}" if details else "")
@dataclass(frozen=True)
class DbConfig:
host: str
port: str
user: str
password: str
name: str
@dataclass(frozen=True)
class FeishuFieldSpec:
field_name: str # 飞书字段名
source_name: str # DataFrame 列名
field_type: int
formatter: str | None = None
BATCH_FEISHU_PRIMARY_FIELD = "库存日期"
BATCH_FEISHU_FIELDS = [
FeishuFieldSpec("货品编码", "商家编码", 1),
FeishuFieldSpec("货品名称", "货品名称", 1),
FeishuFieldSpec("批次号", "批次号", 1),
FeishuFieldSpec("生产日期", "生产日期", 1),
FeishuFieldSpec("过期日期", "过期日期", 1),
FeishuFieldSpec("批次库存", "正常库存", 2, "0"),
FeishuFieldSpec("有赞_大眼睛买买买商店", "有赞_大眼睛买买买商店", 2, "0"),
FeishuFieldSpec("店铺-有赞-麦漾", "店铺-有赞-麦漾", 2, "0"),
FeishuFieldSpec("店铺-天猫-羽心堂旗舰店", "店铺-天猫-羽心堂旗舰店", 2, "0"),
FeishuFieldSpec("店铺-抖音-羽心堂旗舰店", "店铺-抖音-羽心堂旗舰店", 2, "0"),
FeishuFieldSpec("保质期", "保质期", 2, "0"),
FeishuFieldSpec("距离到期天数", "距离到期天数", 2, "0"),
FeishuFieldSpec("效期比例", "效期比例", 2, "0.00"),
FeishuFieldSpec("近30天销售数量", "近30天销售数量", 2, "0"),
FeishuFieldSpec("分类", "一级分类", 1),
]
def load_db_config(prefix: str) -> tuple[DbConfig, Path | None]:
config_map, env_path = load_env_candidates()
return (
DbConfig(
host=require_config(config_map, f"{prefix}_DB_HOST"),
port=require_config(config_map, f"{prefix}_DB_PORT"),
user=require_config(config_map, f"{prefix}_DB_USER"),
password=require_config(config_map, f"{prefix}_DB_PASSWORD"),
name=require_config(config_map, f"{prefix}_DB_NAME"),
),
env_path,
)
def format_feishu_value(column: str, value: Any) -> Any:
if pd.isna(value):
return None
if isinstance(value, pd.Timestamp):
if column == "库存日期":
return value.strftime("%Y-%m-%d")
return value.strftime("%Y-%m-%d %H:%M:%S")
if isinstance(value, datetime):
if column == "库存日期":
return value.strftime("%Y-%m-%d")
return value.strftime("%Y-%m-%d %H:%M:%S")
if column in {
"正常库存",
"有赞_大眼睛买买买商店",
"店铺-有赞-麦漾",
"店铺-天猫-羽心堂旗舰店",
"店铺-抖音-羽心堂旗舰店",
"近30天销售数量",
"保质期",
"距离到期天数",
"效期比例",
}:
return float(value)
return str(value)
def build_batch_feishu_records(df: pd.DataFrame) -> list[dict[str, Any]]:
records: list[dict[str, Any]] = []
for _, row in df.iterrows():
fields: dict[str, Any] = {
BATCH_FEISHU_PRIMARY_FIELD: format_feishu_value("库存日期", row.get("库存日期")),
}
for spec in BATCH_FEISHU_FIELDS:
fields[spec.field_name] = format_feishu_value(spec.source_name, row.get(spec.source_name))
records.append({"fields": fields})
return records
def ensure_batch_inventory_fields(
client: FeishuClient,
access_token: str,
app_token: str,
table_id: str,
) -> list[str]:
managed_names = {BATCH_FEISHU_PRIMARY_FIELD, *(field.field_name for field in BATCH_FEISHU_FIELDS)}
fields = client.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") == BATCH_FEISHU_PRIMARY_FIELD
),
None,
)
if duplicate_primary:
client.delete_field(access_token, app_token, table_id, duplicate_primary["field_id"])
time.sleep(0.2)
fields = client.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") != BATCH_FEISHU_PRIMARY_FIELD:
payload = {"field_name": BATCH_FEISHU_PRIMARY_FIELD, "type": primary_field.get("type")}
if primary_field.get("property") is not None:
payload["property"] = primary_field.get("property")
client.update_field(access_token, app_token, table_id, primary_field["field_id"], payload)
time.sleep(0.2)
fields = client.list_fields(access_token, app_token, table_id)
for field in fields:
field_name = field.get("field_name")
field_id = field.get("field_id")
if field.get("is_primary") or not field_name or not field_id:
continue
if field_name in managed_names:
client.delete_field(access_token, app_token, table_id, field_id)
time.sleep(0.2)
for spec in BATCH_FEISHU_FIELDS:
payload: dict[str, Any] = {"field_name": spec.field_name, "type": spec.field_type}
if spec.formatter is not None:
payload["property"] = {"formatter": spec.formatter}
client.create_field(access_token, app_token, table_id, payload)
time.sleep(0.2)
return [
field.get("field_name")
for field in client.list_fields(access_token, app_token, table_id)
if field.get("field_name")
]
def sync_report_to_feishu(df: pd.DataFrame, table_name: str = BATCH_FEISHU_TABLE_NAME) -> dict[str, Any]:
app_config = build_app_config()
client = FeishuClient(app_config)
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)
target_table = next((table for table in tables if table.get("name") == table_name), None)
if not target_table:
raise FeishuApiError(f"未找到目标飞书表: {table_name}")
table_id = target_table.get("table_id")
if not table_id:
raise FeishuApiError(f"目标飞书表缺少 table_id: {table_name}")
log_event(logging.INFO, "开始同步飞书批次库存表", table_name=table_name, table_id=table_id, report_rows=len(df))
available_fields = ensure_batch_inventory_fields(client, 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
)
records = build_batch_feishu_records(df)
created_count = (
client.batch_create_records(access_token, app_token, table_id, records)
if records else 0
)
log_event(
logging.INFO,
"飞书批次库存表同步完成",
table_name=table_name,
table_id=table_id,
existing_records=len(existing_record_ids),
deleted_records=deleted_count,
created_records=created_count,
available_fields=", ".join(available_fields),
)
return {
"table_name": table_name,
"table_id": table_id,
"available_fields": available_fields,
"existing_records": len(existing_record_ids),
"deleted_records": deleted_count,
"created_records": created_count,
}
def query_dataframe(db_config: DbConfig, sql: str, *, database_name: str) -> pd.DataFrame:
started_at = time.monotonic()
log_event(logging.INFO, "开始查询数据库", database=database_name)
with psycopg2.connect(
host=db_config.host,
port=db_config.port,
user=db_config.user,
password=db_config.password,
dbname=db_config.name,
) as conn:
with conn.cursor() as cur:
cur.execute("SET TIME ZONE 'Asia/Shanghai';")
df = pd.read_sql_query(sql, conn)
log_event(
logging.INFO,
"数据库查询完成",
database=database_name,
rows=len(df),
elapsed_ms=int((time.monotonic() - started_at) * 1000),
)
return df
def normalize_code(value: Any) -> str | None:
if value is None:
return None
text = str(value).strip()
return text or None
def quote_sql_literal(value: str) -> str:
return "'" + value.replace("'", "''") + "'"
def extract_raw_codes(df: pd.DataFrame) -> list[str]:
codes = (
df["商家编码"]
.map(normalize_code)
.dropna()
.drop_duplicates()
.tolist()
)
log_event(logging.INFO, "已提取Raw商家编码集合", rows=len(codes), join_key="商家编码=统一规格编码")
return codes
def build_analytics_lock_sql(codes: list[str]) -> str:
if not codes:
raise ValueError("RAW_BATCH_SQL 未查到可用于查询 Analytics 的商家编码")
code_values = ", ".join(f"({quote_sql_literal(code)})" for code in codes)
return ANALYTICS_LOCK_SQL_TEMPLATE.format(code_values=code_values)
def filter_batch_inventory(df: pd.DataFrame) -> pd.DataFrame:
filtered = df.copy()
filtered["货品名称"] = filtered["货品名称"].fillna("")
filtered["仓库"] = filtered["仓库"].fillna("")
filtered = filtered[
filtered["货品名称"].str.contains("羽心堂", na=False)
& filtered["仓库"].isin(WAREHOUSE_FILTER)
].copy()
filtered["关联编码"] = filtered["商家编码"].map(normalize_code)
missing_join_key = filtered["关联编码"].isna().sum()
if missing_join_key:
filtered.loc[filtered["关联编码"].isna(), "关联编码"] = filtered.loc[
filtered["关联编码"].isna(), "规格码"
].map(normalize_code)
log_event(
logging.INFO,
"Raw批次库存筛选完成",
raw_rows=len(df),
filtered_rows=len(filtered),
warehouse_count=len(WAREHOUSE_FILTER),
)
return filtered
def prepare_analytics_stock(df: pd.DataFrame) -> pd.DataFrame:
prepared = df.copy()
prepared["关联编码"] = prepared["统一规格编码"].map(normalize_code)
log_event(logging.INFO, "店铺锁定库存整理完成", rows=len(prepared), join_key="统一规格编码")
return prepared
def build_report(batch_df: pd.DataFrame, analytics_df: pd.DataFrame) -> pd.DataFrame:
merged = batch_df.merge(
analytics_df,
how="left",
on="关联编码",
suffixes=("", "_锁定库存"),
)
matched_rows = merged["统一规格编码"].notna().sum()
unmatched_rows = merged["统一规格编码"].isna().sum()
group_columns = [
"库存日期",
"商家编码",
"货品名称",
"一级分类",
"批次号",
"生产日期",
"过期日期",
"保质期",
"距离到期天数",
]
inventory_columns = [
"有赞_大眼睛买买买商店",
"店铺-有赞-麦漾",
"店铺-天猫-羽心堂旗舰店",
"店铺-抖音-羽心堂旗舰店",
]
aggregation_map: dict[str, str] = {"正常库存": "sum"}
for column in inventory_columns:
if column in merged.columns:
aggregation_map[column] = "max"
if "近30天销售数量" in merged.columns:
aggregation_map["近30天销售数量"] = "max"
grouped = (
merged.groupby(group_columns, dropna=False, as_index=False)
.agg(aggregation_map)
)
grouped["效期比例"] = grouped.apply(
lambda row: (
float(row["距离到期天数"]) / float(row["保质期"])
if pd.notna(row["距离到期天数"]) and pd.notna(row["保质期"]) and float(row["保质期"]) != 0
else None
),
axis=1,
)
report_columns = [
"库存日期",
"商家编码",
"货品名称",
"一级分类",
"批次号",
"生产日期",
"过期日期",
"正常库存",
"有赞_大眼睛买买买商店",
"店铺-有赞-麦漾",
"店铺-天猫-羽心堂旗舰店",
"店铺-抖音-羽心堂旗舰店",
"近30天销售数量",
"保质期",
"距离到期天数",
"效期比例",
]
available_columns = [column for column in report_columns if column in grouped.columns]
report = grouped[available_columns].sort_values(
by=["商家编码", "批次号", "生产日期"],
na_position="last",
).reset_index(drop=True)
log_event(
logging.INFO,
"批次库存与店铺锁定库存汇总完成",
report_rows=len(report),
matched_rows=int(matched_rows),
unmatched_rows=int(unmatched_rows),
join_key="关联编码",
)
return report
def save_report(df: pd.DataFrame, output_path: Path) -> None:
output_path.parent.mkdir(parents=True, exist_ok=True)
started_at = time.monotonic()
suffix = output_path.suffix.lower()
if suffix == ".csv":
df.to_csv(output_path, index=False, encoding="utf-8-sig")
else:
df.to_excel(output_path, index=False)
log_event(
logging.INFO,
"报表输出完成",
output=output_path,
report_rows=len(df),
elapsed_ms=int((time.monotonic() - started_at) * 1000),
)
def default_output_path() -> Path:
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
return REPORT_DIR / f"羽心堂批次库存与店铺锁定库存_{timestamp}.xlsx"
def parse_args() -> argparse.Namespace:
parser = argparse.ArgumentParser(description="生成羽心堂批次库存与店铺锁定库存合并报表")
parser.add_argument(
"--output",
default=str(default_output_path()),
help="输出文件路径，支持 .xlsx 或 .csv，默认写入 reports/ 目录",
)
parser.add_argument(
"--feishu-table",
default=BATCH_FEISHU_TABLE_NAME,
help="飞书多维表中的目标表名，默认 批次库存表",
)
parser.add_argument(
"--skip-feishu-sync",
action="store_true",
help="只生成本地报表，不同步到飞书",
)
return parser.parse_args()
def main() -> int:
configure_logging()
started_at = time.monotonic()
args = parse_args()
output_path = Path(args.output).expanduser().resolve()
raw_config, env_path = load_db_config("RAW")
analytics_config, _ = load_db_config("ANALYTICS")
log_event(
logging.INFO,
"羽心堂批次库存报表任务启动",
env_path=env_path,
output=output_path,
log_file=LOG_FILE,
)
raw_df = query_dataframe(raw_config, RAW_BATCH_SQL, database_name="Raw")
raw_codes = extract_raw_codes(raw_df)
analytics_sql = build_analytics_lock_sql(raw_codes)
analytics_df = query_dataframe(analytics_config, analytics_sql, database_name="Analytics")
filtered_batch_df = filter_batch_inventory(raw_df)
prepared_analytics_df = prepare_analytics_stock(analytics_df)
report_df = build_report(filtered_batch_df, prepared_analytics_df)
save_report(report_df, output_path)
if args.skip_feishu_sync:
log_event(logging.INFO, "已跳过飞书批次库存表同步", table_name=args.feishu_table)
else:
sync_report_to_feishu(report_df, table_name=args.feishu_table)
log_event(
logging.INFO,
"羽心堂批次库存报表任务完成",
output=output_path,
report_rows=len(report_df),
elapsed_ms=int((time.monotonic() - started_at) * 1000),
)
return 0
from inventory_analytics.batch_inventory_report.cli import main
if __name__ == "__main__":
inventory_analytics/__init__.py
+2
View File
@@ -0,0 +1,2 @@
"""Inventory analytics application package."""
inventory_analytics/batch_inventory_report/__init__.py
+1
View File
@@ -0,0 +1 @@
"""Batch inventory report workflow."""
inventory_analytics/batch_inventory_report/cli.py
+3
View File
@@ -0,0 +1,3 @@
from .service import main
__all__ = ["main"]
inventory_analytics/batch_inventory_report/service.py
+679
View File
@@ -0,0 +1,679 @@
#!/usr/bin/env python3
"""Generate batch inventory report by joining Raw batch stock with Analytics shop stock."""
from __future__ import annotations
import argparse
from datetime import datetime
import logging
from pathlib import Path
import time
from typing import Any
import pandas as pd
import psycopg2
from ..config import DbConfig, build_app_config, load_analytics_db_config, load_raw_db_config
from ..feishu.client import FeishuApiError, FeishuClient
from ..logging_utils import configure_logging as configure_app_logging
from ..logging_utils import log_event as emit_log_event
from ..paths import LOG_DIR, REPORT_DIR
LOG_FILE = LOG_DIR / "generate_batch_inventory_report.log"
LOGGER_NAME = "batch_inventory_report"
BATCH_FEISHU_TABLE_NAME = "批次库存"
RAW_BATCH_SQL = """
WITH latest_batch AS (
SELECT MAX(created_at) AS latest_created_at
FROM raw_data.wangdian_batch_inventory
),
latest_raw AS (
SELECT r.*
FROM raw_data.wangdian_batch_inventory r
JOIN latest_batch lb
ON r.created_at = lb.latest_created_at
)
SELECT
r.payload -> 'inventory_data' ->> '仓库' AS "仓库",
NULLIF(r.payload -> 'inventory_data' ->> '保质期', '')::integer AS "保质期",
r.payload -> 'inventory_data' ->> '批次号' AS "批次号",
r.payload -> 'inventory_data' ->> '规格码' AS "规格码",
NULLIF(r.payload -> 'inventory_data' ->> '冻结库存', '')::numeric(18,4) AS "冻结库存",
NULLIF(
r.payload -> 'inventory_data' ->> '创建时间',
'0000-00-00 00:00:00'
)::timestamp AS "创建时间",
r.payload -> 'inventory_data' ->> '商家编码' AS "商家编码",
NULLIF(r.payload -> 'inventory_data' ->> '库存日期', '0000-00-00')::date AS "库存日期",
NULLIF(r.payload -> 'inventory_data' ->> '次品标记', '')::integer AS "次品标记",
NULLIF(r.payload -> 'inventory_data' ->> '正常库存', '')::numeric(18,4) AS "正常库存",
NULLIF(
r.payload -> 'inventory_data' ->> '生产日期',
'0000-00-00 00:00:00'
)::timestamp AS "生产日期",
r.payload -> 'inventory_data' ->> '规格名称' AS "规格名称",
r.payload -> 'inventory_data' ->> '货品名称' AS "货品名称",
NULLIF(
r.payload -> 'inventory_data' ->> '过期日期',
'0000-00-00 00:00:00'
)::timestamp AS "过期日期",
NULLIF(r.payload -> 'inventory_data' ->> '距离到期天数', '')::integer AS "距离到期天数",
r.payload -> 'inventory_data' ->> '保质期计算方式' AS "保质期计算方式"
FROM latest_raw r
WHERE COALESCE(
r.payload -> 'inventory_data' ->> '货品名称',
''
) NOT LIKE '%测试%'
ORDER BY
r.payload -> 'inventory_data' ->> '商家编码',
r.payload -> 'inventory_data' ->> '仓库',
r.payload -> 'inventory_data' ->> '批次号';
"""
ANALYTICS_LOCK_SQL_TEMPLATE = """
WITH raw_codes AS (
SELECT code AS "统一规格编码"
FROM (
VALUES {code_values}
) AS t(code)
),
latest_shop_stock 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
AND TRIM(f."统一规格编码") IN (
SELECT "统一规格编码" FROM raw_codes
)
) t
WHERE t.rn = 1
AND t."状态" <> '已失效'
GROUP BY
t."统一规格编码",
t."店铺主键"
),
product_dedup AS (
SELECT DISTINCT ON (TRIM("统一规格编码"))
TRIM("统一规格编码") AS "统一规格编码",
"品牌名称",
"商品名称",
"一级分类"
FROM analytics_layer.dim_product
WHERE NULLIF(TRIM("统一规格编码"), '') IS NOT NULL
AND TRIM("统一规格编码") IN (
SELECT "统一规格编码" FROM raw_codes
)
ORDER BY TRIM("统一规格编码"), "商品主键" DESC
),
sales_30d AS (
SELECT
TRIM(fs."统一规格编码") AS "统一规格编码",
SUM(fs."销售数量") AS "近30天销售数量"
FROM analytics_layer.fact_sales fs
WHERE NULLIF(TRIM(fs."统一规格编码"), '') IS NOT NULL
AND TRIM(fs."统一规格编码") IN (
SELECT "统一规格编码" FROM raw_codes
)
AND fs."发货时间" >= now() - INTERVAL '30 days'
AND fs."发货时间" < now()
GROUP BY TRIM(fs."统一规格编码")
),
shop_pivot AS (
SELECT
l."统一规格编码",
SUM(CASE WHEN s."店铺名称" = '有赞_大眼睛买买买商店' THEN l."可发库存" ELSE 0 END) AS "有赞_大眼睛买买买商店",
SUM(CASE WHEN s."店铺名称" = '店铺-有赞-麦漾' THEN l."可发库存" ELSE 0 END) AS "店铺-有赞-麦漾",
SUM(CASE WHEN s."店铺名称" = '店铺-天猫-羽心堂旗舰店' THEN l."可发库存" ELSE 0 END) AS "店铺-天猫-羽心堂旗舰店",
SUM(CASE WHEN s."店铺名称" = '店铺-抖音-羽心堂旗舰店' THEN l."可发库存" ELSE 0 END) AS "店铺-抖音-羽心堂旗舰店"
FROM latest_shop_stock l
LEFT JOIN analytics_layer.dim_shop s
ON l."店铺主键" = s."店铺主键"
GROUP BY l."统一规格编码"
)
SELECT
rc."统一规格编码",
pd."品牌名称",
pd."商品名称",
pd."一级分类",
COALESCE(sp."有赞_大眼睛买买买商店", 0) AS "有赞_大眼睛买买买商店",
COALESCE(sp."店铺-有赞-麦漾", 0) AS "店铺-有赞-麦漾",
COALESCE(sp."店铺-天猫-羽心堂旗舰店", 0) AS "店铺-天猫-羽心堂旗舰店",
COALESCE(sp."店铺-抖音-羽心堂旗舰店", 0) AS "店铺-抖音-羽心堂旗舰店",
COALESCE(sa."近30天销售数量", 0) AS "近30天销售数量"
FROM raw_codes rc
LEFT JOIN product_dedup pd
ON rc."统一规格编码" = pd."统一规格编码"
LEFT JOIN shop_pivot sp
ON rc."统一规格编码" = sp."统一规格编码"
LEFT JOIN sales_30d sa
ON rc."统一规格编码" = sa."统一规格编码"
ORDER BY rc."统一规格编码";
"""
WAREHOUSE_FILTER = {
"杭州飞蚁仓",
"飞蚁-大眼睛仓",
"北京配思云仓",
"上海顺丰松江仓",
}
logger = logging.getLogger(LOGGER_NAME)
LOG_FIELD_LABELS = {
"available_fields": "可用字段",
"created_records": "新增记录数",
"database": "数据库",
"deleted_records": "删除记录数",
"elapsed_ms": "耗时毫秒",
"env_path": "环境文件",
"existing_records": "已有记录数",
"filtered_rows": "筛选后行数",
"join_key": "关联字段",
"log_file": "日志文件",
"matched_rows": "匹配行数",
"output": "输出文件",
"raw_rows": "Raw行数",
"report_rows": "报表行数",
"rows": "行数",
"table": "表",
"table_id": "表ID",
"table_name": "表名",
"unmatched_rows": "未匹配行数",
"warehouse_count": "仓库数量",
}
def configure_logging() -> None:
configure_app_logging(LOG_FILE)
def log_event(level: int, event: str, **fields: Any) -> None:
emit_log_event(logger, LOG_FIELD_LABELS, level, event, **fields)
@dataclass(frozen=True)
class FeishuFieldSpec:
field_name: str # 飞书字段名
source_name: str # DataFrame 列名
field_type: int
formatter: str | None = None
BATCH_FEISHU_PRIMARY_FIELD = "库存日期"
BATCH_FEISHU_FIELDS = [
FeishuFieldSpec("货品编码", "商家编码", 1),
FeishuFieldSpec("货品名称", "货品名称", 1),
FeishuFieldSpec("批次号", "批次号", 1),
FeishuFieldSpec("生产日期", "生产日期", 1),
FeishuFieldSpec("过期日期", "过期日期", 1),
FeishuFieldSpec("批次库存", "正常库存", 2, "0"),
FeishuFieldSpec("有赞_大眼睛买买买商店", "有赞_大眼睛买买买商店", 2, "0"),
FeishuFieldSpec("店铺-有赞-麦漾", "店铺-有赞-麦漾", 2, "0"),
FeishuFieldSpec("店铺-天猫-羽心堂旗舰店", "店铺-天猫-羽心堂旗舰店", 2, "0"),
FeishuFieldSpec("店铺-抖音-羽心堂旗舰店", "店铺-抖音-羽心堂旗舰店", 2, "0"),
FeishuFieldSpec("保质期", "保质期", 2, "0"),
FeishuFieldSpec("距离到期天数", "距离到期天数", 2, "0"),
FeishuFieldSpec("效期比例", "效期比例", 2, "%"),
FeishuFieldSpec("近30天销售数量", "近30天销售数量", 2, "0"),
FeishuFieldSpec("分类", "一级分类", 1),
]
def format_feishu_value(column: str, value: Any) -> Any:
if pd.isna(value):
return None
if isinstance(value, pd.Timestamp):
if column == "库存日期":
return value.strftime("%Y-%m-%d")
return value.strftime("%Y-%m-%d %H:%M:%S")
if isinstance(value, datetime):
if column == "库存日期":
return value.strftime("%Y-%m-%d")
return value.strftime("%Y-%m-%d %H:%M:%S")
if column in {
"正常库存",
"有赞_大眼睛买买买商店",
"店铺-有赞-麦漾",
"店铺-天猫-羽心堂旗舰店",
"店铺-抖音-羽心堂旗舰店",
"近30天销售数量",
"保质期",
"距离到期天数",
"效期比例",
}:
return float(value)
return str(value)
def build_batch_feishu_records(df: pd.DataFrame) -> list[dict[str, Any]]:
records: list[dict[str, Any]] = []
for _, row in df.iterrows():
fields: dict[str, Any] = {
BATCH_FEISHU_PRIMARY_FIELD: format_feishu_value("库存日期", row.get("库存日期")),
}
for spec in BATCH_FEISHU_FIELDS:
fields[spec.field_name] = format_feishu_value(spec.source_name, row.get(spec.source_name))
records.append({"fields": fields})
return records
def build_feishu_field_payload(spec: FeishuFieldSpec) -> dict[str, Any]:
payload: dict[str, Any] = {"field_name": spec.field_name, "type": spec.field_type}
if spec.formatter is not None:
payload["property"] = {"formatter": spec.formatter}
return payload
def feishu_field_matches_spec(field: dict[str, Any], spec: FeishuFieldSpec) -> bool:
if field.get("type") != spec.field_type:
return False
if spec.formatter is None:
return True
return (field.get("property") or {}).get("formatter") == spec.formatter
def ensure_batch_inventory_fields(
client: FeishuClient,
access_token: str,
app_token: str,
table_id: str,
) -> list[str]:
fields = client.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 = {field.field_name for field in BATCH_FEISHU_FIELDS}
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
}
fields_by_name = {
field.get("field_name"): field
for field in fields
if not field.get("is_primary") and field.get("field_name") in expected_field_names
}
primary_matches = primary_field.get("field_name") == BATCH_FEISHU_PRIMARY_FIELD
fields_match = actual_managed_field_names == expected_field_names
field_specs_match = all(
feishu_field_matches_spec(fields_by_name.get(spec.field_name) or {}, spec)
for spec in BATCH_FEISHU_FIELDS
)
if primary_matches and fields_match and field_specs_match:
return [field.get("field_name") for field in fields if field.get("field_name")]
if primary_matches and fields_match:
for spec in BATCH_FEISHU_FIELDS:
field = fields_by_name.get(spec.field_name)
if not field or feishu_field_matches_spec(field, spec):
continue
client.update_field(
access_token,
app_token,
table_id,
field["field_id"],
build_feishu_field_payload(spec),
)
time.sleep(0.2)
return [
field.get("field_name")
for field in client.list_fields(access_token, app_token, table_id)
if field.get("field_name")
]
duplicate_primary = next(
(
field for field in fields
if not field.get("is_primary") and field.get("field_name") == BATCH_FEISHU_PRIMARY_FIELD
),
None,
)
if duplicate_primary:
client.delete_field(access_token, app_token, table_id, duplicate_primary["field_id"])
time.sleep(0.2)
fields = client.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") != BATCH_FEISHU_PRIMARY_FIELD:
payload = {"field_name": BATCH_FEISHU_PRIMARY_FIELD, "type": primary_field.get("type")}
if primary_field.get("property") is not None:
payload["property"] = primary_field.get("property")
client.update_field(access_token, app_token, table_id, primary_field["field_id"], payload)
time.sleep(0.2)
fields = client.list_fields(access_token, app_token, table_id)
for field in fields:
field_name = field.get("field_name")
field_id = field.get("field_id")
if field.get("is_primary") or not field_name or not field_id:
continue
if field_name in expected_field_names:
client.delete_field(access_token, app_token, table_id, field_id)
time.sleep(0.2)
for spec in BATCH_FEISHU_FIELDS:
client.create_field(access_token, app_token, table_id, build_feishu_field_payload(spec))
time.sleep(0.2)
return [
field.get("field_name")
for field in client.list_fields(access_token, app_token, table_id)
if field.get("field_name")
]
def sync_report_to_feishu(df: pd.DataFrame, table_name: str = BATCH_FEISHU_TABLE_NAME) -> dict[str, Any]:
app_config = build_app_config()
client = FeishuClient(app_config)
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)
target_table = next((table for table in tables if table.get("name") == table_name), None)
if not target_table:
raise FeishuApiError(f"未找到目标飞书表: {table_name}")
table_id = target_table.get("table_id")
if not table_id:
raise FeishuApiError(f"目标飞书表缺少 table_id: {table_name}")
log_event(logging.INFO, "开始同步飞书批次库存表", table_name=table_name, table_id=table_id, report_rows=len(df))
available_fields = ensure_batch_inventory_fields(client, 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
)
records = build_batch_feishu_records(df)
created_count = (
client.batch_create_records(access_token, app_token, table_id, records)
if records else 0
)
log_event(
logging.INFO,
"飞书批次库存表同步完成",
table_name=table_name,
table_id=table_id,
existing_records=len(existing_record_ids),
deleted_records=deleted_count,
created_records=created_count,
available_fields=", ".join(available_fields),
)
return {
"table_name": table_name,
"table_id": table_id,
"available_fields": available_fields,
"existing_records": len(existing_record_ids),
"deleted_records": deleted_count,
"created_records": created_count,
}
def query_dataframe(db_config: DbConfig, sql: str, *, database_name: str) -> pd.DataFrame:
started_at = time.monotonic()
log_event(logging.INFO, "开始查询数据库", database=database_name)
with psycopg2.connect(
host=db_config.host,
port=db_config.port,
user=db_config.user,
password=db_config.password,
dbname=db_config.name,
) as conn:
with conn.cursor() as cur:
cur.execute("SET TIME ZONE 'Asia/Shanghai';")
df = pd.read_sql_query(sql, conn)
log_event(
logging.INFO,
"数据库查询完成",
database=database_name,
rows=len(df),
elapsed_ms=int((time.monotonic() - started_at) * 1000),
)
return df
def normalize_code(value: Any) -> str | None:
if value is None:
return None
text = str(value).strip()
return text or None
def quote_sql_literal(value: str) -> str:
return "'" + value.replace("'", "''") + "'"
def extract_raw_codes(df: pd.DataFrame) -> list[str]:
codes = (
df["商家编码"]
.map(normalize_code)
.dropna()
.drop_duplicates()
.tolist()
)
log_event(logging.INFO, "已提取Raw商家编码集合", rows=len(codes), join_key="商家编码=统一规格编码")
return codes
def build_analytics_lock_sql(codes: list[str]) -> str:
if not codes:
raise ValueError("RAW_BATCH_SQL 未查到可用于查询 Analytics 的商家编码")
code_values = ", ".join(f"({quote_sql_literal(code)})" for code in codes)
return ANALYTICS_LOCK_SQL_TEMPLATE.format(code_values=code_values)
def filter_batch_inventory(df: pd.DataFrame) -> pd.DataFrame:
filtered = df.copy()
filtered["货品名称"] = filtered["货品名称"].fillna("")
filtered["仓库"] = filtered["仓库"].fillna("")
filtered = filtered[
filtered["货品名称"].str.contains("羽心堂", na=False)
& filtered["仓库"].isin(WAREHOUSE_FILTER)
].copy()
filtered["关联编码"] = filtered["商家编码"].map(normalize_code)
missing_join_key = filtered["关联编码"].isna().sum()
if missing_join_key:
filtered.loc[filtered["关联编码"].isna(), "关联编码"] = filtered.loc[
filtered["关联编码"].isna(), "规格码"
].map(normalize_code)
log_event(
logging.INFO,
"Raw批次库存筛选完成",
raw_rows=len(df),
filtered_rows=len(filtered),
warehouse_count=len(WAREHOUSE_FILTER),
)
return filtered
def prepare_analytics_stock(df: pd.DataFrame) -> pd.DataFrame:
prepared = df.copy()
prepared["关联编码"] = prepared["统一规格编码"].map(normalize_code)
log_event(logging.INFO, "店铺锁定库存整理完成", rows=len(prepared), join_key="统一规格编码")
return prepared
def build_report(batch_df: pd.DataFrame, analytics_df: pd.DataFrame) -> pd.DataFrame:
merged = batch_df.merge(
analytics_df,
how="left",
on="关联编码",
suffixes=("", "_锁定库存"),
)
matched_rows = merged["统一规格编码"].notna().sum()
unmatched_rows = merged["统一规格编码"].isna().sum()
group_columns = [
"库存日期",
"商家编码",
"货品名称",
"一级分类",
"批次号",
"生产日期",
"过期日期",
"保质期",
"距离到期天数",
]
inventory_columns = [
"有赞_大眼睛买买买商店",
"店铺-有赞-麦漾",
"店铺-天猫-羽心堂旗舰店",
"店铺-抖音-羽心堂旗舰店",
]
aggregation_map: dict[str, str] = {"正常库存": "sum"}
for column in inventory_columns:
if column in merged.columns:
aggregation_map[column] = "max"
if "近30天销售数量" in merged.columns:
aggregation_map["近30天销售数量"] = "max"
grouped = (
merged.groupby(group_columns, dropna=False, as_index=False)
.agg(aggregation_map)
)
grouped["效期比例"] = grouped.apply(
lambda row: (
float(row["距离到期天数"]) / float(row["保质期"])
if pd.notna(row["距离到期天数"]) and pd.notna(row["保质期"]) and float(row["保质期"]) != 0
else None
),
axis=1,
)
report_columns = [
"库存日期",
"商家编码",
"货品名称",
"一级分类",
"批次号",
"生产日期",
"过期日期",
"正常库存",
"有赞_大眼睛买买买商店",
"店铺-有赞-麦漾",
"店铺-天猫-羽心堂旗舰店",
"店铺-抖音-羽心堂旗舰店",
"近30天销售数量",
"保质期",
"距离到期天数",
"效期比例",
]
available_columns = [column for column in report_columns if column in grouped.columns]
report = grouped[available_columns].sort_values(
by=["商家编码", "批次号", "生产日期"],
na_position="last",
).reset_index(drop=True)
log_event(
logging.INFO,
"批次库存与店铺锁定库存汇总完成",
report_rows=len(report),
matched_rows=int(matched_rows),
unmatched_rows=int(unmatched_rows),
join_key="关联编码",
)
return report
def save_report(df: pd.DataFrame, output_path: Path) -> None:
output_path.parent.mkdir(parents=True, exist_ok=True)
started_at = time.monotonic()
suffix = output_path.suffix.lower()
if suffix == ".csv":
df.to_csv(output_path, index=False, encoding="utf-8-sig")
else:
df.to_excel(output_path, index=False)
log_event(
logging.INFO,
"报表输出完成",
output=output_path,
report_rows=len(df),
elapsed_ms=int((time.monotonic() - started_at) * 1000),
)
def default_output_path() -> Path:
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
return REPORT_DIR / f"批次库存统计_{timestamp}.xlsx"
def parse_args() -> argparse.Namespace:
parser = argparse.ArgumentParser(description="生成批次库存统计合并报表，当前默认按羽心堂业务规则执行")
parser.add_argument(
"--output",
default=str(default_output_path()),
help="输出文件路径，支持 .xlsx 或 .csv，默认写入 reports/ 目录",
)
parser.add_argument(
"--feishu-table",
default=BATCH_FEISHU_TABLE_NAME,
help="飞书多维表中的目标表名，默认 批次库存表",
)
parser.add_argument(
"--skip-feishu-sync",
action="store_true",
help="只生成本地报表，不同步到飞书",
)
return parser.parse_args()
def main() -> int:
configure_logging()
started_at = time.monotonic()
args = parse_args()
output_path = Path(args.output).expanduser().resolve()
raw_config, env_path = load_raw_db_config()
analytics_config, _ = load_analytics_db_config()
log_event(
logging.INFO,
"批次库存统计任务启动",
env_path=env_path,
output=output_path,
log_file=LOG_FILE,
)
raw_df = query_dataframe(raw_config, RAW_BATCH_SQL, database_name="Raw")
raw_codes = extract_raw_codes(raw_df)
analytics_sql = build_analytics_lock_sql(raw_codes)
analytics_df = query_dataframe(analytics_config, analytics_sql, database_name="Analytics")
filtered_batch_df = filter_batch_inventory(raw_df)
prepared_analytics_df = prepare_analytics_stock(analytics_df)
report_df = build_report(filtered_batch_df, prepared_analytics_df)
save_report(report_df, output_path)
if args.skip_feishu_sync:
log_event(logging.INFO, "已跳过飞书批次库存表同步", table_name=args.feishu_table)
else:
sync_report_to_feishu(report_df, table_name=args.feishu_table)
log_event(
logging.INFO,
"批次库存统计任务完成",
output=output_path,
report_rows=len(report_df),
elapsed_ms=int((time.monotonic() - started_at) * 1000),
)
return 0
if __name__ == "__main__":
raise SystemExit(main())
inventory_analytics/batch_upload/__init__.py
+2
View File
@@ -0,0 +1,2 @@
"""Batch inventory upload workflow."""
inventory_analytics/batch_upload/web.py
+1028
View File
File diff suppressed because it is too large Load Diff
inventory_analytics/channel_inventory_sync/__init__.py
+1
View File
@@ -0,0 +1 @@
"""Channel inventory sync workflows."""
inventory_analytics/channel_inventory_sync/cli.py
+3
View File
@@ -0,0 +1,3 @@
from inventory_sync.cli import main
__all__ = ["main"]
inventory_analytics/channel_inventory_sync/entrypoint.py
+54
View File
@@ -0,0 +1,54 @@
from __future__ import annotations
import logging
from pathlib import Path
import sys
import time
from .cli import main as channel_inventory_sync_main
from ..logging_utils import configure_logging, log_event
from ..paths import LOG_DIR
LOGGER_NAME = "inventory_analytics.channel_inventory_sync.entrypoint"
LOG_FILE = LOG_DIR / "sync_channel_inventory.log"
logger = logging.getLogger(LOGGER_NAME)
LOG_FIELD_LABELS = {
"argv": "参数",
"cwd": "工作目录",
"elapsed_ms": "耗时毫秒",
"exit_code": "退出码",
"log_file": "日志文件",
}
def main() -> int:
configure_logging(LOG_FILE, stream=sys.stderr)
started_at = time.monotonic()
log_event(
logger,
LOG_FIELD_LABELS,
logging.INFO,
"飞书同步入口启动",
argv=sys.argv[1:],
cwd=Path.cwd(),
log_file=LOG_FILE,
)
try:
exit_code = channel_inventory_sync_main()
except SystemExit as exc:
exit_code = int(exc.code or 0) if isinstance(exc.code, int) else 1
except Exception:
logger.exception("事件=%s", "飞书同步入口异常")
return 1
log_event(
logger,
LOG_FIELD_LABELS,
logging.INFO if exit_code == 0 else logging.ERROR,
"飞书同步入口完成",
exit_code=exit_code,
elapsed_ms=int((time.monotonic() - started_at) * 1000),
)
return exit_code
inventory_analytics/channel_inventory_sync/queries.py
+3
View File
@@ -0,0 +1,3 @@
from inventory_sync.sql import DEFAULT_SYNC_TARGET, POSTGRES_QUERY, build_postgres_query
__all__ = ["DEFAULT_SYNC_TARGET", "POSTGRES_QUERY", "build_postgres_query"]
inventory_analytics/channel_inventory_sync/service.py
+3
View File
@@ -0,0 +1,3 @@
from inventory_sync.service import build_feishu_record, run_feishu_summary, run_sync_brand, sync_brand_target
__all__ = ["build_feishu_record", "run_feishu_summary", "run_sync_brand", "sync_brand_target"]
inventory_analytics/channel_inventory_sync/specs.py
+21
View File
@@ -0,0 +1,21 @@
from inventory_sync.specs import (
BRAND_SYNC_TARGETS,
FIELD_SPECS,
LEGACY_NAME_MAP,
PRIMARY_FIELD_NAME,
PRIMARY_FIELD_SOURCE,
BrandSyncTarget,
FieldSpec,
month_sales_field_name,
)
__all__ = [
"BRAND_SYNC_TARGETS",
"FIELD_SPECS",
"LEGACY_NAME_MAP",
"PRIMARY_FIELD_NAME",
"PRIMARY_FIELD_SOURCE",
"BrandSyncTarget",
"FieldSpec",
"month_sales_field_name",
]
inventory_analytics/config.py
+117
View File
@@ -0,0 +1,117 @@
from __future__ import annotations
from dataclasses import dataclass
import os
from pathlib import Path
from .paths import PROJECT_ROOT
@dataclass(frozen=True)
class DbConfig:
host: str
port: str
user: str
password: str
name: str
@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]:
candidates = [
PROJECT_ROOT / ".env",
PROJECT_ROOT / "project (1)" / ".env",
PROJECT_ROOT.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 load_db_config(prefix: str) -> tuple[DbConfig, Path | None]:
config_map, env_path = load_env_candidates()
return (
DbConfig(
host=require_config(config_map, f"{prefix}_DB_HOST"),
port=require_config(config_map, f"{prefix}_DB_PORT"),
user=require_config(config_map, f"{prefix}_DB_USER"),
password=require_config(config_map, f"{prefix}_DB_PASSWORD"),
name=require_config(config_map, f"{prefix}_DB_NAME"),
),
env_path,
)
def load_raw_db_config() -> tuple[DbConfig, Path | None]:
return load_db_config("RAW")
def load_analytics_db_config() -> tuple[DbConfig, Path | None]:
return load_db_config("ANALYTICS")
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_analytics/db/__init__.py
+2
View File
@@ -0,0 +1,2 @@
"""Database helpers."""
inventory_analytics/feishu/__init__.py
+2
View File
@@ -0,0 +1,2 @@
"""Feishu integrations."""
inventory_analytics/feishu/client.py
+4
View File
@@ -0,0 +1,4 @@
from inventory_sync.feishu import FeishuApiError, FeishuClient
__all__ = ["FeishuApiError", "FeishuClient"]
inventory_analytics/logging_utils.py
+43
View File
@@ -0,0 +1,43 @@
from __future__ import annotations
import logging
from logging.handlers import RotatingFileHandler
from pathlib import Path
import re
import sys
from typing import Any
SENSITIVE_PATTERN = re.compile(r"(?i)(password|passwd|secret|token|authorization)\s*[:=]\s*([^\s,;]+)")
def sanitize_value(value: Any) -> Any:
if isinstance(value, str):
return SENSITIVE_PATTERN.sub(r"\1=***", value)
return value
def configure_logging(log_file: Path, *, stream = None, level: int = logging.INFO) -> None:
log_file.parent.mkdir(parents=True, exist_ok=True)
formatter = logging.Formatter(
fmt="%(asctime)s %(levelname)s %(name)s %(message)s",
datefmt="%Y-%m-%d %H:%M:%S",
)
stream_handler = logging.StreamHandler(stream or sys.stdout)
stream_handler.setFormatter(formatter)
file_handler = RotatingFileHandler(log_file, maxBytes=5 * 1024 * 1024, backupCount=5, encoding="utf-8")
file_handler.setFormatter(formatter)
root_logger = logging.getLogger()
root_logger.handlers.clear()
root_logger.setLevel(level)
root_logger.addHandler(stream_handler)
root_logger.addHandler(file_handler)
def log_event(logger: logging.Logger, labels: dict[str, str], level: int, event: str, **fields: Any) -> None:
safe_event = sanitize_value(event)
details = " ".join(
f"{labels.get(key, key)}={sanitize_value(value)!r}" for key, value in fields.items()
)
logger.log(level, "事件=%s%s", safe_event, f" {details}" if details else "")
inventory_analytics/paths.py
+11
View File
@@ -0,0 +1,11 @@
from __future__ import annotations
from pathlib import Path
PACKAGE_ROOT = Path(__file__).resolve().parent
PROJECT_ROOT = PACKAGE_ROOT.parent
LOG_DIR = PROJECT_ROOT / "logs"
REPORT_DIR = PROJECT_ROOT / "reports"
BATCH_UPLOAD_DIR = PROJECT_ROOT / "batch_inventory_data"
inventory_sync/feishu.py
+10
View File
@@ -206,6 +206,16 @@ class FeishuClient:
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}
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
}
primary_matches = primary_field.get("field_name") == PRIMARY_FIELD_NAME
fields_match = actual_managed_field_names == expected_field_names
if primary_matches and fields_match:
return [field.get("field_name") for field in fields if field.get("field_name")]
duplicate_primary = next(
(field for field in fields if not field.get("is_primary") and field.get("field_name") == PRIMARY_FIELD_NAME),
list_feishu_base_tables.py
+2 -99
View File
@@ -1,104 +1,7 @@
#!/usr/bin/env python3
"""CLI entrypoint for inventory sync."""
"""Compatibility entrypoint for channel inventory sync and Feishu table inspection."""
from __future__ import annotations
import logging
from logging.handlers import RotatingFileHandler
from pathlib import Path
import re
import sys
import time
from typing import Any
from inventory_sync.cli import main as inventory_sync_main
LOGGER_NAME = "inventory_sync_entrypoint"
logger = logging.getLogger(LOGGER_NAME)
PROJECT_ROOT = Path(__file__).resolve().parent
LOG_DIR = PROJECT_ROOT / "logs"
LOG_FILE = LOG_DIR / "list_feishu_base_tables.log"
SENSITIVE_PATTERN = re.compile(r"(?i)(password|passwd|secret|token|authorization)\s*[:=]\s*([^\s,;]+)")
LOG_FIELD_LABELS = {
"argv": "参数",
"cwd": "工作目录",
"elapsed_ms": "耗时毫秒",
"error": "错误",
"exit_code": "退出码",
"log_file": "日志文件",
}
def configure_logging() -> None:
LOG_DIR.mkdir(parents=True, exist_ok=True)
formatter = logging.Formatter(
fmt="%(asctime)s %(levelname)s %(name)s %(message)s",
datefmt="%Y-%m-%d %H:%M:%S",
)
stream_handler = logging.StreamHandler(sys.stderr)
stream_handler.setFormatter(formatter)
file_handler = RotatingFileHandler(LOG_FILE, maxBytes=5 * 1024 * 1024, backupCount=5, encoding="utf-8")
file_handler.setFormatter(formatter)
root_logger = logging.getLogger()
root_logger.handlers.clear()
root_logger.setLevel(logging.INFO)
root_logger.addHandler(stream_handler)
root_logger.addHandler(file_handler)
def sanitize_value(value: Any) -> Any:
if isinstance(value, str):
return SENSITIVE_PATTERN.sub(r"\1=***", value)
return value
def log_event(level: int, event: str, **fields: Any) -> None:
safe_event = sanitize_value(event)
details = " ".join(
f"{LOG_FIELD_LABELS.get(key, key)}={sanitize_value(value)!r}" for key, value in fields.items()
)
logger.log(level, "事件=%s%s", safe_event, f" {details}" if details else "")
def main() -> int:
configure_logging()
started_at = time.monotonic()
log_event(
logging.INFO,
"飞书同步入口启动",
argv=sys.argv[1:],
cwd=Path.cwd(),
log_file=LOG_FILE,
)
try:
exit_code = inventory_sync_main()
except SystemExit as exc:
exit_code = int(exc.code or 0) if isinstance(exc.code, int) else 1
log_event(
logging.INFO if exit_code == 0 else logging.ERROR,
"飞书同步入口主动退出",
exit_code=exit_code,
elapsed_ms=int((time.monotonic() - started_at) * 1000),
)
return exit_code
except Exception as exc:
logger.exception(
"事件=%s 错误=%r 耗时毫秒=%r",
"飞书同步入口异常",
sanitize_value(str(exc)),
int((time.monotonic() - started_at) * 1000),
)
return 1
log_event(
logging.INFO if exit_code == 0 else logging.ERROR,
"飞书同步入口完成",
exit_code=exit_code,
elapsed_ms=int((time.monotonic() - started_at) * 1000),
)
return exit_code
from inventory_analytics.channel_inventory_sync.entrypoint import main
if __name__ == "__main__":
prefect.yaml
+42
View File
@@ -0,0 +1,42 @@
name: inventory-analytics
prefect-version: "3.0.0"
build: []
push: []
pull: []
deployments:
- name: inventory-channel-inventory-sync-daily-0915
version: "1.0.0"
description: "每天 09:15 执行渠道库存统计并写入飞书多维表格。"
entrypoint: prefect_inventory_scheduler.py:inventory_channel_inventory_sync_flow
parameters: {}
tags:
- inventory
- feishu
- channel-inventory-sync
schedules:
- cron: "15 9 * * *"
timezone: "Asia/Shanghai"
active: true
work_pool:
name: inventory-analytics-pool
- name: inventory-batch-inventory-report-daily-0915
version: "1.0.0"
description: "每天 09:15 生成批次库存统计报表并写入飞书多维表格。"
entrypoint: prefect_inventory_scheduler.py:inventory_batch_inventory_report_flow
parameters:
output: ""
feishu_table: "批次库存"
skip_feishu_sync: false
tags:
- inventory
- feishu
- batch-inventory-report
schedules:
- cron: "15 9 * * *"
timezone: "Asia/Shanghai"
active: true
work_pool:
name: inventory-analytics-pool
prefect_inventory_scheduler.py
+120
View File
@@ -0,0 +1,120 @@
from __future__ import annotations
import argparse
import os
import sys
from pathlib import Path
def bootstrap_prefect_env() -> None:
env_file = Path(__file__).resolve().parent / ".env"
if not env_file.exists():
return
for raw_line in env_file.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)
key = key.strip()
if key.startswith("PREFECT_"):
os.environ.setdefault(key, value.strip())
bootstrap_prefect_env()
from prefect import flow, task
from inventory_analytics.channel_inventory_sync.entrypoint import main as channel_inventory_sync_main
from inventory_analytics.batch_inventory_report.cli import main as batch_inventory_report_main
@task(name="run-channel-inventory-sync")
def run_channel_inventory_sync_task() -> None:
original_argv = sys.argv[:]
try:
sys.argv = ["sync_channel_inventory.py", "--action", "all"]
exit_code = channel_inventory_sync_main()
finally:
sys.argv = original_argv
if exit_code != 0:
raise RuntimeError(f"渠道库存统计执行失败，退出码: {exit_code}")
@task(name="run-batch-inventory-report")
def run_batch_inventory_report_task(output: str = "", feishu_table: str = "批次库存", skip_feishu_sync: bool = False) -> None:
original_argv = sys.argv[:]
try:
sys.argv = ["generate_batch_inventory_report.py"]
if output:
sys.argv.extend(["--output", output])
if feishu_table:
sys.argv.extend(["--feishu-table", feishu_table])
if skip_feishu_sync:
sys.argv.append("--skip-feishu-sync")
exit_code = batch_inventory_report_main()
finally:
sys.argv = original_argv
if exit_code != 0:
raise RuntimeError(f"批次库存统计报表执行失败，退出码: {exit_code}")
@flow(name="inventory-channel-inventory-sync")
def inventory_channel_inventory_sync_flow() -> None:
run_channel_inventory_sync_task()
@flow(name="inventory-batch-inventory-report")
def inventory_batch_inventory_report_flow(
output: str = "",
feishu_table: str = "批次库存",
skip_feishu_sync: bool = False,
) -> None:
run_batch_inventory_report_task(output=output, feishu_table=feishu_table, skip_feishu_sync=skip_feishu_sync)
def parse_args() -> argparse.Namespace:
parser = argparse.ArgumentParser(description="Prefect 调度：每天 09:15 执行渠道库存统计和批次库存统计写飞书。")
parser.add_argument("--serve-channel-inventory-sync", action="store_true", help="启动渠道库存统计 flow 的本地 serve 调度")
parser.add_argument("--serve-batch-inventory-report", action="store_true", help="启动批次库存统计 flow 的本地 serve 调度")
parser.add_argument("--run-channel-inventory-sync", action="store_true", help="立即执行一次渠道库存统计 flow")
parser.add_argument("--run-batch-inventory-report", action="store_true", help="立即执行一次批次库存统计 flow")
parser.add_argument("--output", default="", help="批次库存统计输出路径，可选")
parser.add_argument("--feishu-table", default="批次库存", help="批次库存统计同步目标飞书表名")
parser.add_argument("--skip-feishu-sync", action="store_true", help="批次库存统计只生成本地文件，不写飞书")
return parser.parse_args()
def main() -> None:
args = parse_args()
flow_kwargs = {
"output": args.output,
"feishu_table": args.feishu_table,
"skip_feishu_sync": args.skip_feishu_sync,
}
if args.serve_channel_inventory_sync:
inventory_channel_inventory_sync_flow.serve(
name="inventory-channel-inventory-sync-daily-0915",
cron="15 9 * * *",
tags=["inventory", "feishu", "channel-inventory-sync"],
)
return
if args.serve_batch_inventory_report:
inventory_batch_inventory_report_flow.serve(
name="inventory-batch-inventory-report-daily-0915",
cron="15 9 * * *",
parameters=flow_kwargs,
tags=["inventory", "feishu", "batch-inventory-report"],
)
return
if args.run_channel_inventory_sync:
inventory_channel_inventory_sync_flow()
return
if args.run_batch_inventory_report:
inventory_batch_inventory_report_flow(**flow_kwargs)
return
inventory_channel_inventory_sync_flow()
inventory_batch_inventory_report_flow(**flow_kwargs)
if __name__ == "__main__":
main()
pyproject.toml
+27
View File
@@ -0,0 +1,27 @@
[build-system]
requires = ["setuptools>=68"]
build-backend = "setuptools.build_meta"
[project]
name = "inventory-analytics"
version = "0.1.0"
description = "Internal inventory analytics workflows for batch upload, channel inventory sync, and batch inventory reporting."
requires-python = ">=3.11"
dependencies = [
"openpyxl>=3.1.2",
"pandas",
"prefect>=3.0.0",
"psycopg2-binary",
"python-dotenv",
"xlrd>=2.0.1",
]
[tool.setuptools]
packages = [
"inventory_analytics",
"inventory_analytics.db",
"inventory_analytics.feishu",
"inventory_analytics.batch_upload",
"inventory_analytics.channel_inventory_sync",
"inventory_analytics.batch_inventory_report",
]
scripts/generate_batch_inventory_report.py
+7
View File
@@ -0,0 +1,7 @@
#!/usr/bin/env python3
from inventory_analytics.batch_inventory_report.cli import main
if __name__ == "__main__":
raise SystemExit(main())
scripts/sync_channel_inventory.py
+16
View File
@@ -0,0 +1,16 @@
#!/usr/bin/env python3
import sys
from inventory_analytics.channel_inventory_sync.entrypoint import main
def _ensure_default_action() -> None:
if any(arg == "--action" or arg.startswith("--action=") for arg in sys.argv[1:]):
return
sys.argv.extend(["--action", "all"])
if __name__ == "__main__":
_ensure_default_action()
raise SystemExit(main())
scripts/upload_batch_inventory.py
+7
View File
@@ -0,0 +1,7 @@
#!/usr/bin/env python3
from inventory_analytics.batch_upload.web import main
if __name__ == "__main__":
raise SystemExit(main())
supervisord.conf
+10
View File
@@ -0,0 +1,10 @@
[supervisord]
nodaemon=true
[program:inventory-analytics-worker]
command=prefect worker start --pool "inventory-analytics-pool" --type process
directory=/app
autostart=true
autorestart=true
stderr_logfile=/app/logs/prefect_worker.err.log
stdout_logfile=/app/logs/prefect_worker.out.log
sync_channel_inventory.py
+8
View File
@@ -0,0 +1,8 @@
#!/usr/bin/env python3
"""Compatibility entrypoint for channel inventory sync."""
from inventory_analytics.channel_inventory_sync.entrypoint import main
if __name__ == "__main__":
raise SystemExit(main())
upload_inventory_page.py
+2 -1053
View File
File diff suppressed because it is too large Load Diff