111 lines
3.6 KiB
Python
111 lines
3.6 KiB
Python
#!/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())
|