From 4ec7cf2dd699862e40eb9bd27b3eda762163713a Mon Sep 17 00:00:00 2001 From: yeuimu <2197651308@qq.com> Date: Thu, 23 Jul 2026 09:01:39 +0000 Subject: [PATCH] =?UTF-8?q?feat:=20=E8=AF=A6=E7=BB=86=E6=97=A5=E5=BF=97(?= =?UTF-8?q?=E5=88=86=E9=A1=B5/=E6=98=A0=E5=B0=84/=E6=89=B9=E6=AC=A1?= =?UTF-8?q?=E8=BF=9B=E5=BA=A6)+=20Artifacts=20=E6=91=98=E8=A6=81(sync-summ?= =?UTF-8?q?ary/verify-report)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- flows/riin/_pipeline.py | 6 +++ flows/riin/by_orders.py | 3 +- flows/riin/incremental_sync.py | 4 +- flows/riin/order_sync.py | 5 +- flows/riin/recent_sync.py | 4 +- flows/riin/refresh_pending.py | 3 +- flows/riin/verify.py | 8 ++-- flows/sds/_pipeline.py | 1 + flows/sds/incremental_sync.py | 5 +- flows/sds/order_sync.py | 4 +- flows/sds/recent_sync.py | 5 +- flows/sds/refresh_100_prefix.py | 5 +- flows/sds/refresh_pending.py | 4 +- flows/sds/verify.py | 10 ++-- flows/shared/__init__.py | 4 +- flows/shared/artifacts.py | 82 +++++++++++++++++++++++++++++++++ flows/shared/clients.py | 15 +++++- flows/shared/db.py | 8 ++++ 18 files changed, 154 insertions(+), 22 deletions(-) create mode 100644 flows/shared/artifacts.py diff --git a/flows/riin/_pipeline.py b/flows/riin/_pipeline.py index 0f65f08..c470d6a 100644 --- a/flows/riin/_pipeline.py +++ b/flows/riin/_pipeline.py @@ -14,17 +14,21 @@ def sync_riin_records(raw_orders: list[Any], client: RiinDisClient, db: SyncDB) order_inputs: list[Any] = [] order_meta: dict[str, tuple[str, Any]] = {} # raw.id -> (orderNo, createdAt) goods_failed = 0 + order_map_failed = 0 for raw in raw_orders: try: o = map_riin_order(raw) except Exception as err: # noqa: BLE001 + order_map_failed += 1 print(f"[RIIN] 映射订单失败 {getattr(raw, 'platform_oid', '?')}: {err}") continue order_inputs.append(o) if raw.id: order_meta[raw.id] = (o.orderNo, o.createdAt) + print(f"[RIIN] 订单映射完成: 成功 {len(order_inputs)} / 失败 {order_map_failed} (共 {len(raw_orders)})") + goods_inputs: list[Any] = [] if order_meta: goods_map = fetch_goods_concurrent(client, list(order_meta.keys())) @@ -38,6 +42,8 @@ def sync_riin_records(raw_orders: list[Any], client: RiinDisClient, db: SyncDB) except Exception as err: # noqa: BLE001 print(f"[RIIN] 映射商品失败 order={order_no}: {err}") + print(f"[RIIN] 商品映射完成: {len(goods_inputs)} 条, 商品拉取失败订单 {goods_failed} 个") + stats = db.sync_orders_and_goods(order_inputs, goods_inputs) stats["orderCount"] = len(order_inputs) stats["goodsCount"] = len(goods_inputs) diff --git a/flows/riin/by_orders.py b/flows/riin/by_orders.py index 7f8e18e..d2c9dd6 100644 --- a/flows/riin/by_orders.py +++ b/flows/riin/by_orders.py @@ -11,7 +11,7 @@ from prefect import flow, task from api.riin.models import OrderPageRequest from flows.riin._pipeline import sync_riin_records -from flows.shared import SyncDB, create_riin_client +from flows.shared import SyncDB, build_stats_rows, create_riin_client, emit_sync_summary def _split_order_nos(text: str) -> list[str]: @@ -52,6 +52,7 @@ def riin_by_orders_sync(order_nos: str) -> dict: 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} diff --git a/flows/riin/incremental_sync.py b/flows/riin/incremental_sync.py index 28d7645..65a7025 100644 --- a/flows/riin/incremental_sync.py +++ b/flows/riin/incremental_sync.py @@ -11,7 +11,8 @@ from prefect import flow, task from flows.riin._pipeline import sync_riin_records from flows.shared import ( - SyncDB, create_riin_client, fetch_riin_all_pages, floor_to_hour, format_date, + SyncDB, build_stats_rows, create_riin_client, emit_sync_summary, fetch_riin_all_pages, + floor_to_hour, format_date, ) @@ -55,6 +56,7 @@ def riin_incremental_sync() -> dict: raw = extract_orders(params) stats = upsert_records(raw) print(f"完成: 订单 {stats['orderCount']} / 商品 {stats['goodsCount']}") + emit_sync_summary("RIIN 增量同步", build_stats_rows("RIIN", stats, [("时间范围", f"{params['start_time']} ~ {params['end_time']}")])) return {"syncRange": {"from": params["start_time"], "to": params["end_time"]}, **stats} diff --git a/flows/riin/order_sync.py b/flows/riin/order_sync.py index a57dcac..5288722 100644 --- a/flows/riin/order_sync.py +++ b/flows/riin/order_sync.py @@ -11,7 +11,9 @@ from prefect import flow, task from api.riin.models import OrderPageRequest from flows.riin._pipeline import sync_riin_records -from flows.shared import SyncDB, create_riin_client, fetch_riin_all_pages, format_date +from flows.shared import ( + SyncDB, build_stats_rows, create_riin_client, emit_sync_summary, fetch_riin_all_pages, format_date, +) @task(retries=3, retry_delay_seconds=[3, 6, 9]) @@ -70,6 +72,7 @@ def riin_order_sync( raw = extract_orders(params) stats = upsert_records(raw) print(f"完成: 订单 {stats['orderCount']} / 商品 {stats['goodsCount']} | {stats}") + emit_sync_summary("RIIN 订单同步", build_stats_rows("RIIN", stats, [("时间范围", f"{order_date_begin} ~ {order_date_end}")])) return {"syncRange": {"from": order_date_begin, "to": order_date_end}, **stats} diff --git a/flows/riin/recent_sync.py b/flows/riin/recent_sync.py index ef76a19..581e41c 100644 --- a/flows/riin/recent_sync.py +++ b/flows/riin/recent_sync.py @@ -11,7 +11,8 @@ from prefect import flow, task from flows.riin._pipeline import sync_riin_records from flows.shared import ( - SyncDB, create_riin_client, fetch_riin_all_pages, format_date, start_of_day, + SyncDB, build_stats_rows, create_riin_client, emit_sync_summary, + fetch_riin_all_pages, format_date, start_of_day, ) @@ -49,6 +50,7 @@ def riin_recent_sync(recent_days: int = 7) -> dict: raw = extract_orders(params) stats = upsert_records(raw) print(f"完成: 订单 {stats['orderCount']} / 商品 {stats['goodsCount']}") + emit_sync_summary("RIIN 近期同步", build_stats_rows("RIIN", stats, [("近N天", days), ("时间范围", f"{params['start_time']} ~ {params['end_time']}")])) return {"syncRange": {"from": params["start_time"], "to": params["end_time"]}, "recentDays": days, **stats} diff --git a/flows/riin/refresh_pending.py b/flows/riin/refresh_pending.py index 34f8af8..42433c1 100644 --- a/flows/riin/refresh_pending.py +++ b/flows/riin/refresh_pending.py @@ -9,7 +9,7 @@ from prefect import flow, task from api.riin.models import OrderPageRequest from flows.riin._pipeline import sync_riin_records -from flows.shared import SyncDB, create_riin_client +from flows.shared import SyncDB, build_stats_rows, create_riin_client, emit_sync_summary RIIN_FINISHED_STATUSES = ["已退款", "已关闭", "已发货"] ORDER_BATCH = 500 @@ -51,6 +51,7 @@ def riin_refresh_pending() -> dict: raw = fetch_pending(order_nos) stats = upsert_records(raw) print(f"完成: 匹配 {len(raw)} | {stats}") + emit_sync_summary("RIIN 刷新未完结", build_stats_rows("RIIN", stats, [("未完结订单", len(order_nos)), ("匹配", len(raw))])) return {"pendingCount": len(order_nos), "matchedCount": len(raw), **stats} diff --git a/flows/riin/verify.py b/flows/riin/verify.py index 9a636bd..e3e9f7c 100644 --- a/flows/riin/verify.py +++ b/flows/riin/verify.py @@ -12,7 +12,7 @@ from prefect import flow, task from api.riin.models import OrderPageRequest from flows.riin._pipeline import sync_riin_records from flows.shared import ( - SyncDB, create_riin_client, fetch_riin_all_pages, format_date, start_of_day, + SyncDB, create_riin_client, emit_verify_report, fetch_riin_all_pages, format_date, start_of_day, ) RIIN_COUNT_STATUS_MAP = { @@ -96,7 +96,7 @@ def riin_verify(mode: str = "recent", recent_days: int = 1, date_range: str | No print(f"[核验 {day_str}] 远程 {remote_total} 本地 {sum(local_counts.values())}" + (f" 差异 {diff}" if has_diff else " 一致,跳过")) if not has_diff: - day_reports.append({"date": day_str, "diff": diff, "ok": True}) + day_reports.append({"date": day_str, "remote": remote_total, "local": sum(local_counts.values()), "diff": diff, "ok": True}) continue days_with_diff += 1 @@ -111,11 +111,13 @@ def riin_verify(mode: str = "recent", recent_days: int = 1, date_range: str | No totals["ordersUpdated"] += stats["ordersUpdated"] totals["goodsAdded"] += stats["goodsCreated"] totals["goodsUpdated"] += stats["goodsUpdated"] - day_reports.append({"date": day_str, "diff": diff, "ok": False, + day_reports.append({"date": day_str, "remote": remote_total, "local": sum(local_counts.values()), "diff": diff, "ok": False, "orders": stats["orderCount"], "goods": stats["goodsCount"]}) finally: client.close() + print(f"[核验汇总] 共 {len(days)} 天 / 有差异 {days_with_diff} 天 / {totals}") + emit_verify_report("RIIN 数据核验", day_reports, totals, len(days), days_with_diff) return { "verifyReport": {"days": day_reports, "summary": {"totalDays": len(days), "daysWithDiff": days_with_diff, **totals}}, "syncRange": {"from": format_date(start_of_day(days[0])) if days else "", "to": format_date(start_of_day(days[-1])) if days else ""}, diff --git a/flows/sds/_pipeline.py b/flows/sds/_pipeline.py index e66f8e3..8a887d1 100644 --- a/flows/sds/_pipeline.py +++ b/flows/sds/_pipeline.py @@ -51,6 +51,7 @@ def sync_sds_records( order_inputs, goods_inputs, skipped = map_sds_records( raw_orders, channel_map, channel_refresh, skip_unpaid ) + print(f"[SDS] 映射完成: 订单 {len(order_inputs)} / 商品 {len(goods_inputs)} / 跳过待付款 {skipped} (共 {len(raw_orders)})") stats = db.sync_orders_and_goods(order_inputs, goods_inputs) stats["orderCount"] = len(order_inputs) stats["goodsCount"] = len(goods_inputs) diff --git a/flows/sds/incremental_sync.py b/flows/sds/incremental_sync.py index 1a6232c..f6f373d 100644 --- a/flows/sds/incremental_sync.py +++ b/flows/sds/incremental_sync.py @@ -11,8 +11,8 @@ from prefect import flow, task from flows.sds._pipeline import sync_sds_records from flows.shared import ( - SyncDB, create_sds_client, fetch_sds_all_pages, floor_to_hour, format_date, - get_channel_type_map, refresh_channel_type_map, + 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, ) @@ -52,6 +52,7 @@ def sds_incremental_sync() -> dict: 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} diff --git a/flows/sds/order_sync.py b/flows/sds/order_sync.py index 4edbbed..05fe5d6 100644 --- a/flows/sds/order_sync.py +++ b/flows/sds/order_sync.py @@ -11,7 +11,8 @@ from prefect import flow, task from flows.sds._pipeline import sync_sds_records from flows.shared import ( - SyncDB, create_sds_client, fetch_sds_all_pages, get_channel_type_map, refresh_channel_type_map, + SyncDB, build_stats_rows, create_sds_client, emit_sync_summary, + fetch_sds_all_pages, get_channel_type_map, refresh_channel_type_map, ) @@ -69,6 +70,7 @@ def sds_order_sync( raw = extract_orders(params) stats = upsert_records(raw) print(f"完成: 订单 {stats['orderCount']} / 商品 {stats['goodsCount']} / 跳过待付款 {stats.get('skippedUnpaid',0)}") + emit_sync_summary("SDS 订单同步", build_stats_rows("SDS", stats, [("时间范围", f"{pay_date_begin} ~ {pay_date_end}")])) return {"syncRange": {"from": pay_date_begin, "to": pay_date_end}, **stats} diff --git a/flows/sds/recent_sync.py b/flows/sds/recent_sync.py index 1f7e398..842d2b0 100644 --- a/flows/sds/recent_sync.py +++ b/flows/sds/recent_sync.py @@ -11,8 +11,8 @@ from prefect import flow, task from flows.sds._pipeline import sync_sds_records from flows.shared import ( - SyncDB, create_sds_client, fetch_sds_all_pages, format_date, get_channel_type_map, - refresh_channel_type_map, start_of_day, + SyncDB, build_stats_rows, create_sds_client, emit_sync_summary, fetch_sds_all_pages, + format_date, get_channel_type_map, refresh_channel_type_map, start_of_day, ) @@ -48,6 +48,7 @@ def sds_recent_sync(recent_days: int = 7) -> dict: raw = extract_orders(params) stats = upsert_records(raw) print(f"完成: 订单 {stats['orderCount']} / 商品 {stats['goodsCount']}") + emit_sync_summary("SDS 近期同步", build_stats_rows("SDS", stats, [("近N天", days), ("时间范围", f"{params['created_date_begin']} ~ {params['created_date_end']}")])) return {"syncRange": {"from": params["created_date_begin"], "to": format_date(now)}, "recentDays": days, **stats} diff --git a/flows/sds/refresh_100_prefix.py b/flows/sds/refresh_100_prefix.py index 23f9244..24da863 100644 --- a/flows/sds/refresh_100_prefix.py +++ b/flows/sds/refresh_100_prefix.py @@ -9,8 +9,8 @@ from prefect import flow, task from api.sds.models import OrderPageRequest from flows.shared import ( - SyncDB, create_sds_client, get_channel_type_map, - refresh_channel_type_map, resolve_courier_source, + 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 @@ -99,6 +99,7 @@ def sds_refresh_100_prefix() -> dict: 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} diff --git a/flows/sds/refresh_pending.py b/flows/sds/refresh_pending.py index 1d0250b..73ae03c 100644 --- a/flows/sds/refresh_pending.py +++ b/flows/sds/refresh_pending.py @@ -10,7 +10,8 @@ from prefect import flow, task from api.sds.models import OrderPageRequest from flows.sds._pipeline import sync_sds_records from flows.shared import ( - SyncDB, create_sds_client, get_channel_type_map, refresh_channel_type_map, + SyncDB, build_stats_rows, create_sds_client, emit_sync_summary, + get_channel_type_map, refresh_channel_type_map, ) SDS_FINISHED_STATUSES = ["已完成", "已取消"] @@ -54,6 +55,7 @@ def sds_refresh_pending() -> dict: raw = fetch_pending(order_nos) stats = upsert_records(raw) print(f"完成: 匹配 {len(raw)} | {stats}") + emit_sync_summary("SDS 刷新未完结", build_stats_rows("SDS", stats, [("未完结订单", len(order_nos)), ("匹配", len(raw))])) return {"pendingCount": len(order_nos), "matchedCount": len(raw), **stats} diff --git a/flows/sds/verify.py b/flows/sds/verify.py index 7f65072..693ab9b 100644 --- a/flows/sds/verify.py +++ b/flows/sds/verify.py @@ -12,8 +12,8 @@ from prefect import flow, task from api.sds.models import OrderQueryParams from flows.sds._pipeline import sync_sds_records from flows.shared import ( - SyncDB, create_sds_client, fetch_sds_all_pages, format_date, get_channel_type_map, - refresh_channel_type_map, start_of_day, + SyncDB, create_sds_client, emit_verify_report, fetch_sds_all_pages, format_date, + get_channel_type_map, refresh_channel_type_map, start_of_day, ) @@ -75,7 +75,7 @@ def sds_verify(mode: str = "recent", recent_days: int = 1, date_range: str | Non print(f"[核验 {day_str}] 远程 {remote_total} 本地 {sum(local_counts.values())}" + (f" 差异 {diff}" if has_diff else " 一致,跳过")) if not has_diff: - day_reports.append({"date": day_str, "diff": diff, "ok": True}) + day_reports.append({"date": day_str, "remote": remote_total, "local": sum(local_counts.values()), "diff": diff, "ok": True}) continue days_with_diff += 1 @@ -89,11 +89,13 @@ def sds_verify(mode: str = "recent", recent_days: int = 1, date_range: str | Non totals["ordersUpdated"] += stats["ordersUpdated"] totals["goodsAdded"] += stats["goodsCreated"] totals["goodsUpdated"] += stats["goodsUpdated"] - day_reports.append({"date": day_str, "diff": diff, "ok": False, + day_reports.append({"date": day_str, "remote": remote_total, "local": sum(local_counts.values()), "diff": diff, "ok": False, "orders": stats["orderCount"], "goods": stats["goodsCount"]}) finally: client.close() + print(f"[核验汇总] 共 {len(days)} 天 / 有差异 {days_with_diff} 天 / {totals}") + emit_verify_report("SDS 数据核验", day_reports, totals, len(days), days_with_diff) return { "verifyReport": {"days": day_reports, "summary": {"totalDays": len(days), "daysWithDiff": days_with_diff, **totals}}, "syncRange": {"from": format_date(start_of_day(days[0])) if days else "", "to": format_date(start_of_day(days[-1])) if days else ""}, diff --git a/flows/shared/__init__.py b/flows/shared/__init__.py index 8dd51f4..587e485 100644 --- a/flows/shared/__init__.py +++ b/flows/shared/__init__.py @@ -1,6 +1,7 @@ -"""同步编排共享层:客户端/分页、映射、PG 写入。""" +"""同步编排共享层:客户端/分页、映射、PG 写入、Artifacts。""" from flows.shared import clients, db, mappers +from flows.shared.artifacts import build_stats_rows, emit_sync_summary, emit_verify_report from flows.shared.clients import ( create_riin_client, create_sds_client, @@ -24,4 +25,5 @@ __all__ = [ "fetch_riin_all_pages", "fetch_sds_all_pages", "fetch_goods_concurrent", "get_channel_type_map", "refresh_channel_type_map", "resolve_courier_source", "format_date", "start_of_day", "floor_to_hour", + "build_stats_rows", "emit_sync_summary", "emit_verify_report", ] diff --git a/flows/shared/artifacts.py b/flows/shared/artifacts.py new file mode 100644 index 0000000..e77ba0a --- /dev/null +++ b/flows/shared/artifacts.py @@ -0,0 +1,82 @@ +"""Prefect Artifacts:同步结果摘要(markdown),展示在 flow run 的 Artifacts 标签页。 + +所有函数均容错——artifact 生成失败只打印警告,绝不中断 flow run。 +""" + +from __future__ import annotations + +from typing import Any + +from prefect.artifacts import create_markdown_artifact + +_STATS_LABELS: list[tuple[str, str]] = [ + ("orderCount", "拉取订单"), + ("goodsCount", "拉取商品"), + ("ordersCreated", "订单新增"), + ("ordersUpdated", "订单更新"), + ("ordersDeleted", "订单删除"), + ("goodsCreated", "商品新增"), + ("goodsUpdated", "商品更新"), + ("goodsRemoved", "商品移除"), + ("goodsFetchFailed", "商品拉取失败"), + ("skippedUnpaid", "跳过待付款"), + ("skippedStillPending", "仍待分配"), +] + + +def build_stats_rows( + system: str, stats: dict[str, Any], extra: list[tuple[str, Any]] | None = None +) -> list[tuple[str, Any]]: + """从 stats dict 构造摘要行(仅展示实际存在的键)。""" + rows: list[tuple[str, Any]] = [("系统", system)] + if extra: + rows.extend(extra) + for key, label in _STATS_LABELS: + if key in stats: + rows.append((label, stats[key])) + return rows + + +def emit_sync_summary(title: str, rows: list[tuple[str, Any]]) -> None: + """生成同步结果摘要 markdown artifact(表格)。 + + Args: + title: 摘要标题(如 "RIIN 订单同步")。 + rows: [(标签, 值), ...],每项渲染为表格一行。 + """ + lines = [f"# {title}", "", "| 指标 | 值 |", "| --- | --- |"] + for label, value in rows: + lines.append(f"| {label} | {value} |") + try: + create_markdown_artifact(key="sync-summary", markdown="\n".join(lines), description=title) + except Exception as err: # noqa: BLE001 + print(f"[artifact] 摘要生成失败: {err}") + + +def emit_verify_report( + title: str, + day_reports: list[dict], + totals: dict[str, int], + total_days: int, + days_with_diff: int, +) -> None: + """生成核验报告 markdown artifact(逐天差异表 + 汇总)。""" + lines = [f"# {title}", "", "## 逐天核验", "", "| 日期 | 远程 | 本地 | 差异 | 结果 |", "| --- | --- | --- | --- | --- |"] + for r in day_reports: + diff = r.get("diff", {}) + diff_str = ", ".join(f"{k}:{v:+d}" for k, v in diff.items()) if diff else "—" + ok = r.get("ok") + if ok: + result = "✅ 一致" + else: + result = f"🔧 已修复(订单 {r.get('orders', 0)} / 商品 {r.get('goods', 0)})" + lines.append(f"| {r['date']} | {r.get('remote', '—')} | {r.get('local', '—')} | {diff_str} | {result} |") + lines += ["", "## 汇总", "", "| 指标 | 值 |", "| --- | --- |"] + lines.append(f"| 核验天数 | {total_days} |") + lines.append(f"| 有差异天数 | {days_with_diff} |") + for k, v in totals.items(): + lines.append(f"| {k} | {v} |") + try: + create_markdown_artifact(key="verify-report", markdown="\n".join(lines), description=title) + except Exception as err: # noqa: BLE001 + print(f"[artifact] 核验报告生成失败: {err}") diff --git a/flows/shared/clients.py b/flows/shared/clients.py index d1d8fa4..74f9dc3 100644 --- a/flows/shared/clients.py +++ b/flows/shared/clients.py @@ -108,12 +108,16 @@ def fetch_riin_all_pages(client: RiinDisClient, params: dict[str, Any], max_page total = first.data.total records = list(first.data.records) total_pages = min(-(-total // page_size), max_pages) + print(f"[RIIN sync] 分页开始: 共 {total} 条 / {total_pages} 页 (page_size={page_size})") for page in range(2, total_pages + 1): resp = fetch(page) records.extend(resp.data.records) + if page % 10 == 0 or page == total_pages: + print(f"[RIIN sync] 分页进度: {page}/{total_pages} 页, 累计 {len(records)} 条") time.sleep(PAGE_DELAY_MS / 1000) if total_pages == max_pages: print(f"[RIIN sync] 达到最大页数限制 {max_pages},截断到 {len(records)} 条") + print(f"[RIIN sync] 分页完成: 共拉取 {len(records)} 条") return records @@ -127,10 +131,14 @@ def fetch_sds_all_pages(client: SdsMngClient, params: dict[str, Any]) -> list[An first = fetch(1) records = list(first.records) total_pages = -(-first.total_count // page_size) + print(f"[SDS sync] 分页开始: 共 {first.total_count} 条 / {total_pages} 页 (page_size={page_size})") for page in range(2, total_pages + 1): resp = fetch(page) records.extend(resp.records) + if page % 10 == 0 or page == total_pages: + print(f"[SDS sync] 分页进度: {page}/{total_pages} 页, 累计 {len(records)} 条") time.sleep(PAGE_DELAY_MS / 1000) + print(f"[SDS sync] 分页完成: 共拉取 {len(records)} 条") return records @@ -153,14 +161,19 @@ def fetch_goods_concurrent(client: RiinDisClient, order_ids: list[str]) -> dict[ results: dict[str, list[Any]] = {} if not order_ids: return results + concurrency = min(GOODS_CONCURRENCY, len(order_ids)) + print(f"[RIIN sync] 商品拉取开始: {len(order_ids)} 个订单 (并发={concurrency})") def make(oid: str) -> Callable[[], None]: def run() -> None: results[oid] = fetch_goods_for_order(client, oid) return run - with ThreadPoolExecutor(max_workers=min(GOODS_CONCURRENCY, len(order_ids))) as pool: + with ThreadPoolExecutor(max_workers=concurrency) as pool: list(pool.map(lambda fn: fn(), [make(oid) for oid in order_ids])) + total_goods = sum(len(v) for v in results.values()) + empty_orders = sum(1 for v in results.values() if not v) + print(f"[RIIN sync] 商品拉取完成: 共 {total_goods} 条商品, {empty_orders} 个订单无商品") return results diff --git a/flows/shared/db.py b/flows/shared/db.py index 6d77e05..225cebb 100644 --- a/flows/shared/db.py +++ b/flows/shared/db.py @@ -111,6 +111,7 @@ class SyncDB: "goodsUpdated": 0, "goodsRemoved": 0, "ordersDeleted": 0, } if not orders and not goods and not delete_order_nos: + print("[DB sync] 无数据可写入,跳过") return result conn = self._conn() @@ -163,12 +164,18 @@ class SyncDB: for g in goods_dedup: goods_by_order.setdefault(g.orderNo, []).append(g) + print(f"[DB sync] 准备写入: {len(orders_dedup)} 订单 / {len(goods_dedup)} 商品" + + (f" (系统={system_name})" if system_name else "")) + # 3. 分批 upsert orders + 关联 goods + total_order_batches = -(-len(orders_dedup) // ORDER_CHUNK) for i in range(0, len(orders_dedup), ORDER_CHUNK): chunk = orders_dedup[i : i + ORDER_CHUNK] related: list[GoodsUpsertInput] = [] for o in chunk: related.extend(goods_by_order.get(o.orderNo, [])) + batch_no = i // ORDER_CHUNK + 1 + print(f"[DB sync] 订单批次 {batch_no}/{total_order_batches}: {len(chunk)} 订单 / {len(related)} 商品") stats = self._upsert_order_batch(conn, chunk, related, system_name) for k, v in stats.items(): result[k] += v @@ -199,6 +206,7 @@ class SyncDB: raise finally: conn.close() + print(f"[DB sync] 写入完成: {result}") return result def _upsert_order_batch(