61 lines
1.8 KiB
Python
61 lines
1.8 KiB
Python
"""RIIN 指定单号同步:通过订单号列表(逗号/换行分隔)拉取指定订单。
|
||
|
||
对应 riin-by-orders。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import re
|
||
|
||
from prefect import flow, task
|
||
|
||
from api.riin.models import OrderPageRequest
|
||
from flows.riin._pipeline import sync_riin_records
|
||
from flows.shared import SyncDB, build_stats_rows, create_riin_client, emit_sync_summary
|
||
|
||
|
||
def _split_order_nos(text: str) -> list[str]:
|
||
seen: dict[str, None] = {}
|
||
for part in re.split(r"[\n\r,，\s]+", text or ""):
|
||
part = part.strip()
|
||
if part and part not in seen:
|
||
seen[part] = None
|
||
return list(seen.keys())
|
||
|
||
|
||
@task(retries=3, retry_delay_seconds=[3, 6, 9])
|
||
def fetch_by_orders(order_nos: list[str]) -> list:
|
||
client = create_riin_client()
|
||
try:
|
||
resp = client.get_orders(OrderPageRequest(platform_oid_list=",".join(order_nos), all=True))
|
||
return list(resp.data.records)
|
||
finally:
|
||
client.close()
|
||
|
||
|
||
@task
|
||
def upsert_records(raw_orders: list) -> dict:
|
||
client = create_riin_client()
|
||
try:
|
||
return sync_riin_records(raw_orders, client, SyncDB())
|
||
finally:
|
||
client.close()
|
||
|
||
|
||
@flow(name="riin-by-orders-sync", log_prints=True)
|
||
def riin_by_orders_sync(order_nos: str) -> dict:
|
||
nos = _split_order_nos(order_nos)
|
||
print(f"RIIN 指定单号同步 输入 {len(nos)} 个")
|
||
if not nos:
|
||
return {"inputCount": 0, "matchedCount": 0}
|
||
|
||
raw = fetch_by_orders(nos)
|
||
stats = upsert_records(raw)
|
||
print(f"完成: 匹配 {len(raw)} | {stats}")
|
||
emit_sync_summary("RIIN 指定单号同步", build_stats_rows("RIIN", stats, [("输入单号", len(nos)), ("匹配", len(raw))]))
|
||
return {"inputCount": len(nos), "matchedCount": len(raw), **stats}
|
||
|
||
|
||
if __name__ == "__main__":
|
||
riin_by_orders_sync.serve(name="riin-by-orders-sync")
|