Prefect 同步跳过缺失表并推送飞书提醒

README.md

@@ -107,7 +107,7 @@ python run_project_bitable_pipeline.py \
 
 `--no-notify` 表示只同步数据，不发群通知。需要通知时，配置通知渠道并去掉该参数。
 
-`--skip-missing-tables` 表示找不到对应表时记录日志并继续处理其他项目和日期，不会自动创建月份表。当前实现也会跳过匹配到多个表、映射指定表不存在等表匹配错误；应查看跳过日志确认哪些项目尚未同步。数据库连接或飞书写入失败等其他错误仍会中断运行。
+`--skip-missing-tables` 表示找不到对应表时记录日志并继续处理其他项目和日期，不会自动创建月份表。匹配到多个表、映射指定表不存在等配置错误仍会中断运行，避免写入错误的表。
 
 ### 预览执行
 
@@ -129,7 +129,7 @@ python run_project_bitable_pipeline.py \
 | `--start-date YYYY-MM-DD` | 开始日期，默认读取 `SYNC_START_DATE` |
 | `--end-date YYYY-MM-DD` | 结束日期，默认运行机器本地日期的昨天 |
 | `--no-notify` | 跳过飞书群通知，默认未开启 |
-| `--skip-missing-tables` | 跳过表匹配错误，默认未开启 |
+| `--skip-missing-tables` | 本地流水线跳过缺失月份表，默认未开启；Prefect 默认开启，并在完成后推送缺失表提醒 |
 | `--dry-run` | 预览，不执行飞书写入和通知 |
 | `--mapping-file PATH` | 指定店铺映射 Excel 文件，覆盖环境变量配置 |
 | `--table-map-file PATH` | 指定项目月份到飞书表名的 JSON 映射 |
@@ -190,7 +190,7 @@ python prefect_gmv_scheduler.py \
 | `--mode notify` | 只发送通知 |
 | `--mode all` | 先同步再通知，默认模式 |
 
-Prefect 同步未显式指定日期时，默认处理截至昨天的最近 5 天，并以 `SYNC_START_DATE` 为最早边界。通知未指定日期时，优先读取上次成功同步保存的日期范围。
+Prefect 同步未显式指定日期时，默认处理截至昨天的最近 5 天，并以 `SYNC_START_DATE` 为最早边界。缺失月份表会自动跳过，其余表仍会写入；同步结束后，缺失的“项目 + 月份”会汇总为一条飞书提醒。通知未指定日期时，优先读取上次成功同步保存的日期范围。
 
 `prefect.yaml` 定义了北京时间每天 08:00 同步、09:00 通知两条调度。部署前需准备可访问的 Prefect Server，并配置 `PREFECT_API_URL`。
 
@@ -199,7 +199,7 @@ docker compose up -d --build
 docker exec -it prefect-gmv-worker sh -lc 'cd /app && prefect deploy --all'
 ```
 
-Compose 启动的是等待调度任务的 worker。定时任务需要跳过缺失表时，将 `prefect.yaml` 对应 deployment 的 `parameters.skip_missing_tables` 改为 `true`，再重新执行 `prefect deploy --all` 注册配置。
+Compose 启动的是等待调度任务的 worker。当前 Prefect deployment 默认跳过缺失月份表，并在同步结束后推送一条汇总提醒。修改 `prefect.yaml` 后，执行 `prefect deploy --all` 注册新配置。
 
 更多部署说明见项目中的 `DOCKER_PREFECT_USAGE.md`；其中示例的本机目录需要替换为实际项目目录。
 

prefect.yaml

@@ -14,7 +14,7 @@ deployments:
       start_date: null
       end_date: null
       dry_run: false
-      skip_missing_tables: false
+      skip_missing_tables: true
       table_map_file: ""
     tags:
       - gmv
@@ -36,7 +36,7 @@ deployments:
       start_date: null
       end_date: null
       dry_run: false
-      skip_missing_tables: false
+      skip_missing_tables: true
       table_map_file: ""
     tags:
       - gmv

prefect_gmv_scheduler.py

@@ -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",

project_bitable_common.py

@@ -2,7 +2,7 @@ import json
 import os
 import re
 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
@@ -380,6 +380,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
@@ -400,6 +411,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:
@@ -938,6 +950,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，跳过群消息推送")
@@ -958,6 +971,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 "")
                         ),
                     },
                 }
@@ -1005,6 +1019,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，跳过应用机器人卡片推送")
@@ -1019,6 +1034,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 "")
                 ),
             },
         }
@@ -1118,7 +1134,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(
@@ -1715,8 +1731,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
@@ -1815,3 +1838,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,
+    )