FnMessageBots（日志推送）

self.config = Config()                    # env 覆盖 + config.json
init_push_stats(self.config.cursor_dir)   # 推送统计（Web 页上的数字）
has_webhook = any(10 个渠道参数)           # 没配渠道 → 只起 Web 配置页
self.notifier = UnifiedNotifier(config)
self.event_processor = EventProcessor(notifier, config)
self.log_poller = self._build_log_poller()
self._sync_optional_pollers()             # 按勾选起 0~8 个数据源线程
self._register_db_event_handlers()        # eventId → handler 映射
# 再 start() 各 poller、起 UnifiedPollBatchFlush 线程、巡检线程、Flask

def _run_loop(self):
    last_id = None
    while self.running:
        if last_id is None:                       # 首次启动
            last_id = self._get_max_log_id()      # SELECT COALESCE(MAX(id),0) FROM log
            self._write_last_id(last_id)          # data/cursor/db_poller_cursor.txt
        last_id = self._poll_once(last_id)

SELECT id, serviceId, uid, uname, logtime, loglevel, eventId, parameter, category
FROM log WHERE id > ? ORDER BY id ASC

batch_events = dedupe_poll_batch_items(batch_events)      # 批内指纹去重
summary_event_data = {
  "count": len(batch_events),
  "by_type": {"SSH_AUTH_FAILED": 1, ...},                 # 按类型计数
  "items": preview_items[:10],                            # 前 10 条摘要
  "grouped_events": grouped_events,                       # 勿扰汇总要用
  "_source_key": md5(拼接 source:cursor),                 # 渠道层去重用
}
self.notifier.send_notification(event_type="POLL_BATCH_SUMMARY", ...)

if source == "docker":
    # cid + action + timeNano；漏掉纳秒会把同容器短时间两次同类事件误合并
    return f"docker:{event_type}:{cid}:{action}:{engine_time_nano}"
row_id = item.get("row_id")
return f"{source}:{event_type}:row:{row_id}"

def _is_duplicate(self, fingerprint: str) -> bool:
    with self._dedup_lock:
        if fingerprint in self.sent_events:
            if time.time() - self.sent_events[fingerprint] < self.dedup_window:  # 300
                return True
            del self.sent_events[fingerprint]
    return False

if self.batch_handler and batch_events and batch_delivery_failed:
    self._batch_fail_count += 1
    if persistence_failed or self._batch_fail_count < self.BATCH_MAX_ATTEMPTS:
        return last_id          # 不推进游标 → 下一轮重来
    self.logger.warning("批量处理连续失败 %s 次，放弃本批", self._batch_fail_count)
self._write_last_id(rows[-1]["id"])      # 只有走到这儿才提交游标

self.path = directory / "notification_outbox.db"        # chmod 0600
CREATE TABLE state       (name TEXT PRIMARY KEY, payload TEXT NOT NULL)
CREATE TABLE poll_events (id   TEXT PRIMARY KEY, payload TEXT NOT NULL)
PRAGMA synchronous=FULL
BEGIN IMMEDIATE

if ! pgrep -f "python.*src/main.py" > /dev/null; then echo "Python进程不存在"; exit 1; fi
python - <<'PY'
from urllib.request import ProxyHandler, build_opener     # 显式禁用代理
host = {"0.0.0.0": "127.0.0.1", "::": "::1"}.get(host, host)
PY