feat: 详细日志(分页/映射/批次进度)+ Artifacts 摘要(sync-summary/verify-report)

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",
 ]

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}")

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
 
 

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(