83 lines
3.1 KiB
Python
83 lines
3.1 KiB
Python
"""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}")
|