Add environment configuration, enhance Feishu session handling, and implement tests for TLS 1.2 and retry logic

Prefect 同步跳过缺失表并推送飞书提醒
2026-09-28 11:20:14 +08:00 · 2026-09-11 11:13:16 +08:00 · 2026-09-11 11:13:03 +08:00 · 2026-09-08 11:22:42 +08:00 · 2026-07-06 17:16:32 +08:00 · 2026-06-04 13:56:44 +08:00
11 changed files with 643 additions and 110 deletions
@@ -0,0 +1,32 @@
+# PostgreSQL analytics database (required)
+ANALYTICS_DB_HOST=postgres.example.internal
+ANALYTICS_DB_PORT=5432
+ANALYTICS_DB_USER=your_database_user
+ANALYTICS_DB_PASSWORD=your_database_password
+ANALYTICS_DB_NAME=your_database_name
+
+# Feishu application and Bitable (required)
+FEISHU_APP_ID=cli_your_app_id
+FEISHU_APP_SECRET=your_feishu_app_secret
+FEISHU_BASE_URL=https://example.feishu.cn/base/your_app_token
+
+# Sync settings
+PROJECT_MAPPING_FILE=店铺映射表.xlsx
+FEISHU_STORE_FIELD_NAME=店铺名称
+SYNC_START_DATE=2026-04-01
+
+# Feishu notification (optional; configure one or both as needed)
+FEISHU_BOT_WEBHOOK=
+FEISHU_CHAT_ID=
+FEISHU_REPORT_IMAGE_PATH=
+
+# Prefect (required only for Prefect deployment/worker mode)
+PREFECT_API_URL=http://host.docker.internal:4200/api
+PREFECT_SYNC_STATE_FILE=
+
+# Logging (optional)
+LOG_DIR=
+LOG_FILE_NAME=gmv_sync.log
+LOG_LEVEL=INFO
+LOG_MAX_BYTES=10485760
+LOG_BACKUP_COUNT=7
@@ -1,6 +1,8 @@
 __pycache__/*
 *.pyc
 .vscode/
+
+# Local credentials and machine-specific configuration
 .env
 created/
 创建编组.zip
@@ -0,0 +1,233 @@
+# GMV Analytics
+
+按项目、店铺和日期统计销售金额、退款金额及 GMV，并同步到飞书多维表格。支持本地手动执行、历史日期补数，以及通过 Docker + Prefect 定时运行。
+
+## 项目介绍
+
+数据来自 PostgreSQL，通过 `店铺映射表.xlsx` 将 BI 店铺名称转换为业务店铺名称，再按项目和月份匹配飞书目标表。
+
+- 支持项目：大眼睛、羽心堂、MMMJ、EM、春回盏、BROKEN ZOO。
+- 计算口径：`GMV = 销售金额 - 退款金额`，具体取数 SQL 位于 `project_bitable_common.py`。
+- 本地流水线逐日查询销售额和退款额，计算 GMV，写入对应项目的月份表。
+- 当前流水线使用“日期”“指标”和各店铺数值列，指标包括“销售金额”“退款金额”“GMV”“汇总GMV”。汇总逻辑读取表内当月、不晚于处理日期的 GMV 记录进行累加。
+- 可选择同步完成后向飞书群发送通知。
+
+## 环境准备
+
+需要 Python 3.12 或以上版本、可连接的 PostgreSQL 数据库，以及具有目标多维表格访问和写入权限的飞书应用。
+
+以下命令均在项目根目录执行。已有 Conda / Python 环境时，先激活该环境。
+
+使用 uv 安装依赖并进入虚拟环境：
+
+```bash
+uv sync
+source .venv/bin/activate
+```
+
+也可以在已有 Python 环境中安装项目声明的依赖：
+
+```bash
+python - <<'PY'
+import subprocess
+import sys
+import tomllib
+from pathlib import Path
+
+config = tomllib.loads(Path("pyproject.toml").read_text(encoding="utf-8"))
+subprocess.check_call([sys.executable, "-m", "pip", "install", *config["project"]["dependencies"]])
+PY
+```
+
+## 配置 .env
+
+在项目根目录创建或编辑 `.env`。下面均为占位值，请替换为实际配置：
+
+```dotenv
+ANALYTICS_DB_HOST=your-db-host
+ANALYTICS_DB_PORT=5432
+ANALYTICS_DB_NAME=your-db-name
+ANALYTICS_DB_USER=your-db-user
+ANALYTICS_DB_PASSWORD=your-db-password
+
+FEISHU_APP_ID=your-app-id
+FEISHU_APP_SECRET=your-app-secret
+FEISHU_BASE_URL=https://your-tenant.feishu.cn/base/your-base-token
+
+SYNC_START_DATE=2026-09-01
+```
+
+脚本自动加载项目根目录的 `.env`，已存在的进程环境变量优先于 `.env`。当前加载器按 `KEY=value` 读取，值不要额外加引号，也不要在值后追加行内注释。
+
+`.env` 已加入 `.gitignore`，用于保存本地配置。
+
+可选配置：
+
+| 变量 | 用途与默认行为 |
+| --- | --- |
+| `PROJECT_MAPPING_FILE` | 店铺映射文件路径；不设置时使用脚本目录下的 `店铺映射表.xlsx` |
+| `SYNC_START_DATE` | 本地流水线默认开始日期；未设置时为 `2026-04-01` |
+| `FEISHU_CHAT_ID` | 应用机器人通知的目标群；配置后优先使用此方式 |
+| `FEISHU_BOT_WEBHOOK` | 未设置群 ID 时使用的群机器人 Webhook |
+| `FEISHU_REPORT_IMAGE_PATH` | 应用机器人通知使用的本地预览图路径 |
+| `FEISHU_STORE_FIELD_NAME` | 独立脚本使用的店铺字段名，默认 `店铺名称` |
+| `PREFECT_API_URL` | Prefect Server API 地址，本地流水线不需要 |
+| `LOG_DIR` | 日志目录，默认项目下的 `logs` |
+| `LOG_LEVEL` | 日志级别，默认 `INFO` |
+
+映射文件读取第一个活动工作表，跳过第一行表头。三列及以上时，使用第 2 列作为业务店铺名称、第 3 列作为 BI 店铺名称；只有两列时，依次为业务店铺名称、BI 店铺名称。
+
+## 本地执行
+
+本地入口为 `run_project_bitable_pipeline.py`，可直接使用 Python 运行，无需启动 Docker 或 Prefect Server。未激活 uv 虚拟环境时，也可以将以下命令中的 `python` 替换为 `uv run python`。
+
+### 指定一天
+
+将开始日期和结束日期设为同一天：
+
+```bash
+python run_project_bitable_pipeline.py \
+  --start-date 2026-09-07 \
+  --end-date 2026-09-07 \
+  --no-notify \
+  --skip-missing-tables
+```
+
+### 指定日期范围
+
+例如同步 2026 年 9 月 1 日至 9 月 7 日，包含首尾两天：
+
+```bash
+python run_project_bitable_pipeline.py \
+  --start-date 2026-09-01 \
+  --end-date 2026-09-07 \
+  --no-notify \
+  --skip-missing-tables
+```
+
+`--no-notify` 表示只同步数据，不发群通知。需要通知时，配置通知渠道并去掉该参数。
+
+`--skip-missing-tables` 表示找不到对应表时记录日志并继续处理其他项目和日期，不会自动创建月份表。匹配到多个表、映射指定表不存在等配置错误仍会中断运行，避免写入错误的表。
+
+### 预览执行
+
+```bash
+python run_project_bitable_pipeline.py \
+  --start-date 2026-09-01 \
+  --end-date 2026-09-07 \
+  --dry-run \
+  --no-notify \
+  --skip-missing-tables
+```
+
+`--dry-run` 会查询数据库、获取飞书访问令牌和表列表，并输出计划处理的表、日期和指标，不写入飞书、不发通知。因此仍需有效的数据库和飞书配置。
+
+### 常用参数
+
+| 参数 | 说明 |
+| --- | --- |
+| `--start-date YYYY-MM-DD` | 开始日期，默认读取 `SYNC_START_DATE` |
+| `--end-date YYYY-MM-DD` | 结束日期，默认运行机器本地日期的昨天 |
+| `--no-notify` | 跳过飞书群通知，默认未开启 |
+| `--skip-missing-tables` | 本地流水线跳过缺失月份表，默认未开启；Prefect 默认开启，并在完成后推送缺失表提醒 |
+| `--dry-run` | 预览，不执行飞书写入和通知 |
+| `--mapping-file PATH` | 指定店铺映射 Excel 文件，覆盖环境变量配置 |
+| `--table-map-file PATH` | 指定项目月份到飞书表名的 JSON 映射 |
+| `--base-url URL` | 指定飞书多维表格地址；仍需在环境中配置 `FEISHU_BASE_URL`，当前参数解析会读取其默认值 |
+
+日期范围的结束日期不能早于开始日期。本地流水线未传日期时会从默认开始日期一直同步到昨天；补数时建议明确传入两个日期。
+
+完整参数可在配置好 `.env` 后查看：
+
+```bash
+python run_project_bitable_pipeline.py --help
+```
+
+## 常见问题
+
+### 映射表路径不存在
+
+如果出现 `FileNotFoundError`，且路径仍指向其他机器上的 `/Users/idss/...`，请修改 `.env` 中的 `PROJECT_MAPPING_FILE`，或删除该配置以使用项目自带文件。
+
+也可以在命令中添加以下参数，直接覆盖旧路径：
+
+```bash
+--mapping-file ./店铺映射表.xlsx
+```
+
+### 找不到对应月份的飞书表
+
+例如“羽心堂”只有 3—8 月表，执行 9 月数据时会报 `未找到匹配表`。临时跳过可添加 `--skip-missing-tables`；补齐月份表后，再运行对应日期范围。
+
+### 表名不规范或匹配到多个表
+
+创建 `table_map.json`，键格式为 `项目:YYYY-MM`，值为飞书中实际存在的完整表名：
+
+```json
+{
+  "大眼睛:2026-09": "大眼睛买买买-9月",
+  "羽心堂:2026-09": "羽心堂-9月"
+}
+```
+
+执行时添加 `--table-map-file ./table_map.json`。自动匹配支持项目别名和月份文本，跨年份存在同名月份表时应显式指定映射。
+
+## Prefect 执行与定时调度
+
+手动触发一次 Prefect 数据同步：
+
+```bash
+python prefect_gmv_scheduler.py \
+  --mode sync \
+  --start-date 2026-09-01 \
+  --end-date 2026-09-07 \
+  --skip-missing-tables
+```
+
+| 模式 | 行为 |
+| --- | --- |
+| `--mode sync` | 只同步数据 |
+| `--mode notify` | 只发送通知 |
+| `--mode all` | 先同步再通知，默认模式 |
+
+Prefect 同步未显式指定日期时，默认处理截至昨天的最近 5 天，并以 `SYNC_START_DATE` 为最早边界。缺失月份表会自动跳过，其余表仍会写入；同步结束后，缺失的“项目 + 月份”会汇总为一条飞书提醒。通知未指定日期时，优先读取上次成功同步保存的日期范围。
+
+`prefect.yaml` 定义了北京时间每天 08:00 同步、09:00 通知两条调度。部署前需准备可访问的 Prefect Server，并配置 `PREFECT_API_URL`。
+
+```bash
+docker compose up -d --build
+docker exec -it prefect-gmv-worker sh -lc 'cd /app && prefect deploy --all'
+```
+
+Compose 启动的是等待调度任务的 worker。当前 Prefect deployment 默认跳过缺失月份表，并在同步结束后推送一条汇总提醒。修改 `prefect.yaml` 后，执行 `prefect deploy --all` 注册新配置。
+
+更多部署说明见项目中的 `DOCKER_PREFECT_USAGE.md`；其中示例的本机目录需要替换为实际项目目录。
+
+## 文件说明与日志
+
+| 文件 | 用途 |
+| --- | --- |
+| `run_project_bitable_pipeline.py` | 本地完整流水线入口，使用日期 / 指标行结构 |
+| `project_bitable_common.py` | SQL、店铺映射、表匹配、飞书 API、同步与通知逻辑 |
+| `prefect_gmv_scheduler.py` | Prefect flow、日期窗口和执行模式 |
+| `generate_project_sales_csv.py` | 独立销售额同步脚本 |
+| `generate_project_refund_csv.py` | 独立退款额同步脚本 |
+| `generate_project_gmv_csv.py` | 从飞书销售额和退款额字段计算并回写 GMV |
+| `店铺映射表.xlsx` | BI 店铺与业务店铺名称映射 |
+| `logging_config.py` | 控制台与轮转文件日志 |
+| `prefect.yaml` | 定时任务与 work pool 配置 |
+| `docker-compose.yaml` / `Dockerfile` | 容器运行与构建配置 |
+
+三个 `generate_project_*_csv.py` 文件虽带有 `csv` 后缀，实际执行飞书同步；它们使用以店铺为记录、日期指标为字段的结构，与当前完整流水线的表结构不同。日常运行使用完整流水线入口。
+
+默认日志写入 `logs/gmv_sync.log`，单个文件达到 10 MiB 后轮转，保留 7 个备份。
+
+```bash
+tail -f logs/gmv_sync.log
+```
+
+Docker worker 日志：
+
+```bash
+docker logs -f prefect-gmv-worker
+```
@@ -7,14 +7,14 @@ pull: []

 deployments:
  - name: gmv-daily-data-sync-8am
-    version: "1.0.0"
+    version: "1.0.1"
    description: "每天上午 8 点执行 GMV 数据同步并写入飞书多维表格。"
    entrypoint: prefect_gmv_scheduler.py:gmv_daily_data_sync_flow
    parameters:
      start_date: null
      end_date: null
      dry_run: false
-      skip_missing_tables: false
+      skip_missing_tables: true
      table_map_file: ""
    tags:
      - gmv
@@ -29,14 +29,14 @@ deployments:
      name: gmv-analytics-pool

  - name: gmv-daily-notify-9am
-    version: "1.0.0"
+    version: "1.0.1"
    description: "每天上午 9 点推送 GMV 数据同步结果到飞书群。"
    entrypoint: prefect_gmv_scheduler.py:gmv_daily_notify_flow
    parameters:
      start_date: null
      end_date: null
      dry_run: false
-      skip_missing_tables: false
+      skip_missing_tables: true
      table_map_file: ""
    tags:
      - gmv
@@ -39,6 +39,7 @@ from project_bitable_common import (
    load_dotenv,
    log_step,
    run_daily_sync_for_date,
+    send_missing_table_notification,
    send_sync_notification,
    validate_date,
 )
@@ -72,7 +73,7 @@ def load_sync_config(
    start_date: str | None = None,
    end_date: str | None = None,
    dry_run: bool = False,
-    skip_missing_tables: bool = False,
+    skip_missing_tables: bool = True,
    table_map_file: str = "",
 ) -> SyncConfig:
    load_dotenv()
@@ -174,7 +175,7 @@ def gmv_daily_data_sync_flow(
    start_date: str | None = None,
    end_date: str | None = None,
    dry_run: bool = False,
-    skip_missing_tables: bool = False,
+    skip_missing_tables: bool = True,
    table_map_file: str = "",
 ) -> None:
    config = load_sync_config(
@@ -205,6 +206,14 @@ def gmv_daily_data_sync_flow(
        run_daily_sync_for_date(context, stat_date)
    if not config.dry_run:
        mark_sync_window_completed(sync_window)
+    if context.missing_tables:
+        log_step(f"Prefect 数据同步发现缺失表: {len(context.missing_tables)} 个")
+        send_missing_table_notification(
+            context,
+            context.missing_tables,
+            sync_window.start_date,
+            sync_window.end_date,
+        )
    log_step("Prefect 数据同步完成")


@@ -213,7 +222,7 @@ def gmv_daily_notify_flow(
    start_date: str | None = None,
    end_date: str | None = None,
    dry_run: bool = False,
-    skip_missing_tables: bool = False,
+    skip_missing_tables: bool = True,
    table_map_file: str = "",
 ) -> None:
    config = load_sync_config(
@@ -235,7 +244,12 @@ def parse_args() -> argparse.Namespace:
    parser.add_argument("--end-date", default="", help="结束日期，格式 YYYY-MM-DD，默认昨天")
    parser.add_argument("--table-map-file", default="", help="可选，项目到表名的映射 JSON 文件路径")
    parser.add_argument("--dry-run", action="store_true", help="只输出将要更新的内容，不调用飞书写接口")
-    parser.add_argument("--skip-missing-tables", action="store_true", help="找不到对应月份的飞书表时跳过")
+    parser.add_argument(
+        "--skip-missing-tables",
+        action=argparse.BooleanOptionalAction,
+        default=True,
+        help="找不到对应月份的飞书表时跳过并在同步结束后推送提醒（默认开启）",
+    )
    parser.add_argument("--serve", action="store_true", help="启动 Prefect deployments，按 8 点同步、9 点通知 定时执行")
    parser.add_argument(
        "--mode",
@@ -1,8 +1,9 @@
 import json
 import os
 import re
+import ssl
 from collections import defaultdict
-from dataclasses import dataclass
+from dataclasses import dataclass, field
 from datetime import date, datetime, timedelta, timezone
 from decimal import Decimal
 from pathlib import Path
@@ -12,6 +13,8 @@ from urllib.parse import urlparse
 import psycopg2
 import requests
 from openpyxl import load_workbook
+from requests.adapters import HTTPAdapter
+from urllib3.util.retry import Retry

 from logging_config import get_logger, setup_logging

@@ -23,6 +26,51 @@ DEFAULT_START_DATE = "2026-04-01"
 DEFAULT_STORE_FIELD_NAME = "店铺名称"
 STORE_FIELD_CANDIDATES = ["业务店铺名称", "店铺名称", "门店名称", "店铺", "名称"]
 SUPPORTED_PROJECTS = ["大眼睛", "羽心堂", "MMMJ", "EM", "春回盏", "BROKEN ZOO"]
+FEISHU_HTTP_RETRY_COUNT = 3
+
+
+class TLS12HTTPAdapter(HTTPAdapter):
+    """Requests adapter that keeps the TLS policy consistent for Feishu calls."""
+
+    def __init__(self, *args: Any, **kwargs: Any) -> None:
+        self._ssl_context = ssl.create_default_context()
+        # TLS 1.3 handshakes are reset by the current server-side network path.
+        # Keep certificate verification enabled while using the compatible TLS 1.2.
+        self._ssl_context.minimum_version = ssl.TLSVersion.TLSv1_2
+        self._ssl_context.maximum_version = ssl.TLSVersion.TLSv1_2
+        super().__init__(*args, **kwargs)
+
+    def init_poolmanager(self, connections: int, maxsize: int, block: bool = False, **pool_kwargs: Any) -> None:
+        pool_kwargs["ssl_context"] = self._ssl_context
+        super().init_poolmanager(connections, maxsize, block=block, **pool_kwargs)
+
+    def proxy_manager_for(self, proxy: str, **proxy_kwargs: Any) -> Any:
+        proxy_kwargs["ssl_context"] = self._ssl_context
+        return super().proxy_manager_for(proxy, **proxy_kwargs)
+
+
+def create_feishu_session(*, retry_post: bool = False) -> requests.Session:
+    """Create a verified TLS 1.2 session for Feishu APIs with bounded retries.
+
+    Only the token endpoint opts into POST retries because obtaining a token is
+    idempotent. Data-writing POST requests retain their original at-most-once
+    behaviour, while all calls still use TLS 1.2.
+    """
+
+    retry = Retry(
+        total=FEISHU_HTTP_RETRY_COUNT,
+        connect=FEISHU_HTTP_RETRY_COUNT,
+        read=FEISHU_HTTP_RETRY_COUNT,
+        status=0,
+        other=0,
+        backoff_factor=1,
+        allowed_methods=None if retry_post else frozenset({"GET", "HEAD", "OPTIONS"}),
+        raise_on_status=False,
+    )
+    session = requests.Session()
+    adapter = TLS12HTTPAdapter(max_retries=retry)
+    session.mount("https://", adapter)
+    return session


@dataclass(frozen=True)
@@ -46,6 +94,7 @@ WITH base_data AS (
        f.店铺主键,
        s.店铺名称 AS 原始店铺名称,
        p.品牌名称,
+        p.商品名称,
        f.业务模式,
        f.实付分摊金额
    FROM analytics_layer.fact_sales f
@@ -64,23 +113,30 @@ brand_split AS (
        原始店铺名称,
        业务模式,
        CASE
+            WHEN COALESCE(原始店铺名称, '') LIKE '%%羽心堂分销%%'
+                THEN '羽心堂分销'

-            WHEN 店铺主键 IN (1,28,29) AND 品牌名称 LIKE '羽心堂'
+            WHEN 店铺主键 IN (1,28,29) AND COALESCE(品牌名称, '') LIKE '羽心堂'
                THEN 原始店铺名称 || '_羽心堂'
-            WHEN 店铺主键 IN (1,28,29) AND 品牌名称 LIKE '春回盏'
+            WHEN 店铺主键 IN (1,28,29) AND COALESCE(品牌名称, '') LIKE '春回盏'
                THEN 原始店铺名称 || '_春回盏'
-            WHEN 店铺主键 IN (1,28,29) AND 品牌名称 LIKE 'Mama Mujer'
+            WHEN 店铺主键 IN (1,28,29) AND COALESCE(品牌名称, '') LIKE 'Mama Mujer'
                THEN 原始店铺名称 || '_Mama Mujer'
-            WHEN 店铺主键 IN (1,28,29) AND 品牌名称 LIKE 'ERBA MIGINA'
+            WHEN 店铺主键 IN (1,28,29)
+                AND (
+                    COALESCE(品牌名称, '') LIKE 'ERBA MIGINA'
+                    OR COALESCE(商品名称, '') ILIKE '%%ERBA MIGINA%%'
+                    OR COALESCE(商品名称, '') ~* '(^|[^A-Za-z])EM([^A-Za-z]|$)'
+                )
                THEN 原始店铺名称 || '_ERBA MIGINA'
-            WHEN 店铺主键 IN (1,28,29) AND 品牌名称 LIKE 'BROKEN ZOO'
+            WHEN 店铺主键 IN (1,28,29) AND COALESCE(品牌名称, '') LIKE 'BROKEN ZOO'
                THEN 原始店铺名称 || '_BROKEN ZOO'
            WHEN 店铺主键 IN (1,28,29)
                THEN 原始店铺名称 || '_其他'

-            WHEN 店铺主键 IN (3,9) AND 品牌名称 LIKE '羽心堂'
+            WHEN 店铺主键 IN (3,9) AND COALESCE(品牌名称, '') LIKE '羽心堂'
                THEN 原始店铺名称 || '_羽心堂'
-            WHEN 店铺主键 IN (3,9) AND 品牌名称 LIKE 'Mama Mujer'
+            WHEN 店铺主键 IN (3,9) AND COALESCE(品牌名称, '') LIKE 'Mama Mujer'
                THEN 原始店铺名称 || '_Mama Mujer'
            WHEN 店铺主键 IN (3,9)
                THEN 原始店铺名称 || '_其他'
@@ -103,8 +159,8 @@ shop_brand_detail AS (
        SUM(实付分摊金额) AS 总销售额
    FROM brand_split
    GROUP BY 1,2,3
-)
-
+),
+union_data AS (
    SELECT
        统计日期,
        店铺名称,
@@ -117,18 +173,16 @@ WHERE NOT (

    UNION ALL

-- 店铺合计（使用 dim_shop 店铺名）
    SELECT
        统计日期,
        原始店铺名称 AS 店铺名称,
-    SUM(实付分摊金额)
+        SUM(实付分摊金额) AS 总销售额
    FROM base_data
    WHERE 店铺主键 IN (1,28,29,3)
    GROUP BY 1,2

    UNION ALL

-- 权益金（充值购物金单独统计）
    SELECT
        统计日期,
        原始店铺名称 || '_权益金' AS 店铺名称,
@@ -137,6 +191,14 @@ FROM base_data
    WHERE 店铺主键 IN (5,11,14,22,26)
      AND 业务模式 = '充值购物金'
    GROUP BY 1,2
+)
+
+SELECT
+    统计日期,
+    店铺名称,
+    SUM(总销售额) AS 总销售额
+FROM union_data
+GROUP BY 1,2

 ORDER BY 统计日期 DESC, 总销售额 DESC;
 """,
@@ -147,13 +209,53 @@ ORDER BY 统计日期 DESC, 总销售额 DESC;
        field_suffix="退款额",
        sql_template="""
 SET TIMEZONE = 'Asia/Shanghai';
-WITH base_data AS (
+WITH sales_base AS (
+    SELECT
+        fs.*,
+        TRIM(fs."平台订单号") AS "原平台订单号",
+        split_part(TRIM(fs."平台订单号"), '-', 1) AS "主订单号"
+    FROM analytics_layer.fact_sales fs
+    WHERE fs."订单状态" IN ('已关闭', '已取消', '交易关闭')
+      AND NULLIF(TRIM(fs."平台订单号"), '') IS NOT NULL
+      AND fs."支付时间" IS NOT NULL
+      AND fs."支付时间" >= %(start_ts)s
+      AND fs."支付时间" < %(end_ts_exclusive)s
+),
+
+missing_order_keys AS (
+    SELECT DISTINCT
+        sb."主订单号"
+    FROM sales_base sb
+    WHERE NOT EXISTS (
+        SELECT 1
+        FROM analytics_layer.fact_returns fr
+        WHERE split_part(TRIM(fr."平台订单号"), '-', 1) = sb."主订单号"
+          AND NULLIF(TRIM(fr."平台订单号"), '') IS NOT NULL
+    )
+),
+
+missing_return_sales AS (
+    SELECT
+        sb."支付时间"::DATE AS 统计日期,
+        sb."店铺主键",
+        SUM(sb."实付分摊金额") AS 补充退款金额
+    FROM sales_base sb
+    JOIN missing_order_keys mok
+        ON sb."主订单号" = mok."主订单号"
+    WHERE sb."店铺主键" NOT IN (5,11,14,22,26)
+    GROUP BY
+        sb."支付时间"::DATE,
+        sb."店铺主键"
+),
+
+base_data AS (
    SELECT
        f.完结时间::DATE AS 统计日期,
        f.店铺主键,
        f.平台订单号,
        s.店铺名称 AS 原始店铺名称,
        p.品牌名称,
+        p.商品名称,
        f.退款金额
    FROM analytics_layer.fact_returns f
    LEFT JOIN analytics_layer.dim_shop s
@@ -165,10 +267,25 @@ WITH base_data AS (
      AND f.完结时间 < %(end_ts_exclusive)s
 ),

-recharge_orders AS (
-    SELECT DISTINCT 平台订单号, 店铺主键
-    FROM analytics_layer.fact_sales
-    WHERE 业务模式 = '充值购物金'
+sales_order_flag AS (
+    SELECT
+        fs.平台订单号,
+        fs.店铺主键,
+        TRUE AS 订单存在于销售表,
+        BOOL_OR(fs.业务模式 = '充值购物金') AS 是否充值购物金订单
+    FROM analytics_layer.fact_sales fs
+    INNER JOIN (
+        SELECT DISTINCT
+            平台订单号,
+            店铺主键
+        FROM base_data
+        WHERE 店铺主键 IN (5,11,14,22,26)
+    ) r
+        ON fs.平台订单号 = r.平台订单号
+       AND fs.店铺主键 = r.店铺主键
+    GROUP BY
+        fs.平台订单号,
+        fs.店铺主键
 ),

 brand_split AS (
@@ -178,22 +295,27 @@ brand_split AS (
        原始店铺名称,
        CASE

-            WHEN 店铺主键 IN (1,28,29) AND 品牌名称 LIKE '羽心堂'
+            WHEN 店铺主键 IN (1,28,29) AND COALESCE(品牌名称, '') LIKE '羽心堂'
                THEN 原始店铺名称 || '_羽心堂'
-            WHEN 店铺主键 IN (1,28,29) AND 品牌名称 LIKE '春回盏'
+            WHEN 店铺主键 IN (1,28,29) AND COALESCE(品牌名称, '') LIKE '春回盏'
                THEN 原始店铺名称 || '_春回盏'
-            WHEN 店铺主键 IN (1,28,29) AND 品牌名称 LIKE 'Mama Mujer'
+            WHEN 店铺主键 IN (1,28,29) AND COALESCE(品牌名称, '') LIKE 'Mama Mujer'
                THEN 原始店铺名称 || '_Mama Mujer'
-            WHEN 店铺主键 IN (1,28,29) AND 品牌名称 LIKE 'ERBA MIGINA'
+            WHEN 店铺主键 IN (1,28,29)
+                AND (
+                    COALESCE(品牌名称, '') LIKE 'ERBA MIGINA'
+                    OR COALESCE(商品名称, '') ILIKE '%%ERBA MIGINA%%'
+                    OR COALESCE(商品名称, '') ~* '(^|[^A-Za-z])EM([^A-Za-z]|$)'
+                )
                THEN 原始店铺名称 || '_ERBA MIGINA'
-            WHEN 店铺主键 IN (1,28,29) AND 品牌名称 LIKE 'BROKEN ZOO'
+            WHEN 店铺主键 IN (1,28,29) AND COALESCE(品牌名称, '') LIKE 'BROKEN ZOO'
                THEN 原始店铺名称 || '_BROKEN ZOO'
            WHEN 店铺主键 IN (1,28,29)
                THEN 原始店铺名称 || '_其他'

-            WHEN 店铺主键 IN (3,9) AND 品牌名称 LIKE '羽心堂'
+            WHEN 店铺主键 IN (3,9) AND COALESCE(品牌名称, '') LIKE '羽心堂'
                THEN 原始店铺名称 || '_羽心堂'
-            WHEN 店铺主键 IN (3,9) AND 品牌名称 LIKE 'Mama Mujer'
+            WHEN 店铺主键 IN (3,9) AND COALESCE(品牌名称, '') LIKE 'Mama Mujer'
                THEN 原始店铺名称 || '_Mama Mujer'
            WHEN 店铺主键 IN (3,9)
                THEN 原始店铺名称 || '_其他'
@@ -215,8 +337,9 @@ shop_brand_detail AS (
        SUM(退款金额) AS 总退款金额
    FROM brand_split
    GROUP BY 1,2,3
-)
+),

+union_data AS (
    SELECT
        统计日期,
        店铺名称,
@@ -229,7 +352,6 @@ WHERE NOT (

    UNION ALL

-- 店铺合计
    SELECT
        统计日期,
        原始店铺名称 AS 店铺名称,
@@ -240,18 +362,42 @@ GROUP BY 1,2

    UNION ALL

-- 权益金退款
    SELECT
        r.统计日期,
        r.原始店铺名称 || '_权益金' AS 店铺名称,
        SUM(r.退款金额) AS 总退款金额
    FROM base_data r
-JOIN recharge_orders ro
-  ON r.平台订单号 = ro.平台订单号
- AND r.店铺主键 = ro.店铺主键
+    LEFT JOIN sales_order_flag sof
+      ON r.平台订单号 = sof.平台订单号
+     AND r.店铺主键 = sof.店铺主键
    WHERE r.店铺主键 IN (5,11,14,22,26)
+      AND (
+            sof.订单存在于销售表 IS NULL
+            OR sof.是否充值购物金订单 = TRUE
+          )
    GROUP BY 1,2

+    UNION ALL
+
+    SELECT
+        mrs.统计日期,
+        COALESCE(s.店铺名称, '未知店铺') AS 店铺名称,
+        SUM(mrs.补充退款金额) AS 总退款金额
+    FROM missing_return_sales mrs
+    LEFT JOIN analytics_layer.dim_shop s
+        ON mrs.店铺主键 = s.店铺主键
+    GROUP BY 1,2
+)
+
+SELECT
+    统计日期,
+    店铺名称,
+    SUM(总退款金额) AS 总退款金额
+FROM union_data
+GROUP BY
+    统计日期,
+    店铺名称
+
 ORDER BY 统计日期 DESC, 总退款金额 DESC;
 """,
    ),
@@ -282,6 +428,17 @@ class MetricSyncArtifacts:
    created_fields: set[str]


+@dataclass(frozen=True, order=True)
+class MissingTable:
+    project: str
+    year: int
+    month: int
+
+
+class TargetTableNotFoundError(ValueError):
+    """Raised when a project's monthly Feishu table does not exist."""
+
+
@dataclass(frozen=True)
 class DailySyncConfig:
    base_url: str
@@ -302,6 +459,7 @@ class DailySyncContext:
    mapping: dict[str, str]
    table_overrides: dict[str, str]
    config: DailySyncConfig
+    missing_tables: set[MissingTable] = field(default_factory=set)


 def _log_to_prefect(level_name: str, message: str) -> None:
@@ -430,7 +588,7 @@ def classify_project(store_name: str) -> str:

 def get_project_table_aliases(project: str) -> list[str]:
    aliases = {
-        "大眼睛": ["大眼睛", "大眼睛买买买"],
+        "大眼睛": ["大眼睛买买买"],
        "羽心堂": ["羽心堂"],
        "MMMJ": ["MMMJ", "Mama Mujer", "MAMA MUJER"],
        "EM": ["EM", "ERBA MIGINA"],
@@ -569,8 +727,9 @@ def get_feishu_token() -> str:
    app_secret = os.getenv("FEISHU_APP_SECRET")
    if not app_id or not app_secret:
        raise ValueError("缺少飞书环境变量 FEISHU_APP_ID / FEISHU_APP_SECRET")
-    log_step("开始获取飞书 tenant_access_token")
-    resp = requests.post(
+    log_step(f"开始获取飞书 tenant_access_token（TLS 1.2，最多重试 {FEISHU_HTTP_RETRY_COUNT} 次）")
+    with create_feishu_session(retry_post=True) as session:
+        resp = session.post(
            "https://open.feishu.cn/open-apis/auth/v3/tenant_access_token/internal",
            json={"app_id": app_id, "app_secret": app_secret},
            timeout=30,
@@ -592,7 +751,7 @@ class FeishuBitableClient:
        self._fields_cache: dict[str, list[dict[str, Any]]] = {}
        self._records_cache: dict[str, list[dict[str, Any]]] = {}
        self._tables_cache: list[TableInfo] | None = None
-        self.session = requests.Session()
+        self.session = create_feishu_session()
        self.session.headers.update(
            {
                "Authorization": f"Bearer {self.token}",
@@ -840,6 +999,7 @@ def send_feishu_sync_card(
    base_url: str,
    start_date: date,
    end_date: date,
+    extra_content: str = "",
 ) -> None:
    if not webhook_url:
        log_step("未配置飞书群机器人 webhook，跳过群消息推送")
@@ -860,6 +1020,7 @@ def send_feishu_sync_card(
                        "content": (
                            f"**同步范围**\n{start_date.isoformat()} 至 {end_date.isoformat()}\n\n"
                            f"**多维表格**\n[点击打开多维表格]({base_url})"
+                            + (f"\n\n{extra_content}" if extra_content else "")
                        ),
                    },
                }
@@ -867,7 +1028,8 @@ def send_feishu_sync_card(
        },
    }
    log_step("开始推送飞书群卡片消息")
-    resp = requests.post(webhook_url, json=card, timeout=30)
+    with create_feishu_session() as session:
+        resp = session.post(webhook_url, json=card, timeout=30)
    resp.raise_for_status()
    payload = resp.json()
    if payload.get("code") not in (0, None):
@@ -883,11 +1045,18 @@ def upload_feishu_image(image_path: str, token: str | None = None) -> str:
        raise FileNotFoundError(f"未找到预览图文件: {path}")
    token = token or get_feishu_token()
    headers = {"Authorization": f"Bearer {token}"}
+    with create_feishu_session() as session:
        with path.open("rb") as image_file:
            files = {"image": (path.name, image_file, "application/octet-stream")}
            data = {"image_type": "message"}
            log_step(f"开始上传飞书消息图片: {path}")
-        resp = requests.post("https://open.feishu.cn/open-apis/im/v1/images", headers=headers, data=data, files=files, timeout=60)
+            resp = session.post(
+                "https://open.feishu.cn/open-apis/im/v1/images",
+                headers=headers,
+                data=data,
+                files=files,
+                timeout=60,
+            )
    resp.raise_for_status()
    payload = resp.json()
    if payload.get("code") != 0:
@@ -907,6 +1076,7 @@ def send_feishu_app_card(
    end_date: date,
    image_path: str = "",
    token: str | None = None,
+    extra_content: str = "",
 ) -> None:
    if not chat_id:
        log_step("未配置 FEISHU_CHAT_ID，跳过应用机器人卡片推送")
@@ -921,6 +1091,7 @@ def send_feishu_app_card(
                "content": (
                    f"**同步范围**\n{start_date.isoformat()} 至 {end_date.isoformat()}\n\n"
                    f"**多维表格**\n[{title}]({base_url})"
+                    + (f"\n\n{extra_content}" if extra_content else "")
                ),
            },
        }
@@ -957,7 +1128,8 @@ def send_feishu_app_card(
        "Content-Type": "application/json; charset=utf-8",
    }
    log_step(f"开始通过应用机器人发送群卡片: chat_id={chat_id}")
-    resp = requests.post(
+    with create_feishu_session() as session:
+        resp = session.post(
            "https://open.feishu.cn/open-apis/im/v1/messages?receive_id_type=chat_id",
            headers=headers,
            json=message,
@@ -1020,7 +1192,7 @@ def find_target_table(
        return candidates[0]
    if not candidates:
        all_table_names = [table.name for table in tables]
-        raise ValueError(
+        raise TargetTableNotFoundError(
            f"未找到匹配表: 项目={project}, 月份={year:04d}-{month:02d}, 项目相关表={project_related_tables or '[]'}, 全部表={all_table_names}"
        )
    raise ValueError(
@@ -1617,8 +1789,15 @@ def run_daily_sync_for_date(context: DailySyncContext, stat_date: date) -> None:
        }
        try:
            table = find_target_table(context.tables, project, stat_date.year, stat_date.month, context.table_overrides)
-        except ValueError as exc:
+        except TargetTableNotFoundError as exc:
            if context.config.skip_missing_tables:
+                missing_table = MissingTable(project=project, year=stat_date.year, month=stat_date.month)
+                if missing_table not in context.missing_tables:
+                    context.missing_tables.add(missing_table)
+                    log_step(
+                        "记录缺失表: "
+                        f"项目={project}, 月份={stat_date.year:04d}-{stat_date.month:02d}"
+                    )
                log_step(f"跳过: {exc}")
                continue
            raise
@@ -1717,3 +1896,43 @@ def send_sync_notification(
        start_date=start_date,
        end_date=end_date,
    )
+
+
+def send_missing_table_notification(
+    context: DailySyncContext,
+    missing_tables: set[MissingTable],
+    start_date: date,
+    end_date: date,
+) -> None:
+    """Notify Feishu once after a sync run about tables skipped because they do not exist."""
+    if context.config.dry_run or not missing_tables:
+        return
+
+    table_lines = "\n".join(
+        f"- {item.project}：{item.year:04d}-{item.month:02d}" for item in sorted(missing_tables)
+    )
+    extra_content = (
+        "**以下月份表不存在，已跳过写入**\n"
+        f"{table_lines}\n\n"
+        "请创建对应月份表后重新执行补数。"
+    )
+    title = "GMV 数据同步：缺失表提醒"
+    if context.config.chat_id:
+        send_feishu_app_card(
+            chat_id=context.config.chat_id,
+            title=title,
+            base_url=context.config.base_url,
+            start_date=start_date,
+            end_date=end_date,
+            token=context.client.token,
+            extra_content=extra_content,
+        )
+        return
+    send_feishu_sync_card(
+        webhook_url=context.config.group_webhook,
+        title=title,
+        base_url=context.config.base_url,
+        start_date=start_date,
+        end_date=end_date,
+        extra_content=extra_content,
+    )
@@ -34,6 +34,7 @@ def parse_args() -> argparse.Namespace:
    parser.add_argument("--table-map-file", default="", help="可选，项目到表名的映射 JSON 文件路径")
    parser.add_argument("--dry-run", action="store_true", help="只输出将要更新的内容，不调用飞书写接口")
    parser.add_argument("--skip-missing-tables", action="store_true", help="找不到对应月份的飞书表时跳过")
+    parser.add_argument("--no-notify", action="store_true", help="只写入多维表格，不推送飞书群通知")
    return parser.parse_args()


@@ -61,6 +62,9 @@ def main() -> None:
    )
    for stat_date in iter_dates(start_date, end_date):
        run_daily_sync_for_date(context, stat_date)
+    if args.no_notify:
+        log_step("已跳过飞书群通知")
+    else:
        send_sync_notification(context, start_date, end_date)
    log_step("流水线完成")

@@ -0,0 +1,29 @@
+import ssl
+
+from project_bitable_common import FEISHU_HTTP_RETRY_COUNT, TLS12HTTPAdapter, create_feishu_session
+
+
+def test_feishu_session_forces_tls_12() -> None:
+    session = create_feishu_session()
+    try:
+        adapter = session.get_adapter("https://open.feishu.cn")
+        assert isinstance(adapter, TLS12HTTPAdapter)
+        context = adapter._ssl_context
+        assert context.minimum_version == ssl.TLSVersion.TLSv1_2
+        assert context.maximum_version == ssl.TLSVersion.TLSv1_2
+    finally:
+        session.close()
+
+
+def test_token_session_can_retry_post_but_regular_session_cannot() -> None:
+    regular_session = create_feishu_session()
+    token_session = create_feishu_session(retry_post=True)
+    try:
+        regular_retry = regular_session.get_adapter("https://open.feishu.cn").max_retries
+        token_retry = token_session.get_adapter("https://open.feishu.cn").max_retries
+        assert regular_retry.total == FEISHU_HTTP_RETRY_COUNT
+        assert regular_retry.allowed_methods == frozenset({"GET", "HEAD", "OPTIONS"})
+        assert token_retry.allowed_methods is None
+    finally:
+        regular_session.close()
+        token_session.close()
Author	SHA1	Message	Date
BXY	152cb5a65b	Add environment configuration, enhance Feishu session handling, and implement tests for TLS 1.2 and retry logic	2026-09-28 11:20:14 +08:00
BXY	a71dd66504	Prefect 同步跳过缺失表并推送飞书提醒	2026-09-11 11:13:16 +08:00
BXY	0d1ff763d7	Prefect 同步跳过缺失表并推送飞书提醒	2026-09-11 11:13:03 +08:00
BXY	89dfc33c24	更新文档	2026-09-08 11:22:42 +08:00
BXY0414	b5ee0c490b	添加大眼睛美美美	2026-07-06 17:16:32 +08:00
BXY0414	8e93355183	添加店铺	2026-06-04 13:56:44 +08:00
BXY0414	d5af7aaf6c	调整	2026-06-04 10:55:50 +08:00
BXY0414	04c87306f0	调整天猫退款的计算	2026-06-04 10:45:24 +08:00
BXY0414	8a6a0e3858	调整大眼睛EM的统计添加羽心堂分销	2026-05-07 17:55:42 +08:00
BXY0414	6f38209efe	调整退款额计算添加销售为已取消、已关闭、交易关闭情况下退款中未计算到的金额	2026-04-28 18:21:54 +08:00