"""数据核验日报(每日 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、失败几次、何时失败(事件报警的兜底汇总)。 排版:标题 + 单行结论 + 普通明细行,状态用 三色文字表达,不用 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: """单系统一行结论:数量(远程 vs 本地,一致只显示一个数)+商品;状态不符进引用行。""" 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 # 数量一致只看总量;状态差异属第二段,不混入数量结论 g = goods or {} zero = g.get("zeroGoodsOrders", 0) goods_txt = f"{g.get('goods', 0):,} 行" + (f'(缺 {zero:,})' if zero else "") if ok: order_txt = f"订单 {remote_t:,} 一致" else: order_txt = f"订单 远程 {remote_t:,} / 本地 {local_t:,} 不一致" lines.append(f"**{name}** {order_txt} · 商品 {goods_txt}") 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)} 天),已自动刷新修复") return (not ok) or bool(zero) 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():%m-%d %H:%M}(近 {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 processed, failed = (ocr_today or {}).get("processed", 0), (ocr_today or {}).get("failed", 0) ocr_txt = f"**OCR** 今日处理 {processed:,} · 识别失败 " ocr_txt += f'{failed:,}' if failed else "0" if ocr_pending: ocr_txt += f" · 待处理 {ocr_pending:,}" lines.append(ocr_txt) if failures: attention = True lines.append(f'**任务** 失败 {len(failures)} 次(24h)') for f in failures[:6]: start = (f.get("start") or "")[5:] # MM-DD HH:MM lines.append(f"{f.get('flow', '?')}:{f.get('name', '?')} {f.get('state', '?')} {start}") if len(failures) > 6: lines.append(f"…等共 {len(failures)} 次") else: lines.append("**任务** 失败 0 次(24h)") lines.append('存在需关注项,见上' if attention else '数据全部一致,无异常') 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 * * *")