"""SDS 刷新未完结:查 DB 未完结订单号,按 keywords 批量刷新。 对应 sds-refresh-pending。终态:已完成 / 已取消。注意不跳过待付款。 """ from __future__ import annotations from prefect import flow, task from api.sds.models import OrderPageRequest from flows.sds._pipeline import sync_sds_records from flows.shared import ( SyncDB, build_stats_rows, create_sds_client, emit_sync_summary, get_channel_type_map, refresh_channel_type_map, ) SDS_FINISHED_STATUSES = ["已完成", "已取消"] KEYWORDS_BATCH = 500 @task(retries=3, retry_delay_seconds=[3, 6, 9]) def fetch_pending(order_nos: list[str]) -> list: client = create_sds_client() try: all_raw: list = [] 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)) all_raw.extend(resp.records) return all_raw finally: client.close() @task(retries=3, retry_delay_seconds=[3, 6, 9]) 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), skip_unpaid=False) finally: client.close() @flow(name="sds-refresh-pending", log_prints=True) def sds_refresh_pending() -> dict: db = SyncDB() order_nos = db.get_pending_orders("SDS", SDS_FINISHED_STATUSES) no_label_nos = db.get_orders_without_label("SDS", days=7) all_nos = list(set(order_nos) | set(no_label_nos)) print(f"SDS 未完结 {len(order_nos)} + 无面单(近7天) {len(no_label_nos)} → 合并 {len(all_nos)}") if not all_nos: return {"pendingCount": 0, "matchedCount": 0} raw = fetch_pending(all_nos) stats = upsert_records(raw) print(f"完成: 匹配 {len(raw)} | {stats}") emit_sync_summary("SDS 刷新未完结", build_stats_rows("SDS", stats, [ ("未完结订单", len(order_nos)), ("无面单(近7天)", len(no_label_nos)), ("合并去重", len(all_nos)), ("匹配", len(raw)), ])) return {"pendingCount": len(all_nos), "matchedCount": len(raw), **stats} if __name__ == "__main__": sds_refresh_pending.serve(name="sds-refresh-pending", cron="0 */2 * * *")