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

还没有上传文件。上传成功后,最近文件会显示在这里。

' + return ( + '' + "" + f"{''.join(rows)}" + "
文件名大小上传时间
" + ) + + +def render_page(message: str = "", is_error: bool = False) -> bytes: + message_html = "" + if message: + message_class = "message error" if is_error else "message success" + message_html = f'
{escape(message)}
' + + html = f""" + + + + + 委外批次库存文件上传 + + + +
+
+
+ +
+
+

投递委外批次库存文件

+

拖入委外仓、加工方或批次盘点文件,系统会按上传时间自动归档,便于后续同步、核对和追溯。

+
+
+ ARCHIVE PATH + {escape(str(UPLOAD_DIR))} +
+
+
+ +
+
+ {message_html} +
+
+
⇣
+
DROP FILE HERE
+

拖入文件,立即上传

+

支持 Excel、CSV 等委外批次库存文件。松开鼠标后自动保存到 batch_inventory_data,无需点击选择文件。

+
等待拖入文件
+
+
+
+ + + +
+

最近上传的委外批次文件

+ {render_recent_files(list_uploaded_files())} +
+
+
+
+ + + +""" + return html.encode("utf-8") + + +class UploadHandler(BaseHTTPRequestHandler): + server_version = "InventoryUpload/1.0" + + 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 + + params = parse_qs(parsed.query) + message = params.get("message", [""])[0] + is_error = params.get("error", ["0"])[0] == "1" + self.respond_html(render_page(message, is_error)) + + 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 + + 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(upload.filename) + with target_path.open("wb") as output: + output.write(upload.content) + log_event( + logging.INFO, + "上传文件已保存", + original_filename=upload.filename, + stored_file=target_path.name, + path=target_path, + bytes=upload.size, + ) + + 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) + self.send_header("Content-Type", "text/html; charset=utf-8") + self.send_header("Content-Length", str(len(body))) + self.end_headers() + self.wfile.write(body) + + 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: + 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) + log_event(logging.INFO, "上传服务已启动", url=f"http://{HOST}:{PORT}", upload_dir=UPLOAD_DIR, log_file=LOG_FILE) + try: + server.serve_forever() + except KeyboardInterrupt: + log_event(logging.INFO, "收到键盘中断准备停止服务") + finally: + server.server_close() + log_event(logging.INFO, "上传服务已停止", host=HOST, port=PORT) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/inventory_analytics/channel_inventory_sync/__init__.py b/inventory_analytics/channel_inventory_sync/__init__.py new file mode 100644 index 0000000..36852de --- /dev/null +++ b/inventory_analytics/channel_inventory_sync/__init__.py @@ -0,0 +1 @@ +"""Channel inventory sync workflows.""" diff --git a/inventory_analytics/channel_inventory_sync/cli.py b/inventory_analytics/channel_inventory_sync/cli.py new file mode 100644 index 0000000..37fe155 --- /dev/null +++ b/inventory_analytics/channel_inventory_sync/cli.py @@ -0,0 +1,3 @@ +from inventory_sync.cli import main + +__all__ = ["main"] diff --git a/inventory_analytics/channel_inventory_sync/entrypoint.py b/inventory_analytics/channel_inventory_sync/entrypoint.py new file mode 100644 index 0000000..ae8699d --- /dev/null +++ b/inventory_analytics/channel_inventory_sync/entrypoint.py @@ -0,0 +1,54 @@ +from __future__ import annotations + +import logging +from pathlib import Path +import sys +import time + +from .cli import main as channel_inventory_sync_main +from ..logging_utils import configure_logging, log_event +from ..paths import LOG_DIR + + +LOGGER_NAME = "inventory_analytics.channel_inventory_sync.entrypoint" +LOG_FILE = LOG_DIR / "sync_channel_inventory.log" +logger = logging.getLogger(LOGGER_NAME) + +LOG_FIELD_LABELS = { + "argv": "参数", + "cwd": "工作目录", + "elapsed_ms": "耗时毫秒", + "exit_code": "退出码", + "log_file": "日志文件", +} + + +def main() -> int: + configure_logging(LOG_FILE, stream=sys.stderr) + started_at = time.monotonic() + log_event( + logger, + LOG_FIELD_LABELS, + logging.INFO, + "飞书同步入口启动", + argv=sys.argv[1:], + cwd=Path.cwd(), + log_file=LOG_FILE, + ) + try: + exit_code = channel_inventory_sync_main() + except SystemExit as exc: + exit_code = int(exc.code or 0) if isinstance(exc.code, int) else 1 + except Exception: + logger.exception("事件=%s", "飞书同步入口异常") + return 1 + + log_event( + logger, + LOG_FIELD_LABELS, + logging.INFO if exit_code == 0 else logging.ERROR, + "飞书同步入口完成", + exit_code=exit_code, + elapsed_ms=int((time.monotonic() - started_at) * 1000), + ) + return exit_code diff --git a/inventory_analytics/channel_inventory_sync/queries.py b/inventory_analytics/channel_inventory_sync/queries.py new file mode 100644 index 0000000..9ef05f5 --- /dev/null +++ b/inventory_analytics/channel_inventory_sync/queries.py @@ -0,0 +1,3 @@ +from inventory_sync.sql import DEFAULT_SYNC_TARGET, POSTGRES_QUERY, build_postgres_query + +__all__ = ["DEFAULT_SYNC_TARGET", "POSTGRES_QUERY", "build_postgres_query"] diff --git a/inventory_analytics/channel_inventory_sync/service.py b/inventory_analytics/channel_inventory_sync/service.py new file mode 100644 index 0000000..f567569 --- /dev/null +++ b/inventory_analytics/channel_inventory_sync/service.py @@ -0,0 +1,3 @@ +from inventory_sync.service import build_feishu_record, run_feishu_summary, run_sync_brand, sync_brand_target + +__all__ = ["build_feishu_record", "run_feishu_summary", "run_sync_brand", "sync_brand_target"] diff --git a/inventory_analytics/channel_inventory_sync/specs.py b/inventory_analytics/channel_inventory_sync/specs.py new file mode 100644 index 0000000..77833b1 --- /dev/null +++ b/inventory_analytics/channel_inventory_sync/specs.py @@ -0,0 +1,21 @@ +from inventory_sync.specs import ( + BRAND_SYNC_TARGETS, + FIELD_SPECS, + LEGACY_NAME_MAP, + PRIMARY_FIELD_NAME, + PRIMARY_FIELD_SOURCE, + BrandSyncTarget, + FieldSpec, + month_sales_field_name, +) + +__all__ = [ + "BRAND_SYNC_TARGETS", + "FIELD_SPECS", + "LEGACY_NAME_MAP", + "PRIMARY_FIELD_NAME", + "PRIMARY_FIELD_SOURCE", + "BrandSyncTarget", + "FieldSpec", + "month_sales_field_name", +] diff --git a/inventory_analytics/config.py b/inventory_analytics/config.py new file mode 100644 index 0000000..eedf42a --- /dev/null +++ b/inventory_analytics/config.py @@ -0,0 +1,117 @@ +from __future__ import annotations + +from dataclasses import dataclass +import os +from pathlib import Path + +from .paths import PROJECT_ROOT + + +@dataclass(frozen=True) +class DbConfig: + host: str + port: str + user: str + password: str + name: str + + +@dataclass(frozen=True) +class AppConfig: + env_path: Path | None + feishu_app_id: str + feishu_app_secret: str + feishu_base_url: str + feishu_app_token: str | None + analytics_db_host: str + analytics_db_port: str + analytics_db_user: str + analytics_db_password: str + analytics_db_name: str + + +def load_env_file(env_path: Path) -> dict[str, str]: + values: dict[str, str] = {} + if not env_path.exists(): + return values + + for raw_line in env_path.read_text(encoding="utf-8").splitlines(): + line = raw_line.strip() + if not line or line.startswith("#") or "=" not in line: + continue + key, value = line.split("=", 1) + values[key.strip()] = value.strip().strip("'").strip('"') + return values + + +def load_env_candidates() -> tuple[dict[str, str], Path | None]: + candidates = [ + PROJECT_ROOT / ".env", + PROJECT_ROOT / "project (1)" / ".env", + PROJECT_ROOT.parent / ".env", + Path.cwd() / ".env", + ] + seen: set[Path] = set() + for candidate in candidates: + candidate = candidate.resolve() + if candidate in seen: + continue + seen.add(candidate) + values = load_env_file(candidate) + if values: + return values, candidate + return {}, None + + +def get_config(config: dict[str, str], *keys: str) -> str | None: + for key in keys: + value = config.get(key) or os.getenv(key) + if value: + return value + return None + + +def require_config(config: dict[str, str], *keys: str) -> str: + value = get_config(config, *keys) + if value: + return value + raise KeyError(f"缺少必填配置: {' / '.join(keys)}") + + +def load_db_config(prefix: str) -> tuple[DbConfig, Path | None]: + config_map, env_path = load_env_candidates() + return ( + DbConfig( + host=require_config(config_map, f"{prefix}_DB_HOST"), + port=require_config(config_map, f"{prefix}_DB_PORT"), + user=require_config(config_map, f"{prefix}_DB_USER"), + password=require_config(config_map, f"{prefix}_DB_PASSWORD"), + name=require_config(config_map, f"{prefix}_DB_NAME"), + ), + env_path, + ) + + +def load_raw_db_config() -> tuple[DbConfig, Path | None]: + return load_db_config("RAW") + + +def load_analytics_db_config() -> tuple[DbConfig, Path | None]: + return load_db_config("ANALYTICS") + + +def build_app_config() -> AppConfig: + config, env_path = load_env_candidates() + return AppConfig( + env_path=env_path, + feishu_app_id=require_config(config, "FEISHU_APP_ID", "APP_ID"), + feishu_app_secret=require_config(config, "FEISHU_APP_SECRET", "FEISHU_APP_SECRE", "APP_SECRET"), + feishu_base_url=require_config(config, "FEISHU_BASE_URL", "BASE_URL").rstrip("/"), + feishu_app_token=get_config(config, "FEISHU_APP_TOKEN"), + analytics_db_host=require_config(config, "ANALYTICS_DB_HOST"), + analytics_db_port=require_config(config, "ANALYTICS_DB_PORT"), + analytics_db_user=require_config(config, "ANALYTICS_DB_USER"), + analytics_db_password=require_config(config, "ANALYTICS_DB_PASSWORD"), + analytics_db_name=require_config(config, "ANALYTICS_DB_NAME"), + ) + diff --git a/inventory_analytics/db/__init__.py b/inventory_analytics/db/__init__.py new file mode 100644 index 0000000..5b6f4bb --- /dev/null +++ b/inventory_analytics/db/__init__.py @@ -0,0 +1,2 @@ +"""Database helpers.""" + diff --git a/inventory_analytics/feishu/__init__.py b/inventory_analytics/feishu/__init__.py new file mode 100644 index 0000000..a085158 --- /dev/null +++ b/inventory_analytics/feishu/__init__.py @@ -0,0 +1,2 @@ +"""Feishu integrations.""" + diff --git a/inventory_analytics/feishu/client.py b/inventory_analytics/feishu/client.py new file mode 100644 index 0000000..de90de8 --- /dev/null +++ b/inventory_analytics/feishu/client.py @@ -0,0 +1,4 @@ +from inventory_sync.feishu import FeishuApiError, FeishuClient + +__all__ = ["FeishuApiError", "FeishuClient"] + diff --git a/inventory_analytics/logging_utils.py b/inventory_analytics/logging_utils.py new file mode 100644 index 0000000..5c9f485 --- /dev/null +++ b/inventory_analytics/logging_utils.py @@ -0,0 +1,43 @@ +from __future__ import annotations + +import logging +from logging.handlers import RotatingFileHandler +from pathlib import Path +import re +import sys +from typing import Any + + +SENSITIVE_PATTERN = re.compile(r"(?i)(password|passwd|secret|token|authorization)\s*[:=]\s*([^\s,;]+)") + + +def sanitize_value(value: Any) -> Any: + if isinstance(value, str): + return SENSITIVE_PATTERN.sub(r"\1=***", value) + return value + + +def configure_logging(log_file: Path, *, stream = None, level: int = logging.INFO) -> None: + log_file.parent.mkdir(parents=True, exist_ok=True) + formatter = logging.Formatter( + fmt="%(asctime)s %(levelname)s %(name)s %(message)s", + datefmt="%Y-%m-%d %H:%M:%S", + ) + stream_handler = logging.StreamHandler(stream or sys.stdout) + stream_handler.setFormatter(formatter) + file_handler = RotatingFileHandler(log_file, maxBytes=5 * 1024 * 1024, backupCount=5, encoding="utf-8") + file_handler.setFormatter(formatter) + root_logger = logging.getLogger() + root_logger.handlers.clear() + root_logger.setLevel(level) + root_logger.addHandler(stream_handler) + root_logger.addHandler(file_handler) + + +def log_event(logger: logging.Logger, labels: dict[str, str], level: int, event: str, **fields: Any) -> None: + safe_event = sanitize_value(event) + details = " ".join( + f"{labels.get(key, key)}={sanitize_value(value)!r}" for key, value in fields.items() + ) + logger.log(level, "事件=%s%s", safe_event, f" {details}" if details else "") + diff --git a/inventory_analytics/paths.py b/inventory_analytics/paths.py new file mode 100644 index 0000000..0ee9e67 --- /dev/null +++ b/inventory_analytics/paths.py @@ -0,0 +1,11 @@ +from __future__ import annotations + +from pathlib import Path + + +PACKAGE_ROOT = Path(__file__).resolve().parent +PROJECT_ROOT = PACKAGE_ROOT.parent +LOG_DIR = PROJECT_ROOT / "logs" +REPORT_DIR = PROJECT_ROOT / "reports" +BATCH_UPLOAD_DIR = PROJECT_ROOT / "batch_inventory_data" + diff --git a/inventory_sync/feishu.py b/inventory_sync/feishu.py index b265549..a8e786b 100644 --- a/inventory_sync/feishu.py +++ b/inventory_sync/feishu.py @@ -206,6 +206,16 @@ class FeishuClient: primary_field = next((field for field in fields if field.get("is_primary")), None) if not primary_field: raise FeishuApiError("未找到主字段") + expected_field_names = {spec.feishu_name for spec in FIELD_SPECS} + actual_managed_field_names = { + field.get("field_name") + for field in fields + if not field.get("is_primary") and field.get("field_name") in expected_field_names + } + primary_matches = primary_field.get("field_name") == PRIMARY_FIELD_NAME + fields_match = actual_managed_field_names == expected_field_names + if primary_matches and fields_match: + return [field.get("field_name") for field in fields if field.get("field_name")] duplicate_primary = next( (field for field in fields if not field.get("is_primary") and field.get("field_name") == PRIMARY_FIELD_NAME), diff --git a/list_feishu_base_tables.py b/list_feishu_base_tables.py index d986eb8..b54f0d5 100644 --- a/list_feishu_base_tables.py +++ b/list_feishu_base_tables.py @@ -1,104 +1,7 @@ #!/usr/bin/env python3 -"""CLI entrypoint for inventory sync.""" +"""Compatibility entrypoint for channel inventory sync and Feishu table inspection.""" -from __future__ import annotations - -import logging -from logging.handlers import RotatingFileHandler -from pathlib import Path -import re -import sys -import time -from typing import Any - -from inventory_sync.cli import main as inventory_sync_main - - -LOGGER_NAME = "inventory_sync_entrypoint" -logger = logging.getLogger(LOGGER_NAME) -PROJECT_ROOT = Path(__file__).resolve().parent -LOG_DIR = PROJECT_ROOT / "logs" -LOG_FILE = LOG_DIR / "list_feishu_base_tables.log" -SENSITIVE_PATTERN = re.compile(r"(?i)(password|passwd|secret|token|authorization)\s*[:=]\s*([^\s,;]+)") - -LOG_FIELD_LABELS = { - "argv": "参数", - "cwd": "工作目录", - "elapsed_ms": "耗时毫秒", - "error": "错误", - "exit_code": "退出码", - "log_file": "日志文件", -} - - -def configure_logging() -> None: - LOG_DIR.mkdir(parents=True, exist_ok=True) - formatter = logging.Formatter( - fmt="%(asctime)s %(levelname)s %(name)s %(message)s", - datefmt="%Y-%m-%d %H:%M:%S", - ) - stream_handler = logging.StreamHandler(sys.stderr) - stream_handler.setFormatter(formatter) - file_handler = RotatingFileHandler(LOG_FILE, maxBytes=5 * 1024 * 1024, backupCount=5, encoding="utf-8") - file_handler.setFormatter(formatter) - root_logger = logging.getLogger() - root_logger.handlers.clear() - root_logger.setLevel(logging.INFO) - root_logger.addHandler(stream_handler) - root_logger.addHandler(file_handler) - - -def sanitize_value(value: Any) -> Any: - if isinstance(value, str): - return SENSITIVE_PATTERN.sub(r"\1=***", value) - return value - - -def log_event(level: int, event: str, **fields: Any) -> None: - safe_event = sanitize_value(event) - details = " ".join( - f"{LOG_FIELD_LABELS.get(key, key)}={sanitize_value(value)!r}" for key, value in fields.items() - ) - logger.log(level, "事件=%s%s", safe_event, f" {details}" if details else "") - - -def main() -> int: - configure_logging() - started_at = time.monotonic() - log_event( - logging.INFO, - "飞书同步入口启动", - argv=sys.argv[1:], - cwd=Path.cwd(), - log_file=LOG_FILE, - ) - try: - exit_code = inventory_sync_main() - except SystemExit as exc: - exit_code = int(exc.code or 0) if isinstance(exc.code, int) else 1 - log_event( - logging.INFO if exit_code == 0 else logging.ERROR, - "飞书同步入口主动退出", - exit_code=exit_code, - elapsed_ms=int((time.monotonic() - started_at) * 1000), - ) - return exit_code - except Exception as exc: - logger.exception( - "事件=%s 错误=%r 耗时毫秒=%r", - "飞书同步入口异常", - sanitize_value(str(exc)), - int((time.monotonic() - started_at) * 1000), - ) - return 1 - - log_event( - logging.INFO if exit_code == 0 else logging.ERROR, - "飞书同步入口完成", - exit_code=exit_code, - elapsed_ms=int((time.monotonic() - started_at) * 1000), - ) - return exit_code +from inventory_analytics.channel_inventory_sync.entrypoint import main if __name__ == "__main__": diff --git a/prefect.yaml b/prefect.yaml new file mode 100644 index 0000000..16cdb1d --- /dev/null +++ b/prefect.yaml @@ -0,0 +1,42 @@ +name: inventory-analytics +prefect-version: "3.0.0" + +build: [] +push: [] +pull: [] + +deployments: + - name: inventory-channel-inventory-sync-daily-0915 + version: "1.0.0" + description: "每天 09:15 执行渠道库存统计并写入飞书多维表格。" + entrypoint: prefect_inventory_scheduler.py:inventory_channel_inventory_sync_flow + parameters: {} + tags: + - inventory + - feishu + - channel-inventory-sync + schedules: + - cron: "15 9 * * *" + timezone: "Asia/Shanghai" + active: true + work_pool: + name: inventory-analytics-pool + + - name: inventory-batch-inventory-report-daily-0915 + version: "1.0.0" + description: "每天 09:15 生成批次库存统计报表并写入飞书多维表格。" + entrypoint: prefect_inventory_scheduler.py:inventory_batch_inventory_report_flow + parameters: + output: "" + feishu_table: "批次库存" + skip_feishu_sync: false + tags: + - inventory + - feishu + - batch-inventory-report + schedules: + - cron: "15 9 * * *" + timezone: "Asia/Shanghai" + active: true + work_pool: + name: inventory-analytics-pool diff --git a/prefect_inventory_scheduler.py b/prefect_inventory_scheduler.py new file mode 100644 index 0000000..7aeae27 --- /dev/null +++ b/prefect_inventory_scheduler.py @@ -0,0 +1,120 @@ +from __future__ import annotations + +import argparse +import os +import sys +from pathlib import Path + + +def bootstrap_prefect_env() -> None: + env_file = Path(__file__).resolve().parent / ".env" + if not env_file.exists(): + return + for raw_line in env_file.read_text(encoding="utf-8").splitlines(): + line = raw_line.strip() + if not line or line.startswith("#") or "=" not in line: + continue + key, value = line.split("=", 1) + key = key.strip() + if key.startswith("PREFECT_"): + os.environ.setdefault(key, value.strip()) + + +bootstrap_prefect_env() + +from prefect import flow, task + +from inventory_analytics.channel_inventory_sync.entrypoint import main as channel_inventory_sync_main +from inventory_analytics.batch_inventory_report.cli import main as batch_inventory_report_main + + +@task(name="run-channel-inventory-sync") +def run_channel_inventory_sync_task() -> None: + original_argv = sys.argv[:] + try: + sys.argv = ["sync_channel_inventory.py", "--action", "all"] + exit_code = channel_inventory_sync_main() + finally: + sys.argv = original_argv + if exit_code != 0: + raise RuntimeError(f"渠道库存统计执行失败,退出码: {exit_code}") + + +@task(name="run-batch-inventory-report") +def run_batch_inventory_report_task(output: str = "", feishu_table: str = "批次库存", skip_feishu_sync: bool = False) -> None: + original_argv = sys.argv[:] + try: + sys.argv = ["generate_batch_inventory_report.py"] + if output: + sys.argv.extend(["--output", output]) + if feishu_table: + sys.argv.extend(["--feishu-table", feishu_table]) + if skip_feishu_sync: + sys.argv.append("--skip-feishu-sync") + exit_code = batch_inventory_report_main() + finally: + sys.argv = original_argv + if exit_code != 0: + raise RuntimeError(f"批次库存统计报表执行失败,退出码: {exit_code}") + + +@flow(name="inventory-channel-inventory-sync") +def inventory_channel_inventory_sync_flow() -> None: + run_channel_inventory_sync_task() + + +@flow(name="inventory-batch-inventory-report") +def inventory_batch_inventory_report_flow( + output: str = "", + feishu_table: str = "批次库存", + skip_feishu_sync: bool = False, +) -> None: + run_batch_inventory_report_task(output=output, feishu_table=feishu_table, skip_feishu_sync=skip_feishu_sync) + + +def parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser(description="Prefect 调度:每天 09:15 执行渠道库存统计和批次库存统计写飞书。") + parser.add_argument("--serve-channel-inventory-sync", action="store_true", help="启动渠道库存统计 flow 的本地 serve 调度") + parser.add_argument("--serve-batch-inventory-report", action="store_true", help="启动批次库存统计 flow 的本地 serve 调度") + parser.add_argument("--run-channel-inventory-sync", action="store_true", help="立即执行一次渠道库存统计 flow") + parser.add_argument("--run-batch-inventory-report", action="store_true", help="立即执行一次批次库存统计 flow") + parser.add_argument("--output", default="", help="批次库存统计输出路径,可选") + parser.add_argument("--feishu-table", default="批次库存", help="批次库存统计同步目标飞书表名") + parser.add_argument("--skip-feishu-sync", action="store_true", help="批次库存统计只生成本地文件,不写飞书") + return parser.parse_args() + + +def main() -> None: + args = parse_args() + flow_kwargs = { + "output": args.output, + "feishu_table": args.feishu_table, + "skip_feishu_sync": args.skip_feishu_sync, + } + if args.serve_channel_inventory_sync: + inventory_channel_inventory_sync_flow.serve( + name="inventory-channel-inventory-sync-daily-0915", + cron="15 9 * * *", + tags=["inventory", "feishu", "channel-inventory-sync"], + ) + return + if args.serve_batch_inventory_report: + inventory_batch_inventory_report_flow.serve( + name="inventory-batch-inventory-report-daily-0915", + cron="15 9 * * *", + parameters=flow_kwargs, + tags=["inventory", "feishu", "batch-inventory-report"], + ) + return + if args.run_channel_inventory_sync: + inventory_channel_inventory_sync_flow() + return + if args.run_batch_inventory_report: + inventory_batch_inventory_report_flow(**flow_kwargs) + return + inventory_channel_inventory_sync_flow() + inventory_batch_inventory_report_flow(**flow_kwargs) + + +if __name__ == "__main__": + main() diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..ccf190f --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,27 @@ +[build-system] +requires = ["setuptools>=68"] +build-backend = "setuptools.build_meta" + +[project] +name = "inventory-analytics" +version = "0.1.0" +description = "Internal inventory analytics workflows for batch upload, channel inventory sync, and batch inventory reporting." +requires-python = ">=3.11" +dependencies = [ + "openpyxl>=3.1.2", + "pandas", + "prefect>=3.0.0", + "psycopg2-binary", + "python-dotenv", + "xlrd>=2.0.1", +] + +[tool.setuptools] +packages = [ + "inventory_analytics", + "inventory_analytics.db", + "inventory_analytics.feishu", + "inventory_analytics.batch_upload", + "inventory_analytics.channel_inventory_sync", + "inventory_analytics.batch_inventory_report", +] diff --git a/scripts/generate_batch_inventory_report.py b/scripts/generate_batch_inventory_report.py new file mode 100644 index 0000000..79f7b69 --- /dev/null +++ b/scripts/generate_batch_inventory_report.py @@ -0,0 +1,7 @@ +#!/usr/bin/env python3 + +from inventory_analytics.batch_inventory_report.cli import main + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/scripts/sync_channel_inventory.py b/scripts/sync_channel_inventory.py new file mode 100644 index 0000000..03b797c --- /dev/null +++ b/scripts/sync_channel_inventory.py @@ -0,0 +1,16 @@ +#!/usr/bin/env python3 + +import sys + +from inventory_analytics.channel_inventory_sync.entrypoint import main + + +def _ensure_default_action() -> None: + if any(arg == "--action" or arg.startswith("--action=") for arg in sys.argv[1:]): + return + sys.argv.extend(["--action", "all"]) + + +if __name__ == "__main__": + _ensure_default_action() + raise SystemExit(main()) diff --git a/scripts/upload_batch_inventory.py b/scripts/upload_batch_inventory.py new file mode 100644 index 0000000..5e0ab63 --- /dev/null +++ b/scripts/upload_batch_inventory.py @@ -0,0 +1,7 @@ +#!/usr/bin/env python3 + +from inventory_analytics.batch_upload.web import main + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/supervisord.conf b/supervisord.conf new file mode 100644 index 0000000..f30d535 --- /dev/null +++ b/supervisord.conf @@ -0,0 +1,10 @@ +[supervisord] +nodaemon=true + +[program:inventory-analytics-worker] +command=prefect worker start --pool "inventory-analytics-pool" --type process +directory=/app +autostart=true +autorestart=true +stderr_logfile=/app/logs/prefect_worker.err.log +stdout_logfile=/app/logs/prefect_worker.out.log diff --git a/sync_channel_inventory.py b/sync_channel_inventory.py new file mode 100644 index 0000000..245427a --- /dev/null +++ b/sync_channel_inventory.py @@ -0,0 +1,8 @@ +#!/usr/bin/env python3 +"""Compatibility entrypoint for channel inventory sync.""" + +from inventory_analytics.channel_inventory_sync.entrypoint import main + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/upload_inventory_page.py b/upload_inventory_page.py index bf05a15..3887dac 100644 --- a/upload_inventory_page.py +++ b/upload_inventory_page.py @@ -1,1058 +1,7 @@ #!/usr/bin/env python3 -"""Local inventory file upload page. +"""Compatibility entrypoint for batch inventory upload.""" -Run: - python3 upload_inventory_page.py - -Then open: - http://127.0.0.1:8080 -""" - -from __future__ import annotations - -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 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: - name = Path(filename or "inventory_file").name.strip() - name = re.sub(r"[^\w.\-\u4e00-\u9fff()() ]+", "_", name) - name = re.sub(r"\s+", "_", name) - return name or "inventory_file" - - -def unique_upload_path(filename: str) -> Path: - UPLOAD_DIR.mkdir(parents=True, exist_ok=True) - timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") - base_name = safe_filename(filename) - candidate = UPLOAD_DIR / f"{timestamp}_{base_name}" - counter = 1 - 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(".")] - return sorted(files, key=lambda item: item.stat().st_mtime, reverse=True)[:limit] - - -def human_size(size: int) -> str: - units = ["B", "KB", "MB", "GB"] - value = float(size) - for unit in units: - if value < 1024 or unit == units[-1]: - return f"{value:.1f} {unit}" if unit != "B" else f"{int(value)} {unit}" - value /= 1024 - return f"{size} B" - - -def render_recent_files(files: Iterable[Path]) -> str: - rows = [] - for path in files: - stat = path.stat() - uploaded_at = datetime.fromtimestamp(stat.st_mtime).strftime("%Y-%m-%d %H:%M:%S") - rows.append( - "" - f"{escape(path.name)}" - f"{escape(human_size(stat.st_size))}" - f"{escape(uploaded_at)}" - "" - ) - if not rows: - return '

还没有上传文件。上传成功后,最近文件会显示在这里。

' - return ( - '' - "" - f"{''.join(rows)}" - "
文件名大小上传时间
" - ) - - -def render_page(message: str = "", is_error: bool = False) -> bytes: - message_html = "" - if message: - message_class = "message error" if is_error else "message success" - message_html = f'
{escape(message)}
' - - html = f""" - - - - - 委外批次库存文件上传 - - - -
-
-
- -
-
-

投递委外批次库存文件

-

拖入委外仓、加工方或批次盘点文件,系统会按上传时间自动归档,便于后续同步、核对和追溯。

-
-
- ARCHIVE PATH - {escape(str(UPLOAD_DIR))} -
-
-
- -
-
- {message_html} -
-
-
⇣
-
DROP FILE HERE
-

拖入文件,立即上传

-

支持 Excel、CSV 等委外批次库存文件。松开鼠标后自动保存到 batch_inventory_data,无需点击选择文件。

-
等待拖入文件
-
-
-
- - - -
-

最近上传的委外批次文件

- {render_recent_files(list_uploaded_files())} -
-
-
-
- - - -""" - return html.encode("utf-8") - - -class UploadHandler(BaseHTTPRequestHandler): - server_version = "InventoryUpload/1.0" - - 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 - - params = parse_qs(parsed.query) - message = params.get("message", [""])[0] - is_error = params.get("error", ["0"])[0] == "1" - self.respond_html(render_page(message, is_error)) - - 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 - - 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(upload.filename) - with target_path.open("wb") as output: - output.write(upload.content) - log_event( - logging.INFO, - "上传文件已保存", - original_filename=upload.filename, - stored_file=target_path.name, - path=target_path, - bytes=upload.size, - ) - - 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) - self.send_header("Content-Type", "text/html; charset=utf-8") - self.send_header("Content-Length", str(len(body))) - self.end_headers() - self.wfile.write(body) - - 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: - 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) - log_event(logging.INFO, "上传服务已启动", url=f"http://{HOST}:{PORT}", upload_dir=UPLOAD_DIR, log_file=LOG_FILE) - try: - server.serve_forever() - except KeyboardInterrupt: - log_event(logging.INFO, "收到键盘中断准备停止服务") - finally: - server.server_close() - log_event(logging.INFO, "上传服务已停止", host=HOST, port=PORT) - return 0 +from inventory_analytics.batch_upload.web import main if __name__ == "__main__":