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-24 19:00:07 +08:00
parent 3e1d479643
commit 65216cd4cf
11 changed files with 1809 additions and 313 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.gitignore
+2
View File
@@ -3,3 +3,5 @@
inventory_sync/__pycache__
.DS_Store
batch_inventory_data/*
logs
reports
PROJECT_GUIDE.md
+256
View File
@@ -0,0 +1,256 @@
# Inventory Analytics 工程说明
## 项目概览
当前工程包含两条独立业务链路：
- 委外批次库存上传：通过本地网页上传 Excel/CSV 文件，保存原始文件，并逐行写入 Raw PostgreSQL 表。
- 飞书库存同步：从 Analytics PostgreSQL 查询库存、挂单量、销售和商品信息，并同步到飞书多维表格。
## 目录结构
```text
inventory_analytics/
├── .env
├── PROJECT_GUIDE.md
├── REFACTOR_PLAN.md
├── upload_inventory_page.py
├── list_feishu_base_tables.py
└── inventory_sync/
├── __init__.py
├── cli.py
├── feishu.py
├── postgres.py
├── service.py
├── settings.py
├── specs.py
└── sql.py
```
运行后会自动创建或使用：
```text
batch_inventory_data/
```
该目录保存上传的委外批次库存原始文件，不纳入业务代码结构。
## 环境变量
### 委外批次库存上传
上传页使用 `.env` 中的 Raw 数据库配置：
```text
RAW_DB_HOST
RAW_DB_PORT
RAW_DB_USER
RAW_DB_PASSWORD
RAW_DB_NAME
```
写入目标：
```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
ANALYTICS_DB_PASSWORD
ANALYTICS_DB_NAME
```
`FEISHU_APP_TOKEN` 可选；当 `FEISHU_BASE_URL` 已包含 `/base/` 或 `/wiki/` 时，可以不单独配置。
## 委外批次库存上传
启动上传页：
```bash
python3 upload_inventory_page.py
```
打开页面：
```text
http://127.0.0.1:8080
```
支持文件：
- `.xlsx`
- `.xlsm`
- `.xls`
- `.csv`
单文件大小限制：
```text
200 MB
```
上传成功后会执行：
1. 保存文件到 `batch_inventory_data/`。
2. 解析 Excel/CSV 为行数据。
3. 创建 Raw schema/table 和 `created_at` 索引。
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 事件=中文事件描述 中文字段=字段值 ...
```
上传页日志输出到 stdout，典型事件包括：
```text
事件=上传服务启动中
事件=收到上传请求
事件=已解析上传文件
事件=上传文件已保存
事件=开始解析库存文件
事件=已解析Excel工作表
事件=已加载Raw数据库配置
事件=Raw数据表确认完成
事件=Raw数据写入完成
事件=上传请求处理完成
```
飞书同步日志输出到 stderr，JSON 结果仍输出到 stdout，便于命令行管道消费。典型事件包括：
```text
事件=飞书同步入口启动
事件=飞书同步动作开始
事件=已加载环境变量文件
事件=开始读取飞书Base信息
事件=开始查询PostgreSQL
事件=开始同步品牌数据
事件=开始同步目标飞书表
事件=目标飞书表同步完成
事件=飞书同步入口完成
```
## 查看和关闭上传服务
查看 8080 端口：
```bash
lsof -nP -iTCP:8080 -sTCP:LISTEN
```
关闭服务：
```bash
kill <PID>
```
如果普通关闭失败：
```bash
kill -9 <PID>
```
## 验证命令
语法检查：
```bash
python3 -m py_compile upload_inventory_page.py list_feishu_base_tables.py inventory_sync/*.py
```
查看飞书同步帮助：
```bash
python3 list_feishu_base_tables.py --help
```
本地启动上传页：
```bash
python3 upload_inventory_page.py
```
## 维护原则
- 上传 Raw 数据保持原始粒度，每行一条 payload，不在上传阶段做复杂业务清洗。
- 飞书同步使用 `inventory_sync/specs.py` 管理品牌目标和字段映射。
- SQL 调整优先修改 `inventory_sync/sql.py`。
- 入口脚本保持稳定，便于人工执行和后续定时任务接入。
- 日志新增时优先使用 `事件=<中文描述>`，并输出足够定位问题的关键字段。
REFACTOR_PLAN.md
+217 -287
View File
@@ -1,24 +1,28 @@
# 飞书库存同步项目重构方案
# Inventory Analytics 重构记录与后续计划
## 1. 重构目标
## 1. 当前目标
当前项目的核心能力是从 PostgreSQL 查询库存、挂单量、销售和商品信息，按品牌筛选数据，并同步写入对应的飞书多维表格。
本次整理后的工程目标是将项目收束为两条清晰链路：
本次重构的目标是将原先集中在单个脚本中的逻辑拆分为更清晰、可维护、可扩展的工程结构：
- 委外批次库存上传：本地页面上传文件，归档原始文件，并写入 Raw PostgreSQL。
- 飞书库存同步：从 Analytics PostgreSQL 查询库存数据，并同步到飞书多维表格。
- 将配置读取、数据库查询、飞书 API、业务同步、命令行入口拆分到独立模块。
- 将飞书字段名称、SQL 字段来源、字段类型、数字格式统一维护在映射表中。
- 保留原有脚本入口，降低使用成本，避免改变已有执行习惯。
- 提高后续新增品牌、新增字段、新增同步动作时的可读性和安全性。
重构重点：
- 删除与当前工程目标无关的旧脚本。
- 将上传页入库逻辑整理为清晰的函数边界。
- 添加工程化日志，保证关键执行信息可追踪。
- 保留飞书同步原入口，并补齐入口、CLI、业务同步阶段日志。
## 2. 当前目录结构
```text
inventory_analytics/
├── .env
├── .gitignore
├── list_feishu_base_tables.py
├── PROJECT_GUIDE.md
├── REFACTOR_PLAN.md
├── upload_inventory_page.py
├── list_feishu_base_tables.py
└── inventory_sync/
├── __init__.py
├── cli.py
@@ -30,59 +34,206 @@ inventory_analytics/
└── sql.py
```
## 3. 模块职责
已删除旧脚本：
### 3.1 入口脚本
```text
crud.py
base_xlsx_to_db.py
save_orders.py
session.py
tmall_orders_flow.py
```
文件：`list_feishu_base_tables.py`
## 3. 上传页重构结果
文件：`upload_inventory_page.py`
职责：
- 作为项目的兼容入口。
- 将执行逻辑转交给 `inventory_sync.cli.main()`。
- 保持原命令可继续使用。
- 启动本地上传页面。
- 接收拖拽上传的 Excel/CSV 文件。
- 保存原始文件到 `batch_inventory_data/`。
- 读取 `.env` 中的 Raw 数据库配置。
- 创建 `raw_data.wangdian_batch_inventory`。
- 按文件行粒度写入 JSONB payload。
- 输出结构化日志。
示例：
关键常量：
```python
HOST = "127.0.0.1"
PORT = 8080
UPLOAD_DIR = PROJECT_ROOT / "batch_inventory_data"
MAX_UPLOAD_BYTES = 200 * 1024 * 1024
DB_SCHEMA = "raw_data"
DB_TABLE = "wangdian_batch_inventory"
DATA_FIELD = "inventory_data"
```
payload 结构：
```json
{
"source_file": "原始文件名.xlsx",
"stored_file": "带时间戳的归档文件名.xlsx",
"uploaded_at": "上传时间",
"total_count": 100,
"inventory_data": {
"每一行的字段": "每一行的值"
}
}
```
说明：
- `total_count` 表示本批上传文件解析出的总行数。
- `inventory_data` 保存当前行的完整数据。
- Excel 多 sheet 文件会在行数据中补充 `_sheet_name`。
- 空值会写入 JSON `null`。
- 日期类单元格会转成 ISO 字符串。
## 4. 飞书同步重构结果
入口文件：`list_feishu_base_tables.py`
模块职责：
| 文件 | 职责 |
| --- | --- |
| `list_feishu_base_tables.py` | 兼容入口、配置日志、记录启动和退出 |
| `inventory_sync/cli.py` | 解析 action、编排顶层执行、输出 JSON 结果 |
| `inventory_sync/settings.py` | 加载 `.env`，构建配置对象 |
| `inventory_sync/postgres.py` | 执行 Analytics PostgreSQL 查询 |
| `inventory_sync/sql.py` | 维护库存同步 SQL |
| `inventory_sync/feishu.py` | 封装飞书 API |
| `inventory_sync/service.py` | 编排品牌同步业务流程 |
| `inventory_sync/specs.py` | 维护品牌目标和字段映射 |
当前 action：
```text
feishu 查看飞书 Base 表信息
postgres 查询 Analytics PostgreSQL
sync-brand 按配置同步品牌数据到飞书
all 执行完整流程
```
## 5. 日志方案
日志统一采用事件式结构：
```text
YYYY-MM-DD HH:MM:SS LEVEL logger_name 事件=中文事件描述 中文字段=字段值 ...
```
上传页 logger：
```text
inventory_upload
```
飞书同步 logger：
```text
inventory_sync_entrypoint
inventory_sync.cli
inventory_sync.service
```
上传页关键事件：
```text
上传服务启动中
上传服务已启动
收到GET请求
开始处理POST请求
收到上传请求
已解析上传文件
已分配上传归档路径
上传文件已保存
开始解析库存文件
已解析Excel工作表
已解析CSV文件
已加载Raw数据库配置
开始连接Raw数据库
开始确认Raw数据表
Raw数据表确认完成
开始写入Raw数据
Raw数据写入完成
上传请求处理完成
上传文件写入Raw数据库失败
上传服务已停止
```
飞书同步关键事件：
```text
飞书同步入口启动
飞书同步动作开始
已加载环境变量文件
开始读取飞书Base信息
飞书Base信息读取完成
开始查询PostgreSQL
PostgreSQL查询完成
开始准备品牌同步
开始同步目标飞书表
目标飞书表已有记录读取完成
目标飞书表同步完成
飞书同步动作完成
飞书同步入口完成
```
## 6. 执行方式
启动上传页：
```bash
python3 upload_inventory_page.py
```
查看飞书表：
```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
```
### 3.2 命令行模块
执行完整飞书同步：
文件：`inventory_sync/cli.py`
```bash
python3 list_feishu_base_tables.py --action all
```
职责：
## 7. 配置维护
- 解析 `--action` 参数。
- 初始化配置和飞书客户端。
- 根据 action 调用对应服务。
- 统一输出 JSON 格式结果。
### 7.1 上传页 Raw 数据库配置
当前支持的 action：
```text
RAW_DB_HOST
RAW_DB_PORT
RAW_DB_USER
RAW_DB_PASSWORD
RAW_DB_NAME
```
- `feishu`：读取飞书 Base 下的表信息。
- `postgres`：执行 PostgreSQL 查询并输出结果。
- `sync-brand`：按照 `BRAND_SYNC_TARGETS` 中的品牌和仓库主键组合配置查询数据，并同步到对应飞书表。
- `all`：依次执行飞书读取、数据库查询和品牌同步。
### 3.3 配置模块
文件：`inventory_sync/settings.py`
职责：
- 自动查找并加载 `.env` 文件。
- 从 `.env` 或当前 shell 环境变量中读取配置。
- 校验必填配置。
- 构建统一的 `AppConfig` 配置对象。
当前必填配置：
### 7.2 飞书同步配置
```text
FEISHU_APP_ID
FEISHU_APP_SECRET 或 FEISHU_APP_SECRE
FEISHU_APP_SECRET
FEISHU_BASE_URL
FEISHU_APP_TOKEN
ANALYTICS_DB_HOST
ANALYTICS_DB_PORT
ANALYTICS_DB_USER
@@ -90,197 +241,9 @@ ANALYTICS_DB_PASSWORD
ANALYTICS_DB_NAME
```
可选配置：
### 7.3 品牌和字段配置
```text
FEISHU_APP_TOKEN
```
### 3.4 字段映射模块
文件：`inventory_sync/specs.py`
职责：
- 定义品牌、目标飞书表和仓库主键的组合同步配置。
- 定义主字段名称和来源字段。
- 定义 SQL 查询结果字段到飞书表字段的映射关系。
- 定义飞书字段类型和数字格式。
- 定义历史字段名到新字段名的重命名关系。
当前同步目标：
```python
BRAND_SYNC_TARGETS = [
BrandSyncTarget(brand_name="羽心堂", table_name="羽心堂", warehouse_keys=(2, 3, 7, 13)),
]
PRIMARY_FIELD_NAME = "货品编码"
PRIMARY_FIELD_SOURCE = "统一规格编码"
```
`BrandSyncTarget` 中三个字段含义：
- `brand_name`：SQL 中用于筛选 `pd."品牌名称"` 的品牌名称。
- `table_name`：该品牌同步写入的飞书目标表名。
- `warehouse_keys`：该品牌参与总库存汇总的仓库主键列表。
当前字段映射和展示顺序：
| 飞书字段名 | SQL 来源字段 | 飞书字段类型 | 显示格式 |
| --- | --- | --- | --- |
| 货品编码 | 统一规格编码 | 文本主字段 | 文本 |
| 货品名称 | 商品名称 | 文本 | 文本 |
| 分类 | 一级分类 | 文本 | 文本 |
| 条形码 | 条形码 | 文本 | 文本 |
| 零售价 | 零售价 | 数字 | 保留 2 位小数 |
| 店铺名称 | 店铺名称 | 文本 | 文本 |
| 店铺可用库存 | 店铺挂单量 | 数字 | 不保留小数 |
| 总库存 | 总库存 | 数字 | 不保留小数 |
| 当前月份销售数量，例如 202604销售数量 | 本月销售数量 | 数字 | 不保留小数 |
| 上一月份销售数量，例如 202603销售数量 | 上月销售数量 | 数字 | 不保留小数 |
| 上上月份销售数量，例如 202602销售数量 | 上上月销售数量 | 数字 | 不保留小数 |
| 品牌 | 品牌名称 | 文本 | 文本 |
| 产品类别 | 货品简称 | 文本 | 文本 |
### 3.5 SQL 模块
文件：`inventory_sync/sql.py`
职责：
- 集中维护 PostgreSQL 查询 SQL。
- 避免业务代码中混入大段 SQL。
- 后续如需调整库存、销量、商品维度逻辑，只需要修改该文件。
当前 SQL 逻辑包括：
- 取每个统一规格编码、仓库的最新库存快照。
- 按当前品牌配置的 `warehouse_keys` 筛选仓库，并汇总统一规格编码维度的总库存。
- 取每个统一规格编码、店铺的最新可发库存快照，并排除已失效状态。
- 按统一规格编码汇总本月、上月、上上月销售数量。
- 关联商品维表和店铺维表。
- 按当前品牌配置的 `brand_name` 筛选品牌数据。
- 输出用于飞书同步的明细结果。
### 3.6 PostgreSQL 模块
文件：`inventory_sync/postgres.py`
职责：
- 使用 `.env` 中的数据库配置连接 PostgreSQL。
- 根据 `BrandSyncTarget` 中的品牌和仓库主键配置生成并执行 SQL。
- 将查询结果解析为结构化字典列表。
- 对外返回统一结构，供 CLI 或同步服务使用。
### 3.7 飞书 API 模块
文件：`inventory_sync/feishu.py`
职责：
- 获取飞书 `tenant_access_token`。
- 解析飞书 Base 地址和 `app_token`。
- 查询飞书 Base 下的表列表。
- 查询、创建、更新飞书字段。
- 根据字段映射重建受管字段，保证字段类型和展示顺序一致。
- 分页读取记录。
- 批量删除记录。
- 批量创建记录。
该模块只关心飞书接口本身，不关心具体业务品牌和 SQL 字段。
### 3.8 业务服务模块
文件：`inventory_sync/service.py`
职责：
- 编排完整业务流程。
- 按 `BRAND_SYNC_TARGETS` 逐个读取品牌和仓库主键配置。
- 根据每个品牌目标的配置查询 PostgreSQL。
- 根据 `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`：
品牌同步目标维护在 `inventory_sync/specs.py`：
```python
BRAND_SYNC_TARGETS = [
@@ -288,72 +251,39 @@ BRAND_SYNC_TARGETS = [
]
```
后续新增品牌时，只需要在列表中增加一行，并确保飞书 Base 中存在同名或指定名称的目标表。每个品牌可以配置自己的仓库主键列表：
字段映射维护在 `FIELD_SPECS`：
```python
BRAND_SYNC_TARGETS = [
BrandSyncTarget(brand_name="羽心堂", table_name="羽心堂", warehouse_keys=(2, 3, 7, 13)),
BrandSyncTarget(brand_name="其他品牌", table_name="其他品牌", warehouse_keys=(9, 10)),
]
FieldSpec("飞书字段名", "SQL 输出字段名", field_type, formatter, scale)
```
执行 `python3 list_feishu_base_tables.py --action sync-brand` 时，程序会遍历该列表。每个目标都会使用自己的 `brand_name` 和 `warehouse_keys` 生成 SQL，再同步到对应的飞书表。
## 8. 验证命令
## 8. 当前设计取舍
### 8.1 保留原脚本入口
保留 `list_feishu_base_tables.py` 是为了不改变已有使用方式。后续如果项目继续扩大，可以新增更标准的入口，例如：
语法检查：
```bash
python3 -m inventory_sync.cli --action sync-brand
python3 -m py_compile upload_inventory_page.py list_feishu_base_tables.py inventory_sync/*.py
```
### 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
python3 list_feishu_base_tables.py --help
```
如果 `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
}
]
}
```bash
lsof -nP -iTCP:8080 -sTCP:LISTEN
```
## 9. 后续计划
- 为上传页增加 dry-run 或预览模式，只解析文件不写库。
- 为上传页增加重复文件或重复批次检测。
- 为 Raw 表增加按 `stored_file` 或 `uploaded_at` 的查询辅助索引。
- 为飞书同步增加 dry-run，预览删除和新增数量。
- 为飞书同步增加 upsert 模式，降低全量删除带来的风险。
- 增加单元测试，覆盖文件解析、payload 构造、字段映射、数字转换。
- 将日志接入文件轮转或统一日志采集系统。
- 将运行命令接入 Makefile 或任务脚本，减少人工输入成本。
__pycache__/generate_yxt_batch_inventory_report.cpython-313.pyc
BIN
View File
Binary file not shown.
__pycache__/list_feishu_base_tables.cpython-313.pyc
BIN
View File
Binary file not shown.
generate_yxt_batch_inventory_report.py
+685
View File
@@ -0,0 +1,685 @@
#!/usr/bin/env python3
"""Generate YXT batch inventory report by joining Raw batch stock with Analytics shop stock."""
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
if __name__ == "__main__":
raise SystemExit(main())
inventory_sync/cli.py
+69 -5
View File
@@ -2,7 +2,10 @@ from __future__ import annotations
import argparse
import json
import sys
import logging
import re
import time
from typing import Any
from .feishu import FeishuClient
from .postgres import query_postgres
@@ -10,6 +13,36 @@ from .service import run_feishu_summary, run_sync_brand
from .settings import build_app_config
logger = logging.getLogger("inventory_sync.cli")
SENSITIVE_PATTERN = re.compile(r"(?i)(password|passwd|secret|token|authorization)\s*[:=]\s*([^\s,;]+)")
LOG_FIELD_LABELS = {
"action": "动作",
"elapsed_ms": "耗时毫秒",
"env_path": "环境文件",
"error": "错误",
"fallback": "兜底方式",
"row_count": "行数",
"source_row_count": "源数据行数",
"table_count": "表数量",
"target_count": "目标数量",
}
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 parse_args() -> argparse.Namespace:
parser = argparse.ArgumentParser(description="Feishu Base and PostgreSQL helper")
parser.add_argument(
@@ -22,28 +55,59 @@ def parse_args() -> argparse.Namespace:
def main() -> int:
started_at = time.monotonic()
try:
args = parse_args()
log_event(logging.INFO, "飞书同步动作开始", action=args.action)
config = build_app_config()
if config.env_path:
print(f"已加载环境变量文件: {config.env_path}", file=sys.stderr)
log_event(logging.INFO, "已加载环境变量文件", env_path=config.env_path)
else:
print("WARNING: 未找到 .env 文件，将仅使用当前 shell 环境变量", file=sys.stderr)
log_event(logging.WARNING, "未找到环境变量文件", fallback="当前Shell环境变量")
client = FeishuClient(config)
result: dict[str, object] = {}
if args.action in {"feishu", "all"}:
log_event(logging.INFO, "开始读取飞书Base信息")
result["feishu"] = run_feishu_summary(client)
log_event(
logging.INFO,
"飞书Base信息读取完成",
table_count=result["feishu"].get("table_count") if isinstance(result["feishu"], dict) else None,
)
if args.action in {"postgres", "all"}:
log_event(logging.INFO, "开始查询PostgreSQL")
result["postgres"] = query_postgres(config)
log_event(
logging.INFO,
"PostgreSQL查询完成",
row_count=result["postgres"].get("row_count") if isinstance(result["postgres"], dict) else None,
)
if args.action in {"sync-brand", "all"}:
log_event(logging.INFO, "开始同步品牌数据")
postgres_result = result.get("postgres")
result["sync_brand"] = run_sync_brand(config, client, postgres_result if isinstance(postgres_result, dict) else None)
log_event(
logging.INFO,
"品牌数据同步完成",
target_count=result["sync_brand"].get("target_count") if isinstance(result["sync_brand"], dict) else None,
source_row_count=result["sync_brand"].get("source_row_count") if isinstance(result["sync_brand"], dict) else None,
)
print(json.dumps(result, ensure_ascii=False, indent=2))
log_event(
logging.INFO,
"飞书同步动作完成",
action=args.action,
elapsed_ms=int((time.monotonic() - started_at) * 1000),
)
return 0
except Exception as exc: # noqa: BLE001
print(f"执行失败: {exc}", file=sys.stderr)
log_event(
logging.ERROR,
"飞书同步动作失败",
error=exc,
elapsed_ms=int((time.monotonic() - started_at) * 1000),
)
return 1
inventory_sync/service.py
+74
View File
@@ -1,5 +1,7 @@
from __future__ import annotations
import logging
import re
from typing import Any
from .feishu import FeishuApiError, FeishuClient
@@ -8,6 +10,38 @@ from .settings import AppConfig
from .specs import BRAND_SYNC_TARGETS, FIELD_SPECS, PRIMARY_FIELD_NAME, PRIMARY_FIELD_SOURCE, BrandSyncTarget
logger = logging.getLogger("inventory_sync.service")
SENSITIVE_PATTERN = re.compile(r"(?i)(password|passwd|secret|token|authorization)\s*[:=]\s*([^\s,;]+)")
LOG_FIELD_LABELS = {
"advanced_permission_enabled": "高级权限已开启",
"brand_name": "品牌",
"created_records": "新增记录数",
"deleted_records": "删除记录数",
"existing_records": "已有记录数",
"rows": "行数",
"selected_table_id": "选中表ID",
"source_rows": "源数据行数",
"table_count": "表数量",
"table_name": "表名",
"target_count": "目标数量",
}
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 build_feishu_record(row: dict[str, Any]) -> dict[str, Any]:
if PRIMARY_FIELD_SOURCE not in row:
raise KeyError(f"映射缺少主字段来源列: {PRIMARY_FIELD_SOURCE}")
@@ -30,10 +64,19 @@ def build_feishu_record(row: dict[str, Any]) -> dict[str, Any]:
def run_feishu_summary(client: FeishuClient) -> dict[str, Any]:
log_event(logging.INFO, "开始获取飞书租户访问令牌")
access_token = client.get_tenant_access_token()
log_event(logging.INFO, "飞书租户访问令牌获取完成")
base_info = client.parse_base_info(client.build_base_url(), access_token)
log_event(logging.INFO, "飞书Base信息解析完成", selected_table_id=base_info["selected_table_id"])
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 "")
log_event(
logging.INFO,
"飞书Base表列表读取完成",
table_count=len(tables),
advanced_permission_enabled=advanced_permission_enabled,
)
return {
"base_url": base_info["base_url"],
"app_token": base_info["app_token"],
@@ -61,6 +104,13 @@ def sync_brand_target(
target: BrandSyncTarget,
postgres_rows: list[dict[str, Any]] | None = None,
) -> dict[str, Any]:
log_event(
logging.INFO,
"开始同步目标飞书表",
brand_name=target.brand_name,
table_name=target.table_name,
source_rows=len(postgres_rows or []),
)
target_table = next(
(table for table in tables if table.get("name") == target.table_name),
None,
@@ -75,6 +125,13 @@ def sync_brand_target(
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")]
log_event(
logging.INFO,
"目标飞书表已有记录读取完成",
brand_name=target.brand_name,
table_name=target.table_name,
existing_records=len(existing_record_ids),
)
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,
@@ -82,6 +139,14 @@ def sync_brand_target(
table_id,
[build_feishu_record(row) for row in target_rows],
) if target_rows else 0
log_event(
logging.INFO,
"目标飞书表同步完成",
brand_name=target.brand_name,
table_name=target.table_name,
deleted_records=deleted_count,
created_records=created_count,
)
return {
"brand_name": target.brand_name,
@@ -97,11 +162,13 @@ def sync_brand_target(
def run_sync_brand(config: AppConfig, client: FeishuClient, postgres_result: dict[str, Any] | None = None) -> dict[str, Any]:
log_event(logging.INFO, "开始准备品牌同步", target_count=len(BRAND_SYNC_TARGETS))
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)
log_event(logging.INFO, "品牌同步准备完成", table_count=len(tables))
results = []
source_row_count = 0
@@ -111,6 +178,13 @@ def run_sync_brand(config: AppConfig, client: FeishuClient, postgres_result: dic
else:
postgres_payload = query_postgres(config, target=target)
postgres_rows = postgres_payload["rows"]
log_event(
logging.INFO,
"目标品牌PostgreSQL数据已准备",
brand_name=target.brand_name,
table_name=target.table_name,
rows=len(postgres_rows or []),
)
source_row_count += len(postgres_rows or [])
results.append(sync_brand_target(client, access_token, app_token, tables, target, postgres_rows))
inventory_sync/sql.py
+9 -2
View File
@@ -61,16 +61,20 @@ shop_stock_latest AS (
SELECT
t."统一规格编码",
t."店铺主键",
t."可发库存"
SUM(t."可发库存") AS "可发库存"
FROM (
SELECT
f."统一规格编码",
f."店铺主键",
f."仓库主键",
f."快照时间",
f."可发库存",
f."状态",
ROW_NUMBER() OVER (
PARTITION BY f."统一规格编码", f."店铺主键"
PARTITION BY
f."统一规格编码",
f."店铺主键",
f."仓库主键"
ORDER BY f."快照时间" DESC
) AS rn
FROM analytics_layer.fact_shop_stock_snapshot f
@@ -79,6 +83,9 @@ shop_stock_latest AS (
) t
WHERE t.rn = 1
AND t."状态" <> '已失效'
GROUP BY
t."统一规格编码",
t."店铺主键"
),
sales_summary AS (
list_feishu_base_tables.py
+98 -1
View File
@@ -1,7 +1,104 @@
#!/usr/bin/env python3
"""CLI entrypoint for inventory sync."""
from inventory_sync.cli import main
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
if __name__ == "__main__":
upload_inventory_page.py
+399 -18
View File
@@ -10,22 +10,186 @@ Then open:
from __future__ import annotations
import cgi
from dataclasses import dataclass
from email import policy
from email.parser import BytesParser
import json
import logging
from logging.handlers import RotatingFileHandler
from datetime import datetime
from html import escape
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
import os
from pathlib import Path
import re
import shutil
from typing import Iterable
import sys
import time
from typing import Any, Iterable
from urllib.parse import parse_qs, urlparse
import pandas as pd
import psycopg2
from dotenv import load_dotenv
HOST = "127.0.0.1"
PORT = 8080
PROJECT_ROOT = Path(__file__).resolve().parent
UPLOAD_DIR = PROJECT_ROOT / "batch_inventory_data"
LOG_DIR = PROJECT_ROOT / "logs"
LOG_FILE = LOG_DIR / "upload_inventory_page.log"
MAX_UPLOAD_BYTES = 200 * 1024 * 1024
DB_SCHEMA = "raw_data"
DB_TABLE = "wangdian_batch_inventory"
DATA_FIELD = "inventory_data"
LOGGER_NAME = "inventory_upload"
logger = logging.getLogger(LOGGER_NAME)
SENSITIVE_PATTERN = re.compile(r"(?i)(password|passwd|secret|token|authorization)\s*[:=]\s*([^\s,;]+)")
LOG_FIELD_LABELS = {
"advanced_permission_enabled": "高级权限已开启",
"app_token": "应用Token",
"bytes": "字节数",
"client": "客户端",
"columns": "列数",
"content_type": "内容类型",
"database": "数据库",
"db_schema": "数据库Schema",
"db_table": "数据库表",
"elapsed_ms": "耗时毫秒",
"encoding": "编码",
"env_path": "环境文件",
"error": "错误",
"field_name": "字段名",
"host": "主机",
"is_error": "是否错误",
"max_bytes": "最大字节数",
"max_upload_bytes": "最大上传字节数",
"message": "消息",
"log_file": "日志文件",
"original_filename": "原始文件名",
"path": "路径",
"port": "端口",
"rows": "行数",
"payload": "写入内容",
"schema": "Schema",
"sheet_name": "工作表",
"sheets": "工作表数",
"source_file": "源文件",
"stored_file": "归档文件",
"stored_filename": "归档文件名",
"suffix": "文件后缀",
"table": "表名",
"target_path": "目标路径",
"upload_dir": "上传目录",
"url": "地址",
}
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 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 "")
@dataclass(frozen=True)
class UploadedFile:
filename: str
content: bytes
@property
def size(self) -> int:
return len(self.content)
@dataclass(frozen=True)
class RawDbConfig:
host: str
port: str
user: str
password: str
name: str
@classmethod
def from_env(cls) -> "RawDbConfig":
env_path = PROJECT_ROOT / ".env"
load_dotenv(env_path)
values = {
"host": os.getenv("RAW_DB_HOST", ""),
"port": os.getenv("RAW_DB_PORT", ""),
"user": os.getenv("RAW_DB_USER", ""),
"password": os.getenv("RAW_DB_PASSWORD", ""),
"name": os.getenv("RAW_DB_NAME", ""),
}
missing = [f"RAW_DB_{key.upper()}" for key, value in values.items() if not value]
if missing:
raise RuntimeError(f"缺少 Raw 数据库配置: {', '.join(missing)}")
log_event(
logging.INFO,
"已加载Raw数据库配置",
env_path=env_path,
database=values["name"],
)
return cls(**values)
def connect(self):
log_event(
logging.INFO,
"开始连接Raw数据库",
database=self.name,
)
return psycopg2.connect(
host=self.host,
port=self.port,
user=self.user,
password=self.password,
dbname=self.name,
)
def get_create_table_sql(schema_name: str, table_name: str) -> str:
full_table_name = f"{schema_name}.{table_name}"
return f"""
CREATE SCHEMA IF NOT EXISTS {schema_name};
CREATE TABLE IF NOT EXISTS {full_table_name} (
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
);
CREATE INDEX IF NOT EXISTS idx_{table_name}_created_at ON {full_table_name} (created_at);
"""
def safe_filename(filename: str) -> str:
@@ -44,9 +208,176 @@ def unique_upload_path(filename: str) -> Path:
while candidate.exists():
candidate = UPLOAD_DIR / f"{timestamp}_{counter}_{base_name}"
counter += 1
log_event(
logging.INFO,
"已分配上传归档路径",
original_filename=filename,
stored_filename=candidate.name,
target_path=candidate,
)
return candidate
def parse_inventory_upload(headers: Any, body: bytes) -> UploadedFile | None:
content_type = headers.get("Content-Type", "")
if "multipart/form-data" not in content_type:
log_event(logging.WARNING, "上传请求内容类型不支持", content_type=content_type)
return None
message = BytesParser(policy=policy.default).parsebytes(
b"Content-Type: " + content_type.encode("utf-8") + b"\r\n"
b"MIME-Version: 1.0\r\n\r\n"
+ body
)
for part in message.iter_parts():
if part.get_param("name", header="content-disposition") != "inventory_file":
continue
filename = part.get_filename()
content = part.get_payload(decode=True)
if filename and content is not None:
upload = UploadedFile(filename=filename, content=content)
log_event(
logging.INFO,
"已解析上传文件",
filename=upload.filename,
bytes=upload.size,
content_type=part.get_content_type(),
)
return upload
log_event(logging.WARNING, "上传表单缺少文件字段", field_name="inventory_file")
return None
def normalize_cell(value: Any) -> Any:
if pd.isna(value):
return None
if hasattr(value, "isoformat"):
return value.isoformat()
return value
def dataframe_to_rows(frame: pd.DataFrame, sheet_name: str | None = None) -> list[dict[str, Any]]:
frame = frame.dropna(how="all")
rows: list[dict[str, Any]] = []
for record in frame.to_dict(orient="records"):
row = {str(key): normalize_cell(value) for key, value in record.items()}
if sheet_name is not None:
row["_sheet_name"] = sheet_name
rows.append(row)
return rows
def read_inventory_rows(file_path: Path) -> list[dict[str, Any]]:
suffix = file_path.suffix.lower()
log_event(logging.INFO, "开始解析库存文件", path=file_path, suffix=suffix)
if suffix in {".xlsx", ".xlsm", ".xls"}:
sheets = pd.read_excel(file_path, sheet_name=None, dtype=object)
include_sheet_name = len(sheets) > 1
rows: list[dict[str, Any]] = []
for sheet_name, frame in sheets.items():
sheet_rows = dataframe_to_rows(frame, sheet_name if include_sheet_name else None)
rows.extend(sheet_rows)
log_event(
logging.INFO,
"已解析Excel工作表",
path=file_path.name,
sheet_name=sheet_name,
rows=len(sheet_rows),
columns=len(frame.columns),
)
log_event(logging.INFO, "库存文件解析完成", path=file_path.name, rows=len(rows), sheets=len(sheets))
return rows
if suffix == ".csv":
last_error: Exception | None = None
for encoding in ("utf-8-sig", "gb18030"):
try:
frame = pd.read_csv(file_path, dtype=object, encoding=encoding)
rows = dataframe_to_rows(frame)
log_event(
logging.INFO,
"已解析CSV文件",
path=file_path.name,
rows=len(rows),
columns=len(frame.columns),
encoding=encoding,
)
return rows
except UnicodeDecodeError as exc:
last_error = exc
log_event(logging.WARNING, "CSV编码解析失败", path=file_path.name, encoding=encoding, error=exc)
if last_error is not None:
raise last_error
raise ValueError("仅支持 Excel 或 CSV 文件写入数据库。")
def save_inventory_file_to_raw_db(file_path: Path, original_filename: str) -> int:
started_at = time.monotonic()
rows = read_inventory_rows(file_path)
if not rows:
log_event(logging.WARNING, "文件无数据跳过Raw写入", path=file_path.name)
return 0
config = RawDbConfig.from_env()
conn = config.connect()
try:
with conn:
with conn.cursor() as cur:
log_event(logging.INFO, "开始确认Raw数据表", schema=DB_SCHEMA, table=DB_TABLE)
cur.execute(get_create_table_sql(DB_SCHEMA, DB_TABLE))
log_event(logging.INFO, "Raw数据表确认完成", schema=DB_SCHEMA, table=DB_TABLE)
upload_time = datetime.now().isoformat(timespec="seconds")
total_count = len(rows)
log_event(
logging.INFO,
"开始写入Raw数据",
schema=DB_SCHEMA,
table=DB_TABLE,
source_file=original_filename,
stored_file=file_path.name,
rows=total_count,
)
for item in rows:
payload = {
"source_file": original_filename,
"stored_file": file_path.name,
"uploaded_at": upload_time,
"total_count": total_count,
DATA_FIELD: item,
}
log_event(
logging.INFO,
"写入Raw数据库单行数据",
stored_file=file_path.name,
payload=json.dumps(payload, ensure_ascii=False),
)
cur.execute(
f"""
INSERT INTO {DB_SCHEMA}.{DB_TABLE} (
payload, start_time, end_time
) VALUES (
%s::jsonb, %s, %s
);
""",
(json.dumps(payload, ensure_ascii=False), None, None),
)
elapsed_ms = int((time.monotonic() - started_at) * 1000)
log_event(
logging.INFO,
"Raw数据写入完成",
schema=DB_SCHEMA,
table=DB_TABLE,
stored_file=file_path.name,
rows=len(rows),
elapsed_ms=elapsed_ms,
)
return len(rows)
finally:
conn.close()
log_event(logging.INFO, "Raw数据库连接已关闭", schema=DB_SCHEMA, table=DB_TABLE)
def list_uploaded_files(limit: int = 20) -> list[Path]:
UPLOAD_DIR.mkdir(parents=True, exist_ok=True)
files = [path for path in UPLOAD_DIR.iterdir() if path.is_file() and not path.name.startswith(".")]
@@ -597,11 +928,13 @@ class UploadHandler(BaseHTTPRequestHandler):
def do_GET(self) -> None:
parsed = urlparse(self.path)
log_event(logging.INFO, "收到GET请求", client=self.client_address[0], path=parsed.path)
if parsed.path == "/favicon.ico":
self.send_response(204)
self.end_headers()
return
if parsed.path != "/":
log_event(logging.WARNING, "请求路径不存在", client=self.client_address[0], path=parsed.path)
self.send_error(404, "Not Found")
return
@@ -612,33 +945,69 @@ class UploadHandler(BaseHTTPRequestHandler):
def do_POST(self) -> None:
path = urlparse(self.path).path
request_started_at = time.monotonic()
log_event(logging.INFO, "开始处理POST请求", client=self.client_address[0], path=path)
if path != "/upload":
log_event(logging.WARNING, "请求路径不存在", client=self.client_address[0], path=path)
self.send_error(404, "Not Found")
return
content_length = int(self.headers.get("Content-Length", "0") or "0")
log_event(logging.INFO, "收到上传请求", client=self.client_address[0], bytes=content_length)
if content_length <= 0:
log_event(logging.WARNING, "上传请求为空", client=self.client_address[0])
self.redirect("/", "没有收到上传文件。", is_error=True)
return
if content_length > MAX_UPLOAD_BYTES:
log_event(
logging.WARNING,
"上传文件超过大小限制",
client=self.client_address[0],
bytes=content_length,
max_bytes=MAX_UPLOAD_BYTES,
)
self.redirect("/", f"文件太大，最大支持 {human_size(MAX_UPLOAD_BYTES)}。", is_error=True)
return
form = cgi.FieldStorage(fp=self.rfile, headers=self.headers, environ={
"REQUEST_METHOD": "POST",
"CONTENT_TYPE": self.headers.get("Content-Type", ""),
"CONTENT_LENGTH": str(content_length),
})
file_item = form["inventory_file"] if "inventory_file" in form else None
if file_item is None or not getattr(file_item, "filename", ""):
upload = parse_inventory_upload(self.headers, self.rfile.read(content_length))
if upload is None:
self.redirect("/", "请选择需要上传的文件。", is_error=True)
return
target_path = unique_upload_path(file_item.filename)
target_path = unique_upload_path(upload.filename)
with target_path.open("wb") as output:
shutil.copyfileobj(file_item.file, output)
output.write(upload.content)
log_event(
logging.INFO,
"上传文件已保存",
original_filename=upload.filename,
stored_file=target_path.name,
path=target_path,
bytes=upload.size,
)
self.redirect("/", f"上传成功：{target_path.name}")
try:
saved_rows = save_inventory_file_to_raw_db(target_path, upload.filename)
except Exception as exc:
log_event(
logging.ERROR,
"上传文件写入Raw数据库失败",
stored_file=target_path.name,
error=exc,
elapsed_ms=int((time.monotonic() - request_started_at) * 1000),
)
self.redirect("/", f"文件已保存，但写入数据库失败：{exc}", is_error=True)
return
log_event(
logging.INFO,
"上传请求处理完成",
original_filename=upload.filename,
stored_file=target_path.name,
rows=saved_rows,
elapsed_ms=int((time.monotonic() - request_started_at) * 1000),
)
self.redirect("/", f"上传成功：{target_path.name}，已写入 {saved_rows} 行到 {DB_SCHEMA}.{DB_TABLE}")
def respond_html(self, body: bytes) -> None:
self.send_response(200)
@@ -650,27 +1019,39 @@ class UploadHandler(BaseHTTPRequestHandler):
def redirect(self, path: str, message: str, is_error: bool = False) -> None:
from urllib.parse import urlencode
log_event(logging.INFO, "页面重定向", path=path, is_error=is_error, message=message)
query = urlencode({"message": message, "error": "1" if is_error else "0"})
self.send_response(303)
self.send_header("Location", f"{path}?{query}")
self.end_headers()
def log_message(self, format: str, *args: object) -> None:
timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
print(f"[{timestamp}] {self.address_string()} - {format % args}")
log_event(logging.INFO, "HTTP服务日志", client=self.address_string(), message=format % args)
def main() -> int:
configure_logging()
UPLOAD_DIR.mkdir(parents=True, exist_ok=True)
log_event(
logging.INFO,
"上传服务启动中",
host=HOST,
port=PORT,
upload_dir=UPLOAD_DIR,
max_upload_bytes=MAX_UPLOAD_BYTES,
db_schema=DB_SCHEMA,
db_table=DB_TABLE,
log_file=LOG_FILE,
)
server = ThreadingHTTPServer((HOST, PORT), UploadHandler)
print(f"上传页面已启动: http://{HOST}:{PORT}")
print(f"文件保存目录: {UPLOAD_DIR}")
log_event(logging.INFO, "上传服务已启动", url=f"http://{HOST}:{PORT}", upload_dir=UPLOAD_DIR, log_file=LOG_FILE)
try:
server.serve_forever()
except KeyboardInterrupt:
print("\n上传页面已停止。")
log_event(logging.INFO, "收到键盘中断准备停止服务")
finally:
server.server_close()
log_event(logging.INFO, "上传服务已停止", host=HOST, port=PORT)
return 0