列表分页与服务端搜索：分页控件、全量导出、排序索引

graylog_ops/watch.py

@@ -362,6 +362,9 @@ class Store:
                        fused_reason TEXT, updated_at TEXT
                    )"""
             )
+            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 events (
                        id INTEGER PRIMARY KEY AUTOINCREMENT,
@@ -579,7 +582,29 @@ class Store:
                 "SELECT * FROM ip_fused WHERE ip = ?", (ip,)).fetchone()
         return dict(row) if row else None
 
-    def fused_items(self, limit=500):
+    @staticmethod
+    def _fused_filter(q):
+        if not q:
+            return "", []
+        like = "%" + q.replace("%", "").replace("_", "") + "%"
+        fields = ("f.ip", "f.fused_reason", "f.sources",
+                  "s.country_name", "s.region", "s.asn_brand", "s.asn_org",
+                  "b.country_name", "b.region", "b.asn_brand", "b.asn_org")
+        clause = " WHERE (" + " OR ".join("%s LIKE ?" % field
+                                          for field in fields) + ")"
+        return clause, [like] * len(fields)
+
+    def fused_count(self, q=None):
+        where, params = self._fused_filter(q)
+        with self.lock:
+            return self.conn.execute(
+                "SELECT COUNT(*) FROM ip_fused f "
+                "LEFT JOIN ip_stats s ON s.ip = f.ip "
+                "LEFT JOIN behavior_profiles b ON b.ip = f.ip" + where,
+                params).fetchone()[0]
+
+    def fused_items(self, limit=500, offset=0, q=None):
+        where, params = self._fused_filter(q)
         with self.lock:
             rows = self.conn.execute(
                 """SELECT f.*, s.tier AS tier, s.total AS total, s.rules AS s_rules,
@@ -597,8 +622,10 @@ class Store:
                      FROM ip_fused f
                      LEFT JOIN ip_stats s ON s.ip = f.ip
                      LEFT JOIN behavior_profiles b ON b.ip = f.ip
+                    """ + where + """
                     ORDER BY f.first_banned_at DESC, f.ip ASC
-                    LIMIT ?""", (int(limit),)).fetchall()
+                    LIMIT ? OFFSET ?""",
+                params + [int(limit), int(offset)]).fetchall()
         items = []
         for row in rows:
             item = dict(row)
@@ -1837,6 +1864,7 @@ a:hover{text-decoration:underline}
       </table>
       <div class="empty" id="empty" hidden>没有匹配的记录</div>
     </div>
+    <div class="pager" id="pager"></div>
     <div class="hint" id="listHint"></div>
   </section>
 
@@ -1901,7 +1929,8 @@ const RULE_TITLES = {
   exploit_payload: '注入/穿越载荷', login_attack: '登录失败尝试', login_burst: '登录接口高频',
   bad_ua: '扫描器 UA', bot_ua: '爬虫 UA', not_found: '404 探测'
 };
-const state = { q: '', items: [], timer: null, inDetail: false, detailSamples: [] };
+const state = { q: '', items: [], page: 1, pageSize: 50, total: 0, pages: 1,
+                 timer: null, inDetail: false, detailSamples: [] };
 
 const modal = $('#modal');
 function openModal(title, text) {
@@ -1974,26 +2003,47 @@ async function loadStats() {
 
 async function loadList() {
   try {
-    const data = await jget(BASE + '/api/ips?limit=500');
+    const params = new URLSearchParams({
+      page: state.page, page_size: state.pageSize,
+    });
+    if (state.q.trim()) params.set('q', state.q.trim());
+    const data = await jget(BASE + '/api/ips?' + params.toString());
+    state.total = data.total || 0;
+    state.pages = data.pages || 1;
+    if (state.page > state.pages) { state.page = state.pages; return loadList(); }
     state.items = data.items || [];
     render();
+    renderPager();
     $('#listHint').textContent =
-      `共 ${state.items.length} 条 · 点击任意一行查看融合证据 · 按发现时间倒序（最新在前）· 每 60 秒更新`;
+      `共 ${fmt(state.total)} 条 · 第 ${state.page}/${state.pages} 页 · ` +
+      `按发现时间倒序（最新在前）· 每 30 秒更新`;
   } catch (err) {
     $('#rows').innerHTML = `<tr><td colspan="10" class="empty">加载失败：${esc(err.message)}</td></tr>`;
   }
 }
 
+async function fetchAllItems() {
+  const params = new URLSearchParams({ limit: 2000 });
+  if (state.q.trim()) params.set('q', state.q.trim());
+  const data = await jget(BASE + '/api/ips?' + params.toString());
+  return data.items || [];
+}
+
 function filteredItems() {
-  const q = state.q.trim().toLowerCase();
-  if (!q) return state.items;
-  return state.items.filter(item => {
-    const hay = [item.ip, item.country_name, item.region, item.asn_brand,
-                 item.asn_org, item.asn, item.sources,
-                 ...(item.reasons || []),
-                 ...Object.keys(item.rules || {})].join(' ').toLowerCase();
-    return hay.includes(q);
-  });
+  return state.items;
+}
+
+function renderPager() {
+  const pager = $('#pager');
+  if (!pager) return;
+  const sizes = [20, 50, 100].map(n =>
+    `<option value="${n}" ${n === state.pageSize ? 'selected' : ''}>${n} 条/页</option>`
+  ).join('');
+  pager.innerHTML =
+    `<button class="btn" data-page="${state.page - 1}" ${state.page <= 1 ? 'disabled' : ''}>‹ 上一页</button>` +
+    `<span class="pagemeta">第 <b>${state.page}</b> / ${state.pages} 页 · 共 ${fmt(state.total)} 条</span>` +
+    `<button class="btn" data-page="${state.page + 1}" ${state.page >= state.pages ? 'disabled' : ''}>下一页 ›</button>` +
+    `<select id="pageSize">${sizes}</select>`;
 }
 
 let toastTimer = null;
@@ -2032,7 +2082,7 @@ async function copyText(text, message) {
 }
 
 async function copyIps(separator) {
-  const rows = filteredItems();
+  const rows = await fetchAllItems();
   if (!rows.length) { toast('当前列表为空'); return; }
   const newline = separator === 'newline';
   const text = rows.map(item => item.ip).join(newline ? '\n' : ',');
@@ -2061,8 +2111,8 @@ function downloadFile(content, filename, mime) {
   URL.revokeObjectURL(link.href);
 }
 
-function exportList(format) {
-  const rows = filteredItems();
+async function exportList(format) {
+  const rows = await fetchAllItems();
   if (!rows.length) { toast('当前列表为空'); return; }
   const stamp = new Date().toISOString().slice(0, 10);
   if (format === 'txt') {
@@ -2372,7 +2422,12 @@ function route() {
   else showList();
 }
 
-$('#search').addEventListener('input', e => { state.q = e.target.value; render(); });
+let searchTimer = null;
+$('#search').addEventListener('input', e => {
+  state.q = e.target.value;
+  clearTimeout(searchTimer);
+  searchTimer = setTimeout(() => { state.page = 1; loadList(); }, 250);
+});
 document.addEventListener('click', e => {
   const target = e.target.closest('[data-copyip]');
   if (!target) return;
@@ -2399,6 +2454,18 @@ $('#exportMenu').addEventListener('click', e => {
   closeMenus();
   exportList(item.dataset.export);
 });
+$('#pager').addEventListener('click', e => {
+  const btn = e.target.closest('button[data-page]');
+  if (!btn || btn.disabled) return;
+  state.page = Number(btn.dataset.page);
+  loadList();
+});
+$('#pager').addEventListener('change', e => {
+  if (e.target.id !== 'pageSize') return;
+  state.pageSize = Number(e.target.value);
+  state.page = 1;
+  loadList();
+});
 $('#auto').addEventListener('change', e => {
   if (e.target.checked) startTimer(); else clearInterval(state.timer);
 });
@@ -2485,8 +2552,14 @@ def make_handler(store, window, interval, detector=None, path_prefix=""):
                 return self._send(200, status)
 
             if path == "/api/ips":
-                limit = min(2000, int(query.get("limit", 500)))
-                fused = store.fused_items(limit)
+                page_size = int(query.get("page_size", query.get("limit", 50)) or 50)
+                page_size = max(1, min(2000, page_size))
+                page = max(1, int(query.get("page", 1) or 1))
+                offset = max(0, int(query.get("offset",
+                                              (page - 1) * page_size) or 0))
+                search_q = (query.get("q") or "").strip()
+                total = store.fused_count(search_q)
+                fused = store.fused_items(page_size, offset, search_q)
                 window = detector.window if detector is not None else 3600
                 items = []
                 for row in fused:
@@ -2519,9 +2592,12 @@ def make_handler(store, window, interval, detector=None, path_prefix=""):
                         "asn_type": row.get("asn_type") or "",
                         "asn_icon": row.get("asn_icon") or "",
                     })
-                return self._send(200, {"count": len(items),
-                                        "updated_at": store.stats().get("updated_at"),
-                                        "items": items})
+                pages = max(1, (total + page_size - 1) // page_size)
+                return self._send(200, {
+                    "count": len(items), "total": total, "page": page,
+                    "page_size": page_size, "pages": pages, "q": search_q,
+                    "updated_at": store.stats().get("updated_at"),
+                    "items": items})
             if path in ("/api/export.csv", "/api/export.txt"):
                 scope = query.get("scope", "core")
                 if scope not in ("core", "all"):

tests/test_watch.py

@@ -341,6 +341,44 @@ class PathPrefixTests(unittest.TestCase):
                 store.close()
 
 
+class PaginationTests(unittest.TestCase):
+    def test_pagination_and_search(self):
+        with tempfile.TemporaryDirectory() as directory:
+            store = watch.Store(os.path.join(directory, "ips.db"))
+            for index, ip in enumerate(("203.0.113.9", "203.0.113.10",
+                                        "198.51.100.77")):
+                store.upsert({"ip": ip, "tier": watch.TIER_MALICIOUS,
+                              "score": 90.0, "total": 100 + index,
+                              "window": 3600, "first_seen":
+                              "2026-10-17T16:0%d:00Z" % index,
+                              "rules": {"probe_java": 2},
+                              "reasons": ["敏感路径探测"]},
+                             "2026-10-17T16:30: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 + "/api/ips?page_size=2&page=2", timeout=5) as response:
+                    page = json.loads(response.read().decode("utf-8"))
+                self.assertEqual(page["total"], 3)
+                self.assertEqual(page["pages"], 2)
+                self.assertEqual(page["page"], 2)
+                self.assertEqual(len(page["items"]), 1)
+
+                with urllib.request.urlopen(
+                        base + "/api/ips?q=203.0.113", timeout=5) as response:
+                    filtered = json.loads(response.read().decode("utf-8"))
+                self.assertEqual(filtered["total"], 2)
+                self.assertEqual(len(filtered["items"]), 2)
+            finally:
+                server.shutdown()
+                server.server_close()
+                store.close()
+
+
 class StatusIndicatorTests(unittest.TestCase):
     def test_status_endpoint(self):
         with tempfile.TemporaryDirectory() as directory: