gmv-analytics/prefect_gmv_scheduler.py

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 project_bitable_common import (
    DailySyncConfig,
    build_daily_sync_context,
    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_daily_sync_for_date,
    send_sync_notification,
    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="notify-feishu")
def notify_feishu(context: SyncConfig, start_date_text: str, end_date_text: str) -> None:
    context_obj = build_daily_sync_context(
        DailySyncConfig(
            base_url=context.base_url,
            mapping_file=Path(context.mapping_file),
            store_field_name=context.store_field_name,
            table_map_file=context.table_map_file,
            dry_run=context.dry_run,
            skip_missing_tables=context.skip_missing_tables,
            chat_id=context.chat_id,
            report_image_path=context.report_image_path,
            group_webhook=context.group_webhook,
        )
    )
    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(
    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,
    )

    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")
    context = build_daily_sync_context(
        DailySyncConfig(
            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,
            chat_id=config.chat_id,
            report_image_path=config.report_image_path,
            group_webhook=config.group_webhook,
        )
    )

    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

    notify_feishu(config, config.start_date, config.end_date)
    log_step("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()