From c23e21cf3e1072e6e29c253b3f36b3d7e20b5887 Mon Sep 17 00:00:00 2001 From: BXY0414 <1079823339@qq.com> Date: Wed, 1 Apr 2026 17:21:52 +0800 Subject: [PATCH] =?UTF-8?q?=E5=9B=9E=E8=A1=A5=E5=89=8D5=E5=A4=A9=E7=9A=84?= =?UTF-8?q?=E6=95=B0=E6=8D=AE?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- prefect_gmv_scheduler.py | 14 ++++++-------- 1 file changed, 6 insertions(+), 8 deletions(-) diff --git a/prefect_gmv_scheduler.py b/prefect_gmv_scheduler.py index 3dfb5d2..3aa1603 100644 --- a/prefect_gmv_scheduler.py +++ b/prefect_gmv_scheduler.py @@ -2,7 +2,7 @@ import argparse import json import os from dataclasses import dataclass -from datetime import date +from datetime import date, timedelta from pathlib import Path @@ -125,11 +125,12 @@ def resolve_sync_window( raise ValueError("end-date 不能早于 start-date") return SyncWindow(start_date=start_date_value, end_date=end_date_value) - state = load_prefect_sync_state() configured_start_date = validate_date(config.start_date) yesterday = validate_date(config.end_date) + rolling_start_date = max(configured_start_date, yesterday - timedelta(days=4)) if mode == "notify": + state = load_prefect_sync_state() last_sync_start = state.get("last_sync_start_date") last_sync_end = state.get("last_sync_end_date") if last_sync_start and last_sync_end: @@ -138,16 +139,13 @@ def resolve_sync_window( if end_date_value < start_date_value: raise ValueError("prefect_sync_state.json 中的日期范围无效") return SyncWindow(start_date=start_date_value, end_date=end_date_value) - return SyncWindow(start_date=configured_start_date, end_date=yesterday) + return SyncWindow(start_date=rolling_start_date, end_date=yesterday) - if state.get("bootstrap_completed") == "true": - return SyncWindow(start_date=yesterday, end_date=yesterday) - return SyncWindow(start_date=configured_start_date, end_date=yesterday) + return SyncWindow(start_date=rolling_start_date, end_date=yesterday) def mark_sync_window_completed(window: SyncWindow) -> None: state = load_prefect_sync_state() - state["bootstrap_completed"] = "true" state["last_sync_start_date"] = window.start_date.isoformat() state["last_sync_end_date"] = window.end_date.isoformat() save_prefect_sync_state(state) @@ -232,7 +230,7 @@ def gmv_daily_notify_flow( def parse_args() -> argparse.Namespace: - parser = argparse.ArgumentParser(description="Prefect 调度：8点执行数据同步，9点执行飞书通知。") + parser = argparse.ArgumentParser(description="Prefect 调度：每天 8 点同步前 5 天 GMV 数据，9 点发送飞书通知。") parser.add_argument("--start-date", default="", help="开始日期，格式 YYYY-MM-DD，默认读取 .env 中的 SYNC_START_DATE") parser.add_argument("--end-date", default="", help="结束日期，格式 YYYY-MM-DD，默认昨天") parser.add_argument("--table-map-file", default="", help="可选，项目到表名的映射 JSON 文件路径")