From a71dd66504320b03ba8cb4b37c3bcd00ed0b8a3f Mon Sep 17 00:00:00 2001 From: BXY <103352887+BXY0414@users.noreply.github.com> Date: Fri, 11 Sep 2026 11:13:16 +0800 Subject: [PATCH] =?UTF-8?q?Prefect=20=E5=90=8C=E6=AD=A5=E8=B7=B3=E8=BF=87?= =?UTF-8?q?=E7=BC=BA=E5=A4=B1=E8=A1=A8=E5=B9=B6=E6=8E=A8=E9=80=81=E9=A3=9E?= =?UTF-8?q?=E4=B9=A6=E6=8F=90=E9=86=92?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- README.md | 8 ++--- prefect.yaml | 4 +-- prefect_gmv_scheduler.py | 22 ++++++++++--- project_bitable_common.py | 69 +++++++++++++++++++++++++++++++++++++-- 4 files changed, 90 insertions(+), 13 deletions(-) diff --git a/README.md b/README.md index 4dce4d1..18e964a 100644 --- a/README.md +++ b/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`；其中示例的本机目录需要替换为实际项目目录。 diff --git a/prefect.yaml b/prefect.yaml index ce982c4..30ee8fe 100644 --- a/prefect.yaml +++ b/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 diff --git a/prefect_gmv_scheduler.py b/prefect_gmv_scheduler.py index 3aa1603..0a1164a 100644 --- a/prefect_gmv_scheduler.py +++ b/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", diff --git a/project_bitable_common.py b/project_bitable_common.py index acbf9e0..fd6d619 100644 --- a/project_bitable_common.py +++ b/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, + )