229 lines
7.2 KiB
Python
229 lines
7.2 KiB
Python
|
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()
|