自动补测、推送审计/退避重试、监控指标与监控页面

README.md

@@ -327,6 +327,12 @@ cd ../.. && python3 -m graylog_ops.geo \
 [查看详情](http://179.253.234.124:12139/s/301b.../#/ip/112.32.1.211)
 ```
 
+### 自动补测 / 推送审计 / 监控
+
+- **启动自动补测**：启动时读取上次成功检测时间，缺口超过 30 分钟自动后台回扫（可用 `--backfill-threshold`、`--backfill-max-gap`、`--no-auto-backfill` 控制）
+- **推送审计**：`notifications` 表记录每个 IP 的推送成功/失败、错误信息；失败按 60s→120s→…→1h 指数退避重试；连续失败 3 次会记录告警（可选 `NOTIFY_ALERT_URL` 发到独立 webhook）
+- **监控**：页面右上角「监控」入口，展示 Graylog 状态、检测轮次/耗时、行为分析、推送成功率、名单和补测记录；`/metrics` 输出 Prometheus 格式，可直接被任意监控系统抓取
+
 ### 状态指示灯
 
 页面右上角有一个圆点，表示后端与 Graylog 的连接状态：

graylog_ops/backfill.py

@@ -0,0 +1,224 @@
+"""Backfill detection for a historical time window.
+
+Used by the startup auto-backfill thread and by tools/backfill_*.py.
+"""
+import datetime
+import ipaddress
+import time
+
+from . import behavior, watch
+from .asn import AsnDB
+from .geo import GeoIP
+
+
+def parse_iso(value):
+    try:
+        return datetime.datetime.fromisoformat(
+            str(value).replace("Z", "+00:00")).astimezone(datetime.timezone.utc)
+    except (TypeError, ValueError):
+        return None
+
+
+def iso(stamp):
+    return stamp.strftime("%Y-%m-%dT%H:%M:%S.000Z")
+
+
+def _suppress(store, ips):
+    if not ips:
+        return
+    stamp = watch.now_iso()
+    with store.lock, store.conn:
+        store.conn.executemany(
+            "UPDATE ip_fused SET last_pushed_at=?, push_fail_count=0, "
+            "push_next_at=NULL WHERE ip=?", [(stamp, ip) for ip in ips])
+
+
+def feature_backfill(client, store, fields, filter_str, from_dt, to_dt,
+                     geo=None, asn=None, size=10000):
+    """Signature-rule backfill with recursive time slicing."""
+    ip_field = fields["client_ip"]
+    span = int((to_dt - from_dt).total_seconds())
+    counts, rule_totals = {}, {}
+
+    def collect(query, start, end):
+        result = client.terms(ip_field, query=query, size=size,
+                              from_ts=iso(start), to_ts=iso(end),
+                              filter_str=filter_str)
+        terms = result.get("terms") or []
+        if len(terms) >= size and (end - start).total_seconds() > 300:
+            middle = start + (end - start) / 2
+            return collect(query, start, middle) + collect(query, middle, end)
+        return [(str(item.get("term") or ""), int(item.get("count") or 0))
+                for item in terms if item.get("term")]
+
+    for rule in watch.RULE_DEFS:
+        if rule.get("weight", 0) <= 0 and rule["name"] != "not_found":
+            continue
+        try:
+            pairs = collect(rule["query"](fields), from_dt, to_dt)
+        except Exception:
+            continue
+        merged = {}
+        for ip, count in pairs:
+            merged[ip] = merged.get(ip, 0) + count
+        rule_totals[rule["name"]] = sum(merged.values())
+        for ip, count in merged.items():
+            counts.setdefault(ip, {})[rule["name"]] = count
+
+    now_iso = watch.now_iso()
+    candidates = []
+    for ip, hits in counts.items():
+        try:
+            if not ipaddress.ip_address(ip).is_global:
+                continue
+        except ValueError:
+            continue
+        total = sum(hits.values())
+        tier, score, reasons = watch.classify(total, hits, window_scale=span / 900.0)
+        if tier != watch.TIER_MALICIOUS:
+            continue
+        candidates.append({"ip": ip, "hits": hits, "total": total,
+                           "score": score, "reasons": reasons})
+
+    candidates.sort(key=lambda item: (-item["score"], -item["total"], item["ip"]))
+    touched = []
+    created = upgraded = extended = 0
+    for item in candidates:
+        ip = item["ip"]
+        first = last = None
+        try:
+            asc = client.search_absolute(query='%s:"%s"' % (ip_field, ip),
+                                         from_ts=iso(from_dt), to_ts=iso(to_dt),
+                                         limit=1, sort="timestamp:asc",
+                                         fields=["timestamp"], filter_str=filter_str)
+            desc = client.search_absolute(query='%s:"%s"' % (ip_field, ip),
+                                          from_ts=iso(from_dt), to_ts=iso(to_dt),
+                                          limit=1, sort="timestamp:desc",
+                                          fields=["timestamp"], filter_str=filter_str)
+            first = (asc or [{}])[0].get("timestamp")
+            last = (desc or [{}])[0].get("timestamp")
+            first = iso(parse_iso(first)) if parse_iso(first) else None
+            last = iso(parse_iso(last)) if parse_iso(last) else None
+        except Exception:
+            pass
+        first = first or iso(from_dt)
+        last = last or iso(to_dt)
+        record = {
+            "ip": ip, "tier": watch.TIER_MALICIOUS, "score": item["score"],
+            "total": item["total"], "rules": item["hits"],
+            "reasons": item["reasons"], "first_seen": first, "last_seen": last,
+            "last_malicious_at": last,
+            "risk": "high" if item["score"] >= 85 else "medium",
+            "window": span,
+            "geo": geo.lookup(ip) if geo else {},
+            "asn": asn.lookup(ip) if asn else {},
+        }
+        existing = store.get(ip)
+        if existing is None:
+            store.upsert(record, now_iso)
+            created += 1
+            touched.append(ip)
+        elif existing.get("tier") == watch.TIER_MALICIOUS:
+            with store.lock, store.conn:
+                store.conn.execute(
+                    "UPDATE ip_stats SET first_seen=?, last_seen=?, "
+                    "last_malicious_at=?, ever_malicious=1 WHERE ip=?",
+                    (min(first, existing.get("first_seen") or first),
+                     max(last, existing.get("last_seen") or last),
+                     max(last, existing.get("last_malicious_at") or last), ip))
+            extended += 1
+            touched.append(ip)
+        elif float(existing.get("score") or 0) < item["score"]:
+            store.upsert(record, now_iso)
+            upgraded += 1
+            touched.append(ip)
+        else:
+            extended += 1
+            touched.append(ip)
+
+    if touched:
+        store.sync_fused()
+        _suppress(store, touched)
+        with store.lock, store.conn:
+            store.conn.executemany(
+                """UPDATE ip_fused SET first_banned_at = (
+                       SELECT s.first_seen FROM ip_stats s WHERE s.ip = ip_fused.ip)
+                   WHERE ip=? AND EXISTS (
+                       SELECT 1 FROM ip_stats s WHERE s.ip = ip_fused.ip
+                          AND s.first_seen < ip_fused.first_banned_at)""",
+                [(ip,) for ip in touched])
+
+    return {"candidates": len(candidates), "created": created,
+            "upgraded": upgraded, "extended": extended,
+            "rule_totals": rule_totals,
+            "ips": [item["ip"] for item in candidates]}
+
+
+def behavior_backfill(client, store, fields, filter_str, from_dt, to_dt,
+                      root, geo=None, asn=None):
+    """Run the independent behaviour analyzer over an absolute range."""
+    behavior.set_range(iso(from_dt), iso(to_dt))
+    try:
+        analyzer = behavior.BehaviorAnalyzer(client, store, fields, filter_str,
+                                             root=root, notifier=None)
+        items = analyzer.analyze()
+    finally:
+        behavior.clear_range()
+    for item in items:
+        try:
+            last = client.search_absolute(
+                query='%s:"%s"' % (fields["client_ip"], item["ip"]),
+                from_ts=iso(from_dt), to_ts=iso(to_dt), limit=1,
+                sort="timestamp:desc", fields=["timestamp"],
+                filter_str=filter_str)
+            stamp = (last or [{}])[0].get("timestamp")
+            parsed = parse_iso(stamp)
+            if parsed:
+                item["last_ts"] = parsed.strftime("%Y-%m-%dT%H:%M:%SZ")
+        except Exception:
+            pass
+    flagged = [item for item in items if item.get("verdict") == "建议封禁"]
+    store.upsert_historical_profiles(items)
+    before = {row["ip"] for row in store.conn.execute("SELECT ip FROM ip_fused")}
+    store.sync_fused()
+    added = [item["ip"] for item in flagged if item["ip"] not in before]
+    _suppress(store, added)
+    return {"violations": len(items), "flagged": len(flagged),
+            "newly_fused": len(added), "new_ips": sorted(added)}
+
+
+def auto_backfill_gap(store, client, fields, filter_str, from_dt, to_dt,
+                      root, behavior_enabled=True):
+    started = time.time()
+    geo = asn = None
+    try:
+        import os
+        geo_path = os.path.join(root, "data", "geo", "geo.db")
+        asn_path = os.path.join(root, "data", "geo", "asn.db")
+        if os.path.exists(geo_path):
+            geo = GeoIP(geo_path)
+        if os.path.exists(asn_path):
+            asn = AsnDB(asn_path)
+    except Exception:
+        pass
+    summary = {"from": iso(from_dt), "to": iso(to_dt)}
+    try:
+        summary["feature"] = feature_backfill(
+            client, store, fields, filter_str, from_dt, to_dt, geo=geo, asn=asn)
+        gap_hours = (to_dt - from_dt).total_seconds() / 3600.0
+        if behavior_enabled and 1 <= gap_hours <= 48:
+            summary["behavior"] = behavior_backfill(
+                client, store, fields, filter_str, from_dt, to_dt, root,
+                geo=geo, asn=asn)
+        store.incr_meta("backfill_runs_total")
+        store.set_meta("backfill_last_at", watch.now_iso())
+        store.set_meta("backfill_last_from", iso(from_dt))
+        store.set_meta("backfill_last_to", iso(to_dt))
+        store.set_meta("backfill_last_added",
+                       len(summary["feature"].get("ips") or []))
+        store.set_meta("backfill_last_error", "")
+    except Exception as exc:
+        summary["error"] = str(exc)
+        store.set_meta("backfill_last_error", str(exc))
+    summary["seconds"] = round(time.time() - started, 1)
+    return summary

graylog_ops/behavior.py

@@ -134,15 +134,28 @@ def asn_info(ip, root):
     return {"asn": "", "org": "未知", "shared": "unknown"}
 
 
-RANGE_OVERRIDE = None
+_RANGE_LOCAL = threading.local()
+
+
+def set_range(from_ts, to_ts):
+    """Set an absolute analysis range for the current thread only."""
+    _RANGE_LOCAL.value = {"from": from_ts, "to": to_ts}
+
+
+def clear_range():
+    _RANGE_LOCAL.value = None
+
+
+def current_range():
+    return getattr(_RANGE_LOCAL, "value", None)
 
 
 def pivot(client, series, row_groups, column_groups=None, query="*",
           limit=300, timeout=180000):
-    if RANGE_OVERRIDE:
+    override = current_range()
+    if override:
         timerange = {"type": "absolute",
-                     "from": RANGE_OVERRIDE["from"],
-                     "to": RANGE_OVERRIDE["to"]}
+                     "from": override["from"], "to": override["to"]}
     else:
         timerange = {"type": "relative", "range": WINDOW}
     body = {"queries": [{
@@ -367,6 +380,9 @@ class BehaviorAnalyzer:
         self.notifier = notifier
         self.last_run = None
         self.last_error = None
+        self.last_duration = None
+        self.last_count = 0
+        self.error_count = 0
         self.runs = 0
         self.asn = None
         try:
@@ -411,9 +427,10 @@ class BehaviorAnalyzer:
                 pass
             try:
                 extra = {}
-                if RANGE_OVERRIDE:
-                    extra = {"from_ts": RANGE_OVERRIDE["from"],
-                             "to_ts": RANGE_OVERRIDE["to"]}
+                override = current_range()
+                if override:
+                    extra = {"from_ts": override["from"],
+                             "to_ts": override["to"]}
                 hist = self.client.histogram(query=q, range_seconds=self.window,
                                              interval="5m",
                                              filter_str=self.filter_str, **extra)
@@ -623,7 +640,14 @@ class BehaviorAnalyzer:
             self.store.sync_fused()
             self.last_run = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())
             self.last_error = None
+            self.last_duration = round(time.time() - started, 2)
+            self.last_count = len(items)
             self.runs += 1
+            self.store.set_meta("behavior_last_success", self.last_run)
+            self.store.set_meta("behavior_last_duration", self.last_duration)
+            self.store.set_meta("behavior_last_count", self.last_count)
+            self.store.set_meta("behavior_last_error", "")
+            self.store.incr_meta("behavior_runs_total")
             print("[%s] 行为分析: %d 个违规 IP，%d 个建议封禁，耗时 %.1fs"
                   % (self.last_run, len(items),
                      sum(1 for item in items if item["verdict"] == "建议封禁"),
@@ -633,6 +657,9 @@ class BehaviorAnalyzer:
             return items
         except Exception as exc:  # keep the service alive
             self.last_error = str(exc)
+            self.error_count += 1
+            self.store.set_meta("behavior_last_error", str(exc))
+            self.store.incr_meta("behavior_errors_total")
             print("[行为分析失败] %s" % exc, flush=True)
             return []
 
@@ -694,15 +721,46 @@ def notify_fused(store, notifier, root=None):
         if link:
             lines.append("[查看详情](%s)" % link)
         lines.append("")
+    ips = [row["ip"] for row in due[:20]]
     try:
         notifier.post("\n".join(lines).strip())
     except Exception as exc:
-        # 失败时不刷屏：每 5 分钟最多提示一次；下一轮会自动重试
+        store.mark_push_failed(ips, str(exc), stamp=now)
+        fails = store.incr_meta("push_consecutive_failures")
+        store.set_meta("push_last_error", str(exc))
+        store.set_meta("push_last_error_at", now)
+        # 失败日志限流：每 5 分钟最多提示一次
         global _PUSH_ERROR_AT
-        now = time.time()
-        if now - _PUSH_ERROR_AT > 300:
-            print("推送失败（每 5 分钟最多提示一次）: %s" % exc, flush=True)
-            _PUSH_ERROR_AT = now
+        now_epoch = time.time()
+        if now_epoch - _PUSH_ERROR_AT > 300:
+            print("推送失败（重试 #%d，每 5 分钟最多提示一次）: %s"
+                  % (fails, exc), flush=True)
+            _PUSH_ERROR_AT = now_epoch
+        if fails == 3 or (fails >= 3 and fails % 10 == 0):
+            print("[推送告警] 连续 %d 次推送失败，请检查 webhook / 网络" % fails,
+                  flush=True)
+            _send_alert("钉钉推送连续 %d 次失败：%s" % (fails, exc))
         return 0
-    store.mark_pushed([row["ip"] for row in due[:20]], now)
-    return len(due[:20])
+    store.mark_pushed(ips, now, channel=getattr(notifier, "kind", "dingtalk"))
+    store.set_meta("push_consecutive_failures", 0)
+    store.set_meta("push_last_error", "")
+    return len(ips)
+
+
+def _send_alert(text):
+    """Optional independent alert webhook (NOTIFY_ALERT_URL)."""
+    import os
+    url = (os.environ.get("NOTIFY_ALERT_URL") or "").strip()
+    if not url:
+        return
+    try:
+        from .notify import Notifier
+        alert = Notifier({
+            "type": os.environ.get("NOTIFY_ALERT_TYPE", "dingtalk"),
+            "url": url,
+            "secret": os.environ.get("NOTIFY_ALERT_SECRET", ""),
+            "title": "推送告警",
+        })
+        alert.post(text, title="推送告警")
+    except Exception as exc:
+        print("告警发送失败: %s" % exc, flush=True)

graylog_ops/cli.py

@@ -148,6 +148,12 @@ def build_parser():
     watch.add_argument("--host", default="127.0.0.1", help="面板监听地址")
     watch.add_argument("--port", type=int, default=8088, help="面板端口，默认 8088")
     watch.add_argument("--db", default="data/ips.db", help="SQLite 路径")
+    watch.add_argument("--backfill-threshold", type=int, default=1800,
+                       help="启动时检测缺口超过多少秒自动补测（默认 1800）")
+    watch.add_argument("--backfill-max-gap", type=int, default=7 * 86400,
+                       help="自动补测允许的最大缺口秒数（默认 7 天）")
+    watch.add_argument("--no-auto-backfill", action="store_true",
+                       help="启动时不自动补测历史缺口")
     watch.add_argument("--behavior-interval", type=int, default=900,
                        help="行为检测刷新间隔（秒，默认 900）")
     watch.add_argument("--no-behavior", action="store_true",

graylog_ops/watch.py

@@ -21,6 +21,7 @@ Design
 """
 
 import csv
+import datetime
 import html
 import io
 import ipaddress
@@ -362,9 +363,31 @@ class Store:
                        fused_reason TEXT, updated_at TEXT
                    )"""
             )
+            for column in ("push_fail_count INTEGER DEFAULT 0",
+                           "push_next_at TEXT"):
+                try:
+                    self.conn.execute(
+                        "ALTER TABLE ip_fused ADD COLUMN %s" % column)
+                except sqlite3.OperationalError:
+                    pass
             self.conn.execute(
                 "CREATE INDEX IF NOT EXISTS idx_ip_fused_first "
                 "ON ip_fused(first_banned_at DESC, ip)")
+            self.conn.execute(
+                """CREATE TABLE IF NOT EXISTS meta (
+                       key TEXT PRIMARY KEY, value TEXT
+                   )"""
+            )
+            self.conn.execute(
+                """CREATE TABLE IF NOT EXISTS notifications (
+                       id INTEGER PRIMARY KEY AUTOINCREMENT,
+                       ip TEXT, channel TEXT, status TEXT, attempt INTEGER,
+                       error TEXT, created_at TEXT
+                   )"""
+            )
+            self.conn.execute(
+                "CREATE INDEX IF NOT EXISTS idx_notifications_created "
+                "ON notifications(created_at DESC)")
             self.conn.execute(
                 """CREATE TABLE IF NOT EXISTS events (
                        id INTEGER PRIMARY KEY AUTOINCREMENT,
@@ -746,6 +769,85 @@ class Store:
                 self.conn.execute("DELETE FROM ip_fused WHERE ip = ?", (ip,))
             return len(wanted)
 
+    # ------------------------------------------------------------- meta
+    def get_meta(self, key, default=None):
+        with self.lock:
+            row = self.conn.execute(
+                "SELECT value FROM meta WHERE key=?", (key,)).fetchone()
+        return row["value"] if row else default
+
+    def set_meta(self, key, value):
+        with self.lock, self.conn:
+            self.conn.execute(
+                "INSERT INTO meta(key, value) VALUES(?,?) "
+                "ON CONFLICT(key) DO UPDATE SET value=excluded.value",
+                (key, str(value)))
+
+    def incr_meta(self, key, amount=1):
+        with self.lock, self.conn:
+            self.conn.execute(
+                "INSERT INTO meta(key, value) VALUES(?,?) "
+                "ON CONFLICT(key) DO UPDATE SET value=CAST("
+                "CAST(value AS INTEGER)+? AS TEXT)",
+                (key, str(amount), int(amount)))
+            row = self.conn.execute(
+                "SELECT value FROM meta WHERE key=?", (key,)).fetchone()
+        return int(row["value"] or 0)
+
+    def meta_dict(self):
+        with self.lock:
+            rows = self.conn.execute("SELECT key, value FROM meta").fetchall()
+        return {row["key"]: row["value"] for row in rows}
+
+    # ------------------------------------------------------- notifications
+    def log_notification(self, ip, channel, status, attempt, error=None,
+                        stamp=None):
+        stamp = stamp or now_iso()
+        with self.lock, self.conn:
+            self.conn.execute(
+                "INSERT INTO notifications(ip, channel, status, attempt, error, "
+                "created_at) VALUES(?,?,?,?,?,?)",
+                (ip, channel, status, int(attempt or 0), error, stamp))
+
+    def mark_push_failed(self, ips, error, stamp=None, base=60, cap=3600):
+        stamp = stamp or now_iso()
+        base_time = time.time()
+        with self.lock, self.conn:
+            for ip in ips:
+                row = self.conn.execute(
+                    "SELECT COALESCE(push_fail_count,0) AS n FROM ip_fused "
+                    "WHERE ip=?", (ip,)).fetchone()
+                fails = int(row["n"] if row else 0) + 1
+                delay = min(cap, base * (2 ** (fails - 1)))
+                next_at = time.strftime(
+                    "%Y-%m-%dT%H:%M:%SZ",
+                    time.gmtime(base_time + delay))
+                self.conn.execute(
+                    "UPDATE ip_fused SET push_fail_count=?, push_next_at=? "
+                    "WHERE ip=?", (fails, next_at, ip))
+                self.conn.execute(
+                    "INSERT INTO notifications(ip, channel, status, attempt, "
+                    "error, created_at) VALUES(?,?,?,?,?,?)",
+                    (ip, "dingtalk", "failed", fails, error, stamp))
+        return len(ips)
+
+    def notification_stats(self, hours=24):
+        since = time.strftime("%Y-%m-%dT%H:%M:%SZ",
+                              time.gmtime(time.time() - hours * 3600))
+        with self.lock:
+            rows = self.conn.execute(
+                "SELECT status, COUNT(*) AS n FROM notifications "
+                "WHERE created_at >= ? GROUP BY status", (since,)).fetchall()
+            recent = self.conn.execute(
+                "SELECT * FROM notifications ORDER BY id DESC LIMIT 30").fetchall()
+        counts = {row["status"]: int(row["n"]) for row in rows}
+        sent = counts.get("sent", 0)
+        failed = counts.get("failed", 0)
+        total = sent + failed
+        return {"sent": sent, "failed": failed,
+                "success_ratio": round(sent / total, 4) if total else None,
+                "recent": [dict(row) for row in recent]}
+
     def due_pushes(self, active_hours=24, cooldown_hours=24):
         now = time.time()
         active_cut = time.strftime("%Y-%m-%dT%H:%M:%SZ",
@@ -767,8 +869,9 @@ class Store:
                      LEFT JOIN behavior_profiles b ON b.ip = f.ip
                     WHERE f.last_activity_at >= ?
                       AND (f.last_pushed_at IS NULL OR f.last_pushed_at <= ?)
+                      AND (f.push_next_at IS NULL OR f.push_next_at <= ?)
                     ORDER BY f.first_banned_at DESC LIMIT 50""",
-                (active_cut, push_cut)).fetchall()
+                (active_cut, push_cut, now_iso())).fetchall()
         items = []
         for row in rows:
             item = dict(row)
@@ -785,12 +888,18 @@ class Store:
             items.append(item)
         return items
 
-    def mark_pushed(self, ips, stamp):
+    def mark_pushed(self, ips, stamp, channel="dingtalk"):
         with self.lock, self.conn:
             for ip in ips:
                 self.conn.execute(
-                    "UPDATE ip_fused SET last_pushed_at=?, push_count=push_count+1 "
-                    "WHERE ip=?", (stamp, ip))
+                    "UPDATE ip_fused SET last_pushed_at=?, "
+                    "push_count=push_count+1, push_fail_count=0, "
+                    "push_next_at=NULL WHERE ip=?", (stamp, ip))
+                self.conn.execute(
+                    "INSERT INTO notifications(ip, channel, status, attempt, "
+                    "error, created_at) VALUES(?,?,?,?,?,?)",
+                    (ip, channel, "sent", 1, None, stamp))
+        self.incr_meta("push_sent_total", len(ips))
 
     def stats(self):
         with self.lock:
@@ -946,6 +1055,8 @@ class Detector:
         self.last_run = None
         self.last_success = None
         self.last_error = None
+        self.last_duration = None
+        self.error_count = 0
         self.runs = 0
 
     def _terms(self, query, window_cap=None):
@@ -1232,7 +1343,13 @@ class Detector:
 
         self.last_run = stamp
         self.last_success = time.time()
+        self.last_duration = round(time.time() - started, 2)
+        self.last_error = None
         self.runs += 1
+        self.store.set_meta("detect_last_success", stamp)
+        self.store.set_meta("detect_last_duration", self.last_duration)
+        self.store.set_meta("detect_last_error", "")
+        self.store.incr_meta("detect_rounds_total")
         return {
             "window": self.window,
             "request_seconds": round(time.time() - started, 2),
@@ -1779,6 +1896,32 @@ a:hover{text-decoration:underline}
 .notes div{font-size:12.5px;color:var(--ink-2);background:var(--surface-2);
   border:1px solid var(--line);border-radius:9px;padding:9px 12px;line-height:1.55}
 .hint{color:var(--muted);font-size:12px;margin-top:11px;line-height:1.7}
+.monitor-head{display:flex;align-items:center;gap:14px;margin:6px 0 18px}
+.monitor-title{font-size:21px;font-weight:600;color:var(--ink);margin-right:auto}
+.mgrid{display:grid;grid-template-columns:repeat(3,minmax(0,1fr));
+  gap:12px;margin-bottom:18px}
+@media (max-width:1100px){.mgrid{grid-template-columns:repeat(2,minmax(0,1fr))}}
+@media (max-width:700px){.mgrid{grid-template-columns:1fr}}
+.mcard{background:var(--surface);border:1px solid var(--line);
+  border-radius:14px;padding:15px 17px;box-shadow:0 1px 2px rgba(36,33,25,.03)}
+.mcard h3{font-size:11.5px;color:var(--muted);font-weight:600;
+  letter-spacing:.06em;margin:0 0 7px}
+.mbig{font-family:"Iowan Old Style",Palatino,Georgia,serif;font-size:27px;
+  color:var(--ink);line-height:1.15;margin-bottom:6px}
+.mbig.mok{color:#3E8E63}.mbig.mbad{color:#C0503A}.mbig.mwarn{color:var(--amber)}
+.mrow{display:flex;justify-content:space-between;gap:10px;font-size:12px;
+  color:var(--ink-2);margin-top:5px}
+.mrow span:last-child{color:var(--muted);text-align:right}
+.mtablewrap{overflow:auto;border:1px solid var(--line);border-radius:14px;
+  background:var(--surface);box-shadow:0 1px 2px rgba(36,33,25,.03)}
+.mtable{width:100%;border-collapse:collapse;font-size:12.5px}
+.mtable th{color:var(--muted);font-size:11.5px;font-weight:600;text-align:left;
+  padding:10px 14px;border-bottom:1px solid var(--line);background:var(--surface-2)}
+.mtable td{padding:9px 14px;border-bottom:1px solid #EFE9DD;color:var(--ink-2)}
+.mtable tr:last-child td{border-bottom:0}
+.nbadge{display:inline-block;padding:2px 9px;border-radius:99px;font-size:11px}
+.nbadge.ok{background:var(--green-soft,#E4EFE7);color:#2E6B4F}
+.nbadge.bad{background:var(--red-soft);color:#93402D}
 .pager{display:flex;align-items:center;justify-content:center;gap:18px;
   margin:18px 0 2px;color:var(--muted);font-size:12.5px;flex-wrap:wrap}
 .pg-total b{color:var(--ink);font-weight:600}
@@ -1895,6 +2038,7 @@ a:hover{text-decoration:underline}
   </section>
 
   <section id="detailView" hidden></section>
+  <section id="monitorView" hidden></section>
 </main>
 
 <div class="toast" id="toast"></div>
@@ -1956,7 +2100,7 @@ const RULE_TITLES = {
   bad_ua: '扫描器 UA', bot_ua: '爬虫 UA', not_found: '404 探测'
 };
 const state = { q: '', items: [], page: 1, pageSize: 20, total: 0, pages: 1,
-                 timer: null, inDetail: false, detailSamples: [] };
+                 timer: null, inDetail: false, inMonitor: false, detailSamples: [] };
 
 const modal = $('#modal');
 function openModal(title, text) {
@@ -2020,8 +2164,9 @@ async function loadStats() {
       <div class="stat cool"><div class="n">${fmt(s.fused_both)}</div><div class="k">双来源</div></div>
       <div class="stat"><div class="n">${fmt(s.fused_behavior_only)}</div><div class="k">行为新发现</div></div>`;
     $('#meta').innerHTML = `最近更新 <b>${localTime(s.updated_at)}</b><br>
+      <a href="#/monitor">监控</a> ·
       <a href="${BASE}/api/ips" target="_blank">JSON</a> ·
-      <a href="${BASE}/api/stats" target="_blank">stats</a>`;
+      <a href="${BASE}/api/metrics" target="_blank">metrics</a>`;
   } catch (err) {
     $('#meta').textContent = '统计加载失败: ' + err.message;
   }
@@ -2281,6 +2426,8 @@ function advice(item) {
 
 async function showDetail(ip) {
   state.inDetail = true;
+  state.inMonitor = false;
+  $('#monitorView').hidden = true;
   $('#listView').hidden = true;
   const view = $('#detailView');
   view.hidden = false;
@@ -2458,7 +2605,9 @@ async function showDetail(ip) {
 
 function showList() {
   state.inDetail = false;
+  state.inMonitor = false;
   document.title = '恶意 IP 清单';
+  $('#monitorView').hidden = true;
   $('#detailView').hidden = true;
   $('#detailView').innerHTML = '';
   $('#listView').hidden = false;
@@ -2467,7 +2616,101 @@ function showList() {
   loadList();
 }
 
+
+function monitorCard(title, value, rows, valueClass) {
+  const body = rows.map(row =>
+    `<div class="mrow"><span>${esc(row[0])}</span><span>${esc(row[1])}</span></div>`
+  ).join('');
+  return `<div class="mcard"><h3>${esc(title)}</h3>
+    <div class="mbig ${valueClass || ''}">${esc(value)}</div>${body}</div>`;
+}
+
+function monitorHtml(m) {
+  const d = m.detect, b = m.behavior, p = m.push, l = m.list, bf = m.backfill;
+  const stateText = m.graylog === 'ok' ? '正常' : m.graylog === 'error' ? '失败'
+                  : m.graylog === 'stale' ? '过期' : '未知';
+  const stateCls = m.graylog === 'ok' ? 'mok' : m.graylog === 'error' ? 'mbad'
+                 : 'mwarn';
+  const ratio = p.success_ratio_24h == null ? '-' :
+                (p.success_ratio_24h * 100).toFixed(1) + '%';
+  const cards = [
+    monitorCard('Graylog 连接', stateText, [
+      ['最近成功', d.last_success ? localTime(d.last_success) : '-'],
+      ['距上次成功', d.gap_seconds == null ? '-' : Math.round(d.gap_seconds) + ' 秒'],
+      ['最近失败', d.last_error ? d.last_error.slice(0, 24) : '无'],
+    ], stateCls),
+    monitorCard('特征检测', fmt(d.rounds) + ' 轮', [
+      ['失败次数', fmt(d.errors)],
+      ['最近耗时', d.last_duration ? d.last_duration + ' 秒' : '-'],
+      ['检测间隔', '60 秒'],
+    ]),
+    monitorCard('行为分析', fmt(b.runs) + ' 次', [
+      ['最近耗时', b.last_duration ? b.last_duration + ' 秒' : '-'],
+      ['最近违规', fmt(b.last_count)],
+      ['最近运行', b.last_success ? localTime(b.last_success) : '-'],
+    ], b.last_error ? 'mbad' : ''),
+    monitorCard('推送（24h）', ratio, [
+      ['成功 / 失败', fmt(p.sent_24h) + ' / ' + fmt(p.failed_24h)],
+      ['连续失败', fmt(p.consecutive_failures)],
+      ['最近错误', p.last_error ? p.last_error.slice(0, 24) : '无'],
+    ], (p.consecutive_failures > 0) ? 'mbad' : 'mok'),
+    monitorCard('名单', fmt(l.fused_total) + ' 个', [
+      ['今日新增', fmt(l.fused_new_24h)],
+      ['双来源', fmt(l.fused_both)],
+      ['行为新发现', fmt(l.fused_behavior_only)],
+    ]),
+    monitorCard('自动补测', fmt(bf.runs) + ' 次', [
+      ['上次补测', bf.last_at ? localTime(bf.last_at) : '无'],
+      ['补测区间', bf.last_from ? (localTime(bf.last_from) + ' → ' + localTime(bf.last_to)) : '-'],
+      ['上次新增', fmt(bf.last_added)],
+    ], bf.last_error ? 'mbad' : ''),
+  ].join('');
+  const rows = (m.recent_notifications || []).map(row => {
+    const cls = row.status === 'sent' ? 'ok' : 'bad';
+    return `<tr><td class="muted">${localTime(row.created_at)}</td>
+      <td>${esc(row.ip)}</td>
+      <td><span class="nbadge ${cls}">${row.status === 'sent' ? '成功' : '失败'}</span></td>
+      <td class="muted">${esc(row.error || '-')}</td></tr>`;
+  }).join('');
+  return `<div class="monitor-head">
+      <a class="back" href="#/">← 返回名单</a>
+      <div class="monitor-title">运行监控</div>
+      <span class="muted">每 30 秒刷新</span>
+    </div>
+    <div class="mgrid">${cards}</div>
+    <div class="mtablewrap">
+      <table class="mtable">
+        <thead><tr><th>时间</th><th>IP</th><th>结果</th><th>错误</th></tr></thead>
+        <tbody>${rows || '<tr><td colspan="4" class="empty">暂无推送记录</td></tr>'}</tbody>
+      </table>
+    </div>`;
+}
+
+async function showMonitor() {
+  state.inMonitor = true;
+  state.inDetail = false;
+  document.title = '运行监控';
+  $('#listView').hidden = true;
+  $('#detailView').hidden = true;
+  $('#monitorView').hidden = false;
+  loadStatus();
+  loadStats();
+  await loadMonitor();
+}
+
+async function loadMonitor() {
+  const view = $('#monitorView');
+  try {
+    const data = await jget(BASE + '/api/monitor');
+    view.innerHTML = monitorHtml(data);
+  } catch (err) {
+    view.innerHTML = `<a class="back" href="#/">← 返回名单</a>
+      <div class="empty">加载失败：${esc(err.message)}</div>`;
+  }
+}
+
 function route() {
+  if (location.hash === '#/monitor') { showMonitor(); return; }
   const match = location.hash.match(/^#\/ip\/(.+)$/);
   if (match) showDetail(decodeURIComponent(match[1]));
   else showList();
@@ -2524,7 +2767,8 @@ function startTimer() {
   clearInterval(state.timer);
   state.timer = setInterval(() => {
     loadStatus();
-    if (!state.inDetail) { loadStats(); loadList(); }
+    if (state.inMonitor) { loadMonitor(); }
+    else if (!state.inDetail) { loadStats(); loadList(); }
   }, 30000);
 }
 window.addEventListener('hashchange', route);
@@ -2584,6 +2828,133 @@ def make_handler(store, window, interval, detector=None, path_prefix=""):
                 return self._send(200, {"ok": True})
             if path == "/api/stats":
                 return self._send(200, store.stats())
+            if path in ("/metrics", "/api/monitor"):
+                stats = store.stats()
+                meta = store.meta_dict()
+                notif = store.notification_stats(24)
+                now_epoch = time.time()
+
+                def stamp_epoch(value):
+                    try:
+                        return datetime.strptime(
+                            value, "%Y-%m-%dT%H:%M:%SZ").replace(
+                                tzinfo=timezone.utc).timestamp()
+                    except (TypeError, ValueError):
+                        return None
+
+                last_success = stamp_epoch(meta.get("detect_last_success"))
+                last_behavior = stamp_epoch(meta.get("behavior_last_success"))
+                age = round(now_epoch - last_success, 1) if last_success else None
+                behavior_age = (round(now_epoch - last_behavior, 1)
+                                if last_behavior else None)
+                if age is None:
+                    graylog = "unknown"
+                elif meta.get("detect_last_error") and age > interval * 2:
+                    graylog = "error"
+                elif age <= max(120, interval * 2.5):
+                    graylog = "ok"
+                else:
+                    graylog = "stale"
+
+                payload = {
+                    "time": now_iso(),
+                    "graylog": graylog,
+                    "detect": {
+                        "rounds": int(meta.get("detect_rounds_total") or 0),
+                        "errors": int(meta.get("detect_errors_total") or 0),
+                        "last_success": meta.get("detect_last_success"),
+                        "gap_seconds": age,
+                        "last_duration": float(meta.get("detect_last_duration") or 0),
+                        "last_error": meta.get("detect_last_error") or "",
+                    },
+                    "behavior": {
+                        "runs": int(meta.get("behavior_runs_total") or 0),
+                        "errors": int(meta.get("behavior_errors_total") or 0),
+                        "last_success": meta.get("behavior_last_success"),
+                        "age_seconds": behavior_age,
+                        "last_duration": float(meta.get("behavior_last_duration") or 0),
+                        "last_count": int(meta.get("behavior_last_count") or 0),
+                        "last_error": meta.get("behavior_last_error") or "",
+                    },
+                    "push": {
+                        "sent_total": int(meta.get("push_sent_total") or 0),
+                        "failed_total": int(meta.get("push_failed_total") or 0),
+                        "sent_24h": notif["sent"],
+                        "failed_24h": notif["failed"],
+                        "success_ratio_24h": notif["success_ratio"],
+                        "consecutive_failures": int(
+                            meta.get("push_consecutive_failures") or 0),
+                        "last_error": meta.get("push_last_error") or "",
+                        "last_error_at": meta.get("push_last_error_at") or "",
+                    },
+                    "list": {
+                        "fused_total": stats.get("fused_total"),
+                        "fused_new_24h": stats.get("fused_new_24h"),
+                        "fused_both": stats.get("fused_both"),
+                        "fused_behavior_only": stats.get("fused_behavior_only"),
+                        "updated_at": stats.get("updated_at"),
+                    },
+                    "backfill": {
+                        "runs": int(meta.get("backfill_runs_total") or 0),
+                        "last_at": meta.get("backfill_last_at") or "",
+                        "last_from": meta.get("backfill_last_from") or "",
+                        "last_to": meta.get("backfill_last_to") or "",
+                        "last_added": int(meta.get("backfill_last_added") or 0),
+                        "last_error": meta.get("backfill_last_error") or "",
+                    },
+                    "recent_notifications": notif["recent"],
+                }
+                if path == "/api/monitor":
+                    return self._send(200, payload)
+
+                lines = [
+                    "# HELP ipsec_detect_rounds_total Successful detection rounds",
+                    "# TYPE ipsec_detect_rounds_total counter",
+                    "ipsec_detect_rounds_total %d" % payload["detect"]["rounds"],
+                    "# TYPE ipsec_detect_errors_total counter",
+                    "ipsec_detect_errors_total %d" % payload["detect"]["errors"],
+                    "# TYPE ipsec_detect_last_success_timestamp gauge",
+                    "ipsec_detect_last_success_timestamp %s"
+                    % (last_success or 0),
+                    "# TYPE ipsec_detect_gap_seconds gauge",
+                    "ipsec_detect_gap_seconds %s" % (age if age is not None else -1),
+                    "# TYPE ipsec_detect_last_duration_seconds gauge",
+                    "ipsec_detect_last_duration_seconds %s"
+                    % payload["detect"]["last_duration"],
+                    "# TYPE ipsec_behavior_runs_total counter",
+                    "ipsec_behavior_runs_total %d" % payload["behavior"]["runs"],
+                    "# TYPE ipsec_behavior_errors_total counter",
+                    "ipsec_behavior_errors_total %d" % payload["behavior"]["errors"],
+                    "# TYPE ipsec_behavior_last_duration_seconds gauge",
+                    "ipsec_behavior_last_duration_seconds %s"
+                    % payload["behavior"]["last_duration"],
+                    "# TYPE ipsec_behavior_last_violations gauge",
+                    "ipsec_behavior_last_violations %d"
+                    % payload["behavior"]["last_count"],
+                    "# TYPE ipsec_fused_total gauge",
+                    "ipsec_fused_total %s" % (payload["list"]["fused_total"] or 0),
+                    "# TYPE ipsec_fused_new_24h gauge",
+                    "ipsec_fused_new_24h %s" % (payload["list"]["fused_new_24h"] or 0),
+                    "# TYPE ipsec_fused_both gauge",
+                    "ipsec_fused_both %s" % (payload["list"]["fused_both"] or 0),
+                    "# TYPE ipsec_push_sent_total counter",
+                    "ipsec_push_sent_total %d" % payload["push"]["sent_total"],
+                    "# TYPE ipsec_push_failed_total counter",
+                    "ipsec_push_failed_total %d" % payload["push"]["failed_total"],
+                    "# TYPE ipsec_push_consecutive_failures gauge",
+                    "ipsec_push_consecutive_failures %d"
+                    % payload["push"]["consecutive_failures"],
+                    "# TYPE ipsec_push_success_ratio_24h gauge",
+                    "ipsec_push_success_ratio_24h %s"
+                    % (payload["push"]["success_ratio_24h"]
+                       if payload["push"]["success_ratio_24h"] is not None else -1),
+                    "# TYPE ipsec_graylog_up gauge",
+                    "ipsec_graylog_up %d" % (1 if graylog == "ok" else 0),
+                    "",
+                ]
+                return self._send(200, "\n".join(lines),
+                                  content_type="text/plain; version=0.0.4; charset=utf-8")
+
             if path == "/api/status":
                 status = {"graylog": "unknown", "last_run": None,
                           "last_error": None, "age_seconds": None}
@@ -2838,6 +3209,47 @@ def run(ctx, args):
     # Fused block list: existing feature detections first (historical rows keep
     # their original discovery time), then the behaviour thread keeps adding.
     store.sync_fused()
+    filter_str = ("streams:" + ctx.stream_id) if ctx.stream_id else None
+
+    def auto_backfill_worker(from_dt, to_dt):
+        from .backfill import auto_backfill_gap
+        summary = auto_backfill_gap(
+            store, ctx.client, fields, filter_str, from_dt, to_dt,
+            os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+        print("[自动补测] %s" % json.dumps(summary, ensure_ascii=False),
+              flush=True)
+
+    auto_enabled = not getattr(args, "no_auto_backfill", False)
+    threshold = int(getattr(args, "backfill_threshold", 1800) or 0)
+    max_gap = int(getattr(args, "backfill_max_gap", 7 * 86400) or 0)
+    if (auto_enabled and threshold > 0
+            and not getattr(args, "serve_only", False)):
+        from .backfill import parse_iso
+        last_meta = store.get_meta("detect_last_success")
+        if not last_meta:
+            row = store.conn.execute(
+                "SELECT MAX(updated_at) AS t FROM ip_stats").fetchone()
+            last_meta = row["t"] if row and row["t"] else None
+        last_dt = parse_iso(last_meta) if last_meta else None
+        if last_dt is not None:
+            now_dt = datetime.now(timezone.utc)
+            gap = (now_dt - last_dt).total_seconds()
+            if gap > max_gap:
+                print("检测缺口 %.1f 小时，超过自动补测上限 %d 小时，已跳过"
+                      % (gap / 3600.0, max_gap // 3600), flush=True)
+            elif gap > threshold:
+                print("检测缺口 %.1f 小时，后台自动补测 %s → %s"
+                      % (gap / 3600.0, last_dt.strftime("%m-%d %H:%M"),
+                         now_dt.astimezone().strftime("%m-%d %H:%M")),
+                      flush=True)
+                threading.Thread(target=auto_backfill_worker,
+                                 args=(last_dt, now_dt), daemon=True).start()
+            else:
+                print("检测连续（上次成功 %s），无需补测"
+                      % last_dt.strftime("%m-%d %H:%M"), flush=True)
+        else:
+            print("没有历史检测记录，跳过自动补测", flush=True)
+
     behavior_analyzer = None
     if not getattr(args, "no_behavior", False):
         behavior_analyzer = BehaviorAnalyzer(
@@ -2915,9 +3327,15 @@ def run(ctx, args):
                     print("  封禁名单: %d 个 -> %s" % (len(chosen), args.export), flush=True)
             except GraylogError as exc:
                 detector.last_error = str(exc)
+                detector.error_count += 1
+                detector.store.set_meta("detect_last_error", str(exc))
+                detector.store.incr_meta("detect_errors_total")
                 print("[%s] 检测失败: %s" % (now_iso(), exc), flush=True)
             except Exception as exc:
                 detector.last_error = str(exc)
+                detector.error_count += 1
+                detector.store.set_meta("detect_last_error", str(exc))
+                detector.store.incr_meta("detect_errors_total")
                 print("[%s] 检测异常: %s" % (now_iso(), exc), flush=True)
             time.sleep(max(5, int(args.interval)))
     except KeyboardInterrupt:

tests/test_behavior.py

@@ -8,6 +8,7 @@ HERE = os.path.dirname(os.path.abspath(__file__))
 sys.path.insert(0, os.path.dirname(HERE))
 
 from graylog_ops import watch  # noqa: E402
+from graylog_ops.behavior import notify_fused  # noqa: E402
 
 NOW = "2026-10-17T16:00:00Z"
 
@@ -99,6 +100,54 @@ class FusedTests(unittest.TestCase):
                 store.close()
 
 
+class PushRetryTests(unittest.TestCase):
+    def test_failure_backoff_and_audit(self):
+        with tempfile.TemporaryDirectory() as directory:
+            store = watch.Store(os.path.join(directory, "ips.db"))
+            try:
+                store.replace_behavior_profiles([_behavior("198.51.100.8")])
+                store.sync_fused()
+                with store.lock, store.conn:
+                    store.conn.execute(
+                        "UPDATE ip_fused SET last_pushed_at=NULL WHERE ip=?",
+                        ("198.51.100.8",))
+
+                class Failing:
+                    kind = "dingtalk"
+                    def link_for(self, ip): return ""
+                    def post(self, text, title=None): raise RuntimeError("boom")
+
+                notify_fused(store, Failing())
+                notif = store.notification_stats(24)
+                self.assertEqual(notif["failed"], 1)
+                self.assertEqual(notif["sent"], 0)
+                row = store.fused_row("198.51.100.8")
+                self.assertEqual(row["push_fail_count"], 1)
+                self.assertTrue(row["push_next_at"])
+                self.assertEqual(store.due_pushes(24, 24), [])  # backoff active
+                self.assertEqual(store.get_meta("push_consecutive_failures"), "1")
+
+                with store.lock, store.conn:
+                    store.conn.execute(
+                        "UPDATE ip_fused SET push_next_at=NULL WHERE ip=?",
+                        ("198.51.100.8",))
+
+                class Working:
+                    kind = "dingtalk"
+                    def link_for(self, ip): return ""
+                    def post(self, text, title=None): return None
+
+                notify_fused(store, Working())
+                row = store.fused_row("198.51.100.8")
+                self.assertEqual(row["push_fail_count"], 0)
+                self.assertIsNone(row["push_next_at"])
+                self.assertEqual(store.get_meta("push_consecutive_failures"), "0")
+                notif = store.notification_stats(24)
+                self.assertEqual(notif["sent"], 1)
+            finally:
+                store.close()
+
+
 class HistoricalBehaviourTests(unittest.TestCase):
     def test_historical_profile_survives_live_refresh(self):
         with tempfile.TemporaryDirectory() as directory:

tests/test_watch.py

@@ -341,6 +341,53 @@ class PathPrefixTests(unittest.TestCase):
                 store.close()
 
 
+class MetricsTests(unittest.TestCase):
+    def test_metrics_and_monitor(self):
+        with tempfile.TemporaryDirectory() as directory:
+            store = watch.Store(os.path.join(directory, "ips.db"))
+            now = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())
+            store.set_meta("detect_last_success", now)
+            store.set_meta("detect_last_error", "")
+            store.incr_meta("detect_rounds_total", 3)
+            store.incr_meta("detect_errors_total", 1)
+            store.set_meta("behavior_last_success", now)
+            store.set_meta("behavior_last_duration", 12.5)
+            store.set_meta("behavior_last_count", 7)
+            store.incr_meta("behavior_runs_total", 2)
+            store.incr_meta("push_sent_total", 5)
+            store.incr_meta("push_failed_total", 2)
+            store.upsert({"ip": "203.0.113.9", "tier": watch.TIER_MALICIOUS,
+                          "score": 91.0, "total": 120, "window": 3600,
+                          "rules": {"probe_java": 2},
+                          "reasons": ["探测"]}, "2026-10-17T16:00:00Z")
+            store.sync_fused()
+            server = ThreadingHTTPServer(("127.0.0.1", 0),
+                                         watch.make_handler(store, 3600, 60))
+            threading.Thread(target=server.serve_forever, daemon=True).start()
+            base = "http://127.0.0.1:%d" % server.server_address[1]
+            try:
+                with urllib.request.urlopen(base + "/metrics", timeout=5) as response:
+                    text = response.read().decode("utf-8")
+                self.assertIn("ipsec_detect_rounds_total 3", text)
+                self.assertIn("ipsec_detect_errors_total 1", text)
+                self.assertIn("ipsec_behavior_last_duration_seconds 12.5", text)
+                self.assertIn("ipsec_fused_total 1", text)
+                self.assertIn("ipsec_push_sent_total 5", text)
+                self.assertIn("ipsec_graylog_up 1", text)
+
+                with urllib.request.urlopen(base + "/api/monitor", timeout=5) as response:
+                    payload = json.loads(response.read().decode("utf-8"))
+                self.assertEqual(payload["graylog"], "ok")
+                self.assertEqual(payload["detect"]["rounds"], 3)
+                self.assertEqual(payload["behavior"]["last_duration"], 12.5)
+                self.assertEqual(payload["push"]["sent_total"], 5)
+                self.assertEqual(payload["list"]["fused_total"], 1)
+            finally:
+                server.shutdown()
+                server.server_close()
+                store.close()
+
+
 class PaginationTests(unittest.TestCase):
     def test_pagination_and_search(self):
         with tempfile.TemporaryDirectory() as directory:

tools/backfill_behavior.py

@@ -59,14 +59,14 @@ def main():
     fields = build_fields(load_profile(args.profile))
     filter_str = "streams:" + args.stream
 
-    behavior.RANGE_OVERRIDE = {"from": utc_iso(from_dt), "to": utc_iso(to_dt)}
+    behavior.set_range(utc_iso(from_dt), utc_iso(to_dt))
     store = watch.Store(args.db)
     analyzer = behavior.BehaviorAnalyzer(client, store, fields, filter_str,
                                          root=ROOT, notifier=None)
     try:
         items = analyzer.analyze()
     finally:
-        behavior.RANGE_OVERRIDE = None
+        behavior.clear_range()
 
     flagged = [item for item in items if item.get("verdict") == "建议封禁"]
     for item in items: