108 lines
3.8 KiB
Python
108 lines
3.8 KiB
Python
"""SDS 100前缀刷新:备货中/待确认且 productionNo 以 100 开头的订单,工厂分配后用新数据替换。
|
|
|
|
对应 sds-refresh-100-prefix。
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from prefect import flow, task
|
|
|
|
from api.sds.models import OrderPageRequest
|
|
from flows.shared import (
|
|
SyncDB, build_stats_rows, create_sds_client, emit_sync_summary,
|
|
get_channel_type_map, refresh_channel_type_map, resolve_courier_source,
|
|
)
|
|
from flows.shared.mappers import map_sds_goods, map_sds_order
|
|
|
|
TARGET_STATUSES = ["备货中", "待确认"]
|
|
KEYWORDS_BATCH = 200
|
|
|
|
|
|
def _has_100_prefix_item(raw) -> bool:
|
|
for item in raw.items or []:
|
|
prod_no = (item.no or item.id) if hasattr(item, "no") else None
|
|
if prod_no and str(prod_no).startswith("100"):
|
|
return True
|
|
return False
|
|
|
|
|
|
@task(retries=3, retry_delay_seconds=[3, 6, 9])
|
|
def fetch_remote(order_nos: list[str]) -> dict:
|
|
client = create_sds_client()
|
|
try:
|
|
remote: dict[str, object] = {}
|
|
for i in range(0, len(order_nos), KEYWORDS_BATCH):
|
|
batch = order_nos[i : i + KEYWORDS_BATCH]
|
|
resp = client.get_orders(OrderPageRequest(keywords="\n".join(batch), all=True))
|
|
for o in resp.records:
|
|
if o.no:
|
|
remote[o.no] = o
|
|
return remote
|
|
finally:
|
|
client.close()
|
|
|
|
|
|
@task
|
|
def upsert_records(remote_map: dict, target_order_nos: list[str]) -> dict:
|
|
client = create_sds_client()
|
|
try:
|
|
channel_map = get_channel_type_map(client)
|
|
refreshed: set[int] = set()
|
|
|
|
order_inputs: list = []
|
|
goods_inputs: list = []
|
|
skipped_pending = skipped_unpaid = 0
|
|
|
|
for no in target_order_nos:
|
|
raw = remote_map.get(no)
|
|
if not raw:
|
|
continue
|
|
if _has_100_prefix_item(raw):
|
|
skipped_pending += 1
|
|
continue
|
|
courier = resolve_courier_source(raw.logistics_id, channel_map,
|
|
lambda: refresh_channel_type_map(client), refreshed)
|
|
try:
|
|
o = map_sds_order(raw, channel_map, courier)
|
|
except Exception as err: # noqa: BLE001
|
|
print(f"[SDS 100] 映射失败 {no}: {err}")
|
|
continue
|
|
if o.systemStatus == "待付款":
|
|
skipped_unpaid += 1
|
|
continue
|
|
order_inputs.append(o)
|
|
for item in raw.items or []:
|
|
try:
|
|
goods_inputs.append(map_sds_goods(item, o.orderNo, o.createdAt))
|
|
except Exception: # noqa: BLE001
|
|
pass
|
|
|
|
stats = SyncDB().sync_orders_and_goods(order_inputs, goods_inputs)
|
|
stats["orderCount"] = len(order_inputs)
|
|
stats["goodsCount"] = len(goods_inputs)
|
|
stats["skippedStillPending"] = skipped_pending
|
|
stats["skippedUnpaid"] = skipped_unpaid
|
|
return stats
|
|
finally:
|
|
client.close()
|
|
|
|
|
|
@flow(name="sds-refresh-100-prefix", log_prints=True)
|
|
def sds_refresh_100_prefix() -> dict:
|
|
db = SyncDB()
|
|
targets = db.get_orders_with_100_prefix("SDS", TARGET_STATUSES)
|
|
order_nos = list({o for o, _ in targets})
|
|
print(f"SDS 100前缀目标订单 {len(order_nos)} 个")
|
|
if not order_nos:
|
|
return {"targetCount": 0, "refreshed": 0}
|
|
|
|
remote = fetch_remote(order_nos)
|
|
stats = upsert_records(remote, order_nos)
|
|
print(f"完成: 刷新 {stats['orderCount']} / 仍待分配 {stats.get('skippedStillPending',0)} / 待付款 {stats.get('skippedUnpaid',0)}")
|
|
emit_sync_summary("SDS 100前缀刷新", build_stats_rows("SDS", stats, [("目标订单", len(order_nos))]))
|
|
return {"targetCount": len(order_nos), **stats}
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sds_refresh_100_prefix.serve(name="sds-refresh-100-prefix", cron="30 */2 * * *")
|