diff --git a/.gitignore b/.gitignore index de3a51b..cfb410b 100644 --- a/.gitignore +++ b/.gitignore @@ -3,3 +3,5 @@ inventory_sync/__pycache__ .DS_Store batch_inventory_data/* +logs +reports diff --git a/PROJECT_GUIDE.md b/PROJECT_GUIDE.md new file mode 100644 index 0000000..827fd95 --- /dev/null +++ b/PROJECT_GUIDE.md @@ -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 +``` + +如果普通关闭失败: + +```bash +kill -9 +``` + +## 验证命令 + +语法检查: + +```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`。 +- 入口脚本保持稳定,便于人工执行和后续定时任务接入。 +- 日志新增时优先使用 `事件=<中文描述>`,并输出足够定位问题的关键字段。 diff --git a/REFACTOR_PLAN.md b/REFACTOR_PLAN.md index 883c6f2..3ddf2f0 100644 --- a/REFACTOR_PLAN.md +++ b/REFACTOR_PLAN.md @@ -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 或任务脚本,减少人工输入成本。 diff --git a/__pycache__/generate_yxt_batch_inventory_report.cpython-313.pyc b/__pycache__/generate_yxt_batch_inventory_report.cpython-313.pyc new file mode 100644 index 0000000..9ed75b7 Binary files /dev/null and b/__pycache__/generate_yxt_batch_inventory_report.cpython-313.pyc differ diff --git a/__pycache__/list_feishu_base_tables.cpython-313.pyc b/__pycache__/list_feishu_base_tables.cpython-313.pyc new file mode 100644 index 0000000..20b305e Binary files /dev/null and b/__pycache__/list_feishu_base_tables.cpython-313.pyc differ diff --git a/generate_yxt_batch_inventory_report.py b/generate_yxt_batch_inventory_report.py new file mode 100644 index 0000000..716010a --- /dev/null +++ b/generate_yxt_batch_inventory_report.py @@ -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()) diff --git a/inventory_sync/cli.py b/inventory_sync/cli.py index 5aea783..16cebaa 100644 --- a/inventory_sync/cli.py +++ b/inventory_sync/cli.py @@ -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 - diff --git a/inventory_sync/service.py b/inventory_sync/service.py index e51d2b6..1c6f0f0 100644 --- a/inventory_sync/service.py +++ b/inventory_sync/service.py @@ -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)) diff --git a/inventory_sync/sql.py b/inventory_sync/sql.py index ff1494f..fd8415e 100644 --- a/inventory_sync/sql.py +++ b/inventory_sync/sql.py @@ -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 ( diff --git a/list_feishu_base_tables.py b/list_feishu_base_tables.py index 23d4b41..d986eb8 100644 --- a/list_feishu_base_tables.py +++ b/list_feishu_base_tables.py @@ -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__": diff --git a/upload_inventory_page.py b/upload_inventory_page.py index 232c063..bf05a15 100644 --- a/upload_inventory_page.py +++ b/upload_inventory_page.py @@ -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