feat: 四份报告按用户口径重做

- 核验日报: 先报订单数量(远程vs本地, verify 修复后重数 verifyReport.post)/商品行数一致性, 再报状态不符单数; SDS 统计口径排除'售后'与本地对称
- OCR: Orders 加 ocrProcessedAt 列(写入自动盖时间戳), 日报报'今日处理 X · 识别失败 Y'(db.count_ocr_today)
- 失败任务: check_prefect_failures 补 workflow 名, 日报/设备报列 run 名+状态+时间
- 设备日报: 新增 metrics-agent 侧车(每5分钟采样内存/swap/GPU+每日 du 快照), 报告改 平均/峰值/最低 + 总体/根目录/home 余量 + 增长最快目录 + GPU 温度; 采样缺失降级瞬时值
- Dockerfile 加 tzdata(TZ=Asia/Shanghai); du 用实际占块且容错 rc=1

.dockerignore

@@ -0,0 +1,6 @@
+data/
+.git/
+.venv/
+__pycache__/
+*.log
+.env

.gitignore

@@ -6,3 +6,5 @@ __pycache__/
 external_projects/
 *.log
 .prefect/
+data/
+.zcode/

AGENTS.md

@@ -84,9 +84,10 @@ uv run prefect <cmd>         # 运行 prefect CLI
 - 发送层 `flows/shared/wecom.py`:`send_markdown/send_text`,wecom-aibot-sdk 按需连接(连接→认证→发送→断开),全容错。
   凭据在 `Secret` Block `wecom-aibot-credentials`(JSON:`bot_id/secret/chat_id`);chat_id 用 `scripts/wecom_capture_chatid.py` 捕获。
 - `flows/monitor/alert_failure.py`(wecom-alert):prefect.yaml 里声明的事件触发器,任意 flow run `Failed/Crashed` 自动运行并发群报警,零代码侵入,新 flow 自动覆盖。
-- `flows/monitor/daily_report.py`(daily-report,18:30):近 7 天 RIIN/SDS 核验(subflow 复用 verify)+商品完整性(`db.get_goods_integrity`)+OCR+失败回顾。
+- `flows/monitor/daily_report.py`(daily-report,18:30):近 7 天 RIIN/SDS 核验(subflow 复用 verify,先报订单数量/商品行数是否一致,再报状态不符单数"已自动刷新修复",verify 有差异时修复后重数即 `verifyReport.post`)+OCR 当日处理量(按 `Orders.ocrProcessedAt` 盖章,`db.count_ocr_today`)+近 24h 失败任务明细(workflow/run/次数/时间)。
-- `flows/monitor/device_check.py`(device-check,09:00):内存/swap/宿主机磁盘(`/` 挂载在容器 `/host`)/GPU(nvidia-smi)阈值 80% 报警 + Docker 容器异常(docker.sock)+Prefect 失败+OCR/PG 探活。
+- `flows/monitor/device_check.py`(device-check,09:00):按 24h 采样报内存/swap/GPU 的平均·峰值·最低 + 磁盘 总体/根目录/home 余量与"增长最快目录"(du 快照对比)+ GPU 温度;另查 Docker 容器异常(docker.sock)、Prefect 失败、OCR/PG 探活。阈值 80% 报警。
-- worker 容器已挂载 `/:/host:ro`、`/var/run/docker.sock`、GPU(见 docker-compose.yml),勿移除,否则对应检查降级。
+- **`metrics-agent` 侧车容器**(docker-compose 服务,非 Prefect flow):`scripts/metrics_agent.py` 每 5 分钟采样内存/swap/GPU 写 `data/metrics/metrics-YYYY-MM-DD.jsonl`,每日 7 点后 du 快照 `data/metrics/disk-*.json`(实际占块;容错个别文件消失的 rc=1);保留采样 14 天/快照 7 份。device-check 经 `/data/metrics:ro` 读取,采样缺失自动降级为瞬时值。
+- worker 容器已挂载 `/:/host:ro`、`/var/run/docker.sock`、GPU、`./data/metrics:ro`(见 docker-compose.yml),勿移除,否则对应检查降级;镜像含 tzdata(`ENV TZ=Asia/Shanghai`)。
 
 ## 参考文档
 

Dockerfile

@@ -1,8 +1,10 @@
 FROM python:3.12-slim
 
-RUN apt-get update && apt-get install -y --no-install-recommends git curl ca-certificates \
+RUN apt-get update && apt-get install -y --no-install-recommends git curl ca-certificates tzdata \
     && rm -rf /var/lib/apt/lists/*
 
+ENV TZ=Asia/Shanghai
+
 RUN curl -LsSf https://astral.sh/uv/install.sh | sh
 ENV PATH="/root/.local/bin:$PATH"
 

docker-compose.yml

@@ -21,9 +21,10 @@ services:
     depends_on:
       - prefect-server
     volumes:
-      # 设备监控(device-check flow):宿主机磁盘使用率 + Docker 容器状态
+      # 设备监控(device-check flow):宿主机磁盘使用率 + Docker 容器状态 + 指标采样
       - /var/run/docker.sock:/var/run/docker.sock
       - /:/host:ro
+      - ./data/metrics:/data/metrics:ro
     deploy:
       resources:
         reservations:
@@ -34,5 +35,26 @@ services:
               capabilities: [gpu]
     restart: unless-stopped
 
+  # 宿主机指标采样侧车:每 5 分钟写内存/swap/GPU,每日 du 快照(供 device-check 日报)
+  metrics-agent:
+    build: .
+    container_name: metrics-agent
+    command: python /opt/metrics_agent.py
+    environment:
+      TZ: Asia/Shanghai
+      METRICS_DIR: /data/metrics
+    volumes:
+      - /:/host:ro
+      - ./data/metrics:/data/metrics
+      - ./scripts/metrics_agent.py:/opt/metrics_agent.py:ro
+    deploy:
+      resources:
+        reservations:
+          devices:
+            - driver: nvidia
+              count: all
+              capabilities: [gpu]
+    restart: unless-stopped
+
 volumes:
   prefect-data:

flows/monitor/daily_report.py

@@ -1,12 +1,12 @@
-"""数据核验日报(每日 18:30):近 N 天 RIIN/SDS 订单+商品数据准确性核验并上报企微。
+"""数据核验日报(每日 18:30):RIIN/SDS 数量一致性 + 状态差异 + OCR 当日 + 失败任务回顾。
 
-对应 daily-report deployment,cron 30 18 * * *。内容:
+对应 daily-report deployment,cron 30 18 * * *。报告结构:
-1. 复用 riin-verify / sds-verify 逻辑(subflow 调用):远程 vs 本地逐天逐状态计数比对,
+1. RIIN/SDS:先报订单数量(远程 vs 本地,有差异修复后重数)与商品行数是否一致;
-   有差异自动拉取修复——与早 6/7 点的每日核验互补,二次确认白天同步质量。
+   再报核验时发现多少单状态不符(已自动刷新修复)。subflow 复用 riin/sds-verify。
-2. 商品数据完整性:窗口内零商品订单 / 商品无名称 / 全库孤儿商品(本地 SQL)。
+2. OCR:今天处理了多少订单、识别失败多少(按 ocrProcessedAt 盖章统计)。
-3. OCR 处理状态与当日失败 flow runs 回顾(事件报警的兜底汇总)。
+3. 失败任务:近 24h 哪些 workflow、哪些 run、失败几次、何时失败(事件报警的兜底汇总)。
 
-消息排版遵循 WeCom markdown 子集(无表格,引用块+彩色字体)。
+消息排版遵循 WeCom aibot markdown 子集(无表格、无 font 标签,引用块 + emoji)。
 """
 
 from __future__ import annotations
@@ -21,70 +21,79 @@ from flows.sds.verify import sds_verify
 from flows.shared import SyncDB, emit_sync_summary, send_markdown, start_of_day
 
 
-
-def _short_date(date_str: str) -> str:
-    return date_str[5:] if len(date_str) == 10 and date_str[:4].isdigit() else date_str
-
-
 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, ok = post["remote"], post["local"], post.get("daysWithDiff", 0) == 0
+    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 = all(d.get("ok") for d in days_r)
+
+    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: dict, failures: list[dict],
+    goods: dict, ocr_today: dict, ocr_pending: int, failures: list[dict],
 ) -> tuple[str, bool]:
-    """拼装日报 markdown,返回 (消息, 是否需关注)。结论式极简排版,不暴露内部术语。"""
+    """拼装日报 markdown,返回 (消息, 是否需关注)。"""
     lines = [f"📊 **数据核验日报** {datetime.now().strftime('%m-%d %H:%M')}"
              f"（近 {days} 天，{window}）", ""]
     attention = False
-    diff_names: list[str] = []
 
     for name, report in (("RIIN", riin), ("SDS", sds)):
         try:
-            days_r = (report.get("verifyReport") or {}).get("days") or []
+            attention |= _system_lines(name, report, (goods or {}).get(name), lines)
-            summary = (report.get("verifyReport") or {}).get("summary") or {}
-            diff_days = [d for d in days_r if not d.get("ok")]
-            total = summary.get("totalDays", len(days_r))
-            if not diff_days:
-                lines.append(f"**{name}** {total}/{total} 天与远程一致 ✅")
-            else:
-                diff_names.append(name)
-                attention = True
-                lines.append(f"**{name}** {total - len(diff_days)}/{total} 天与远程一致 ⚠️")
-                if len(diff_days) <= 4:
-                    per_day = " · ".join(
-                        f"{_short_date(d['date'])} {_affected_units(d.get('diff'))}单" for d in diff_days)
-                    lines.append(f"> {per_day}状态不符，已自动重新同步")
-                else:
-                    total_affected = sum(_affected_units(d.get("diff")) for d in diff_days)
-                    lines.append(f"> {len(diff_days)} 天共约 {total_affected:,} 单状态不符，已自动重新同步")
         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("")
 
-    zero_total = sum(g["zeroGoodsOrders"] for g in goods.values())
-    lines += ["", f"**商品** RIIN {goods['RIIN']['orders']:,} 单/{goods['RIIN']['goods']:,} 行"
-              f" · SDS {goods['SDS']['orders']:,} 单/{goods['SDS']['goods']:,} 行"
-              + (" ⚠️" if zero_total else " ✅")]
-    if zero_total:
-        lines.append(f"> 有 {zero_total} 单缺商品数据")
-        attention = True
-    lines.append(f"**OCR** 已完成 {ocr.get('done', 0):,} · 待处理 {ocr.get('pending', 0):,}"
-                 f" · 失败 {ocr.get('error', 0):,}")
     if failures:
         attention = True
-        names = "、".join(f"{f['name']}（{f['state']}）" for f in failures[:4])
+        lines.append(f"**失败任务(24h)** 共 {len(failures)} 次 🔴")
-        lines.append(f"**失败任务(24h)** {len(failures)} 个：{names}")
+        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("**失败任务(24h)** 0 次 ✅")
 
     lines.append("")
     if attention:
         lines.append("⚠️ **存在需关注项，见上**")
-        if diff_names:
-            lines.append("> 每天都出现的相同差异属远程统计口径问题，非数据丢失")
     else:
         lines.append("✅ **数据全部一致，无异常**")
     return "\n".join(lines), attention
@@ -102,11 +111,13 @@ 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_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, failures)
+    content, attention = build_daily_message(days, window, riin, sds, goods,
+                                             ocr_today, ocr.get("pending", 0), failures)
 
     sent = send_markdown(content)
     emit_sync_summary("数据核验日报", [
@@ -116,11 +127,12 @@ def daily_report(days: int = 7) -> dict:
         ("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, "ocr": ocr, "failures24h": len(failures)}
+            "goods": goods, "ocrToday": ocr_today, "failures24h": len(failures)}
 
 
 if __name__ == "__main__":

flows/monitor/device_check.py

@@ -1,19 +1,22 @@
-"""设备/系统健康日报(每日一次):内存/磁盘/GPU 使用率 + 系统级错误检查。
+"""设备/系统健康日报(每日 09:00):内存/swap/GPU 平均·峰值·最低 + 磁盘余量与目录增长 + 系统级错误。
 
-对应 device-check deployment,每天 09:00(可调)。检查项:
+对应 device-check deployment。数据来源:
-- 内存/swap、宿主机磁盘(经 /:/host:ro 挂载)、GPU(nvidia-smi)——使用率 ≥ 阈值(默认 80%)标 🔴
+- 24h 指标:metrics-agent 侧车容器每 5 分钟采样,写 data/metrics/metrics-YYYY-MM-DD.jsonl(worker 经
-- Docker 容器状态(经 /var/run/docker.sock):应运行但退出 / restarting / unhealthy / OOMKilled
+  /data/metrics:ro 只读挂载读取;本地调试直接读仓库 data/metrics)
-- Prefect 近 24h FAILED/CRASHED flow runs、OCR 服务探活、PG 连通性、系统负载
+- 磁盘增长:metrics-agent 每日 du 快照 data/metrics/disk-*.json,日报对比最近两份得"增长最快目录"
+- 采样缺失时降级为 psutil/nvidia-smi 瞬时值,不阻塞报告
+另检查:Docker 容器异常、Prefect 近 24h 失败明细、OCR/PG 探活。超阈值(默认 80%)标 🔴。
 
-依赖 docker-compose 给 prefect-worker 挂载 /:/host:ro、/var/run/docker.sock、GPU;
+依赖 compose 挂载:/data/metrics:ro、/:/host:ro、/var/run/docker.sock、GPU;缺失项自动降级。
-未挂载时对应检查项自动降级为"不可用",不影响其余检查。
 """
 
 from __future__ import annotations
 
+import json
 import os
 import subprocess
 from datetime import datetime, timedelta
+from pathlib import Path
 
 import httpx
 import psutil
@@ -24,6 +27,8 @@ from flows.shared import SyncDB, emit_sync_summary, send_markdown
 # (容器内路径, 展示名);/host 为 compose 挂载的宿主机根目录
 DISK_PATHS = [("/host", "/"), ("/host/home", "/home")]
 DOCKER_SOCK = "/var/run/docker.sock"
+METRICS_DIR_CANDIDATES = ("/data/metrics", str(Path(__file__).resolve().parents[2] / "data" / "metrics"))
+GPU_TEMP_WARN = 85  # °C
 
 _OCR_BASE = "http://192.168.124.92:9801"
 
@@ -32,6 +37,115 @@ def _gb(nbytes: float) -> str:
     return f"{nbytes / (1 << 30):.1f}G"
 
 
+def _fmt_growth(nbytes: float) -> str:
+    if abs(nbytes) >= 1 << 30:
+        return f"{nbytes / (1 << 30):+.1f}G"
+    return f"{nbytes / (1 << 20):+.0f}M"
+
+
+def _metrics_dir() -> str | None:
+    cands = [os.environ.get("METRICS_DIR"), *METRICS_DIR_CANDIDATES]
+    for p in cands:
+        if p and os.path.isdir(p):
+            return p
+    return None
+
+
+# ---------- 24h 采样聚合 ----------
+
+def read_samples(metrics_dir: str | None, hours: int = 24) -> list[dict]:
+    """读取 metrics-agent 的 JSONL 采样,返回窗口内记录(ts 升序)。"""
+    if not metrics_dir:
+        return []
+    now = datetime.now()
+    cutoff = now - timedelta(hours=hours)
+    out: list[dict] = []
+    for i in range(hours // 24 + 1):
+        day = (now - timedelta(days=i)).strftime("%Y-%m-%d")
+        path = Path(metrics_dir) / f"metrics-{day}.jsonl"
+        if not path.is_file():
+            continue
+        try:
+            rows = [json.loads(l) for l in path.read_text(encoding="utf-8").splitlines() if l.strip()]
+        except (json.JSONDecodeError, OSError):
+            continue
+        for r in rows:
+            try:
+                if datetime.fromisoformat(r["ts"]) >= cutoff:
+                    out.append(r)
+            except (KeyError, ValueError, TypeError):
+                continue
+    out.sort(key=lambda r: r["ts"])
+    return out
+
+
+def _stats(values: list[float]) -> dict | None:
+    if not values:
+        return None
+    return {"avg": sum(values) / len(values), "peak": max(values), "min": min(values)}
+
+
+def aggregate_samples(samples: list[dict]) -> dict:
+    """聚合采样:内存/swap 统计 + 按 GPU 的利用率/显存/温度统计。"""
+    agg: dict = {"count": len(samples)}
+    if not samples:
+        return agg
+    agg["firstTs"] = samples[0]["ts"]
+    mem = [s["memPct"] for s in samples if isinstance(s.get("memPct"), (int, float))]
+    swap = [s["swapPct"] for s in samples if isinstance(s.get("swapPct"), (int, float))]
+    agg["mem"], agg["swap"] = _stats(mem), _stats(swap)
+    agg["memTotal"] = next((s.get("memTotal") for s in reversed(samples) if s.get("memTotal")), None)
+    agg["swapTotal"] = next((s.get("swapTotal") for s in reversed(samples) if s.get("swapTotal")), None)
+
+    gpus: dict[int, dict] = {}
+    for s in samples:
+        for g in s.get("gpus") or []:
+            slot = gpus.setdefault(g.get("index", 0), {"name": g.get("name", "?"), "util": [], "memUsed": [], "memTotal": g.get("memTotal"), "temp": []})
+            for key in ("utilPct", "memUsed", "tempC"):
+                if isinstance(g.get(key), (int, float)):
+                    slot["util" if key == "utilPct" else "memUsed" if key == "memUsed" else "temp"].append(g[key])
+    agg["gpus"] = [
+        {"index": idx, "name": slot["name"], "util": _stats(slot["util"]),
+         "memPeakUsed": max(slot["memUsed"]) if slot["memUsed"] else None,
+         "memTotal": slot["memTotal"], "tempPeak": max(slot["temp"]) if slot["temp"] else None}
+        for idx, slot in sorted(gpus.items())
+    ]
+    return agg
+
+
+# ---------- 磁盘目录增长(du 快照对比) ----------
+
+def disk_growth(metrics_dir: str | None) -> dict:
+    """对比最近两份 disk-*.json 快照,返回各挂载点 {dir, growth, snapshotTs}。"""
+    out: dict = {}
+    if not metrics_dir:
+        return out
+    snaps = sorted(Path(metrics_dir).glob("disk-*.json"))
+    if len(snaps) < 2:
+        return out
+    try:
+        new = json.loads(snaps[-1].read_text(encoding="utf-8"))
+        old = json.loads(snaps[-2].read_text(encoding="utf-8"))
+    except (json.JSONDecodeError, OSError):
+        return out
+    for key in ("root", "home"):
+        nd = (new.get(key) or {}).get("dirs") or {}
+        od = (old.get(key) or {}).get("dirs") or {}
+        best: tuple[str, int] | None = None
+        for name, nb in nd.items():
+            ob = od.get(name)
+            if ob is None:
+                continue
+            if best is None or nb - ob > best[1]:
+                best = (name, nb - ob)
+        out[key] = {"dir": best[0] if best else None,
+                    "growth": best[1] if best else 0,
+                    "snapshotTs": new.get("ts", "")}
+    return out
+
+
+# ---------- 瞬时兜底检查(采样缺失时用) ----------
+
 def _check_memory() -> dict:
     vm = psutil.virtual_memory()
     sw = psutil.swap_memory()
@@ -131,6 +245,7 @@ def _exited_recently(finished_at: str, hours: int = 24) -> bool:
 
 
 def check_prefect_failures(hours: int = 24) -> list[dict]:
+    """近 N 小时 FAILED/CRASHED 的 flow run,附 workflow 名(供日报/设备报告)。"""
     api_url = os.environ.get("PREFECT_API_URL", "http://localhost:4200/api")
     since = (datetime.now().astimezone() - timedelta(hours=hours)).isoformat()
     resp = httpx.post(
@@ -143,10 +258,18 @@ def check_prefect_failures(hours: int = 24) -> list[dict]:
         timeout=15,
     )
     resp.raise_for_status()
+    runs = resp.json()
+    flow_names: dict[str, str] = {}
+    flow_ids = list({r.get("flow_id") for r in runs if r.get("flow_id")})
+    if flow_ids:
+        fresp = httpx.post(f"{api_url}/flows/filter", json={"flows": {"id": {"any_": flow_ids}}}, timeout=15)
+        fresp.raise_for_status()
+        flow_names = {f["id"]: f["name"] for f in fresp.json()}
     out = []
-    for r in resp.json():
+    for r in runs:
         start = (r.get("start_time") or "")[:16].replace("T", " ")
-        out.append({"name": r.get("name", "?"), "state": (r.get("state") or {}).get("type", "?"),
+        out.append({"flow": flow_names.get(r.get("flow_id"), "?"), "name": r.get("name", "?"),
+                    "state": (r.get("state") or {}).get("type", "?"),
                     "message": ((r.get("state") or {}).get("message") or "").strip()[:100], "start": start})
     return out
 
@@ -176,48 +299,124 @@ def _mark(pct: float | None, threshold: int) -> str:
     return f"{pct:.0f}%"
 
 
+def _live_disks() -> list[dict]:
+    """实时磁盘(含设备去重汇总),供 总体/根目录/home 行。"""
+    paths = DISK_PATHS if os.path.isdir(DISK_PATHS[0][0]) else [("/", "/")]
+    mounts = []
+    seen: set[tuple[int, int]] = set()
+    total_used = total_avail = 0
+    for raw, label in paths:
+        try:
+            du = psutil.disk_usage(raw)
+        except Exception:  # noqa: BLE001
+            continue
+        key = (du.total, du.free)
+        if key not in seen:
+            seen.add(key)
+            total_used += du.used
+            total_avail += du.free
+        mounts.append({"label": label, "pct": du.percent, "used": du.used,
+                       "free": du.free, "total": du.total})
+    overall = {"used": total_used, "free": total_avail,
+               "pct": total_used / (total_used + total_avail) * 100 if total_used + total_avail else 0.0}
+    return mounts, overall
+
+
 @flow(name="device-check", log_prints=True)
 def device_check(alarm_threshold: int = 80) -> dict:
-    """设备/系统健康日报:检查并直接发企微群。单项检查失败不影响其余项。"""
+    """设备/系统健康日报:按 24h 采样报 平均/峰值/最低,直接发企微群。单项失败不影响其余。"""
+    mdir = _metrics_dir()
+    agg = aggregate_samples(read_samples(mdir))
+    growth = disk_growth(mdir)
     checks: dict[str, object] = {}
-    for name, fn in [
+    for name, fn in [("docker", _check_docker), ("prefectFailures", check_prefect_failures),
-        ("memory", _check_memory), ("disks", _check_disks), ("gpus", _check_gpus),
+                     ("services", _check_services)]:
-        ("docker", _check_docker), ("prefectFailures", check_prefect_failures), ("services", _check_services),
-    ]:
         try:
             checks[name] = fn()
         except Exception as err:  # noqa: BLE001
             checks[name] = None
             print(f"[device-check] {name} 检查失败: {err}")
+    try:
+        mounts, overall = _live_disks()
+    except Exception as err:  # noqa: BLE001
+        mounts, overall = [], None
+        print(f"[device-check] 磁盘检查失败: {err}")
 
     threshold = alarm_threshold
-    lines = [f"🖥 **设备健康日报** {datetime.now().strftime('%m-%d %H:%M')}", ""]
+    header = f"🖥 **设备日报** {datetime.now():%m-%d %H:%M}（过去 24h · 采样 {agg['count']} 次"
+    if agg.get("firstTs"):
+        first = datetime.fromisoformat(agg["firstTs"])
+        if first > datetime.now() - timedelta(hours=23):
+            header += f" · 自 {first:%m-%d %H:%M} 起"
+    lines = [header + "）", ""]
 
-    mem = checks.get("memory") or {}
+    # ---- 内存 / swap ----
+    mem, swap = agg.get("mem"), agg.get("swap")
     if mem:
-        lines.append(f"**内存** {_mark(mem['pct'], threshold)}（{mem['used']}/{mem['total']}）"
+        lines.append(f"**内存** 平均 {mem['avg']:.0f}% · 峰值 {_mark(mem['peak'], threshold)} · 最低 {mem['min']:.0f}%"
-                     f"｜swap {_mark(mem['swapPct'], threshold)}（{mem['swapUsed']}/{mem['swapTotal']}）")
+                     + (f"（共 {_gb((agg.get('memTotal') or 0) * (1 << 30))}）" if agg.get("memTotal") else ""))
     else:
+        try:
+            now_mem = _check_memory()
+            lines.append(f"**内存** 当前 {_mark(now_mem['pct'], threshold)}（{now_mem['used']}/{now_mem['total']}，无采样数据）")
+            mem = {"peak": now_mem["pct"]}
+        except Exception:  # noqa: BLE001
             lines.append("**内存** 检查失败")
+            mem = None
+    if swap:
+        lines.append(f"**交换内存** 平均 {swap['avg']:.0f}% · 峰值 {_mark(swap['peak'], threshold)} · 最低 {swap['min']:.0f}%"
+                     + (f"（共 {_gb((agg.get('swapTotal') or 0) * (1 << 30))}）" if agg.get("swapTotal") else ""))
+    else:
+        lines.append("**交换内存** 无采样数据")
+        swap = None
 
-    disks = checks.get("disks") or []
+    # ---- 磁盘:总体 / 根目录 / home ----
-    if disks:
     lines.append("**磁盘**")
-    for d in disks:
+    if overall:
-        if "error" in d:
+        lines.append(f"> 总体 已用 {_gb(overall['used'])} · 剩余 {_gb(overall['free'])}（{_mark(overall['pct'], threshold)}）")
-            lines.append(f"> {d['label']} 检查失败：{d['error']}")
+    label_map = {"/": "root", "/home": "home"}
-        else:
+    for m in mounts:
-            lines.append(f"> {d['label']} {_mark(d['pct'], threshold)}（{d['used']}/{d['total']}）")
+        g = growth.get(label_map.get(m["label"], ""))
+        growth_txt = ""
+        if g and g.get("dir"):
+            growth_txt = f"｜增长最快 {g['dir']} {_fmt_growth(g['growth'])}"
+        elif m["label"] in ("/", "/home"):
+            growth_txt = "｜增长数据不足" if not g else ""
+        lines.append(f"> {m['label']} 已用 {_gb(m['used'])} · 剩余 {_gb(m['free'])}"
+                     f"（{_mark(m['pct'], threshold)}）{growth_txt}")
+    if not mounts:
+        lines.append("> 磁盘检查不可用")
 
-    gpus = checks.get("gpus")
+    # ---- GPU ----
-    if isinstance(gpus, list) and gpus:
+    gpu_stats = agg.get("gpus") or []
-        lines.append("**GPU**")
+    if gpu_stats:
-        for g in gpus:
+        for g in gpu_stats:
-            lines.append(f"> {g['name']}｜利用率 {_mark(g['utilPct'], threshold)}"
+            name = f"{g['name']}（GPU{g['index']}）" if g.get("name") else f"GPU{g['index']}"
-                         f"｜显存 {_mark(g['memPct'], threshold)}（{g['memUsed']}/{g['memTotal']}）")
+            util = g.get("util")
+            util_txt = f"利用率 平均 {util['avg']:.0f}% · 峰值 {util['peak']:.0f}% · 最低 {util['min']:.0f}%" if util else "利用率 无数据"
+            mem_peak = g.get("memPeakUsed")
+            mem_total = g.get("memTotal")
+            if mem_peak is not None and mem_total:
+                mem_pct = mem_peak / mem_total * 100
+                mem_txt = f"显存峰值 {_gb(mem_peak * (1 << 30))}/{_gb(mem_total * (1 << 30))}（{_mark(mem_pct, threshold)}）"
             else:
-        lines.append(f"**GPU** {gpus if isinstance(gpus, str) else '未检测到'}")
+                mem_txt = "显存 无数据"
+            temp_peak = g.get("tempPeak")
+            if temp_peak is not None:
+                temp_txt = f"温度峰值 {temp_peak:.0f}°C" + ("🔴 过热" if temp_peak >= GPU_TEMP_WARN else " 正常 ✅")
+            else:
+                temp_txt = "温度 无数据"
+            lines.append(f"**GPU** {name}｜{util_txt}｜{mem_txt}｜{temp_txt}")
+    else:
+        gpus_now = _check_gpus()
+        if isinstance(gpus_now, list) and gpus_now:
+            for g in gpus_now:
+                lines.append(f"**GPU** {g['name']}｜利用率 {g['utilPct']:.0f}%｜显存 {g['memUsed']}/{g['memTotal']}"
+                             f"（无采样数据）")
+        else:
+            lines.append(f"**GPU** {gpus_now if isinstance(gpus_now, str) else '未检测到'}")
 
+    # ---- Docker / 失败任务 / 探活 ----
     docker = checks.get("docker") or {}
     issues = docker.get("issues") or []
     if docker.get("running") is None:
@@ -226,27 +425,31 @@ def device_check(alarm_threshold: int = 80) -> dict:
         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)}**🔴" if issues else " · 异常 0"
-            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)} 个" + (" ✅" if not failures else ""))
+    lines.append(f"**失败任务(24h)** {len(failures)} 次" + (" ✅" if not failures else "🔴"))
     for f in failures[:5]:
-        lines.append(f"> 🔴 {f['name']}（{f['state']}）{f['start']}")
+        lines.append(f"> 🔴 {f['flow']} ·「{f['name']}」{f['state']} · {f['start']}")
+    if len(failures) > 5:
+        lines.append(f"> …共 {len(failures)} 次")
 
     svc = checks.get("services") or {}
+    if svc:
         lines.append(f"**探活** OCR {'✅' if svc.get('ocr') else '🔴'}｜PG {'✅' if svc.get('pg') else '🔴'}"
                      f"｜负载 {os.getloadavg()[0]:.1f}")
 
+    mem_peak_for_alarm = mem.get("peak") if mem else None
+    swap_peak_for_alarm = swap.get("peak") if swap else None
     alarms = (
-        (mem and (mem["pct"] >= threshold or mem["swapPct"] >= threshold))
+        (mem_peak_for_alarm is not None and mem_peak_for_alarm >= threshold)
-        or any(d.get("pct", 0) >= threshold for d in (checks.get("disks") or []))
+        or (swap_peak_for_alarm is not None and swap_peak_for_alarm >= threshold)
-        or (isinstance(gpus, list) and any(g["utilPct"] >= threshold or g["memPct"] >= threshold for g in gpus))
+        or any(m["pct"] >= threshold for m in mounts)
+        or any((g.get("memPeakUsed") or 0) / (g.get("memTotal") or 1) * 100 >= threshold for g in gpu_stats)
+        or any((g.get("tempPeak") or 0) >= GPU_TEMP_WARN for g in gpu_stats)
         or bool(issues) or bool(failures) or not svc.get("ocr") or not svc.get("pg")
     )
     lines += ["", ("🔴 **有项目需要处理，见上**" if alarms else "✅ **一切正常**")]
@@ -254,13 +457,15 @@ def device_check(alarm_threshold: int = 80) -> dict:
     content = "\n".join(lines)
     sent = send_markdown(content)
     emit_sync_summary("设备健康日报", [
-        ("内存", f"{_mark(mem['pct'], threshold)}" if mem else "—"),
+        ("采样次数", agg["count"]),
+        ("内存峰值", f"{mem_peak_for_alarm:.0f}%" if mem_peak_for_alarm is not None else "—"),
         ("Docker 异常", len(issues)),
-        ("Prefect 失败(24h)", len(failures)),
+        ("失败任务(24h)", len(failures)),
         ("OCR", "✅" if svc.get("ocr") else "🔴"), ("PG", "✅" if svc.get("pg") else "🔴"),
         ("已发企微", "是" if sent else "否"),
     ])
-    return {"alarms": bool(alarms), "sent": sent, "dockerIssues": len(issues), "prefectFailures": len(failures)}
+    return {"alarms": bool(alarms), "sent": sent, "samples": agg["count"],
+            "dockerIssues": len(issues), "prefectFailures": len(failures)}
 
 
 if __name__ == "__main__":

flows/riin/verify.py

@@ -105,21 +105,39 @@ def riin_verify(mode: str = "recent", recent_days: int = 1, date_range: str | No
                 {"time_type": 3, "start_time": format_date(day_start), "end_time": format_date(day_end), "page_size": 200},
             )
             stats = sync_riin_records(raw_orders, client, db)
-            for k in ("ordersCreated", "ordersUpdated", "goodsCreated", "goodsUpdated"):
-                pass
             totals["ordersAdded"] += stats["ordersCreated"]
             totals["ordersUpdated"] += stats["ordersUpdated"]
             totals["goodsAdded"] += stats["goodsCreated"]
             totals["goodsUpdated"] += stats["goodsUpdated"]
             day_reports.append({"date": day_str, "remote": remote_total, "local": sum(local_counts.values()), "diff": diff, "ok": False,
                                 "orders": stats["orderCount"], "goods": stats["goodsCount"]})
+
+        # 修复后重数:数量 API + 本地 SQL 很便宜,给出修复后的数量一致性结论(口径与 _diff 相同)
+        post = None
+        if days_with_diff:
+            post_days, post_diff_days = [], 0
+            for day in days:
+                day_start, day_end = start_of_day(day), start_of_day(day) + timedelta(days=1)
+                count_resp = client.get_order_count(
+                    OrderPageRequest(time_type=3, start_time=format_date(day_start), end_time=format_date(day_end))
+                )
+                remote_map = _count_to_status_map(count_resp.data.model_dump(by_alias=False))
+                local_counts = db.get_order_count_by_status("RIIN", day_start, day_end)
+                diff, has_diff = _diff(remote_map, local_counts)
+                post_diff_days += int(has_diff)
+                post_days.append({"date": format_date(day).split(" ")[0], "remote": sum(remote_map.values()),
+                                  "local": sum(local_counts.values()), "ok": not has_diff})
+            post = {"days": post_days, "daysWithDiff": post_diff_days,
+                    "remote": sum(d["remote"] for d in post_days), "local": sum(d["local"] for d in post_days)}
+            print(f"[核验重数] 修复后 远程 {post['remote']} 本地 {post['local']} / 有差异 {post_diff_days} 天")
     finally:
         client.close()
 
     print(f"[核验汇总] 共 {len(days)} 天 / 有差异 {days_with_diff} 天 / {totals}")
     emit_verify_report("RIIN 数据核验", day_reports, totals, len(days), days_with_diff)
     return {
-        "verifyReport": {"days": day_reports, "summary": {"totalDays": len(days), "daysWithDiff": days_with_diff, **totals}},
+        "verifyReport": {"days": day_reports, "post": post,
+                         "summary": {"totalDays": len(days), "daysWithDiff": days_with_diff, **totals}},
         "syncRange": {"from": format_date(start_of_day(days[0])) if days else "", "to": format_date(start_of_day(days[-1])) if days else ""},
     }
 

flows/sds/verify.py

@@ -48,6 +48,14 @@ def _diff(remote_stats: list, local: dict[str, int]) -> tuple[dict[str, int], bo
     return diff, has_diff
 
 
+def _remote_map(remote_stats: list) -> dict[str, int]:
+    """远程状态→数量映射(排除"售后"和无名状态,与本地口径对称)。"""
+    out = {getattr(s, "name", None): (getattr(s, "orderNum", 0) or 0) for s in remote_stats}
+    out.pop("售后", None)
+    out.pop(None, None)
+    return out
+
+
 @flow(name="sds-verify", log_prints=True)
 def sds_verify(mode: str = "recent", recent_days: int = 1, date_range: str | None = None) -> dict:
     days = _parse_days(mode, recent_days, date_range)
@@ -68,7 +76,8 @@ def sds_verify(mode: str = "recent", recent_days: int = 1, date_range: str | Non
             remote_stats = client.get_status(
                 OrderQueryParams(pay_date_begin=format_date(day_start), pay_date_end=format_date(day_end))
             )
-            remote_total = sum(getattr(s, "orderNum", 0) or 0 for s in remote_stats)
+            remote_map = _remote_map(remote_stats)
+            remote_total = sum(remote_map.values())
             local_counts = db.get_order_count_by_status("SDS", day_start, day_end)
             diff, has_diff = _diff(remote_stats, local_counts)
 
@@ -91,13 +100,37 @@ def sds_verify(mode: str = "recent", recent_days: int = 1, date_range: str | Non
             totals["goodsUpdated"] += stats["goodsUpdated"]
             day_reports.append({"date": day_str, "remote": remote_total, "local": sum(local_counts.values()), "diff": diff, "ok": False,
                                 "orders": stats["orderCount"], "goods": stats["goodsCount"]})
+
+        # 修复后重数:数量接口 + 本地 SQL 很便宜,给出修复后的数量一致性结论(口径与 _diff 相同,排除"售后")
+        post = None
+        if days_with_diff:
+            post_days, post_diff_days = [], 0
+            for day in days:
+                day_start, day_end = start_of_day(day), start_of_day(day) + timedelta(days=1)
+                remote_stats = client.get_status(
+                    OrderQueryParams(pay_date_begin=format_date(day_start), pay_date_end=format_date(day_end))
+                )
+                remote_map = _remote_map(remote_stats)
+                local_counts = db.get_order_count_by_status("SDS", day_start, day_end)
+                statuses = set(remote_map) | set(local_counts)
+                statuses.discard("售后")
+                has_diff = any(remote_map.get(st, 0) != local_counts.get(st, 0) for st in statuses)
+                post_diff_days += int(has_diff)
+                post_days.append({"date": format_date(day).split(" ")[0],
+                                  "remote": sum(remote_map.values()),
+                                  "local": sum(v for k, v in local_counts.items() if k != "售后"),
+                                  "ok": not has_diff})
+            post = {"days": post_days, "daysWithDiff": post_diff_days,
+                    "remote": sum(d["remote"] for d in post_days), "local": sum(d["local"] for d in post_days)}
+            print(f"[核验重数] 修复后 远程 {post['remote']} 本地 {post['local']} / 有差异 {post_diff_days} 天")
     finally:
         client.close()
 
     print(f"[核验汇总] 共 {len(days)} 天 / 有差异 {days_with_diff} 天 / {totals}")
     emit_verify_report("SDS 数据核验", day_reports, totals, len(days), days_with_diff)
     return {
-        "verifyReport": {"days": day_reports, "summary": {"totalDays": len(days), "daysWithDiff": days_with_diff, **totals}},
+        "verifyReport": {"days": day_reports, "post": post,
+                         "summary": {"totalDays": len(days), "daysWithDiff": days_with_diff, **totals}},
         "syncRange": {"from": format_date(start_of_day(days[0])) if days else "", "to": format_date(start_of_day(days[-1])) if days else ""},
     }
 

flows/shared/db.py

@@ -200,9 +200,10 @@ class SyncDB:
     def update_ocr_fields(
         self, order_no: str, created_at: datetime, fields: dict[str, str | None]
     ) -> None:
-        """更新单个订单的 ocr_* 字段。fields 为 {ocr_xxx: value} 字典。"""
+        """更新单个订单的 ocr_* 字段。fields 为 {ocr_xxx: value} 字典。自动盖处理时间戳。"""
         if not fields:
             return
+        fields = {"ocrProcessedAt": datetime.now(), **fields}
         assignments = ", ".join(f'"{k}"=%s' for k in fields)
         params = list(fields.values()) + [order_no, created_at]
         with self._conn() as conn:
@@ -228,6 +229,7 @@ class SyncDB:
                     for order_no, created_at, fields in updates:
                         if not fields:
                             continue
+                        fields = {"ocrProcessedAt": datetime.now(), **fields}
                         assignments = ", ".join(f'"{k}"=%s' for k in fields)
                         params = list(fields.values()) + [order_no, created_at]
                         cur.execute(
@@ -258,6 +260,21 @@ class SyncDB:
                 row = cur.fetchone()
         return {"done": row[0], "error": row[1], "unsupported": row[2], "pending": row[3]} if row else {}
 
+    def count_ocr_today(self, day_start: datetime) -> dict[str, int]:
+        """今日 OCR 处理量(按 ocrProcessedAt 盖章时间):processed=成功识别, failed=识别失败/不支持。"""
+        with self._conn() as conn:
+            with conn.cursor() as cur:
+                cur.execute(
+                    '''SELECT
+                        COUNT(*) FILTER (WHERE "ocrLayout" IS NOT NULL
+                                          AND "ocrLayout" NOT IN ('ERROR','UNSUPPORTED')) AS processed,
+                        COUNT(*) FILTER (WHERE "ocrLayout" IN ('ERROR','UNSUPPORTED')) AS failed
+                       FROM "Orders" WHERE "ocrProcessedAt">=%s''',
+                    (day_start,),
+                )
+                row = cur.fetchone()
+        return {"processed": row[0], "failed": row[1]} if row else {"processed": 0, "failed": 0}
+
     # ---------- 主 upsert ----------
 
     def sync_orders_and_goods(

scripts/metrics_agent.py

@@ -0,0 +1,175 @@
+"""metrics-agent:宿主机指标采样侧车(独立于 Prefect,常驻循环,compose 服务运行)。
+
+每 5 分钟追加一行 JSONL:data/metrics/metrics-YYYY-MM-DD.jsonl
+  {"ts","memPct","memUsed","memTotal","swapPct","swapUsed","swapTotal",
+   "gpus":[{"index","name","utilPct","memPct","memUsed","memTotal","tempC"}]}
+每日 7 点后做一次 du 快照:data/metrics/disk-YYYYMMDD-HHMM.json
+  (根目录 / home 各自的总量 + 顶层目录字节数,device-check 对比最近两份算"增长最快目录")。
+保留:采样 14 天、快照 7 份。
+
+compose 中挂载 /:/host:ro(du 扫描)、GPU(nvidia-smi)、./data/metrics(writable)。
+调试单次采样:uv run python scripts/metrics_agent.py --once
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+import os
+import subprocess
+import threading
+import time
+from datetime import datetime
+from pathlib import Path
+
+INTERVAL = int(os.environ.get("METRICS_INTERVAL", "300"))
+DATA_DIR = Path(os.environ.get("METRICS_DIR", "data/metrics"))
+HOST_ROOT = os.environ.get("METRICS_HOST_ROOT", "/host")
+SNAPSHOT_AFTER_HOUR = 7
+KEEP_SAMPLE_DAYS = 14
+KEEP_SNAPSHOTS = 7
+# 虚拟文件系统与不关心的挂载点,du 快照跳过
+DU_EXCLUDE = {"proc", "sys", "dev", "run", "lost+found"}
+
+
+def _gb(nbytes: float) -> float:
+    return round(nbytes / (1 << 30), 2)
+
+
+def sample() -> dict:
+    import psutil
+
+    vm, sw = psutil.virtual_memory(), psutil.swap_memory()
+    return {
+        "ts": datetime.now().isoformat(timespec="seconds"),
+        "memPct": vm.percent, "memUsed": _gb(vm.used), "memTotal": _gb(vm.total),
+        "swapPct": sw.percent, "swapUsed": _gb(sw.used), "swapTotal": _gb(sw.total),
+        "gpus": _gpus(),
+    }
+
+
+def _gpus() -> list[dict]:
+    try:
+        proc = subprocess.run(
+            ["nvidia-smi", "--query-gpu=index,name,utilization.gpu,memory.used,memory.total,temperature.gpu",
+             "--format=csv,noheader,nounits"],
+            capture_output=True, text=True, timeout=20,
+        )
+        if proc.returncode != 0:
+            return []
+        out = []
+        for line in proc.stdout.strip().splitlines():
+            p = [x.strip() for x in line.split(",")]
+            if len(p) != 6:
+                continue
+            idx, name, util, mem_used, mem_total, temp = p
+            mt = float(mem_total)
+            out.append({
+                "index": int(idx), "name": name[:24], "utilPct": float(util),
+                "memPct": round(float(mem_used) / mt * 100, 1) if mt > 0 else 0.0,
+                "memUsed": _gb(float(mem_used) * (1 << 20)), "memTotal": _gb(mt * (1 << 20)),
+                "tempC": float(temp),
+            })
+        return out
+    except Exception:  # noqa: BLE001
+        return []
+
+
+def write_sample(rec: dict) -> None:
+    DATA_DIR.mkdir(parents=True, exist_ok=True)
+    path = DATA_DIR / f"metrics-{rec['ts'][:10]}.jsonl"
+    with path.open("a", encoding="utf-8") as f:
+        f.write(json.dumps(rec, ensure_ascii=False) + "\n")
+
+
+def _du_bytes(path: str) -> int:
+    # 实际占块(非表面大小,避免 docker 稀疏文件虚增);rc=1 时是个别文件消失,stdout 仍有效
+    proc = subprocess.run(["du", "-sB1", path], capture_output=True, text=True, timeout=7200)
+    if not proc.stdout:
+        return -1
+    try:
+        return int(proc.stdout.split()[0])
+    except (IndexError, ValueError):
+        return -1
+
+
+def _usage_entry(base: str) -> dict:
+    import shutil
+
+    st = shutil.disk_usage(base)
+    return {"total": st.total, "used": st.used, "free": st.free}
+
+
+def disk_snapshot() -> Path:
+    """du 扫描根目录/home 顶层目录,写 disk-YYYYMMDD-HHMM.json(原子写)。"""
+    now = datetime.now()
+    out: dict = {"ts": now.isoformat(timespec="seconds")}
+    for key, base in (("root", HOST_ROOT), ("home", f"{HOST_ROOT}/home")):
+        if not os.path.isdir(base):
+            continue
+        entry = _usage_entry(base)
+        dirs: dict[str, int] = {}
+        for name in sorted(os.listdir(base)):
+            if name in DU_EXCLUDE:
+                continue
+            p = os.path.join(base, name)
+            if os.path.isdir(p) and not os.path.islink(p) and not os.path.ismount(p):
+                size = _du_bytes(p)
+                if size >= 0:
+                    dirs[name] = size
+        entry["dirs"] = dirs
+        out[key] = entry
+    DATA_DIR.mkdir(parents=True, exist_ok=True)
+    path = DATA_DIR / f"disk-{now:%Y%m%d-%H%M}.json"
+    tmp = path.with_suffix(".tmp")
+    tmp.write_text(json.dumps(out, ensure_ascii=False), encoding="utf-8")
+    tmp.rename(path)
+    print(f"[metrics-agent] 磁盘快照完成: {path.name}", flush=True)
+    return path
+
+
+def snapshot_today_exists() -> bool:
+    today = datetime.now().strftime("%Y%m%d")
+    return any(DATA_DIR.glob(f"disk-{today}-*.json"))
+
+
+def prune() -> None:
+    cutoff = time.time() - KEEP_SAMPLE_DAYS * 86400
+    for p in DATA_DIR.glob("metrics-*.jsonl"):
+        if p.stat().st_mtime < cutoff:
+            p.unlink(missing_ok=True)
+    snaps = sorted(DATA_DIR.glob("disk-*.json"))
+    for p in snaps[:-KEEP_SNAPSHOTS] if len(snaps) > KEEP_SNAPSHOTS else []:
+        p.unlink(missing_ok=True)
+
+
+def _maybe_snapshot_async() -> None:
+    """7 点后补当日快照;du 可能耗时数分钟,放后台线程不阻塞采样。"""
+    if datetime.now().hour < SNAPSHOT_AFTER_HOUR or snapshot_today_exists():
+        return
+    threading.Thread(target=lambda: _safe(disk_snapshot), daemon=True).start()
+
+
+def _safe(fn, *args):
+    try:
+        fn(*args)
+    except Exception as err:  # noqa: BLE001
+        print(f"[metrics-agent] {type(err).__name__}: {err}", flush=True)
+
+
+def main() -> None:
+    ap = argparse.ArgumentParser(description="宿主机指标采样侧车")
+    ap.add_argument("--once", action="store_true", help="只采样一次即退出(调试)")
+    args = ap.parse_args()
+    print(f"[metrics-agent] 启动: interval={INTERVAL}s dir={DATA_DIR} host_root={HOST_ROOT}", flush=True)
+    while True:
+        _safe(lambda: write_sample(sample()))
+        _maybe_snapshot_async()
+        _safe(prune)
+        if args.once:
+            break
+        time.sleep(INTERVAL)
+
+
+if __name__ == "__main__":
+    main()