fix: multi-node online stats, per-node Hy2 ports, and panel reload stability

Co-authored-by: Cursor <cursoragent@cursor.com>

panel/stats.py

@@ -3,18 +3,26 @@ from __future__ import annotations
 
 import json
 import os
+import re
 import sqlite3
+import subprocess
 import time
 import urllib.error
 import urllib.request
 from pathlib import Path
 
 from db import connect, list_nodes
+from nodes_util import hy2_inbound_tag, ordered_nodes
 
 ROOT = Path(os.environ.get("JIEDIAN_ROOT", Path(__file__).resolve().parents[1]))
 ENV_FILE = ROOT / ".env"
 
 CLASH_ADDR = "127.0.0.1:9090"
+_VLESS_INBOUND = "vless-reality-in"
+_LOG_USER_RE = re.compile(
+    r"\[([0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12})\]\s+inbound connection"
+)
+_LOG_INDEX_RE = re.compile(r"\[(\d+)\] inbound connection")
 
 _speed_cache: dict[int, tuple[float, int, int]] = {}
 _conn_cache: dict[str, dict[str, int | str]] = {}
@@ -64,6 +72,43 @@ def fetch_clash_connections() -> tuple[list[dict], bool]:
     return payload.get("connections") or [], True
 
 
+def fetch_recent_log_uuids(nodes: list[dict]) -> set[str]:
+    """sing-box Clash API 不导出 user 字段，VLESS 多用户需从近期日志补全在线 UUID。"""
+    try:
+        proc = subprocess.run(
+            [
+                "journalctl",
+                "-u",
+                "sing-box",
+                "--since",
+                "3 min ago",
+                "--no-pager",
+                "-o",
+                "cat",
+            ],
+            capture_output=True,
+            text=True,
+            timeout=2,
+        )
+    except (OSError, subprocess.TimeoutExpired):
+        return set()
+    if proc.returncode != 0:
+        return set()
+
+    known = {node["uuid"] for node in nodes}
+    index_to_uuid = {i: node["uuid"] for i, node in enumerate(ordered_nodes(nodes))}
+    active: set[str] = set()
+    for match in _LOG_USER_RE.finditer(proc.stdout):
+        uid = match.group(1)
+        if uid in known:
+            active.add(uid)
+    for match in _LOG_INDEX_RE.finditer(proc.stdout):
+        uid = index_to_uuid.get(int(match.group(1)))
+        if uid:
+            active.add(uid)
+    return active
+
+
 def _global_conn_speed(connections: list[dict]) -> tuple[float, float]:
     """从 /connections 汇总字节增量估算全局速率（/traffic 为 WebSocket 流，不能同步 HTTP 读）。"""
     total_up = sum(int(c.get("upload") or 0) for c in connections)
@@ -81,6 +126,14 @@ def _global_conn_speed(connections: list[dict]) -> tuple[float, float]:
     return max(0.0, (total_up - u0) / dt), max(0.0, (total_down - d0) / dt)
 
 
+def _connection_inbound_tag(conn: dict) -> str:
+    meta = conn.get("metadata") or {}
+    inbound_type = str(meta.get("type") or "")
+    if "/" in inbound_type:
+        return inbound_type.split("/", 1)[1]
+    return inbound_type
+
+
 def _connection_user(conn: dict) -> str:
     meta = conn.get("metadata") or {}
     for key in ("user", "uid", "auth_user", "auth", "username"):
@@ -94,17 +147,6 @@ def _connection_user(conn: dict) -> str:
     return ""
 
 
-def _connection_meta(conn: dict) -> str:
-    meta = conn.get("metadata") or {}
-    parts = [
-        str(meta.get("type") or ""),
-        str(meta.get("network") or ""),
-        str(meta.get("inbound") or meta.get("inboundTag") or ""),
-        str(meta.get("inboundType") or ""),
-    ]
-    return " ".join(parts).lower()
-
-
 def _node_auth_keys(node: dict) -> set[str]:
     keys = {node["uuid"]}
     if node.get("hy2_password"):
@@ -112,18 +154,32 @@ def _node_auth_keys(node: dict) -> set[str]:
     return keys
 
 
-def _match_connection(conn: dict, node: dict) -> bool:
+def _match_connection(conn: dict, node: dict, *, single_node: bool = False) -> bool:
     user = _connection_user(conn)
     if user and user in _node_auth_keys(node):
         return True
-    meta = _connection_meta(conn)
-    if "hysteria" in meta and node.get("hy2_password"):
-        # 旧配置未设置 hy2 name 时，Clash API 可能不带 user
-        if user == node["hy2_password"]:
-            return True
+
+    tag = _connection_inbound_tag(conn)
+    node_id = int(node["id"])
+    expected_hy2 = hy2_inbound_tag(node_id)
+    if tag == expected_hy2:
+        return True
+    if tag == "hysteria2-in" and single_node:
+        return True
     return False
 
 
+def _match_vless_connection(conn: dict, node: dict, log_active: set[str]) -> bool:
+    tag = _connection_inbound_tag(conn)
+    if tag != _VLESS_INBOUND:
+        return False
+    user = _connection_user(conn)
+    if user == node["uuid"]:
+        return True
+    # 共享 VLESS inbound 无法从 Clash API 区分用户；仅唯一活跃用户时归因
+    return node["uuid"] in log_active and len(log_active) == 1
+
+
 def _connection_id(conn: dict) -> str:
     return str(conn.get("id") or "")
 
@@ -185,6 +241,7 @@ def _sync_connections(
     connections: list[dict],
     nodes: list[dict],
     uuid_to_node: dict[str, int],
+    log_active: set[str],
 ) -> dict[str, tuple[int, int]]:
     """同步连接缓存，断开连接时写入累计流量，返回各用户当前活跃会话流量。"""
     seen: set[str] = set()
@@ -195,7 +252,9 @@ def _sync_connections(
     for conn in connections:
         matched_uuid = ""
         for node in nodes:
-            if _match_connection(conn, node):
+            if _match_connection(conn, node, single_node=single_node) or _match_vless_connection(
+                conn, node, log_active
+            ):
                 matched_uuid = node["uuid"]
                 break
         if not matched_uuid and single_node and connections:
@@ -252,11 +311,39 @@ def _calc_speed(node_id: int, up: int, down: int) -> tuple[float, float]:
     return max(0.0, (up - u0) / dt), max(0.0, (down - d0) / dt)
 
 
+def _connections_for_node(
+    connections: list[dict],
+    node: dict,
+    nodes: list[dict],
+    log_active: set[str],
+) -> list[dict | None]:
+    single_node = len(nodes) == 1
+    matched = [
+        c
+        for c in connections
+        if _match_connection(c, node, single_node=single_node)
+        or _match_vless_connection(c, node, log_active)
+    ]
+    if matched:
+        return matched
+
+    if single_node and connections:
+        return connections
+
+    if node["uuid"] in log_active:
+        vless_hits = [c for c in connections if _connection_inbound_tag(c) == _VLESS_INBOUND]
+        if vless_hits:
+            return vless_hits
+        return [None]
+    return []
+
+
 def collect_node_stats() -> dict:
     nodes = list_nodes()
     uuid_to_node = {node["uuid"]: int(node["id"]) for node in nodes}
     connections, clash_ok = fetch_clash_connections()
-    active_by_uuid = _sync_connections(connections, nodes, uuid_to_node)
+    log_active = fetch_recent_log_uuids(nodes) if len(nodes) > 1 else set()
+    active_by_uuid = _sync_connections(connections, nodes, uuid_to_node, log_active)
     single_node = len(nodes) == 1
     has_connections = len(connections) > 0
     global_up_speed, global_down_speed = _global_conn_speed(connections) if clash_ok else (0.0, 0.0)
@@ -276,11 +363,7 @@ def collect_node_stats() -> dict:
         display_down = stored_down + session_down
         up_speed, down_speed = _calc_speed(node_id, display_up, display_down)
 
-        matched = [c for c in connections if _match_connection(c, node)]
-        if not matched and single_node and has_connections:
-            matched = connections
-        if not matched and (session_up + session_down) > 0:
-            matched = [None]  # 有活跃会话但 Clash 未返回连接详情
+        matched = _connections_for_node(connections, node, nodes, log_active)
         if not matched and single_node and global_active:
             up_speed = global_up_speed
             down_speed = global_down_speed
@@ -288,6 +371,7 @@ def collect_node_stats() -> dict:
         online = (
             len(matched) > 0
             or (session_up + session_down) > 0
+            or uid in log_active
             or (up_speed + down_speed) > 512
             or (single_node and (global_active or has_connections))
         )