回补前5天的数据

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 文件路径")