"""数据核验日报(每日 18:30):近 N 天 RIIN/SDS 订单+商品数据准确性核验并上报企微。
对应 daily-report deployment,cron 30 18 * * *。内容:
1. 复用 riin-verify / sds-verify 逻辑(subflow 调用):远程 vs 本地逐天逐状态计数比对,
有差异自动拉取修复——与早 6/7 点的每日核验互补,二次确认白天同步质量。
2. 商品数据完整性:窗口内零商品订单 / 商品无名称 / 全库孤儿商品(本地 SQL)。
3. OCR 处理状态与当日失败 flow runs 回顾(事件报警的兜底汇总)。
消息排版遵循 WeCom markdown 子集(无表格,引用块+彩色字体)。
"""
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 _fmt_diff(diff: dict) -> str:
return " ".join(f"{k}{v:+d}" for k, v in (diff or {}).items()) or "—"
def build_daily_message(
days: int, window: str, riin: dict, sds: dict,
goods: dict, ocr: dict, 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:
days_r = (report.get("verifyReport") or {}).get("days") or []
summary = (report.get("verifyReport") or {}).get("summary") or {}
diff_days = [d for d in days_r if not d.get("ok")]
total, n_diff = summary.get("totalDays", len(days_r)), summary.get("daysWithDiff", len(diff_days))
if diff_days:
lines.append(f"**{name}** {total - n_diff}/{total} 天一致,"
f'**{n_diff} 天有差异(已补拉)**')
for d in diff_days[:8]:
lines.append(f"> {d['date']} {_fmt_diff(d.get('diff'))}(补拉 {d.get('orders', 0)} 单)")
if len(diff_days) > 8:
lines.append(f"> …共 {len(diff_days)} 天")
totals = {k: v for k, v in summary.items()
if k not in ("totalDays", "daysWithDiff") and v}
if totals:
lines.append(f"> 补拉量合计 {totals}")
lines.append("> 注:同一差异长期存在多为统计口径/状态映射问题(如发货后退款未同步)")
attention = True
else:
lines.append(f"**{name}** {total}/{total} 天一致 ✅")
except Exception as err: # noqa: BLE001
lines.append(f"**{name}** 核验结果解析失败:{err}")
attention = True
lines.append("")
lines.append("**商品完整性**")
for sys in ("RIIN", "SDS"):
g = goods[sys]
zero = f'**{g["zeroGoodsOrders"]}**' if g["zeroGoodsOrders"] else "0"
lines.append(f"> {sys} 订单 {g['orders']:,} · 商品 {g['goods']:,} · 零商品订单 {zero} · 无名称 {g['goodsNoName']:,}")
attention = attention or g["zeroGoodsOrders"] > 0
lines.append(f"> 全库孤儿商品 {goods['RIIN']['orphanGoods']:,}")
lines.append("")
lines.append(f"**OCR** 完成 {ocr.get('done', 0):,} · 失败 {ocr.get('error', 0):,}"
f" · 不支持 {ocr.get('unsupported', 0):,} · 待处理 {ocr.get('pending', 0):,}")
lines.append(f"**失败 runs(24h)** {len(failures)} 个" + (" ✅" if not failures else ""))
for f in failures[:5]:
lines.append(f"> 🔴 {f['name']}({f['state']}){f['start']}")
if failures:
attention = True
lines += ["", ('⚠️ **存在需关注项,见上**' 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 = 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, failures)
sent = send_markdown(content)
emit_sync_summary("数据核验日报", [
("核验窗口", f"近{days}天"), ("需关注", "是" if attention else "否"),
("RIIN 零商品订单", goods["RIIN"]["zeroGoodsOrders"]),
("SDS 零商品订单", goods["SDS"]["zeroGoodsOrders"]),
("OCR 待处理", ocr.get("pending", 0)), ("失败 runs(24h)", len(failures)),
("已发企微", "是" if sent else "否"),
])
return {"attention": attention, "sent": sent,
"goods": goods, "ocr": ocr, "failures24h": len(failures)}
if __name__ == "__main__":
daily_report.serve(name="daily-report", cron="30 18 * * *")