调整

prefect_gmv_scheduler.py

@@ -1,4 +1,5 @@
 import argparse
+import json
 import os
 from dataclasses import dataclass
 from datetime import date
@@ -42,6 +43,9 @@ from project_bitable_common import (
     validate_date,
 )
 
+BASE_DIR = Path(__file__).resolve().parent
+DEFAULT_PREFECT_SYNC_STATE_FILE = BASE_DIR / "prefect_sync_state.json"
+
 
 @dataclass(frozen=True)
 class SyncConfig:
@@ -58,6 +62,12 @@ class SyncConfig:
     skip_missing_tables: bool
 
 
+@dataclass(frozen=True)
+class SyncWindow:
+    start_date: date
+    end_date: date
+
+
 def load_sync_config(
     start_date: str | None = None,
     end_date: str | None = None,
@@ -81,6 +91,68 @@ def load_sync_config(
     )
 
 
+def get_prefect_sync_state_file() -> Path:
+    return Path(os.getenv("PREFECT_SYNC_STATE_FILE", str(DEFAULT_PREFECT_SYNC_STATE_FILE)))
+
+
+def load_prefect_sync_state() -> dict[str, str]:
+    path = get_prefect_sync_state_file()
+    if not path.exists():
+        return {}
+    try:
+        payload = json.loads(path.read_text(encoding="utf-8"))
+    except Exception as exc:
+        log_step(f"读取 Prefect 同步状态失败，已忽略: path={path}, error={exc}")
+        return {}
+    return {str(key): str(value) for key, value in payload.items() if value is not None}
+
+
+def save_prefect_sync_state(state: dict[str, str]) -> None:
+    path = get_prefect_sync_state_file()
+    path.write_text(json.dumps(state, ensure_ascii=False, indent=2), encoding="utf-8")
+
+
+def resolve_sync_window(
+    config: SyncConfig,
+    explicit_start_date: str | None,
+    explicit_end_date: str | None,
+    mode: str,
+) -> SyncWindow:
+    if explicit_start_date or explicit_end_date:
+        start_date_value = validate_date(explicit_start_date or config.start_date)
+        end_date_value = validate_date(explicit_end_date or config.end_date)
+        if end_date_value < start_date_value:
+            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)
+
+    if mode == "notify":
+        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:
+            start_date_value = validate_date(last_sync_start)
+            end_date_value = validate_date(last_sync_end)
+            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)
+
+    if state.get("bootstrap_completed") == "true":
+        return SyncWindow(start_date=yesterday, end_date=yesterday)
+    return SyncWindow(start_date=configured_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)
+
+
 @task(name="notify-feishu")
 def notify_feishu(context: SyncConfig, start_date_text: str, end_date_text: str) -> None:
     context_obj = build_daily_sync_context(
@@ -99,8 +171,8 @@ def notify_feishu(context: SyncConfig, start_date_text: str, end_date_text: str)
     send_sync_notification(context_obj, validate_date(start_date_text), validate_date(end_date_text))
 
 
-@flow(name="gmv-daily-sync")
-def gmv_daily_sync_flow(
+@flow(name="gmv-daily-data-sync")
+def gmv_daily_data_sync_flow(
     start_date: str | None = None,
     end_date: str | None = None,
     dry_run: bool = False,
@@ -114,11 +186,7 @@ def gmv_daily_sync_flow(
         skip_missing_tables=skip_missing_tables,
         table_map_file=table_map_file,
     )
-
-    start_date_value = validate_date(config.start_date)
-    end_date_value = validate_date(config.end_date)
-    if end_date_value < start_date_value:
-        raise ValueError("end-date 不能早于 start-date")
+    sync_window = resolve_sync_window(config, start_date, end_date, mode="sync")
     context = build_daily_sync_context(
         DailySyncConfig(
             base_url=config.base_url,
@@ -133,27 +201,50 @@ def gmv_daily_sync_flow(
         )
     )
 
-    log_step(f"Prefect 调度开始: {config.start_date} -> {config.end_date}")
-    try:
-        for stat_date in iter_dates(start_date_value, end_date_value):
-            log_step(f"Prefect 开始处理日期: {stat_date.isoformat()}")
-            run_daily_sync_for_date(context, stat_date)
-    except Exception:
-        notify_feishu.with_options(name="notify-feishu-failed")(config, config.start_date, config.end_date)
-        raise
+    log_step(f"Prefect 数据同步开始: {sync_window.start_date.isoformat()} -> {sync_window.end_date.isoformat()}")
+    for stat_date in iter_dates(sync_window.start_date, sync_window.end_date):
+        log_step(f"Prefect 开始处理日期: {stat_date.isoformat()}")
+        run_daily_sync_for_date(context, stat_date)
+    if not config.dry_run:
+        mark_sync_window_completed(sync_window)
+    log_step("Prefect 数据同步完成")
 
-    notify_feishu(config, config.start_date, config.end_date)
-    log_step("Prefect 调度完成")
+
+@flow(name="gmv-daily-notify")
+def gmv_daily_notify_flow(
+    start_date: str | None = None,
+    end_date: str | None = None,
+    dry_run: bool = False,
+    skip_missing_tables: bool = False,
+    table_map_file: str = "",
+) -> None:
+    config = load_sync_config(
+        start_date=start_date,
+        end_date=end_date,
+        dry_run=dry_run,
+        skip_missing_tables=skip_missing_tables,
+        table_map_file=table_map_file,
+    )
+    sync_window = resolve_sync_window(config, start_date, end_date, mode="notify")
+    log_step(f"Prefect 飞书通知开始: {sync_window.start_date.isoformat()} -> {sync_window.end_date.isoformat()}")
+    notify_feishu(config, sync_window.start_date.isoformat(), sync_window.end_date.isoformat())
+    log_step("Prefect 飞书通知完成")
 
 
 def parse_args() -> argparse.Namespace:
-    parser = argparse.ArgumentParser(description="Prefect 调度：每天上午 9 点执行 GMV 数据同步。")
+    parser = argparse.ArgumentParser(description="Prefect 调度：8点执行数据同步，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 文件路径")
     parser.add_argument("--dry-run", action="store_true", help="只输出将要更新的内容，不调用飞书写接口")
     parser.add_argument("--skip-missing-tables", action="store_true", help="找不到对应月份的飞书表时跳过")
-    parser.add_argument("--serve", action="store_true", help="启动 Prefect deployment，并按每天上午 9 点定时执行")
+    parser.add_argument("--serve", action="store_true", help="启动 Prefect deployments，按 8 点同步、9 点通知 定时执行")
+    parser.add_argument(
+        "--mode",
+        choices=("sync", "notify", "all"),
+        default="all",
+        help="单次执行模式：sync=只跑数据同步，notify=只发飞书通知，all=先同步再通知",
+    )
     return parser.parse_args()
 
 
@@ -167,15 +258,29 @@ def main() -> None:
         "table_map_file": args.table_map_file,
     }
     if args.serve:
-        gmv_daily_sync_flow.serve(
-            name="gmv-daily-sync-9am",
-            cron="0 9 * * *",
+        gmv_daily_data_sync_flow.serve(
+            name="gmv-daily-data-sync-8am",
+            cron="0 8 * * *",
             timezone="Asia/Shanghai",
             parameters=flow_kwargs,
             tags=["gmv", "feishu", "bitable"],
         )
+        gmv_daily_notify_flow.serve(
+            name="gmv-daily-notify-9am",
+            cron="0 9 * * *",
+            timezone="Asia/Shanghai",
+            parameters=flow_kwargs,
+            tags=["gmv", "feishu", "notify"],
+        )
         return
-    gmv_daily_sync_flow(**flow_kwargs)
+    if args.mode == "sync":
+        gmv_daily_data_sync_flow(**flow_kwargs)
+        return
+    if args.mode == "notify":
+        gmv_daily_notify_flow(**flow_kwargs)
+        return
+    gmv_daily_data_sync_flow(**flow_kwargs)
+    gmv_daily_notify_flow(**flow_kwargs)
 
 
 if __name__ == "__main__":