61 lines
2.1 KiB
Python
61 lines
2.1 KiB
Python
"""SDS 增量同步:从数据库最近下单时间增量拉取(无参,全自动)。
|
|
|
|
对应 sds-incremental-sync。from = floorToHour(MAX(orderedAt)),to = now,按 payDate。
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from datetime import datetime
|
|
|
|
from prefect import flow, task
|
|
|
|
from flows.sds._pipeline import sync_sds_records
|
|
from flows.shared import (
|
|
SyncDB, build_stats_rows, create_sds_client, emit_sync_summary, fetch_sds_all_pages,
|
|
floor_to_hour, format_date, get_channel_type_map, refresh_channel_type_map,
|
|
)
|
|
|
|
|
|
@task(retries=3, retry_delay_seconds=[3, 6, 9])
|
|
def extract_orders(params: dict) -> list:
|
|
client = create_sds_client()
|
|
try:
|
|
return fetch_sds_all_pages(client, params)
|
|
finally:
|
|
client.close()
|
|
|
|
|
|
@task
|
|
def upsert_records(raw_orders: list) -> dict:
|
|
client = create_sds_client()
|
|
try:
|
|
channel_map = get_channel_type_map(client)
|
|
return sync_sds_records(raw_orders, channel_map, SyncDB(),
|
|
channel_refresh=lambda: refresh_channel_type_map(client))
|
|
finally:
|
|
client.close()
|
|
|
|
|
|
@flow(name="sds-incremental-sync", log_prints=True)
|
|
def sds_incremental_sync() -> dict:
|
|
db = SyncDB()
|
|
latest = db.get_latest_ordered_at("SDS")
|
|
if latest is not None and latest.tzinfo is not None:
|
|
latest = latest.replace(tzinfo=None)
|
|
now = datetime.now()
|
|
from_dt = floor_to_hour(latest) if latest else floor_to_hour(now)
|
|
if from_dt >= now:
|
|
from_dt = floor_to_hour(now)
|
|
|
|
params = {"pay_date_begin": format_date(from_dt), "pay_date_end": format_date(now), "size": 500}
|
|
print(f"SDS 增量同步 {params['pay_date_begin']} ~ {params['pay_date_end']}")
|
|
raw = extract_orders(params)
|
|
stats = upsert_records(raw)
|
|
print(f"完成: 订单 {stats['orderCount']} / 商品 {stats['goodsCount']}")
|
|
emit_sync_summary("SDS 增量同步", build_stats_rows("SDS", stats, [("时间范围", f"{params['pay_date_begin']} ~ {params['pay_date_end']}")]))
|
|
return {"syncRange": {"from": params["pay_date_begin"], "to": params["pay_date_end"]}, **stats}
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sds_incremental_sync.serve(name="sds-incremental-sync", cron="15 * * * *")
|