141 lines
6.2 KiB
Python
141 lines
6.2 KiB
Python
"""数据核验日报(每日 18:30):RIIN/SDS 数量一致性 + 状态差异 + OCR 当日 + 失败任务回顾。
|
||
|
||
对应 daily-report deployment,cron 30 18 * * *。报告结构:
|
||
1. RIIN/SDS:先报订单数量(远程 vs 本地,有差异修复后重数)与商品行数是否一致;
|
||
再报核验时发现多少单状态不符(已自动刷新修复)。subflow 复用 riin/sds-verify。
|
||
2. OCR:今天处理了多少订单、识别失败多少(按 ocrProcessedAt 盖章统计)。
|
||
3. 失败任务:近 24h 哪些 workflow、哪些 run、失败几次、何时失败(事件报警的兜底汇总)。
|
||
|
||
消息排版遵循 WeCom aibot markdown 子集(无表格、无 font 标签,引用块 + emoji)。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
from datetime import datetime, timedelta
|
||
|
||
from prefect import flow
|
||
|
||
from flows.monitor.device_check import check_prefect_failures
|
||
from flows.riin.verify import riin_verify
|
||
from flows.sds.verify import sds_verify
|
||
from flows.shared import SyncDB, emit_sync_summary, send_markdown, start_of_day
|
||
|
||
|
||
def _affected_units(diff: dict) -> int:
|
||
"""状态不符单数 ≈ 本地比远程多的状态计数之和(远程改了状态的单)。"""
|
||
return sum(v for v in (diff or {}).values() if v > 0)
|
||
|
||
|
||
def _system_lines(name: str, report: dict, goods: dict, lines: list[str]) -> bool:
|
||
"""单系统的核验两段:①数量是否一致(订单+商品) ②状态不符多少单。返回是否需关注。"""
|
||
vr = (report.get("verifyReport") or {}) if isinstance(report, dict) else {}
|
||
days_r = vr.get("days") or []
|
||
post = vr.get("post")
|
||
if post:
|
||
remote_t, local_t = post["remote"], post["local"]
|
||
else:
|
||
remote_t = sum(d.get("remote", 0) for d in days_r)
|
||
local_t = sum(d.get("local", 0) for d in days_r)
|
||
# 数量一致只看总量;状态差异属第二段,不混入数量结论
|
||
ok = remote_t == local_t
|
||
|
||
if ok:
|
||
lines.append(f"**{name} 订单** 远程 {remote_t:,} · 本地 {local_t:,} ✅")
|
||
else:
|
||
lines.append(f"**{name} 订单** 远程 {remote_t:,} · 本地 {local_t:,} ⚠️ 不一致")
|
||
|
||
diff_days = [d for d in days_r if not d.get("ok")]
|
||
diff_units = sum(_affected_units(d.get("diff")) for d in diff_days)
|
||
if diff_units:
|
||
lines.append(f"> 核验时发现约 {diff_units:,} 单状态不符（{len(diff_days)} 天），已自动刷新修复")
|
||
|
||
g = goods or {}
|
||
zero = g.get("zeroGoodsOrders", 0)
|
||
if zero:
|
||
lines.append(f"**{name} 商品** {g.get('goods', 0):,} 行 · {zero:,} 单缺商品 ⚠️")
|
||
return True
|
||
lines.append(f"**{name} 商品** {g.get('goods', 0):,} 行 · 无缺漏 ✅")
|
||
return not ok
|
||
|
||
|
||
def build_daily_message(
|
||
days: int, window: str, riin: dict, sds: dict,
|
||
goods: dict, ocr_today: dict, ocr_pending: int, failures: list[dict],
|
||
) -> tuple[str, bool]:
|
||
"""拼装日报 markdown,返回 (消息, 是否需关注)。"""
|
||
lines = [f"📊 **数据核验日报** {datetime.now().strftime('%m-%d %H:%M')}"
|
||
f"（近 {days} 天，{window}）", ""]
|
||
attention = False
|
||
|
||
for name, report in (("RIIN", riin), ("SDS", sds)):
|
||
try:
|
||
attention |= _system_lines(name, report, (goods or {}).get(name), lines)
|
||
except Exception as err: # noqa: BLE001
|
||
lines.append(f"**{name}** 核验结果解析失败：{err}")
|
||
attention = True
|
||
lines.append("")
|
||
|
||
processed, failed = (ocr_today or {}).get("processed", 0), (ocr_today or {}).get("failed", 0)
|
||
lines.append(f"**OCR 今日** 已处理 {processed:,} 单 · 识别失败 {failed:,} 单" + (" ✅" if not failed else ""))
|
||
if ocr_pending:
|
||
lines.append(f"> 累计待处理 {ocr_pending:,} 单")
|
||
lines.append("")
|
||
|
||
if failures:
|
||
attention = True
|
||
lines.append(f"**失败任务(24h)** 共 {len(failures)} 次 🔴")
|
||
for f in failures[:6]:
|
||
start = (f.get("start") or "")[5:] # MM-DD HH:MM
|
||
lines.append(f"> {f.get('flow', '?')} · run「{f.get('name', '?')}」{f.get('state', '?')} · {start}")
|
||
if len(failures) > 6:
|
||
lines.append(f"> …等共 {len(failures)} 次")
|
||
else:
|
||
lines.append("**失败任务(24h)** 0 次 ✅")
|
||
|
||
lines.append("")
|
||
if attention:
|
||
lines.append("⚠️ **存在需关注项，见上**")
|
||
else:
|
||
lines.append("✅ **数据全部一致，无异常**")
|
||
return "\n".join(lines), attention
|
||
|
||
|
||
@flow(name="daily-report", log_prints=True)
|
||
def daily_report(days: int = 7) -> dict:
|
||
"""近 N 天数据核验 + 商品完整性 + OCR/失败回顾,发企微群日报。"""
|
||
today = start_of_day(datetime.now())
|
||
window_start = today - timedelta(days=days)
|
||
window_end = today # 与 verify 的 recent 约定一致:近 N 天、不含今天
|
||
|
||
riin = riin_verify(mode="recent", recent_days=days)
|
||
sds = sds_verify(mode="recent", recent_days=days)
|
||
|
||
db = SyncDB()
|
||
goods = {sys: db.get_goods_integrity(sys, window_start, window_end) for sys in ("RIIN", "SDS")}
|
||
ocr_today = db.count_ocr_today(today)
|
||
ocr = db.count_ocr_status()
|
||
failures = check_prefect_failures(hours=24)
|
||
|
||
window = f"{window_start:%m-%d}~{window_end:%m-%d}"
|
||
content, attention = build_daily_message(days, window, riin, sds, goods,
|
||
ocr_today, ocr.get("pending", 0), failures)
|
||
|
||
sent = send_markdown(content)
|
||
emit_sync_summary("数据核验日报", [
|
||
("核验窗口", f"近{days}天"), ("需关注", "是" if attention else "否"),
|
||
("RIIN 零商品订单", goods["RIIN"]["zeroGoodsOrders"]),
|
||
("RIIN 商品无名称", goods["RIIN"]["goodsNoName"]),
|
||
("SDS 零商品订单", goods["SDS"]["zeroGoodsOrders"]),
|
||
("SDS 商品无名称", goods["SDS"]["goodsNoName"]),
|
||
("全库孤儿商品", goods["RIIN"]["orphanGoods"]),
|
||
("OCR 今日处理", ocr_today.get("processed", 0)), ("OCR 今日失败", ocr_today.get("failed", 0)),
|
||
("OCR 待处理", ocr.get("pending", 0)), ("失败 runs(24h)", len(failures)),
|
||
("已发企微", "是" if sent else "否"),
|
||
])
|
||
return {"attention": attention, "sent": sent,
|
||
"goods": goods, "ocrToday": ocr_today, "failures24h": len(failures)}
|
||
|
||
|
||
if __name__ == "__main__":
|
||
daily_report.serve(name="daily-report", cron="30 18 * * *")
|