From 14321faa9100436c7766f79720e71609f2948326 Mon Sep 17 00:00:00 2001
From: yeuimu <2197651308@qq.com>
Date: Fri, 28 Aug 2026 01:55:01 +0000
Subject: [PATCH] =?UTF-8?q?style:=20=E6=8A=A5=E5=91=8A=E6=B6=88=E6=81=AF?=
=?UTF-8?q?=E6=94=B9=E7=94=A8=20WeCom=20=E5=8F=8B=E5=A5=BD=E6=8E=92?=
=?UTF-8?q?=E7=89=88(=E5=8E=BB=E8=A1=A8=E6=A0=BC/=E5=BD=A9=E8=89=B2?=
=?UTF-8?q?=E9=AB=98=E4=BA=AE/=E5=BC=95=E7=94=A8=E5=9D=97)?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit
---
flows/monitor/alert_failure.py | 14 ++--
flows/monitor/daily_report.py | 122 +++++++++++++++++---------------
flows/monitor/device_check.py | 66 +++++++++--------
scripts/wecom_capture_chatid.py | 2 +-
4 files changed, 108 insertions(+), 96 deletions(-)
diff --git a/flows/monitor/alert_failure.py b/flows/monitor/alert_failure.py
index 86d22ba..0b3612d 100644
--- a/flows/monitor/alert_failure.py
+++ b/flows/monitor/alert_failure.py
@@ -103,20 +103,18 @@ def wecom_alert_failure(event: dict | str | None = None) -> dict:
occurred = occurred or ended
lines = [
- "🔴 **Prefect 流程失败报警**",
- f"- 事件: `{event_name}`",
- f"- Flow: {flow_name or resource_name or '未知'}",
+ f'🔴 **流程失败** {flow_name or resource_name or "未知"}',
]
if run_name and run_name != flow_name:
- lines.append(f"- Run: {run_name}")
+ lines.append(f"> Run：{run_name}")
if state_msg:
- lines.append(f"- 错误: {state_msg[:200].replace(chr(10), ' ')}")
- lines.append(f"- 时间: {_fmt_ts(occurred)}")
+ lines.append(f"> 错误：{state_msg[:150].replace(chr(10), ' ')}")
+ lines.append(f"> {event_name} · {_fmt_ts(occurred)}")
if run_id:
- lines.append(f"- 详情: [Prefect UI]({UI_BASE}/runs/flow-run/{run_id})")
+ lines.append(f"> [查看详情]({UI_BASE}/runs/flow-run/{run_id})")
else:
raw = json.dumps(ev, ensure_ascii=False)[:300] if ev else "(事件内容解析失败)"
- lines.append(f"- 原始事件: {raw}")
+ lines.append(f"> 原始事件：{raw}")
result["sent"] = send_markdown("\n".join(lines))
result.update({"flowName": flow_name or None, "runId": run_id})
diff --git a/flows/monitor/daily_report.py b/flows/monitor/daily_report.py
index 6b44a35..df4cfa1 100644
--- a/flows/monitor/daily_report.py
+++ b/flows/monitor/daily_report.py
@@ -1,10 +1,12 @@
-"""数据核验日报(每日 18:30):近 N 天 RIIN/SDS 订单+商品数据准确性核验并上报企微群。
+"""数据核验日报(每日 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
@@ -19,24 +21,65 @@ from flows.sds.verify import sds_verify
from flows.shared import SyncDB, emit_sync_summary, send_markdown, start_of_day
-def _verify_section(name: str, report: dict) -> tuple[str, list[str], bool]:
- """把 verify flow 返回的 verifyReport 压缩成结论行 + 差异日明细行。"""
- days = (report.get("verifyReport") or {}).get("days") or []
- summary = (report.get("verifyReport") or {}).get("summary") or {}
- diff_days = [d for d in days if not d.get("ok")]
- total, n_diff = summary.get("totalDays", len(days)), summary.get("daysWithDiff", len(diff_days))
- head = f"**{name}**: {total} 天中 {total - n_diff} 天一致" + (" ✅" if not diff_days else f", {n_diff} 天有差异(已补拉)")
- detail = [
- f"- {d['date']}: 本地 {d.get('local', '—')} / 远程 {d.get('remote', '—')},"
- f"差异 {d.get('diff') or '—'},已补拉 订单 {d.get('orders', 0)} / 商品 {d.get('goods', 0)}"
- for d in diff_days
- ]
- totals = {k: v for k, v in summary.items() if k not in ("totalDays", "daysWithDiff")}
- if any(totals.values()):
- detail.append(f"- 补拉量合计: {totals}")
- if diff_days:
- detail.append("- 注: 差异日已重新补拉;若同一差异长期存在,多为远程统计口径与本地状态映射不一致(如发货后退款未同步),需单独排查")
- return head, detail, bool(diff_days)
+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)
@@ -52,47 +95,12 @@ def daily_report(days: int = 7) -> dict:
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)
- lines = [f"📊 **数据核验日报** {datetime.now().strftime('%Y-%m-%d %H:%M')}"
- f"(近 {days} 天,{window_start:%m-%d} ~ {window_end:%m-%d})", ""]
+ window = f"{window_start:%m-%d}~{window_end:%m-%d}"
+ content, attention = build_daily_message(days, window, riin, sds, goods, ocr, failures)
- attention = False
- for name, report in (("RIIN", riin), ("SDS", sds)):
- try:
- head, detail, had_diff = _verify_section(name, report)
- lines.append(head)
- lines.extend(detail)
- attention = attention or had_diff
- except Exception as err: # noqa: BLE001
- lines.append(f"**{name}**: 核验结果解析失败: {err}")
- attention = True
-
- lines.append("**商品完整性**:")
- lines.append("| 系统 | 订单 | 商品行 | 零商品订单 | 商品无名称 | 孤儿商品(全库) |")
- lines.append("| --- | --- | --- | --- | --- | --- |")
- for sys in ("RIIN", "SDS"):
- g = goods[sys]
- flag = " ⚠️" if g["zeroGoodsOrders"] > 0 else ""
- lines.append(f"| {sys} | {g['orders']} | {g['goods']} | {g['zeroGoodsOrders']}{flag} "
- f"| {g['goodsNoName']} | {g['orphanGoods']} |")
- if g["zeroGoodsOrders"] > 0:
- attention = True
-
- lines.append(f"**OCR(SDS)**: 完成 {ocr.get('done', 0)} / 失败 {ocr.get('error', 0)} / "
- f"不支持 {ocr.get('unsupported', 0)} / 待处理 {ocr.get('pending', 0)}")
- lines.append(f"**近24h 失败 flow runs**: {len(failures)} 个" + ("" if failures else " ✅"))
- for f in failures[:8]:
- lines.append(f"- 🔴 {f['name']}({f['state']}) {f['start']} {f['message'][:60]}")
- if len(failures) > 8:
- lines.append(f"- ...共 {len(failures)} 个,详见 Prefect UI")
- if failures:
- attention = True
-
- lines += ["", ("⚠️ **存在需关注项,见上**" if attention else "✅ **数据全部一致,无异常**")]
-
- sent = send_markdown("\n".join(lines))
+ sent = send_markdown(content)
emit_sync_summary("数据核验日报", [
("核验窗口", f"近{days}天"), ("需关注", "是" if attention else "否"),
("RIIN 零商品订单", goods["RIIN"]["zeroGoodsOrders"]),
diff --git a/flows/monitor/device_check.py b/flows/monitor/device_check.py
index 8bde042..e81e0a7 100644
--- a/flows/monitor/device_check.py
+++ b/flows/monitor/device_check.py
@@ -168,9 +168,12 @@ def _check_services() -> dict:
def _mark(pct: float | None, threshold: int) -> str:
+ """百分比展示:WeCom markdown 不支持表格,超阈值用红色字体高亮。"""
if pct is None:
return "—"
- return f"{pct:.0f}%" + (" 🔴" if pct >= threshold else "")
+ if pct >= threshold:
+ return f'**{pct:.0f}%**'
+ return f"{pct:.0f}%"
@flow(name="device-check", log_prints=True)
@@ -188,55 +191,57 @@ def device_check(alarm_threshold: int = 80) -> dict:
print(f"[device-check] {name} 检查失败: {err}")
threshold = alarm_threshold
- lines = [f"🖥 **设备健康日报** {datetime.now().strftime('%Y-%m-%d %H:%M')}", ""]
+ lines = [f"🖥 **设备健康日报** {datetime.now().strftime('%m-%d %H:%M')}", ""]
mem = checks.get("memory") or {}
if mem:
- lines.append(f"**内存**: {_mark(mem['pct'], threshold)}({mem['used']}/{mem['total']})"
- f" | swap {_mark(mem['swapPct'], threshold)}({mem['swapUsed']}/{mem['swapTotal']})")
+ lines.append(f"**内存** {_mark(mem['pct'], threshold)}（{mem['used']}/{mem['total']}）"
+ f"｜swap {_mark(mem['swapPct'], threshold)}（{mem['swapUsed']}/{mem['swapTotal']}）")
else:
- lines.append("**内存**: 检查失败")
+ lines.append("**内存** 检查失败")
- lines.append("**磁盘**:")
- lines.append("| 挂载点 | 使用率 | 已用/总量 |")
- lines.append("| --- | --- | --- |")
- for d in checks.get("disks") or []:
+ disks = checks.get("disks") or []
+ if disks:
+ lines.append("**磁盘**")
+ for d in disks:
if "error" in d:
- lines.append(f"| {d['label']} | 检查失败 | {d['error']} |")
+ lines.append(f"> {d['label']} 检查失败：{d['error']}")
else:
- lines.append(f"| {d['label']} | {_mark(d['pct'], threshold)} | {d['used']}/{d['total']} |")
+ lines.append(f"> {d['label']} {_mark(d['pct'], threshold)}（{d['used']}/{d['total']}）")
- lines.append("**GPU**:")
gpus = checks.get("gpus")
if isinstance(gpus, list) and gpus:
- lines.append("| GPU | 利用率 | 显存 |")
- lines.append("| --- | --- | --- |")
+ lines.append("**GPU**")
for g in gpus:
- lines.append(f"| {g['name']} | {_mark(g['utilPct'], threshold)} | "
- f"{_mark(g['memPct'], threshold)}({g['memUsed']}/{g['memTotal']}) |")
+ lines.append(f"> {g['name']}｜利用率 {_mark(g['utilPct'], threshold)}"
+ f"｜显存 {_mark(g['memPct'], threshold)}（{g['memUsed']}/{g['memTotal']}）")
else:
- lines.append(f"> {gpus if isinstance(gpus, str) else '未检测到'}")
+ lines.append(f"**GPU** {gpus if isinstance(gpus, str) else '未检测到'}")
docker = checks.get("docker") or {}
issues = docker.get("issues") or []
if docker.get("running") is None:
- lines.append(f"**Docker**: {issues[0]['issue'] if issues else '不可用'}")
+ lines.append(f"**Docker** {issues[0]['issue'] if issues else '不可用'}")
else:
- lines.append(f"**Docker 容器**: 运行中 {docker['running']}/{docker.get('total', '?')}"
- + (f",已停用 {docker['stopped']}" if docker.get("stopped") else "")
- + (f",⚠️ 异常 {len(issues)} 个" if issues else ",全部正常"))
- for it in issues[:10]:
- lines.append(f"- 🔴 {it['name']}: {it['issue']}")
+ docker_line = f"**Docker** 运行 {docker['running']}/{docker.get('total', '?')}"
+ if docker.get("stopped"):
+ docker_line += f" · 已停用 {docker['stopped']}"
+ if issues:
+ docker_line += f" · 异常 **{len(issues)}**"
+ else:
+ docker_line += " · 异常 0"
+ lines.append(docker_line)
+ for it in issues[:8]:
+ lines.append(f"> 🔴 {it['name']}：{it['issue']}")
failures = checks.get("prefectFailures") or []
- lines.append(f"**Prefect 近24h 失败**: {len(failures)} 个 flow run"
- + ("" if failures else " ✅"))
- for f in failures[:8]:
- lines.append(f"- 🔴 {f['name']}({f['state']}) {f['start']} {f['message'][:60]}")
+ lines.append(f"**Prefect 失败(24h)** {len(failures)} 个" + (" ✅" if not failures else ""))
+ for f in failures[:5]:
+ lines.append(f"> 🔴 {f['name']}（{f['state']}）{f['start']}")
svc = checks.get("services") or {}
- lines.append(f"**服务探活**: OCR {'✅' if svc.get('ocr') else '🔴'} | "
- f"PG {'✅' if svc.get('pg') else '🔴'} | 负载 {os.getloadavg()[0]:.1f}")
+ lines.append(f"**探活** OCR {'✅' if svc.get('ocr') else '🔴'}｜PG {'✅' if svc.get('pg') else '🔴'}"
+ f"｜负载 {os.getloadavg()[0]:.1f}")
alarms = (
(mem and (mem["pct"] >= threshold or mem["swapPct"] >= threshold))
@@ -244,7 +249,8 @@ def device_check(alarm_threshold: int = 80) -> dict:
or (isinstance(gpus, list) and any(g["utilPct"] >= threshold or g["memPct"] >= threshold for g in gpus))
or bool(issues) or bool(failures) or not svc.get("ocr") or not svc.get("pg")
)
- lines += ["", ("🔴 **有项目需要处理,见上**" if alarms else "✅ **一切正常**")]
+ lines += ["", ('🔴 **有项目需要处理，见上**' if alarms
+ else '✅ **一切正常**')]
content = "\n".join(lines)
sent = send_markdown(content)
diff --git a/scripts/wecom_capture_chatid.py b/scripts/wecom_capture_chatid.py
index fd5355e..bbbbca3 100644
--- a/scripts/wecom_capture_chatid.py
+++ b/scripts/wecom_capture_chatid.py
@@ -63,7 +63,7 @@ def do_save_chatid(chatid: str) -> None:
def do_test(message: str) -> None:
from flows.shared.wecom import send_markdown
- ok = send_markdown(f"# 机器人连通性测试\n\n{message}\n\n来自 inkreach-prefect 通知链路。")
+ ok = send_markdown(f'✅ **机器人通知链路正常**\n\n> {message}')
sys.exit(0 if ok else 1)