feat: SDS 历史数据回填脚本(工厂+确认时间补全)

2026-07-23 10:15:50 +00:00
parent a527ebc1de
commit 2b8a2b65f4
1 changed files with 97 additions and 0 deletions
@@ -0,0 +1,97 @@
+"""SDS 历史数据回填:按天拉取订单,补全 factoryName / confirmedAt。
+
+一次性脚本(不挂 deployment)。按 pay_date 天为粒度拉取 → 修复后 mapper 映射 → upsert,
+每日独立提交,可中断后重跑(幂等)。复用单个 client 避免每天重复登录。
+
+用法: uv run python flows/sds/backfill_fields.py [start_date] [end_date]
+"""
+
+from __future__ import annotations
+
+from datetime import datetime, timedelta
+
+from prefect import flow
+
+from flows.shared import (
+    SyncDB, create_sds_client, emit_sync_summary, fetch_sds_all_pages,
+    format_date, get_channel_type_map, refresh_channel_type_map, start_of_day,
+)
+from flows.sds._pipeline import sync_sds_records
+
+DEFAULT_START = "2025-03-01"
+DEFAULT_END = "2026-07-24"
+
+
+@flow(name="sds-backfill-fields", log_prints=True)
+def backfill_sds_fields(start_date: str = DEFAULT_START, end_date: str = DEFAULT_END) -> dict:
+    start = start_of_day(datetime.fromisoformat(start_date))
+    end = start_of_day(datetime.fromisoformat(end_date))
+    db = SyncDB()
+
+    before = _count_missing(db)
+    print(f"[回填] 区间 {format_date(start)} ~ {format_date(end)} | 缺失前: 工厂={before['factory']} 确认={before['confirm']}")
+
+    client = create_sds_client()
+    try:
+        channel_map = get_channel_type_map(client)
+        totals = {"orders": 0, "goods": 0, "days": 0, "days_failed": 0}
+        cur = start
+        day_count = 0
+        while cur < end:
+            day_str = format_date(cur).split(" ")[0]
+            day_end = cur + timedelta(days=1)
+            day_count += 1
+            try:
+                raw = fetch_sds_all_pages(client, {
+                    "pay_date_begin": format_date(cur),
+                    "pay_date_end": format_date(day_end),
+                    "size": 500,
+                })
+                if not raw:
+                    cur += timedelta(days=1)
+                    continue
+                stats = sync_sds_records(raw, channel_map, db,
+                                         channel_refresh=lambda: refresh_channel_type_map(client),
+                                         skip_unpaid=False)
+                totals["orders"] += stats["orderCount"]
+                totals["goods"] += stats["goodsCount"]
+                totals["days"] += 1
+                print(f"[回填 {day_str}] 订单 {stats['orderCount']:5d} / 商品 {stats['goodsCount']:5d}"
+                      f" | 累计 订单 {totals['orders']:7d} / {totals['days']} 天 "
+                      f"(no.{day_count}, {datetime.now():%H:%M:%S})")
+            except Exception as err:  # noqa: BLE001
+                totals["days_failed"] += 1
+                print(f"[回填 {day_str}] 失败,跳过: {err}")
+
+            cur += timedelta(days=1)
+    finally:
+        client.close()
+
+    after = _count_missing(db)
+    filled = {"factory": before["factory"] - after["factory"], "confirm": before["confirm"] - after["confirm"]}
+    print(f"[回填完成] {totals} | 补全: 工厂 {filled['factory']} / 确认 {filled['confirm']} | 剩余缺失: 工厂 {after['factory']} / 确认 {after['confirm']}")
+    emit_sync_summary("SDS 字段回填", [
+        ("区间", f"{format_date(start)} ~ {format_date(end)}"),
+        ("处理天数", totals["days"]),
+        ("失败天数", totals["days_failed"]),
+        ("累计订单", totals["orders"]),
+        ("补全工厂", filled["factory"]),
+        ("补全确认时间", filled["confirm"]),
+        ("剩余缺工厂", after["factory"]),
+        ("剩余缺确认时间", after["confirm"]),
+    ])
+    return {"totals": totals, "filled": filled, "remaining": after}
+
+
+def _count_missing(db: SyncDB) -> dict[str, int]:
+    with db._conn() as conn:
+        with conn.cursor() as cur:
+            cur.execute('SELECT COUNT(*) FROM "Orders" WHERE "systemName"=\'SDS\' AND "factoryName" IS NULL')
+            f = cur.fetchone()[0]
+            cur.execute('SELECT COUNT(*) FROM "Orders" WHERE "systemName"=\'SDS\' AND "confirmedAt" IS NULL')
+            co = cur.fetchone()[0]
+    return {"factory": int(f), "confirm": int(co)}
+
+
+if __name__ == "__main__":
+    backfill_sds_fields()