#!/usr/bin/env python3 import argparse import json import os from datetime import datetime from prefect import flow, get_run_logger, task import import_logistics_to_feishu_bitable def prefect_log_writer(level: str, message: str) -> None: logger = get_run_logger() level_name = level.upper() if level_name == "ERROR": logger.error(message) elif level_name == "WARN": logger.warning(message) else: logger.info(message) def send_feishu_group_message(text: str) -> None: import_logistics_to_feishu_bitable.load_env(import_logistics_to_feishu_bitable.ENV_PATH) app_id = os.getenv("FEISHU_APP_ID") app_secret = os.getenv("FEISHU_APP_SECRET") chat_id = os.getenv("FEISHU_CHAT_ID") if not app_id or not app_secret or not chat_id: raise RuntimeError("缺少 FEISHU_APP_ID / FEISHU_APP_SECRET / FEISHU_CHAT_ID") token = import_logistics_to_feishu_bitable.get_tenant_access_token(app_id, app_secret) url = "https://open.feishu.cn/open-apis/im/v1/messages?receive_id_type=chat_id" payload = { "receive_id": chat_id, "msg_type": "text", "content": json.dumps({"text": text}, ensure_ascii=False), } import_logistics_to_feishu_bitable.feishu_request("POST", url, token=token, payload=payload) @task(name="notify-feishu-chat", retries=3, retry_delay_seconds=[30, 60, 120], log_prints=True) def notify_feishu_chat_task(text: str) -> None: logger = get_run_logger() logger.info("开始推送飞书群通知") send_feishu_group_message(text) logger.info("飞书群通知推送完成") @task(name="run-logistics-sync", retries=2, retry_delay_seconds=[60, 180], log_prints=True) def run_logistics_sync_task() -> dict: logger = get_run_logger() logger.info("开始执行物流数据同步任务") started_at = datetime.now().strftime("%Y-%m-%d %H:%M:%S") import_logistics_to_feishu_bitable.set_logger_writer(prefect_log_writer) try: summary = import_logistics_to_feishu_bitable.execute_sync() except Exception as exc: finished_at = datetime.now().strftime("%Y-%m-%d %H:%M:%S") logger.error("物流数据同步任务失败: %s", exc) failure_text = ( "物流多维表同步失败\n" f"执行时间: {finished_at}\n" f"错误信息: {exc}" ) send_feishu_group_message(failure_text) raise finally: import_logistics_to_feishu_bitable.set_logger_writer(None) finished_at = datetime.now().strftime("%Y-%m-%d %H:%M:%S") logger.info("物流数据同步任务完成") return { "status": "success", "started_at": started_at, "finished_at": finished_at, **summary, } @flow(name="logistics-bitable-sync-flow", log_prints=True) def logistics_bitable_sync_flow() -> dict: result = run_logistics_sync_task() import_logistics_to_feishu_bitable.load_env(import_logistics_to_feishu_bitable.ENV_PATH) base_url = os.getenv("FEISHU_BASE_URL", "") success_text = ( "截止到中午12点,超48小时未发货的订单数据同步完成\n" f"新增记录数: {result['created']}\n" ) if base_url: success_text += f"\n多维表链接: {base_url}" notify_feishu_chat_task(success_text) return result def main() -> int: parser = argparse.ArgumentParser(description="Prefect scheduler for logistics -> Feishu Bitable sync") parser.add_argument("--mode", choices=["sync"], default="sync") args = parser.parse_args() if args.mode == "sync": logistics_bitable_sync_flow() return 0 return 1 if __name__ == "__main__": raise SystemExit(main())