import argparse import os from dataclasses import dataclass from datetime import date from pathlib import Path def bootstrap_prefect_env() -> None: env_file = Path(__file__).resolve().parent / ".env" if not env_file.exists(): return for raw_line in env_file.read_text(encoding="utf-8").splitlines(): line = raw_line.strip() if not line or line.startswith("#") or "=" not in line: continue key, value = line.split("=", 1) key = key.strip() if key.startswith("PREFECT_"): os.environ.setdefault(key, value.strip()) bootstrap_prefect_env() from prefect import flow, task from prefect.logging import get_run_logger from project_bitable_common import ( get_base_url, get_default_end_date, get_default_start_date, get_feishu_chat_id, get_feishu_report_image_path, get_group_webhook_url, get_mapping_file, get_store_field_name, iter_dates, load_dotenv, log_step, run_gmv_sync, run_metric_sync, send_feishu_app_card, send_feishu_sync_card, validate_date, ) @dataclass(frozen=True) class SyncConfig: start_date: str end_date: str base_url: str mapping_file: str store_field_name: str chat_id: str report_image_path: str group_webhook: str table_map_file: str dry_run: bool skip_missing_tables: bool def load_sync_config( start_date: str | None = None, end_date: str | None = None, dry_run: bool = False, skip_missing_tables: bool = False, table_map_file: str = "", ) -> SyncConfig: load_dotenv() return SyncConfig( start_date=start_date or get_default_start_date(), end_date=end_date or get_default_end_date(), base_url=get_base_url(), mapping_file=str(get_mapping_file()), store_field_name=get_store_field_name(), chat_id=get_feishu_chat_id(), report_image_path=get_feishu_report_image_path(), group_webhook=get_group_webhook_url(), table_map_file=table_map_file, dry_run=dry_run, skip_missing_tables=skip_missing_tables, ) @task(name="sync-sales") def sync_sales_for_day(stat_date_text: str, config: SyncConfig) -> None: stat_date = validate_date(stat_date_text) run_metric_sync( metric_name="sales", start_date=stat_date, end_date=stat_date, base_url=config.base_url, mapping_file=Path(config.mapping_file), store_field_name=config.store_field_name, table_map_file=config.table_map_file, dry_run=config.dry_run, skip_missing_tables=config.skip_missing_tables, ) @task(name="sync-refund") def sync_refund_for_day(stat_date_text: str, config: SyncConfig) -> None: stat_date = validate_date(stat_date_text) run_metric_sync( metric_name="refund", start_date=stat_date, end_date=stat_date, base_url=config.base_url, mapping_file=Path(config.mapping_file), store_field_name=config.store_field_name, table_map_file=config.table_map_file, dry_run=config.dry_run, skip_missing_tables=config.skip_missing_tables, ) @task(name="sync-gmv") def sync_gmv_for_day(stat_date_text: str, config: SyncConfig) -> None: stat_date = validate_date(stat_date_text) run_gmv_sync( start_date=stat_date, end_date=stat_date, base_url=config.base_url, store_field_name=config.store_field_name, dry_run=config.dry_run, skip_missing_tables=config.skip_missing_tables, ) @task(name="notify-feishu") def notify_feishu(title: str, config: SyncConfig) -> None: if config.dry_run: log_step("dry-run 模式下跳过飞书群推送") return start_date = validate_date(config.start_date) end_date = validate_date(config.end_date) if config.chat_id: send_feishu_app_card( chat_id=config.chat_id, title=title, base_url=config.base_url, start_date=start_date, end_date=end_date, image_path=config.report_image_path, ) return send_feishu_sync_card( webhook_url=config.group_webhook, title=title, base_url=config.base_url, start_date=start_date, end_date=end_date, ) @flow(name="gmv-daily-sync") def gmv_daily_sync_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: logger = get_run_logger() 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, ) 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") logger.info("Prefect 调度开始: %s -> %s", config.start_date, config.end_date) try: for stat_date in iter_dates(start_date_value, end_date_value): stat_date_text = stat_date.isoformat() logger.info("开始处理日期: %s", stat_date_text) sync_sales_for_day(stat_date_text, config) sync_refund_for_day(stat_date_text, config) sync_gmv_for_day(stat_date_text, config) logger.info("日期处理完成: %s", stat_date_text) except Exception: notify_feishu.with_options(name="notify-feishu-failed")("GMV数据同步失败", config) raise notify_feishu("GMV数据同步", config) logger.info("Prefect 调度完成") def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser(description="Prefect 调度：每天上午 9 点执行 GMV 数据同步。") 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 点定时执行") return parser.parse_args() def main() -> None: args = parse_args() flow_kwargs = { "start_date": args.start_date or None, "end_date": args.end_date or None, "dry_run": args.dry_run, "skip_missing_tables": args.skip_missing_tables, "table_map_file": args.table_map_file, } if args.serve: gmv_daily_sync_flow.serve( name="gmv-daily-sync-9am", cron="0 9 * * *", timezone="Asia/Shanghai", parameters=flow_kwargs, tags=["gmv", "feishu", "bitable"], ) return gmv_daily_sync_flow(**flow_kwargs) if __name__ == "__main__": main()