Update to v8.16.0

mod/project/node/dbutil/__init__.py

@@ -1,5 +1,6 @@
 from .load_db import LoadSite, HttpNode, TcpNode, NodeDB
 from .node_db import Node, ServerNodeDB, ServerMonitorRepo, NodeAPPKey
+from .node_monitor_db import NodeMonitorDB
 from .file_transfer_db import FileTransfer, FileTransferDB, FileTransferTask
 # from .executor import Script, ScriptGroup, ExecutorDB, ExecutorLog, ExecutorTask
 from .node_task_flow import Script, Flow, CommandTask, CommandLog, TransferFile, TransferLog, TaskFlowsDB, \
@@ -9,6 +10,8 @@ from .node_task_flow import Script, Flow, CommandTask, CommandLog, TransferFile,
 try:
     NodeDB().init_db()
     ServerNodeDB().init_db()
+    # 节点监控管理
+    NodeMonitorDB().init_db()
     FileTransferDB().init_db()
     # ExecutorDB().init_db()
     TaskFlowsDB().init_db()

mod/project/node/dbutil/node_db.py

@@ -430,7 +430,109 @@ class ServerMonitorRepo:
     @staticmethod
     def get_local_server_status():
         from system import system
-        return system().GetNetWork(None)
+        sys_obj = system()
+        try:
+            return sys_obj.GetNetWork(None)
+        except Exception:
+            return ServerMonitorRepo.get_local_server_status_without_request(sys_obj)
+
+    @staticmethod
+    def get_local_server_status_without_request(sys_obj=None):
+        from system import system
+        import psutil
+
+        sys_obj = sys_obj or system()
+        data = {
+            "network": {},
+            "upTotal": 0,
+            "downTotal": 0,
+            "up": 0,
+            "down": 0,
+            "downPackets": 0,
+            "upPackets": 0,
+        }
+        try:
+            for name, io_data in psutil.net_io_counters(pernic=True).items():
+                data["network"][name] = {
+                    "upTotal": io_data.bytes_sent,
+                    "downTotal": io_data.bytes_recv,
+                    "up": 0,
+                    "down": 0,
+                    "downPackets": io_data.packets_recv,
+                    "upPackets": io_data.packets_sent,
+                }
+                data["upTotal"] += io_data.bytes_sent
+                data["downTotal"] += io_data.bytes_recv
+                data["downPackets"] += io_data.packets_recv
+                data["upPackets"] += io_data.packets_sent
+        except Exception:
+            pass
+
+        try:
+            data["cpu"] = sys_obj.GetCpuInfo(1)
+        except Exception:
+            data["cpu"] = [0, 0]
+        try:
+            data["cpu_times"] = sys_obj.get_cpu_times()
+        except Exception:
+            data["cpu_times"] = None
+        try:
+            data["load"] = sys_obj.GetLoadAverage(None)
+        except Exception:
+            data["load"] = {"one": 0, "five": 0, "fifteen": 0}
+        try:
+            data["mem"] = sys_obj.GetMemInfo(None)
+        except Exception:
+            data["mem"] = {}
+        try:
+            data["version"] = public.version()
+        except Exception:
+            data["version"] = ""
+
+        disk_list = []
+        try:
+            for disk in sys_obj.GetDiskInfo2(False):
+                disk = dict(disk)
+                size_data = disk.get("size", [])
+                if isinstance(size_data, list) and len(size_data) >= 3:
+                    try:
+                        size_data.append(int(size_data[0]) - (int(size_data[1]) + int(size_data[2])))
+                    except Exception:
+                        pass
+                    disk["size"] = [
+                        "{}G".format(round(int(num) / 1048576, 2)) if str(num).isdigit() and "G" not in str(num) else num
+                        for num in size_data
+                    ]
+                inode_data = disk.get("inodes", [])
+                if isinstance(inode_data, list):
+                    disk["inodes"] = [
+                        "{}G".format(round(int(num) / 1048576, 2)) if str(num).isdigit() and "G" not in str(num) else num
+                        for num in inode_data
+                    ]
+                disk_list.append(disk)
+        except Exception:
+            pass
+        data["disk"] = disk_list
+
+        for key, default, getter in (
+            ("title", "", sys_obj.GetTitle),
+            ("time", "", sys_obj.GetBootTime),
+            ("system", "", sys_obj.GetSystemVersion),
+            ("installed", False, sys_obj.CheckInstalled),
+            ("iostat", {}, sys_obj.get_disk_iostat),
+        ):
+            try:
+                data[key] = getter()
+            except Exception:
+                data[key] = default
+        try:
+            data["site_total"] = public.M("sites").count()
+            data["ftp_total"] = public.M("ftps").count()
+            data["database_total"] = public.M("databases").count()
+        except Exception:
+            data["site_total"] = data["ftp_total"] = data["database_total"] = 0
+
+        return data
 
     def get_server_status(self, server_id: int) -> Optional[Dict]:
         cache_file = os.path.join(self._REPO_DIR, "server_{}.json".format(server_id))
@@ -459,4 +561,4 @@ class ServerMonitorRepo:
     def remove_cache(self, server_id: int):
         cache_file = os.path.join(self._REPO_DIR, "server_{}.json".format(server_id))
         if os.path.exists(cache_file):
-            os.remove(cache_file)
+            os.remove(cache_file)

mod/project/node/dbutil/node_monitor_db.py

@@ -0,0 +1,955 @@
+# coding: utf-8
+import json
+import os
+import re
+import sqlite3
+import sys
+import time
+from typing import Any, Dict, List, Optional, Tuple
+
+if "/www/server/panel/class" not in sys.path:
+    sys.path.insert(0, "/www/server/panel/class")
+
+import public
+
+
+DEFAULT_SNAPSHOT_RETENTION_DAYS = 7
+MIN_SNAPSHOT_RETENTION_DAYS = 1
+MAX_SNAPSHOT_RETENTION_DAYS = 3650
+
+
+def _json_dumps(data: Any) -> str:
+    try:
+        return json.dumps(data)
+    except Exception:
+        return "{}"
+
+
+def _json_loads(data: Any, default: Any) -> Any:
+    if isinstance(data, (dict, list)):
+        return data
+    if not isinstance(data, str) or not data:
+        return default
+    try:
+        return json.loads(data)
+    except Exception:
+        return default
+
+
+def _to_float(value: Any, default: float = 0.0) -> float:
+    try:
+        if isinstance(value, str):
+            value = value.strip().replace("%", "")
+        return float(value)
+    except Exception:
+        return default
+
+
+def _to_int(value: Any, default: int = 0) -> int:
+    try:
+        return int(float(value))
+    except Exception:
+        return default
+
+
+def _to_bytes(value: Any) -> int:
+    if isinstance(value, (int, float)):
+        return int(value)
+    if not isinstance(value, str):
+        return 0
+    value = value.strip().replace(",", "")
+    if not value:
+        return 0
+    match = re.match(r"^([0-9]+(?:\.[0-9]+)?)\s*([KMGTPE]?)(?:I?B)?$", value, re.I)
+    if not match:
+        return _to_int(value)
+    number = _to_float(match.group(1))
+    unit = match.group(2).upper()
+    units = {
+        "": 1,
+        "K": 1024,
+        "M": 1024 ** 2,
+        "G": 1024 ** 3,
+        "T": 1024 ** 4,
+        "P": 1024 ** 5,
+        "E": 1024 ** 6,
+    }
+    return int(number * units.get(unit, 1))
+
+
+def _to_bytes_from_panel_mem(value: Any) -> int:
+    """面板获取网络内存数值单位通常为 MB,Shell 探针返回单位可能为字节,统一单位"""
+    number = _to_float(value)
+    if number <= 0:
+        return 0
+    if number < 1024 * 1024:
+        return int(number * 1024 * 1024)
+    return int(number)
+
+
+def _percent(value: Any) -> float:
+    return round(_to_float(value), 2)
+
+
+def _normalize_retention_days(value: Any) -> int:
+    days = _to_int(value, DEFAULT_SNAPSHOT_RETENTION_DAYS)
+    if days < MIN_SNAPSHOT_RETENTION_DAYS:
+        return MIN_SNAPSHOT_RETENTION_DAYS
+    if days > MAX_SNAPSHOT_RETENTION_DAYS:
+        return MAX_SNAPSHOT_RETENTION_DAYS
+    return days
+
+
+TABLE_SCHEMAS = {
+    # 保存每个节点的监控总开关、采集周期(默认采集,在线状态检测,详细指标采集 间隔)、费用(金额,币种代码/符号,费用周期)、到期时间
+    "node_monitor_setting": {
+        "columns": [
+            ("id", "INTEGER PRIMARY KEY AUTOINCREMENT"),
+            ("node_id", "INTEGER NOT NULL"),
+            ("enabled", "INTEGER NOT NULL DEFAULT 1"),
+            ("collect_interval", "INTEGER NOT NULL DEFAULT 60"),
+            ("status_interval", "INTEGER NOT NULL DEFAULT 60"),
+            ("detail_interval", "INTEGER NOT NULL DEFAULT 60"),
+            ("snapshot_retention_days", "INTEGER NOT NULL DEFAULT 7"),
+            ("cost_amount", "REAL NOT NULL DEFAULT 0"),
+            ("cost_currency", "TEXT NOT NULL DEFAULT 'CNY'"),
+            ("cost_symbol", "TEXT NOT NULL DEFAULT ''"),
+            ("cost_period", "TEXT NOT NULL DEFAULT 'month'"),
+            ("expire_at", "INTEGER NOT NULL DEFAULT 0"),
+            ("remark", "TEXT NOT NULL DEFAULT ''"),
+            ("created_at", "INTEGER NOT NULL DEFAULT 0"),
+            ("updated_at", "INTEGER NOT NULL DEFAULT 0"),
+        ],
+        "indexes": [
+            ("idx_node_monitor_setting_node", True, ("node_id",)),
+        ],
+    },
+    # 保存节点选择监控哪个磁盘、哪个网卡、哪些服务  target_type:disk,nic,service  target_key:对象唯一标识  extra:JSON 扩展字段  is_primary:标记展示的磁盘网卡
+    "node_monitor_target": {
+        "columns": [
+            ("id", "INTEGER PRIMARY KEY AUTOINCREMENT"),
+            ("node_id", "INTEGER NOT NULL"),
+            ("target_type", "TEXT NOT NULL"),
+            ("target_key", "TEXT NOT NULL"),
+            ("target_name", "TEXT NOT NULL DEFAULT ''"),
+            ("enabled", "INTEGER NOT NULL DEFAULT 1"),
+            ("is_primary", "INTEGER NOT NULL DEFAULT 0"),
+            ("sort", "INTEGER NOT NULL DEFAULT 0"),
+            ("extra", "TEXT NOT NULL DEFAULT '{}'"),
+            ("discovered_at", "INTEGER NOT NULL DEFAULT 0"),
+            ("updated_at", "INTEGER NOT NULL DEFAULT 0"),
+        ],
+        "indexes": [
+            ("idx_node_monitor_target_unique", True, ("node_id", "target_type", "target_key")),
+            ("idx_node_monitor_target_node_type", False, ("node_id", "target_type")),
+        ],
+    },
+    # 保存每个节点最新监控快照 每分钟覆盖写 status:-1在线 -0离线 -2采集失败但未判定离线 -4重启等待中 {}_json:当前节点全部磁盘网卡服务数据
+    "node_monitor_latest": {
+        "columns": [
+            ("node_id", "INTEGER PRIMARY KEY"),
+            ("ts", "INTEGER NOT NULL DEFAULT 0"),
+            ("status", "INTEGER NOT NULL DEFAULT 0"),
+            ("error_msg", "TEXT NOT NULL DEFAULT ''"),
+            ("cpu_usage", "REAL NOT NULL DEFAULT 0"),
+            ("mem_usage", "REAL NOT NULL DEFAULT 0"),
+            ("mem_used", "INTEGER NOT NULL DEFAULT 0"),
+            ("mem_total", "INTEGER NOT NULL DEFAULT 0"),
+            ("load1", "REAL NOT NULL DEFAULT 0"),
+            ("load5", "REAL NOT NULL DEFAULT 0"),
+            ("load15", "REAL NOT NULL DEFAULT 0"),
+            ("disk_json", "TEXT NOT NULL DEFAULT '[]'"),
+            ("nic_json", "TEXT NOT NULL DEFAULT '[]'"),
+            ("service_json", "TEXT NOT NULL DEFAULT '[]'"),
+            ("raw_json", "TEXT NOT NULL DEFAULT '{}'"),
+        ],
+        "indexes": [
+            ("idx_node_monitor_latest_ts", False, ("ts",)),
+        ],
+    },
+    # 保存历史监控数据，用于趋势图、平均值告警、持续时间判断
+    # 每分钟一条，一台节点一天 1440 条，100 台节点一天 14.4 万条。
+    # 保留策略：
+    # 分钟级数据保留 7 天
+    # 5 分钟聚合数据保留 30 天
+    # 超期数据定时清理
+
+    "node_monitor_snapshot": {
+        "columns": [
+            ("id", "INTEGER PRIMARY KEY AUTOINCREMENT"),
+            ("node_id", "INTEGER NOT NULL"),
+            ("ts", "INTEGER NOT NULL"),
+            ("status", "INTEGER NOT NULL DEFAULT 0"),
+            ("error_msg", "TEXT NOT NULL DEFAULT ''"),
+            ("cpu_usage", "REAL NOT NULL DEFAULT 0"),
+            ("mem_usage", "REAL NOT NULL DEFAULT 0"),
+            ("mem_used", "INTEGER NOT NULL DEFAULT 0"),
+            ("mem_total", "INTEGER NOT NULL DEFAULT 0"),
+            ("load1", "REAL NOT NULL DEFAULT 0"),
+            ("load5", "REAL NOT NULL DEFAULT 0"),
+            ("load15", "REAL NOT NULL DEFAULT 0"),
+            ("disk_json", "TEXT NOT NULL DEFAULT '[]'"),
+            ("nic_json", "TEXT NOT NULL DEFAULT '[]'"),
+            ("service_json", "TEXT NOT NULL DEFAULT '[]'"),
+            ("raw_json", "TEXT NOT NULL DEFAULT '{}'"),
+        ],
+        "indexes": [
+            ("idx_node_monitor_snapshot_node_ts", False, ("node_id", "ts")),
+            ("idx_node_monitor_snapshot_ts", False, ("ts",)),
+        ],
+    },
+}
+
+
+class NodeMonitorDB:
+    _DB_FILE = public.get_panel_path() + "/data/db/node_monitor.db"
+
+    def __init__(self):
+        pass
+
+    def close(self):
+        pass
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, exc_type, exc_value, exc_traceback):
+        self.close()
+
+    def __del__(self):
+        self.close()
+
+    def table(self, table_name: str):
+        return public.S(table_name, self._DB_FILE)
+
+    @classmethod
+    def _column_map(cls, table_name: str) -> Dict[str, str]:
+        schema = TABLE_SCHEMAS[table_name]
+        return {name: definition for name, definition in schema["columns"]}
+
+    @staticmethod
+    def _create_table_sql(table_name: str) -> str:
+        columns = ["`{}` {}".format(name, definition) for name, definition in TABLE_SCHEMAS[table_name]["columns"]]
+        return "CREATE TABLE IF NOT EXISTS `{}` (\n    {}\n);".format(table_name, ",\n    ".join(columns))
+
+    @staticmethod
+    def _create_index_sql(table_name: str, index_name: str, unique: bool, columns: Tuple[str, ...]) -> str:
+        unique_sql = "UNIQUE " if unique else ""
+        col_sql = ", ".join("`{}`".format(item) for item in columns)
+        return "CREATE {}INDEX IF NOT EXISTS `{}` ON `{}` ({});".format(unique_sql, index_name, table_name, col_sql)
+
+    # 初始化数据库 支持新增字段(在表定义中添加
+    def init_db(self):
+        db_dir = os.path.dirname(self._DB_FILE)
+        if not os.path.exists(db_dir):
+            os.makedirs(db_dir)
+        if not os.path.exists(self._DB_FILE):
+            public.writeFile(self._DB_FILE, "")
+
+        conn = sqlite3.connect(self._DB_FILE)
+        try:
+            conn.execute("PRAGMA busy_timeout=5000;")
+            try:
+                conn.execute("PRAGMA journal_mode=WAL;")
+            except Exception:
+                pass
+            cur = conn.cursor()
+            for table_name in TABLE_SCHEMAS:
+                cur.execute(self._create_table_sql(table_name))
+                cur.execute("PRAGMA table_info(`{}`)".format(table_name))
+                existing_cols = {row[1] for row in cur.fetchall()}
+                for col_name, col_def in TABLE_SCHEMAS[table_name]["columns"]:
+                    if col_name in existing_cols:
+                        continue
+                    if "PRIMARY KEY" in col_def.upper():
+                        continue
+                    cur.execute("ALTER TABLE `{}` ADD COLUMN `{}` {}".format(table_name, col_name, col_def))
+
+                for index_name, unique, columns in TABLE_SCHEMAS[table_name].get("indexes", []):
+                    cur.execute(self._create_index_sql(table_name, index_name, unique, columns))
+            conn.commit()
+        finally:
+            conn.close()
+
+    def get_setting(self, node_id: int, create: bool = True) -> Dict[str, Any]:
+        node_id = int(node_id)
+        data = self.table("node_monitor_setting").where("node_id=?", (node_id,)).find()
+        if isinstance(data, dict) and data:
+            return data
+        if not create:
+            return {}
+        now = int(time.time())
+        row = {
+            "node_id": node_id,
+            "enabled": 1,
+            "collect_interval": 60,
+            "status_interval": 60,
+            "detail_interval": 60,
+            "snapshot_retention_days": DEFAULT_SNAPSHOT_RETENTION_DAYS,
+            "cost_amount": 0,
+            "cost_currency": "CNY",
+            "cost_symbol": "",
+            "cost_period": "month",
+            "expire_at": 0,
+            "remark": "",
+            "created_at": now,
+            "updated_at": now,
+        }
+        self.table("node_monitor_setting").insert(row)
+        return self.table("node_monitor_setting").where("node_id=?", (node_id,)).find() or row
+
+    def save_setting(self, node_id: int, data: Dict[str, Any]) -> str:
+        node_id = int(node_id)
+        allowed = {
+            "enabled", "collect_interval", "status_interval", "detail_interval",
+            "snapshot_retention_days",
+            "cost_amount", "cost_currency", "cost_symbol", "cost_period",
+            "expire_at", "remark",
+        }
+        row = {}
+        for key, value in data.items():
+            if key not in allowed:
+                continue
+            if key == "snapshot_retention_days":
+                value = _normalize_retention_days(value)
+            row[key] = value
+        if not row:
+            return ""
+        row["updated_at"] = int(time.time())
+        if not self.get_setting(node_id, create=False):
+            self.get_setting(node_id, create=True)
+        res = self.table("node_monitor_setting").where("node_id=?", (node_id,)).update(row)
+        return res if isinstance(res, str) else ""
+
+    def get_targets(self, node_id: int, target_type: str = "") -> List[Dict[str, Any]]:
+        query = self.table("node_monitor_target")
+        if target_type:
+            data = query.where("node_id=? AND target_type=?", (int(node_id), target_type)).order("sort", "ASC").order("id", "ASC").select()
+        else:
+            data = query.where("node_id=?", (int(node_id),)).order("target_type", "ASC").order("sort", "ASC").order("id", "ASC").select()
+        if isinstance(data, list):
+            for item in data:
+                item["extra"] = _json_loads(item.get("extra"), {})
+            return data
+        return []
+
+    def get_enabled_service_targets(self, node_id: int) -> List[Dict[str, Any]]:
+        return [item for item in self.get_targets(node_id, "service") if int(item.get("enabled", 0)) == 1]
+
+    def upsert_target(self, node_id: int, target_type: str, target_key: str, data: Dict[str, Any]) -> str:
+        node_id = int(node_id)
+        target_type = str(target_type)
+        target_key = str(target_key)
+        now = int(time.time())
+        row = {
+            "target_name": data.get("target_name", ""),
+            "enabled": int(data.get("enabled", 1)),
+            "is_primary": int(data.get("is_primary", 0)),
+            "sort": int(data.get("sort", 0)),
+            "extra": _json_dumps(data.get("extra", {})),
+            "updated_at": now,
+        }
+        old = self.table("node_monitor_target").where(
+            "node_id=? AND target_type=? AND target_key=?", (node_id, target_type, target_key)
+        ).find()
+        try:
+            if isinstance(old, dict) and old:
+                self.table("node_monitor_target").where("id=?", (old["id"],)).update(row)
+            else:
+                row.update({
+                    "node_id": node_id,
+                    "target_type": target_type,
+                    "target_key": target_key,
+                    "discovered_at": now,
+                })
+                self.table("node_monitor_target").insert(row)
+        except Exception as e:
+            return str(e)
+        return ""
+
+    def set_primary_target(self, node_id: int, target_type: str, target_key: str) -> str:
+        if target_type not in ("disk", "nic"):
+            return "Only disk or nic can be set as the primary target"
+        node_id = int(node_id)
+        target_key = str(target_key)
+        self.table("node_monitor_target").where(
+            "node_id=? AND target_type=?", (node_id, target_type)
+        ).update({"is_primary": 0, "updated_at": int(time.time())})
+        old = self.table("node_monitor_target").where(
+            "node_id=? AND target_type=? AND target_key=?", (node_id, target_type, target_key)
+        ).find()
+        if not isinstance(old, dict) or not old:
+            err = self.upsert_target(node_id, target_type, target_key, {
+                "target_name": target_key,
+                "enabled": 1,
+                "is_primary": 1,
+            })
+            return err
+        res = self.table("node_monitor_target").where("id=?", (old["id"],)).update({
+            "enabled": 1,
+            "is_primary": 1,
+            "updated_at": int(time.time()),
+        })
+        return res if isinstance(res, str) else ""
+
+    def save_service_targets(self, node_id: int, services: List[Dict[str, Any]]) -> str:
+        node_id = int(node_id)
+        old = self.get_targets(node_id, "service")
+        old_map = {item["target_key"]: item for item in old}
+        seen = set()
+        for idx, item in enumerate(services):
+            key = str(item.get("target_key") or item.get("name") or "").strip()
+            if not key:
+                continue
+            seen.add(key)
+            err = self.upsert_target(node_id, "service", key, {
+                "target_name": item.get("target_name", key),
+                "enabled": int(item.get("enabled", 1)),
+                "is_primary": 0,
+                "sort": int(item.get("sort", idx)),
+                "extra": item.get("extra", {}),
+            })
+            if err:
+                return err
+        now = int(time.time())
+        for key, item in old_map.items():
+            if key not in seen:
+                self.table("node_monitor_target").where("id=?", (item["id"],)).update({
+                    "enabled": 0,
+                    "updated_at": now,
+                })
+        return ""
+
+    def sync_discovered_targets(self, node_id: int, disks: List[Dict[str, Any]], nics: List[Dict[str, Any]]) -> None:
+        self._sync_discovered_by_type(node_id, "disk", disks, "path")
+        self._sync_discovered_by_type(node_id, "nic", nics, "name")
+
+    def _sync_discovered_by_type(self, node_id: int, target_type: str, items: List[Dict[str, Any]], key_name: str) -> None:
+        if not items:
+            return
+        current = self.get_targets(node_id, target_type)
+        current_keys = {item["target_key"] for item in current}
+        has_primary = any(int(item.get("is_primary", 0)) == 1 for item in current)
+        for idx, item in enumerate(items):
+            key = str(item.get(key_name, "")).strip()
+            if not key:
+                continue
+            is_primary = 1 if (not has_primary and idx == 0) else 0
+            has_primary = has_primary or bool(is_primary)
+            if key in current_keys:
+                old = next((row for row in current if row["target_key"] == key), None)
+                if old:
+                    self.table("node_monitor_target").where("id=?", (old["id"],)).update({
+                        "target_name": item.get("name", key),
+                        "extra": _json_dumps(item),
+                        "updated_at": int(time.time()),
+                    })
+                continue
+            self.upsert_target(node_id, target_type, key, {
+                "target_name": item.get("name", key),
+                "enabled": 1,
+                "is_primary": is_primary,
+                "sort": idx,
+                "extra": item,
+            })
+
+    def normalize_monitor_data(self, raw_data: Dict[str, Any], status: int = 1, error_msg: str = "") -> Dict[str, Any]:
+        now = int(time.time())
+        cpu_usage = 0.0
+        cpu_data = raw_data.get("cpu", [])
+        if isinstance(cpu_data, (list, tuple)) and cpu_data:
+            cpu_usage = _percent(cpu_data[0])
+        elif isinstance(cpu_data, dict):
+            cpu_usage = _percent(cpu_data.get("usage", cpu_data.get("used", 0)))
+
+        mem = raw_data.get("mem", {}) if isinstance(raw_data.get("mem", {}), dict) else {}
+        mem_used = _to_bytes_from_panel_mem(mem.get("memRealUsed", mem.get("used", 0)))
+        mem_total = _to_bytes_from_panel_mem(mem.get("memTotal", mem.get("total", 0)))
+        mem_usage = round(mem_used / mem_total * 100, 2) if mem_total else 0.0
+
+        load = raw_data.get("load", {}) if isinstance(raw_data.get("load", {}), dict) else {}
+        disks = self._normalize_disks(raw_data)
+        nics = self._normalize_nics(raw_data)
+        services = raw_data.get("services", [])
+        if not isinstance(services, list):
+            services = []
+        else:
+            services = self._normalize_services(services)
+
+        normalized = {
+            "ts": now,
+            "status": int(status),
+            "error_msg": str(error_msg or ""),
+            "cpu_usage": cpu_usage,
+            "mem_usage": mem_usage,
+            "mem_used": mem_used,
+            "mem_total": mem_total,
+            "load1": _to_float(load.get("one", load.get("load1", 0))),
+            "load5": _to_float(load.get("five", load.get("load5", 0))),
+            "load15": _to_float(load.get("fifteen", load.get("load15", 0))),
+            "disk_json": _json_dumps(disks),
+            "nic_json": _json_dumps(nics),
+            "service_json": _json_dumps(services),
+            "raw_json": _json_dumps(raw_data),
+        }
+        return normalized
+
+    @staticmethod
+    def _normalize_disks(raw_data: Dict[str, Any]) -> List[Dict[str, Any]]:
+        disk_data = raw_data.get("disk", raw_data.get("disks", []))
+        if not isinstance(disk_data, list):
+            return []
+        result = []
+        for item in disk_data:
+            if not isinstance(item, dict):
+                continue
+            byte_size = item.get("byte_size", [])
+            total = used = available = 0
+            if isinstance(byte_size, (list, tuple)) and len(byte_size) >= 3:
+                total, used, available = _to_int(byte_size[0]), _to_int(byte_size[1]), _to_int(byte_size[2])
+            elif isinstance(item.get("size"), (list, tuple)) and len(item.get("size")) >= 3:
+                size_data = item.get("size")
+                total, used, available = _to_bytes(size_data[0]), _to_bytes(size_data[1]), _to_bytes(size_data[2])
+            else:
+                total = _to_int(item.get("total", item.get("size_total", 0)))
+                used = _to_int(item.get("used", item.get("size_used", 0)))
+                available = _to_int(item.get("available", item.get("free", 0)))
+            usage = _percent(item.get("usage", item.get("d_size", 0)))
+            if not usage and total:
+                usage = round(used / total * 100, 2)
+            inode_usage = 0.0
+            inodes = item.get("inodes", [])
+            if isinstance(inodes, (list, tuple)) and len(inodes) >= 4:
+                inode_usage = _percent(inodes[3])
+            else:
+                inode_usage = _percent(item.get("inode_usage", 0))
+            result.append({
+                "path": item.get("path", item.get("mountpoint", "")),
+                "filesystem": item.get("filesystem", item.get("device", "")),
+                "type": item.get("type", item.get("types", "")),
+                "total": total,
+                "used": used,
+                "available": available,
+                "usage": usage,
+                "inode_usage": inode_usage,
+            })
+        return [item for item in result if item["path"]]
+
+    @staticmethod
+    def _normalize_nics(raw_data: Dict[str, Any]) -> List[Dict[str, Any]]:
+        network = raw_data.get("network", raw_data.get("nics", {}))
+        result = []
+        if isinstance(network, list):
+            items = [(item.get("name", ""), item) for item in network if isinstance(item, dict)]
+        elif isinstance(network, dict):
+            items = [(name, data) for name, data in network.items() if isinstance(data, dict)]
+        else:
+            items = []
+        for name, item in items:
+            if not name or name == "lo":
+                continue
+            result.append({
+                "name": name,
+                "up_kbs": _to_float(item.get("up", item.get("up_kbs", 0))),
+                "down_kbs": _to_float(item.get("down", item.get("down_kbs", 0))),
+                "up_total": _to_int(item.get("upTotal", item.get("up_total", 0))),
+                "down_total": _to_int(item.get("downTotal", item.get("down_total", 0))),
+                "up_packets": _to_int(item.get("upPackets", item.get("up_packets", 0))),
+                "down_packets": _to_int(item.get("downPackets", item.get("down_packets", 0))),
+            })
+        return result
+
+    @staticmethod
+    def _normalize_services(services: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
+        result = []
+        for item in services:
+            if not isinstance(item, dict):
+                continue
+            name = str(item.get("name") or item.get("target_key") or "").strip()
+            if not name:
+                continue
+            status = _to_int(item.get("status", 1 if item.get("running") else 0))
+            result.append({
+                "name": name,
+                "target_key": str(item.get("target_key") or name),
+                "target_name": str(item.get("target_name") or item.get("name") or name),
+                "status": status,
+                "running": bool(item.get("running", status == 1)),
+                "state": str(item.get("state") or ("active" if status == 1 else "inactive")),
+                "error_msg": str(item.get("error_msg") or ""),
+                "ts": _to_int(item.get("ts"), int(time.time())),
+            })
+        return result
+
+    def save_node_snapshot(self, node_id: int, raw_data: Dict[str, Any], status: int = 1, error_msg: str = "") -> str:
+        node_id = int(node_id)
+        if not isinstance(raw_data, dict):
+            raw_data = {}
+        row = self.normalize_monitor_data(raw_data, status=status, error_msg=error_msg)
+        latest_row = {"node_id": node_id}
+        latest_row.update(row)
+        snapshot_row = {"node_id": node_id}
+        snapshot_row.update(row)
+        try:
+            exists = self.table("node_monitor_latest").where("node_id=?", (node_id,)).count() > 0
+            if exists:
+                data = latest_row.copy()
+                data.pop("node_id", None)
+                self.table("node_monitor_latest").where("node_id=?", (node_id,)).update(data)
+            else:
+                self.table("node_monitor_latest").insert(latest_row)
+            self.table("node_monitor_snapshot").insert(snapshot_row)
+            self.sync_discovered_targets(
+                node_id,
+                _json_loads(row["disk_json"], []),
+                _json_loads(row["nic_json"], []),
+            )
+        except Exception as e:
+            return str(e)
+        return ""
+
+    def save_latest_status(self, node_id: int, status: int, error_msg: str = "") -> str:
+        node_id = int(node_id)
+        now = int(time.time())
+        row = {
+            "node_id": node_id,
+            "ts": now,
+            "status": int(status),
+            "error_msg": str(error_msg or ""),
+        }
+        try:
+            exists = self.table("node_monitor_latest").where("node_id=?", (node_id,)).count() > 0
+            if exists:
+                data = row.copy()
+                data.pop("node_id", None)
+                self.table("node_monitor_latest").where("node_id=?", (node_id,)).update(data)
+            else:
+                self.table("node_monitor_latest").insert(row)
+        except Exception as e:
+            return str(e)
+        return ""
+
+    def save_latest_services(self, node_id: int, services: List[Dict[str, Any]], error_msg: str = "") -> str:
+        node_id = int(node_id)
+        service_list = self._normalize_services(services if isinstance(services, list) else [])
+        try:
+            latest = self.get_latest(node_id)
+            raw_json = latest.get("raw_json", {}) if latest else {}
+            if not isinstance(raw_json, dict):
+                raw_json = {}
+            raw_json["services"] = service_list
+            row = {
+                "service_json": _json_dumps(service_list),
+                "raw_json": _json_dumps(raw_json),
+            }
+            if latest:
+                self.table("node_monitor_latest").where("node_id=?", (node_id,)).update(row)
+            else:
+                row.update({
+                    "node_id": node_id,
+                    "ts": int(time.time()),
+                    "status": 2,
+                    "error_msg": str(error_msg or ""),
+                })
+                self.table("node_monitor_latest").insert(row)
+        except Exception as e:
+            return str(e)
+        return ""
+
+    def get_latest(self, node_id: int) -> Dict[str, Any]:
+        row = self.table("node_monitor_latest").where("node_id=?", (int(node_id),)).find()
+        if not isinstance(row, dict) or not row:
+            return {}
+        return self._decode_latest_row(row)
+
+    def _decode_latest_row(self, row: Dict[str, Any]) -> Dict[str, Any]:
+        row = dict(row)
+        row["disk_json"] = _json_loads(row.get("disk_json"), [])
+        row["nic_json"] = _json_loads(row.get("nic_json"), [])
+        row["service_json"] = _json_loads(row.get("service_json"), [])
+        row["raw_json"] = _json_loads(row.get("raw_json"), {})
+        return row
+
+    def get_latest_view(self, node_id: int) -> Dict[str, Any]:
+        latest = self.get_latest(node_id)
+        if not latest:
+            return {}
+        targets = self.get_targets(node_id)
+        target_map = {}
+        for item in targets:
+            target_map.setdefault(item["target_type"], []).append(item)
+
+        latest["selected_disk"] = self._select_primary(latest["disk_json"], target_map.get("disk", []), "path")
+        latest["selected_nic"] = self._select_primary(latest["nic_json"], target_map.get("nic", []), "name")
+        service_targets = [item["target_key"] for item in target_map.get("service", []) if int(item.get("enabled", 0)) == 1]
+        if service_targets:
+            latest["service_json"] = [
+                item for item in latest["service_json"]
+                if isinstance(item, dict) and (item.get("target_key") in service_targets or item.get("name") in service_targets)
+            ]
+        return latest
+
+    def get_list_monitor_map(self, node_ids: List[int]) -> Dict[int, Dict[str, Any]]:
+        node_ids = self._normalize_node_ids(node_ids)
+        if not node_ids:
+            return {}
+
+        setting_list = []
+        latest_list = []
+        target_list = []
+        for chunk_ids in self._chunk_node_ids(node_ids):
+            placeholders = ",".join(["?"] * len(chunk_ids))
+            params = tuple(chunk_ids)
+            settings = self.table("node_monitor_setting").where(
+                "node_id IN ({})".format(placeholders), params
+            ).select()
+            latest = self.table("node_monitor_latest").where(
+                "node_id IN ({})".format(placeholders), params
+            ).select()
+            targets = self.table("node_monitor_target").where(
+                "node_id IN ({})".format(placeholders), params
+            ).order("node_id", "ASC").order("target_type", "ASC").order("sort", "ASC").order("id", "ASC").select()
+            if isinstance(settings, list):
+                setting_list.extend(settings)
+            if isinstance(latest, list):
+                latest_list.extend(latest)
+            if isinstance(targets, list):
+                target_list.extend(targets)
+
+        setting_map = {}
+        if isinstance(setting_list, list):
+            setting_map = {int(item["node_id"]): item for item in setting_list if isinstance(item, dict)}
+
+        latest_map = {}
+        if isinstance(latest_list, list):
+            latest_map = {
+                int(item["node_id"]): self._decode_latest_row(item)
+                for item in latest_list if isinstance(item, dict)
+            }
+
+        target_map = {node_id: {"disk": [], "nic": [], "service": []} for node_id in node_ids}
+        if isinstance(target_list, list):
+            for item in target_list:
+                if not isinstance(item, dict):
+                    continue
+                node_id = int(item.get("node_id", 0))
+                target_type = item.get("target_type")
+                if node_id not in target_map or target_type not in target_map[node_id]:
+                    continue
+                item = dict(item)
+                item["extra"] = _json_loads(item.get("extra"), {})
+                target_map[node_id][target_type].append(item)
+
+        return {
+            node_id: self._build_list_monitor_item(
+                node_id,
+                setting_map.get(node_id) or self._default_setting_row(node_id),
+                latest_map.get(node_id) or {},
+                target_map.get(node_id) or {"disk": [], "nic": [], "service": []},
+            )
+            for node_id in node_ids
+        }
+
+    @staticmethod
+    def _normalize_node_ids(node_ids: List[int]) -> List[int]:
+        result = []
+        seen = set()
+        for node_id in node_ids:
+            try:
+                node_id = int(node_id)
+            except Exception:
+                continue
+            if node_id <= 0 or node_id in seen:
+                continue
+            seen.add(node_id)
+            result.append(node_id)
+        return result
+
+    @staticmethod
+    def _chunk_node_ids(node_ids: List[int], chunk_size: int = 500) -> List[List[int]]:
+        return [node_ids[idx:idx + chunk_size] for idx in range(0, len(node_ids), chunk_size)]
+
+    @staticmethod
+    def _default_setting_row(node_id: int) -> Dict[str, Any]:
+        now = int(time.time())
+        return {
+            "id": 0,
+            "node_id": int(node_id),
+            "enabled": 1,
+            "collect_interval": 60,
+            "status_interval": 60,
+            "detail_interval": 60,
+            "snapshot_retention_days": DEFAULT_SNAPSHOT_RETENTION_DAYS,
+            "cost_amount": 0,
+            "cost_currency": "CNY",
+            "cost_symbol": "",
+            "cost_period": "month",
+            "expire_at": 0,
+            "remark": "",
+            "created_at": now,
+            "updated_at": now,
+        }
+
+    def _build_list_monitor_item(
+            self,
+            node_id: int,
+            setting: Dict[str, Any],
+            latest: Dict[str, Any],
+            targets: Dict[str, List[Dict[str, Any]]],
+    ) -> Dict[str, Any]:
+        disks = latest.get("disk_json", []) if isinstance(latest, dict) else []
+        nics = latest.get("nic_json", []) if isinstance(latest, dict) else []
+        services = latest.get("service_json", []) if isinstance(latest, dict) else []
+        if not isinstance(disks, list):
+            disks = []
+        if not isinstance(nics, list):
+            nics = []
+        if not isinstance(services, list):
+            services = []
+
+        selected_disk = self._select_primary(disks, targets.get("disk", []), "path")
+        selected_nic = self._select_primary(nics, targets.get("nic", []), "name")
+        service_targets = [
+            item.get("target_key")
+            for item in targets.get("service", [])
+            if int(item.get("enabled", 0)) == 1 and item.get("target_key")
+        ]
+        if service_targets:
+            service_key_set = set(service_targets)
+            services = [
+                item for item in services
+                if isinstance(item, dict) and (item.get("target_key") in service_key_set or item.get("name") in service_key_set)
+            ]
+        else:
+            services = []
+
+        summary = {
+            "ts": _to_int(latest.get("ts", 0)) if latest else 0,
+            "status": _to_int(latest.get("status", 0)) if latest else 0,
+            "error_msg": str(latest.get("error_msg", "")) if latest else "",
+            "cpu_usage": _percent(latest.get("cpu_usage", 0)) if latest else 0,
+            "mem_usage": _percent(latest.get("mem_usage", 0)) if latest else 0,
+            "mem_used": _to_int(latest.get("mem_used", 0)) if latest else 0,
+            "mem_total": _to_int(latest.get("mem_total", 0)) if latest else 0,
+            "load1": _to_float(latest.get("load1", 0)) if latest else 0,
+            "load5": _to_float(latest.get("load5", 0)) if latest else 0,
+            "load15": _to_float(latest.get("load15", 0)) if latest else 0,
+        }
+        latest_view = dict(summary)
+        latest_view.update({
+            "selected_disk": selected_disk,
+            "selected_nic": selected_nic,
+            "services": services,
+        })
+
+        return {
+            "enabled": int(setting.get("enabled", 1)),
+            "setting": setting,
+            "summary": summary,
+            "latest": latest_view,
+            "selected_disk": selected_disk,
+            "selected_nic": selected_nic,
+            "disks": disks,
+            "nics": nics,
+            "services": services,
+            "targets": targets,
+            "cost": {
+                "amount": _to_float(setting.get("cost_amount", 0)),
+                "currency": str(setting.get("cost_currency", "")),
+                "symbol": str(setting.get("cost_symbol", "")),
+                "period": str(setting.get("cost_period", "")),
+            },
+            "expire": self._build_expire_info(setting.get("expire_at", 0)),
+            "has_latest": bool(latest),
+        }
+
+    @staticmethod
+    def _build_expire_info(expire_at: Any) -> Dict[str, Any]:
+        expire_at = _to_int(expire_at)
+        if expire_at <= 0:
+            return {
+                "expire_at": 0,
+                "days_left": None,
+                "status": "none",
+                "expired": False,
+                "is_soon": False,
+            }
+        now = int(time.time())
+        seconds_left = expire_at - now
+        if seconds_left >= 0:
+            days_left = seconds_left // 86400
+            if seconds_left % 86400:
+                days_left += 1
+        else:
+            days_left = -((-seconds_left) // 86400)
+            if seconds_left % 86400:
+                days_left -= 1
+        expired = seconds_left < 0
+        is_soon = (not expired) and seconds_left <= 7 * 86400
+        return {
+            "expire_at": expire_at,
+            "days_left": int(days_left),
+            "status": "expired" if expired else ("soon" if is_soon else "normal"),
+            "expired": expired,
+            "is_soon": is_soon,
+        }
+
+    @staticmethod
+    def _select_primary(items: List[Dict[str, Any]], targets: List[Dict[str, Any]], item_key: str) -> Dict[str, Any]:
+        if not isinstance(items, list) or not items:
+            return {}
+        primary_key = ""
+        for target in targets:
+            if int(target.get("is_primary", 0)) == 1:
+                primary_key = target.get("target_key", "")
+                break
+        if primary_key:
+            for item in items:
+                if isinstance(item, dict) and item.get(item_key) == primary_key:
+                    return item
+        return items[0] if isinstance(items[0], dict) else {}
+
+    def get_snapshots(self, node_id: int, start_ts: int = 0, end_ts: int = 0, limit: int = 1440) -> List[Dict[str, Any]]:
+        where = ["node_id=?"]
+        params = [int(node_id)]
+        if start_ts:
+            where.append("ts>=?")
+            params.append(int(start_ts))
+        if end_ts:
+            where.append("ts<=?")
+            params.append(int(end_ts))
+        limit = max(1, min(int(limit), 10080))
+        data = self.table("node_monitor_snapshot").where(" AND ".join(where), tuple(params)).order("ts", "DESC").limit(limit).select()
+        if not isinstance(data, list):
+            return []
+        result = [self._decode_latest_row(item) for item in data if isinstance(item, dict)]
+        result.reverse()
+        return result
+
+    def cleanup_snapshots(self, before_ts: int) -> str:
+        try:
+            self.table("node_monitor_snapshot").where("ts<?", (int(before_ts),)).delete()
+        except Exception as e:
+            return str(e)
+        return ""
+
+    def cleanup_expired_snapshots(self) -> str:
+        now = int(time.time())
+        try:
+            settings = self.table("node_monitor_setting").field("node_id,snapshot_retention_days").select()
+            if not isinstance(settings, list) or not settings:
+                return self.cleanup_snapshots(now - DEFAULT_SNAPSHOT_RETENTION_DAYS * 86400)
+            cleaned_nodes = set()
+            for item in settings:
+                if not isinstance(item, dict):
+                    continue
+                node_id = _to_int(item.get("node_id", 0))
+                if node_id <= 0 or node_id in cleaned_nodes:
+                    continue
+                cleaned_nodes.add(node_id)
+                retention_days = _normalize_retention_days(item.get("snapshot_retention_days"))
+                before_ts = now - retention_days * 86400
+                self.table("node_monitor_snapshot").where("node_id=? AND ts<?", (node_id, before_ts)).delete()
+        except Exception as e:
+            return str(e)
+        return ""

mod/project/node/monitorMod.py

@@ -0,0 +1,421 @@
+# coding: utf-8
+import json
+import time
+from datetime import datetime
+from typing import Any, Dict, List
+
+import public
+from mod.project.node.dbutil import NodeMonitorDB, ServerMonitorRepo, ServerNodeDB
+from mod.project.node.nodeutil.monitor_service import (
+    build_service_options_from_file_exists,
+    collect_local_service_status,
+    discover_local_services,
+    get_service_file_probe_paths,
+    keep_single_version_services,
+    normalize_service_targets,
+    service_unknown_statuses,
+)
+
+
+def _json_loads(value: Any, default: Any) -> Any:
+    if isinstance(value, (dict, list)):
+        return value
+    if not isinstance(value, str) or not value:
+        return default
+    try:
+        return json.loads(value)
+    except Exception:
+        return default
+
+
+def _get_int(get, key: str, default: int = 0) -> int:
+    try:
+        return int(get.get("{}/d".format(key), get.get(key, default)))
+    except Exception:
+        return default
+
+
+def _get_str(get, key: str, default: str = "") -> str:
+    try:
+        value = get.get("{}/s".format(key), get.get(key, default))
+    except Exception:
+        value = default
+    return str(value if value is not None else default)
+
+
+def _get_raw(get, key: str, default: Any = None) -> Any:
+    try:
+        return get.get(key, default)
+    except Exception:
+        return default
+
+
+def _parse_expire_at(value: Any) -> int:
+    if value in (None, ""):
+        return 0
+    try:
+        return int(float(value))
+    except Exception:
+        pass
+    value = str(value).strip()
+    for fmt in ("%Y-%m-%d %H:%M:%S", "%Y-%m-%d"):
+        try:
+            return int(time.mktime(datetime.strptime(value, fmt).timetuple()))
+        except Exception:
+            continue
+    return 0
+
+
+def _target_groups(targets: List[Dict[str, Any]]) -> Dict[str, List[Dict[str, Any]]]:
+    data = {"disk": [], "nic": [], "service": []}
+    for item in targets:
+        target_type = item.get("target_type")
+        if target_type in data:
+            data[target_type].append(item)
+    return data
+
+
+def _target_map(targets: List[Dict[str, Any]]) -> Dict[str, Dict[str, Any]]:
+    return {item["target_key"]: item for item in targets if isinstance(item, dict) and item.get("target_key")}
+
+
+def _option_from_item(item: Dict[str, Any], key_name: str, target: Dict[str, Any] = None) -> Dict[str, Any]:
+    target = target or {}
+    key = str(item.get(key_name, ""))
+    data = dict(item)
+    data.update({
+        "target_key": key,
+        "target_name": target.get("target_name") or item.get("name") or key,
+        "enabled": int(target.get("enabled", 1)),
+        "is_primary": int(target.get("is_primary", 0)),
+        "sort": int(target.get("sort", 0)),
+    })
+    return data
+
+
+class main:
+    def __init__(self):
+        self.monitor_db = NodeMonitorDB()
+        self.monitor_db.init_db()
+        self.node_db = ServerNodeDB()
+
+    def _get_node(self, get):
+        node_id = _get_int(get, "node_id")
+        if not node_id:
+            return 0, None, "node_id cannot be empty"
+        node = self.node_db.get_node_by_id(node_id)
+        if not node:
+            return node_id, None, "Node does not exist"
+        if node.get("lpver"):
+            return node_id, None, "1Panel node is not supported"
+        return node_id, node, ""
+
+    def _ensure_latest_from_cache(self, node_id: int, node: Dict[str, Any]):
+        latest = self.monitor_db.get_latest_view(node_id)
+        if latest:
+            return latest
+        try:
+            if node["app_key"] == "local" and node["api_key"] == "local":
+                data = ServerMonitorRepo.get_local_server_status()
+            else:
+                data = ServerMonitorRepo().get_server_status(node_id)
+            if isinstance(data, dict) and data:
+                self.monitor_db.save_node_snapshot(node_id, data, status=1)
+        except Exception:
+            if public.is_debug():
+                public.print_error()
+        return self.monitor_db.get_latest_view(node_id)
+
+    def _build_monitor_data(self, node_id: int, node: Dict[str, Any]) -> Dict[str, Any]:
+        setting = self.monitor_db.get_setting(node_id)
+        latest = self._ensure_latest_from_cache(node_id, node)
+        targets = self.monitor_db.get_targets(node_id)
+        groups = _target_groups(targets)
+        disk_targets = _target_map(groups["disk"])
+        nic_targets = _target_map(groups["nic"])
+
+        disks = []
+        nics = []
+        services = []
+        if latest:
+            for item in latest.get("disk_json", []):
+                if isinstance(item, dict) and item.get("path"):
+                    disks.append(_option_from_item(item, "path", disk_targets.get(item.get("path"))))
+            for item in latest.get("nic_json", []):
+                if isinstance(item, dict) and item.get("name"):
+                    nics.append(_option_from_item(item, "name", nic_targets.get(item.get("name"))))
+
+        latest_services = {}
+        if latest:
+            for item in latest.get("service_json", []):
+                if isinstance(item, dict):
+                    latest_services[item.get("target_key") or item.get("name")] = item
+        for item in groups["service"]:
+            service_item = dict(item)
+            service_item["latest"] = latest_services.get(item.get("target_key"), {})
+            services.append(service_item)
+
+        return {
+            "setting": setting,
+            "latest": latest,
+            "targets": groups,
+            "options": {
+                "disks": disks,
+                "nics": nics,
+                "services": services,
+                "service_options": [],
+                "service_groups": [],
+            },
+        }
+
+    def get_node_monitor(self, get):
+        node_id, node, err = self._get_node(get)
+        if err:
+            return public.return_message(-1, 0, err)
+        return public.return_message(0, 0, self._build_monitor_data(node_id, node))
+
+    def get_monitor_options(self, get):
+        return self.get_node_monitor(get)
+
+    def get_setting(self, get):
+        node_id, node, err = self._get_node(get)
+        if err:
+            return public.return_message(-1, 0, err)
+        return public.return_message(0, 0, self.monitor_db.get_setting(node_id))
+
+    def save_setting(self, get):
+        node_id, node, err = self._get_node(get)
+        if err:
+            return public.return_message(-1, 0, err)
+
+        data = {}
+        if _get_str(get, "enabled", "") != "":
+            data["enabled"] = 1 if _get_int(get, "enabled", 1) else 0
+        for key in ("collect_interval", "status_interval", "detail_interval"):
+            value = _get_int(get, key, 0)
+            if value:
+                data[key] = max(value, 60)
+        if _get_str(get, "snapshot_retention_days", "") != "":
+            retention_days = _get_int(get, "snapshot_retention_days", 7)
+            if retention_days < 1:
+                return public.return_message(-1, 0, "snapshot_retention_days must be greater than 0")
+            data["snapshot_retention_days"] = min(retention_days, 3650)
+        if _get_str(get, "cost_amount", "") != "":
+            try:
+                data["cost_amount"] = max(float(_get_str(get, "cost_amount", "0")), 0)
+            except Exception:
+                return public.return_message(-1, 0, "cost_amount format error")
+        for key, max_len in (("cost_currency", 16), ("cost_symbol", 16), ("cost_period", 32), ("remark", 500)):
+            value = _get_str(get, key, "")
+            if value != "":
+                data[key] = value[:max_len]
+        expire_value = _get_str(get, "expire_at", "")
+        if expire_value != "":
+            data["expire_at"] = _parse_expire_at(expire_value)
+
+        err = self.monitor_db.save_setting(node_id, data)
+        if err:
+            return public.return_message(-1, 0, err)
+        return public.return_message(0, 0, self.monitor_db.get_setting(node_id))
+
+    def get_latest(self, get):
+        node_id, node, err = self._get_node(get)
+        if err:
+            return public.return_message(-1, 0, err)
+        return public.return_message(0, 0, self._ensure_latest_from_cache(node_id, node))
+
+    def get_snapshots(self, get):
+        node_id, node, err = self._get_node(get)
+        if err:
+            return public.return_message(-1, 0, err)
+        start_ts = _get_int(get, "start_ts", 0)
+        end_ts = _get_int(get, "end_ts", 0)
+        limit = _get_int(get, "limit", 1440)
+        return public.return_message(0, 0, self.monitor_db.get_snapshots(node_id, start_ts, end_ts, limit))
+
+    def set_primary_target(self, get):
+        node_id, node, err = self._get_node(get)
+        if err:
+            return public.return_message(-1, 0, err)
+        target_type = _get_str(get, "target_type", "").strip()
+        target_key = _get_str(get, "target_key", "").strip()
+        if target_type not in ("disk", "nic"):
+            return public.return_message(-1, 0, "target_type must be disk or nic")
+        if not target_key:
+            return public.return_message(-1, 0, "target_key cannot be empty")
+        err = self.monitor_db.set_primary_target(node_id, target_type, target_key)
+        if err:
+            return public.return_message(-1, 0, err)
+        return public.return_message(0, 0, "Set successfully")
+
+    def save_service_targets(self, get):
+        node_id, node, err = self._get_node(get)
+        if err:
+            return public.return_message(-1, 0, err)
+        raw_services = _get_raw(get, "services", None)
+        if raw_services is None:
+            raw_services = _get_str(get, "services", "[]")
+        services = _json_loads(raw_services, [])
+        if not isinstance(services, list):
+            return public.return_message(-1, 0, "services format error")
+        service_list = keep_single_version_services(services)
+        err = self.monitor_db.save_service_targets(node_id, service_list)
+        if err:
+            return public.return_message(-1, 0, err)
+        self._refresh_latest_service_status(node_id, node, service_list)
+        # return public.return_message(0, 0, self._build_monitor_data(node_id, node))
+        return public.return_message(0, 0, "Set successfully")
+
+    def _refresh_latest_service_status(self, node_id: int, node: Dict[str, Any], services: List[Dict[str, Any]]):
+        service_list = [item for item in normalize_service_targets(services) if int(item.get("enabled", 1)) == 1]
+        if not service_list:
+            self.monitor_db.save_latest_services(node_id, [])
+            return
+        service_data, err = self._get_node_service_status(node, service_list)
+        if err and not service_data:
+            service_data = service_unknown_statuses(service_list, err)
+        save_err = self.monitor_db.save_latest_services(node_id, service_data, err)
+        if save_err and public.is_debug():
+            public.print_log("Node monitor latest services save failed: {}".format(save_err))
+
+    @staticmethod
+    def _get_node_service_status(node: Dict[str, Any], services: List[Dict[str, Any]]):
+        if node["app_key"] == "local" and node["api_key"] == "local":
+            return collect_local_service_status(services), ""
+        ssh_conf = _json_loads(node.get("ssh_conf"), {})
+        api_err = ""
+        if node.get("api_key") or node.get("app_key"):
+            from mod.project.node.nodeutil import ServerNode
+            srv = ServerNode.new_by_data(dict(node))
+            if not srv:
+                return [], "Node does not exist"
+            service_data, api_err = srv.get_service_status(services)
+            if not api_err:
+                return service_data, ""
+
+        if ssh_conf:
+            try:
+                from mod.project.node.nodeutil.ssh_wrap import SSHApi
+                return SSHApi(**ssh_conf).get_service_status(services)
+            except Exception as e:
+                if api_err:
+                    return [], "{}; SSH fallback failed: {}".format(api_err, str(e))
+                return [], str(e)
+
+        if api_err:
+            return [], main._format_api_service_status_error(api_err)
+        return [], "Service status check is not supported"
+
+    @staticmethod
+    def _format_api_service_status_error(error_msg: str) -> str:
+        if "status code is 404" in error_msg or "status code is incorrect" in error_msg:
+            return (
+                "The remote API node does not support service status monitoring. "
+                "Please update the target panel to a version containing node monitor service APIs, "
+                "or configure SSH for this node. Original error: {}".format(error_msg)
+            )
+        return error_msg
+
+    def get_service_options(self, get):
+        node_id = _get_int(get, "node_id", 0)
+        if not node_id:
+            return public.return_message(0, 0, self._with_service_capability(
+                discover_local_services(), "local", True
+            ))
+        node_id, node, err = self._get_node(get)
+        if err:
+            return public.return_message(-1, 0, err)
+        data, err = self._get_node_service_options(node)
+        if err:
+            return public.return_message(-1, 0, err)
+        return public.return_message(0, 0, data)
+
+    def get_default_services(self, get):
+        return self.get_service_options(get)
+
+    @staticmethod
+    def _get_node_service_options(node: Dict[str, Any]):
+        if node["app_key"] == "local" and node["api_key"] == "local":
+            return main._with_service_capability(discover_local_services(), "local", True), ""
+        ssh_conf = _json_loads(node.get("ssh_conf"), {})
+        if node.get("api_key") or node.get("app_key"):
+            from mod.project.node.nodeutil import ServerNode
+            srv = ServerNode.new_by_data(dict(node))
+            if not srv:
+                return {"services": [], "groups": []}, "Node does not exist"
+            data, err = srv.get_service_options()
+            if not err:
+                return main._with_service_capability(data, "api", True), ""
+
+            api_err = err
+            if ssh_conf:
+                data, ssh_err = main._get_service_options_by_ssh(ssh_conf)
+                if not ssh_err:
+                    return main._with_service_capability(data, "ssh", True), ""
+            else:
+                ssh_err = ""
+
+            data, probe_err = main._get_service_options_by_api_file_probe(srv)
+            if not probe_err:
+                return main._with_service_capability(
+                    data, "api_file_probe", False, main._format_api_service_status_error(api_err)
+                ), ""
+
+            err_list = [item for item in (api_err, ssh_err, probe_err) if item]
+            return {"services": [], "groups": []}, "; ".join(err_list) or "Node does not support service discovery"
+
+        if ssh_conf:
+            data, err = main._get_service_options_by_ssh(ssh_conf)
+            if not err:
+                data = main._with_service_capability(data, "ssh", True)
+            return data, err
+        return {"services": [], "groups": []}, "Node does not support service discovery"
+
+    @staticmethod
+    def _with_service_capability(data: Dict[str, Any], source: str, service_status: bool, message: str = ""):
+        if not isinstance(data, dict):
+            data = {"services": [], "groups": []}
+        data["capability"] = {
+            "source": source,
+            "service_status": bool(service_status),
+            "need_upgrade": source == "api_file_probe" and not service_status,
+            "message": message,
+        }
+        return data
+
+    @staticmethod
+    def _get_service_options_by_ssh(ssh_conf: Dict[str, Any]):
+        try:
+            from mod.project.node.nodeutil.ssh_wrap import SSHApi
+            srv = SSHApi(**ssh_conf)
+            return srv.get_service_options()
+        except Exception as e:
+            return {"services": [], "groups": []}, str(e)
+
+    @staticmethod
+    def _get_service_options_by_api_file_probe(srv):
+        if not hasattr(srv, "upload_check"):
+            return {"services": [], "groups": []}, "Node API file probe is not supported"
+        files, err = srv.upload_check(get_service_file_probe_paths())
+        if err:
+            return {"services": [], "groups": []}, err
+
+        existing_paths = []
+        if isinstance(files, list):
+            for item in files:
+                if not isinstance(item, dict) or not item.get("exists"):
+                    continue
+                filename = item.get("filename") or item.get("path")
+                if filename:
+                    existing_paths.append(str(filename))
+        return build_service_options_from_file_exists(existing_paths), ""
+
+    def service_status(self, get):
+        raw_services = _get_raw(get, "services", None)
+        if raw_services is None:
+            raw_services = _get_str(get, "services", "[]")
+        services = _json_loads(raw_services, [])
+        service_list = normalize_service_targets(services)
+        return public.return_message(0, 0, {
+            "services": collect_local_service_status(service_list),
+        })

mod/project/node/nodeMod.py

@@ -16,7 +16,7 @@ except:
     public.ExecShell(cmd_h11)
     public.ExecShell(cmd_wsproto)
     from mod.project.node.nodeutil import ServerNode, LocalNode, monitor_node_once_with_timeout
-from mod.project.node.dbutil import Node, ServerNodeDB, ServerMonitorRepo
+from mod.project.node.dbutil import Node, ServerNodeDB, ServerMonitorRepo, NodeMonitorDB
 from mod.project.node.task_flow import flow_useful_version
 
 
@@ -175,6 +175,9 @@ class main():
         search = get.get('search', "").strip()
         category_id = get.get('category_id/d', -1)
         refresh = get.get('refresh/s', "")
+        force_refresh = get.get('force_refresh/s', "")
+        refresh_node_ids = self._parse_id_list(get.get('refresh_node_ids', get.get('node_ids', "")))
+        refresh_timeout = min(max(int(get.get('refresh_timeout/d', 5)), 1), 30)
         show_mode = get.get('show_mode/s', "")
         if not show_mode or show_mode not in ["list", "block"]:
             show_mode = self.default_show_mode()
@@ -191,15 +194,18 @@ class main():
         if err:
             return public.return_message(-1, 0, err)
 
-        if refresh and refresh == "1":
-            th_list = []
-            for node in data:
-                th = threading.Thread(target=monitor_node_once_with_timeout, args=(node,5))
-                th.start()
-                th_list.append(th)
+        if self._is_true(refresh) or self._is_true(force_refresh) or refresh_node_ids:
+            self._refresh_monitor_nodes(srv_db, data, refresh_node_ids, refresh_timeout)
+            data, err = srv_db.get_node_list(search, category_id, (page_num - 1) * limit, limit)
+            if err:
+                return public.return_message(-1, 0, err)
 
-            for th in th_list:
-                th.join()
+        monitor_map = {}
+        try:
+            monitor_map = NodeMonitorDB().get_list_monitor_map([int(node.get("id", 0)) for node in data])
+        except Exception:
+            if public.is_debug():
+                public.print_error()
 
         for node in data:
             if isinstance(node["ssh_conf"], str):
@@ -212,12 +218,63 @@ class main():
                 node["remarks"] = node["remarks"] + " | 1Panel"
             node_data = self.get_node_data(node)
             node['data'] = node_data
+            node['monitor'] = monitor_map.get(int(node.get("id", 0)), {})
         count = srv_db.node_count(search, category_id)
         page = public.get_page(count, page_num, limit)
         page["data"] = data
         page["show_mode"] = show_mode
         return public.return_message(0, 0,page)
 
+    @staticmethod
+    def _is_true(value):
+        return str(value).lower() in ("1", "true", "yes", "on")
+
+    @staticmethod
+    def _parse_id_list(value):
+        if value in ("", None):
+            return []
+        if isinstance(value, int):
+            return [value]
+        if isinstance(value, list):
+            return [int(item) for item in value if str(item).strip().isdigit()]
+        if isinstance(value, str):
+            value = value.strip()
+            if not value:
+                return []
+            try:
+                data = json.loads(value)
+                if isinstance(data, list):
+                    return [int(item) for item in data if str(item).strip().isdigit()]
+                if isinstance(data, int):
+                    return [data]
+            except Exception:
+                pass
+            return [int(item.strip()) for item in value.split(",") if item.strip().isdigit()]
+        return []
+
+    @staticmethod
+    def _refresh_monitor_nodes(srv_db, page_nodes, refresh_node_ids, timeout):
+        if refresh_node_ids:
+            nodes = []
+            for node_id in refresh_node_ids:
+                node = srv_db.get_node_by_id(int(node_id))
+                if node:
+                    nodes.append(node)
+        else:
+            nodes = page_nodes
+        if not nodes:
+            return
+
+        max_workers = 10
+        for idx in range(0, len(nodes), max_workers):
+            th_list = []
+            for node in nodes[idx:idx + max_workers]:
+                th = threading.Thread(target=monitor_node_once_with_timeout, args=(dict(node), timeout))
+                th.start()
+                th_list.append(th)
+            for th in th_list:
+                th.join()
+
     @staticmethod
     def get_node_data(node: dict):
         if node["app_key"] == "local" and node["api_key"] == "local":
@@ -558,4 +615,4 @@ class main():
         if ret.get("status"):
             return public.return_message(0, 0, ret.get("msg"))
         else:
-            return public.return_message(-1, 0, ret.get("msg"))
+            return public.return_message(-1, 0, ret.get("msg"))

mod/project/node/nodeutil/base.py

@@ -261,6 +261,34 @@ class ServerNode:
             return data, ""
         return None, "data in wrong format: %s" % str(data)
 
+    def get_service_status(self, services) -> Tuple[List[dict], str]:
+        from mod.project.node.nodeutil.monitor_service import normalize_service_targets
+        service_list = [item for item in normalize_service_targets(services) if int(item.get("enabled", 1)) == 1]
+        if not service_list:
+            return [], ""
+        data, err = self._request("/mod/node/monitor/service_status", "", pdata={
+            "services": json.dumps(service_list),
+        })
+        if err:
+            return [], err
+        if isinstance(data, dict) and isinstance(data.get("services"), list):
+            return data["services"], ""
+        if isinstance(data, list):
+            return data, ""
+        if isinstance(data, str):
+            return [], data
+        return [], "data in wrong format: %s" % str(data)
+
+    def get_service_options(self) -> Tuple[Dict[str, list], str]:
+        data, err = self._request("/mod/node/monitor/get_service_options", "", pdata={})
+        if err:
+            return {"services": [], "groups": []}, err
+        if isinstance(data, dict) and isinstance(data.get("services"), list):
+            if not isinstance(data.get("groups"), list):
+                data["groups"] = []
+            return data, ""
+        return {"services": [], "groups": []}, "data in wrong format: %s" % str(data)
+
     def get_tmp_token(self) -> Tuple[str, str]:
         data, err = self._request("/config", "get_tmp_token")
         if err:
@@ -1047,6 +1075,107 @@ class ServerNode:
         return ""
 
 
+def _monitor_setting_enabled(monitor_db, node_id: int) -> bool:
+    try:
+        setting = monitor_db.get_setting(node_id, create=True)
+        return int(setting.get("enabled", 1)) == 1
+    except Exception:
+        if public.is_debug():
+            public.print_error()
+        return False
+
+
+def _monitor_should_save_snapshot(monitor_db, node_id: int) -> bool:
+    try:
+        setting = monitor_db.get_setting(node_id, create=True)
+        if int(setting.get("enabled", 1)) != 1:
+            return False
+        interval = max(int(setting.get("collect_interval", 60)), 60)
+        latest = monitor_db.get_latest(node_id)
+        if latest and int(time.time()) - int(latest.get("ts", 0)) < interval:
+            return False
+        return True
+    except Exception:
+        if public.is_debug():
+            public.print_error()
+        return False
+
+
+def _attach_monitor_service_data(monitor_db, node_id: int, srv, data: dict) -> dict:
+    if not isinstance(data, dict):
+        return data
+    services = []
+    try:
+        services = monitor_db.get_enabled_service_targets(node_id)
+        if not services:
+            data["services"] = []
+            return data
+
+        from mod.project.node.nodeutil.monitor_service import collect_local_service_status, service_unknown_statuses
+        if getattr(srv, "is_local", False):
+            service_data = collect_local_service_status(services)
+        elif hasattr(srv, "get_service_status"):
+            service_data, err = srv.get_service_status(services)
+            if err and not service_data:
+                service_data, err = _get_monitor_service_status_by_saved_ssh(node_id, services)
+            if err and not service_data:
+                service_data = service_unknown_statuses(services, err)
+        else:
+            service_data = service_unknown_statuses(services, "Service status check is not supported")
+        data["services"] = service_data
+    except Exception as e:
+        from mod.project.node.nodeutil.monitor_service import service_unknown_statuses
+        data["services"] = service_unknown_statuses(services, str(e))
+        if public.is_debug():
+            public.print_error()
+    return data
+
+
+def _get_monitor_service_status_by_saved_ssh(node_id: int, services: List[Dict[str, Any]]) -> Tuple[List[dict], str]:
+    try:
+        node_data = public.S('node', public.get_panel_path() + "/data/db/node.db").where("id=?", (int(node_id),)).find()
+        if not isinstance(node_data, dict) or not node_data:
+            return [], "Node does not exist"
+        ssh_conf = node_data.get("ssh_conf", {})
+        if isinstance(ssh_conf, str):
+            ssh_conf = json.loads(ssh_conf or "{}")
+        if not isinstance(ssh_conf, dict) or not ssh_conf:
+            return [], "Service status check is not supported by the remote node"
+        from mod.project.node.nodeutil.ssh_wrap import SSHApi
+        return SSHApi(**ssh_conf).get_service_status(services)
+    except Exception as e:
+        return [], str(e)
+
+
+def _save_monitor_snapshot(node_id: int, srv, data: dict):
+    try:
+        from mod.project.node.dbutil import NodeMonitorDB
+        monitor_db = NodeMonitorDB()
+        if not _monitor_should_save_snapshot(monitor_db, node_id):
+            return
+        data = _attach_monitor_service_data(monitor_db, node_id, srv, data)
+        err = monitor_db.save_node_snapshot(node_id, data, status=1)
+        if err and public.is_debug():
+            public.print_log("Node monitor snapshot save failed: {}".format(err))
+    except Exception:
+        if public.is_debug():
+            public.print_error()
+
+
+def _save_monitor_status(node_id: int, status: int, error_msg: str):
+    try:
+        from mod.project.node.dbutil import NodeMonitorDB
+        monitor_db = NodeMonitorDB()
+        if not _monitor_setting_enabled(monitor_db, node_id):
+            return
+        err = monitor_db.save_latest_status(node_id, status, error_msg)
+        if err and public.is_debug():
+            public.print_log("Node monitor latest status save failed: {}".format(err))
+    except Exception:
+        if public.is_debug():
+            public.print_error()
+
+
 def monitor_all_node_status():
     all_nodes = public.S('node', public.get_panel_path() + "/data/db/node.db").select()
     if not isinstance(all_nodes, list) or not all_nodes:
@@ -1063,6 +1192,14 @@ def monitor_all_node_status():
     # 等待所有线程完成（可选）
     for t in threads:
         t.join()
+    try:
+        from mod.project.node.dbutil import NodeMonitorDB
+        err = NodeMonitorDB().cleanup_expired_snapshots()
+        if err and public.is_debug():
+            public.print_log("Node monitor snapshot cleanup failed: {}".format(err))
+    except Exception:
+        if public.is_debug():
+            public.print_error()
 
 
 def monitor_node_once(node_data: dict):
@@ -1071,6 +1208,11 @@ def monitor_node_once(node_data: dict):
     from mod.base.ssh_executor import test_ssh_config
     try:
         if node_data["app_key"] == "local" and node_data["api_key"] == "local":
+            data = ServerMonitorRepo.get_local_server_status()
+            if data:
+                _save_monitor_snapshot(int(node_data.get("id", 0)), LocalNode(), data)
+            else:
+                _save_monitor_status(int(node_data.get("id", 0)), 2, "Failed to obtain local node data")
             return
         node_data["error"] = json.loads(node_data["error"])
         node_data["ssh_conf"] = json.loads(node_data["ssh_conf"])
@@ -1098,6 +1240,7 @@ def monitor_node_once(node_data: dict):
                 ServerMonitorRepo().remove_cache(node.id)
 
             ServerNodeDB().update_node(node)
+            _save_monitor_status(node.id, 0 if node.error_num >= 2 else 2, err)
 
         else:
             if node.error_num > 0:
@@ -1109,6 +1252,7 @@ def monitor_node_once(node_data: dict):
         if data:
             repo = ServerMonitorRepo()
             repo.save_server_status(node.id, data)
+            _save_monitor_snapshot(node.id, srv, data)
             if repo.is_reboot_wait(node.server_ip):
                 repo.set_wait_reboot(node.server_ip, False)
         if not node_data["ssh_test"] and not node_data["ssh_conf"] and not isinstance(srv, SSHApi):
@@ -1144,6 +1288,12 @@ def monitor_node_once(node_data: dict):
 
 def monitor_node_once_with_timeout(node_data: dict, timeout: int = 5):
     if node_data["app_key"] == "local" and node_data["api_key"] == "local":
+        from mod.project.node.dbutil import ServerMonitorRepo
+        data = ServerMonitorRepo.get_local_server_status()
+        if data:
+            _save_monitor_snapshot(int(node_data.get("id", 0)), LocalNode(), data)
+        else:
+            _save_monitor_status(int(node_data.get("id", 0)), 2, "Failed to obtain local node data")
         return
     from mod.project.node.dbutil import Node, ServerNodeDB, ServerMonitorRepo
     from mod.project.node.nodeutil.ssh_wrap import SSHApi
@@ -1176,6 +1326,7 @@ def monitor_node_once_with_timeout(node_data: dict, timeout: int = 5):
                 ServerMonitorRepo().remove_cache(node.id)
 
             ServerNodeDB().update_node(node)
+            _save_monitor_status(node.id, 0 if node.error_num >= 2 else 2, err)
 
         elif node.error_num > 0:
             node.error_num = 0
@@ -1186,6 +1337,7 @@ def monitor_node_once_with_timeout(node_data: dict, timeout: int = 5):
         if data:
             repo = ServerMonitorRepo()
             repo.save_server_status(node.id, data)
+            _save_monitor_snapshot(node.id, srv, data)
             if repo.is_reboot_wait(node.server_ip):
                 repo.set_wait_reboot(node.server_ip, False)
     except Exception as e:
@@ -1958,4 +2110,4 @@ API Response Status: 200
         key, err = self.lpc.get_file_body("/root/.ssh/id_ed25519_1panel")
         if err:
             return None, "Failed to read SSH key, error message is: %s" % err
-        return key, ""
+        return key, ""

mod/project/node/nodeutil/monitor_service.py

@@ -0,0 +1,399 @@
+# coding: utf-8
+import os
+import re
+import time
+from typing import Any, Dict, List, Tuple
+
+import public
+
+
+SERVICE_NAME_RE = re.compile(r"^[A-Za-z0-9_.@:+-]+$")
+
+DEFAULT_SERVICES = [
+    {"target_key": "nginx", "target_name": "nginx"},
+    {"target_key": "httpd", "target_name": "apache"},
+    {"target_key": "mysqld", "target_name": "mysql"},
+    {"target_key": "mariadbd", "target_name": "mariadb"},
+    {"target_key": "redis", "target_name": "redis"},
+    {"target_key": "pure-ftpd", "target_name": "pure-ftpd"},
+    {"target_key": "memcached", "target_name": "memcached"},
+    {"target_key": "mongodb", "target_name": "mongodb"},
+    {"target_key": "pgsql", "target_name": "pgsql"},
+]
+
+LOCAL_SERVICE_DEFINITIONS = [
+    {
+        "family": "nginx",
+        "target_key": "nginx",
+        "target_name": "Nginx",
+        "paths": ["/www/server/nginx/sbin/nginx", "/www/server/nginx/nginx/sbin/nginx", "/etc/init.d/nginx"],
+    },
+    {
+        "family": "mysql",
+        "target_key": "mysqld",
+        "target_name": "MySQL",
+        "paths": ["/www/server/mysql/bin/mysqld", "/etc/init.d/mysqld"],
+    },
+    {
+        "family": "apache",
+        "target_key": "httpd",
+        "target_name": "Apache",
+        "paths": ["/www/server/apache/bin/httpd", "/etc/init.d/httpd"],
+    },
+    {
+        "family": "redis",
+        "target_key": "redis",
+        "target_name": "Redis",
+        "paths": ["/www/server/redis/src/redis-server", "/etc/init.d/redis"],
+    },
+    {
+        "family": "pure-ftpd",
+        "target_key": "pure-ftpd",
+        "target_name": "Pure-Ftpd",
+        "paths": ["/www/server/pure-ftpd/sbin/pure-ftpd", "/etc/init.d/pure-ftpd"],
+    },
+    {
+        "family": "memcached",
+        "target_key": "memcached",
+        "target_name": "Memcached",
+        "paths": ["/usr/local/memcached/bin/memcached", "/etc/init.d/memcached"],
+    },
+    {
+        "family": "mongodb",
+        "target_key": "mongodb",
+        "target_name": "MongoDB",
+        "paths": ["/www/server/mongodb/bin/mongod", "/etc/init.d/mongodb"],
+    },
+    {
+        "family": "pgsql",
+        "target_key": "pgsql",
+        "target_name": "PostgreSQL",
+        "paths": ["/www/server/pgsql/bin/postgres", "/etc/init.d/pgsql"],
+    },
+]
+
+PROCESS_CHECK_SERVICES = {
+    "mysqld",
+    "mariadbd",
+    "redis",
+    "pure-ftpd",
+    "nginx",
+    "httpd",
+    "apache",
+    "memcached",
+    "mongodb",
+    "pgsql",
+}
+
+
+def valid_service_name(name: str) -> bool:
+    return bool(name and SERVICE_NAME_RE.match(name))
+
+
+def normalize_service_targets(services: Any) -> List[Dict[str, Any]]:
+    if not isinstance(services, list):
+        return []
+    result = []
+    seen = set()
+    for idx, item in enumerate(services):
+        if isinstance(item, str):
+            key = item.strip()
+            name = key
+            enabled = 1
+            sort = idx
+            extra = {}
+        elif isinstance(item, dict):
+            key = str(item.get("target_key") or item.get("name") or "").strip()
+            name = str(item.get("target_name") or item.get("name") or key).strip()
+            enabled = int(item.get("enabled", 1))
+            sort = int(item.get("sort", idx))
+            extra = item.get("extra", {})
+        else:
+            continue
+        if not valid_service_name(key) or key in seen:
+            continue
+        seen.add(key)
+        service_family = str(item.get("service_family") or item.get("family") or "").strip() if isinstance(item, dict) else ""
+        if not service_family and key.startswith("php-fpm-"):
+            service_family = "php"
+        result.append({
+            "target_key": key,
+            "target_name": name or key,
+            "enabled": 1 if enabled else 0,
+            "sort": sort,
+            "extra": extra if isinstance(extra, dict) else {},
+            "service_family": service_family,
+            "version": str(item.get("version") or "").strip() if isinstance(item, dict) else "",
+        })
+    return result
+
+
+def keep_single_version_services(services: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
+    result = []
+    seen_version_family = set()
+    for item in normalize_service_targets(services):
+        family = item.get("service_family") or ("php" if item["target_key"].startswith("php-fpm-") else "")
+        if family == "php":
+            if family in seen_version_family:
+                continue
+            seen_version_family.add(family)
+        result.append(item)
+    return result
+
+
+def service_unknown_statuses(services: List[Dict[str, Any]], error_msg: str) -> List[Dict[str, Any]]:
+    now = int(time.time())
+    result = []
+    for item in normalize_service_targets(services):
+        result.append({
+            "name": item["target_key"],
+            "target_key": item["target_key"],
+            "target_name": item["target_name"],
+            "status": 2,
+            "running": False,
+            "state": "unknown",
+            "error_msg": error_msg,
+            "ts": now,
+        })
+    return result
+
+
+def collect_local_service_status(services: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
+    now = int(time.time())
+    result = []
+    for item in normalize_service_targets(services):
+        if not int(item.get("enabled", 1)):
+            continue
+        running, err = _check_local_service(item["target_key"])
+        status = 1 if running else 0
+        if err and not running:
+            status = 2
+        result.append({
+            "name": item["target_key"],
+            "target_key": item["target_key"],
+            "target_name": item["target_name"],
+            "status": status,
+            "running": bool(running),
+            "state": "active" if running else ("unknown" if err else "inactive"),
+            "error_msg": err,
+            "ts": now,
+        })
+    return result
+
+
+def discover_local_services() -> Dict[str, Any]:
+    services = []
+    for item in LOCAL_SERVICE_DEFINITIONS:
+        if not any(os.path.exists(path) for path in item["paths"]):
+            continue
+        status = collect_local_service_status([{
+            "target_key": item["target_key"],
+            "target_name": item["target_name"],
+            "service_family": item["family"],
+        }])
+        status_item = status[0] if status else {}
+        services.append(_service_option(
+            target_key=item["target_key"],
+            target_name=item["target_name"],
+            family=item["family"],
+            running=bool(status_item.get("running", False)),
+            status=int(status_item.get("status", 0)),
+            state=str(status_item.get("state") or "inactive"),
+        ))
+
+    for version in _installed_php_versions():
+        key = "php-fpm-{}".format(version)
+        status = collect_local_service_status([{
+            "target_key": key,
+            "target_name": "PHP {}".format(_format_php_version(version)),
+            "service_family": "php",
+            "version": version,
+        }])
+        status_item = status[0] if status else {}
+        services.append(_service_option(
+            target_key=key,
+            target_name="PHP {}".format(_format_php_version(version)),
+            family="php",
+            version=version,
+            version_name=_format_php_version(version),
+            running=bool(status_item.get("running", False)),
+            status=int(status_item.get("status", 0)),
+            state=str(status_item.get("state") or "inactive"),
+            is_versioned=1,
+        ))
+
+    return {
+        "services": services,
+        "groups": build_service_groups(services),
+    }
+
+
+def get_service_file_probe_paths() -> List[str]:
+    paths = []
+    for item in LOCAL_SERVICE_DEFINITIONS:
+        paths.extend(item["paths"])
+    for version in _candidate_php_versions():
+        paths.extend(_php_version_paths(version))
+    return _dedupe(paths)
+
+
+def build_service_options_from_file_exists(existing_paths: Any) -> Dict[str, Any]:
+    if isinstance(existing_paths, (list, tuple, set)):
+        exists_map = {str(path): True for path in existing_paths}
+    elif isinstance(existing_paths, dict):
+        exists_map = {str(path): bool(value) for path, value in existing_paths.items()}
+    else:
+        exists_map = {}
+
+    services = []
+    for item in LOCAL_SERVICE_DEFINITIONS:
+        if not any(exists_map.get(path) for path in item["paths"]):
+            continue
+        services.append(_service_option(
+            target_key=item["target_key"],
+            target_name=item["target_name"],
+            family=item["family"],
+            status=2,
+            state="unknown",
+        ))
+
+    for version in _candidate_php_versions():
+        if not any(exists_map.get(path) for path in _php_version_paths(version)):
+            continue
+        services.append(_service_option(
+            target_key="php-fpm-{}".format(version),
+            target_name="PHP {}".format(_format_php_version(version)),
+            family="php",
+            version=version,
+            version_name=_format_php_version(version),
+            status=2,
+            state="unknown",
+            is_versioned=1,
+        ))
+
+    return {
+        "services": services,
+        "groups": build_service_groups(services),
+    }
+
+
+def build_service_groups(services: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
+    groups = []
+    group_map = {}
+    for item in services:
+        family = item.get("service_family") or item.get("family") or item.get("target_key")
+        if not family:
+            continue
+        if family not in group_map:
+            group_map[family] = {
+                "family": family,
+                "target_name": "PHP" if family == "php" else item.get("target_name", family),
+                "type": "version" if family == "php" else "single",
+                "select_mode": "single" if family == "php" else "toggle",
+                "multiple": 0 if family == "php" else 1,
+                "services": [],
+            }
+            groups.append(group_map[family])
+        group_map[family]["services"].append(item)
+
+    for group in groups:
+        if group["family"] == "php":
+            group["services"].sort(key=lambda x: x.get("version", ""))
+        else:
+            group["services"].sort(key=lambda x: x.get("sort", 0))
+    return groups
+
+
+def _installed_php_versions() -> List[str]:
+    versions = []
+    for version in _candidate_php_versions():
+        for path in _php_version_paths(version):
+            if os.path.exists(path):
+                versions.append(str(version))
+                break
+    return versions
+
+
+def _candidate_php_versions() -> List[str]:
+    try:
+        versions = public.get_php_versions()
+    except Exception:
+        versions = ['52', '53', '54', '55', '56', '70', '71', '72', '73', '74', '80', '81', '82', '83', '84', '85']
+    return [str(version) for version in versions if re.match(r"^\d{2}$", str(version))]
+
+
+def _php_version_paths(version: str) -> List[str]:
+    return [
+        "/www/server/php/{}/sbin/php-fpm".format(version),
+        "/www/server/php/{}/bin/php".format(version),
+        "/etc/init.d/php-fpm-{}".format(version),
+    ]
+
+
+def _dedupe(items: List[str]) -> List[str]:
+    result = []
+    seen = set()
+    for item in items:
+        if item in seen:
+            continue
+        seen.add(item)
+        result.append(item)
+    return result
+
+
+def _format_php_version(version: str) -> str:
+    version = str(version)
+    if len(version) == 2 and version.isdigit():
+        return "{}.{}".format(version[0], version[1])
+    return version
+
+
+def _service_option(
+        target_key: str,
+        target_name: str,
+        family: str,
+        version: str = "",
+        version_name: str = "",
+        running: bool = False,
+        status: int = 0,
+        state: str = "inactive",
+        is_versioned: int = 0,
+) -> Dict[str, Any]:
+    return {
+        "target_type": "service",
+        "target_key": target_key,
+        "target_name": target_name,
+        "service_family": family,
+        "version": version,
+        "version_name": version_name,
+        "enabled": 0,
+        "installed": 1,
+        "is_versioned": int(is_versioned),
+        "status": int(status),
+        "running": bool(running),
+        "state": state,
+        "sort": 0,
+    }
+
+
+def _check_local_service(name: str) -> Tuple[bool, str]:
+    if not valid_service_name(name):
+        return False, "Invalid service name"
+    if name in PROCESS_CHECK_SERVICES or name.find("php-fpm") != -1:
+        try:
+            from system import system
+            return bool(system().check_service_status(name)), ""
+        except Exception as e:
+            return False, str(e)
+
+    cmd = (
+        "if command -v systemctl >/dev/null 2>&1; then "
+        "systemctl is-active --quiet {name}; "
+        "else service {name} status >/dev/null 2>&1; fi; echo $?"
+    ).format(name=name)
+    try:
+        stdout, stderr = public.ExecShell(cmd)
+        lines = str(stdout or "").strip().splitlines()
+        return bool(lines and lines[-1].strip() == "0"), str(stderr or "")
+    except Exception as e:
+        return False, str(e)

mod/project/node/nodeutil/ssh_wrap.py

@@ -1,9 +1,11 @@
 import json
 import os.path
+import time
 import traceback
-from typing import Optional, Tuple, Callable, Union, Dict
+from typing import Optional, Tuple, Callable, Union, Dict, List
 from mod.base.ssh_executor import SSHExecutor, CommandResult
 from mod.project.node.dbutil import ServerNodeDB, Node
+from mod.project.node.nodeutil.monitor_service import build_service_groups, normalize_service_targets, service_unknown_statuses
 
 import public
 
@@ -13,6 +15,11 @@ def is_much_difference(a:float, b:float)->bool:
     ratio = a / b
     return ratio >= 10 or ratio <= 0.1
 
+
+def _shell_single_quote(value: str) -> str:
+    return "'" + value.replace("'", "'\"'\"'") + "'"
+
+
 class SSHApi:
     is_local = False
     _local_scripts_dir = os.path.join(os.path.dirname(__file__), "ssh_warp_scripts")
@@ -67,6 +74,153 @@ class SSHApi:
         except Exception as e:
             return None, str(e)
 
+    def get_service_status(self, services) -> Tuple[List[dict], str]:
+        service_list = [item for item in normalize_service_targets(services) if int(item.get("enabled", 1)) == 1]
+        if not service_list:
+            return [], ""
+        service_args = " ".join(_shell_single_quote(item["target_key"]) for item in service_list)
+        cmd = """
+for svc in {service_args}; do
+  if command -v systemctl >/dev/null 2>&1; then
+    systemctl is-active --quiet "$svc"
+    rc=$?
+    state=$(systemctl is-active "$svc" 2>/dev/null || true)
+  else
+    service "$svc" status >/dev/null 2>&1
+    rc=$?
+    if [ "$rc" = "0" ]; then state="active"; else state="inactive"; fi
+  fi
+  if [ "$rc" = "0" ]; then running=1; else running=0; fi
+  printf '%s\\t%s\\t%s\\n' "$svc" "$running" "$state"
+done
+""".format(service_args=service_args)
+        executor = None
+        try:
+            executor = self._get_ssh_executor()
+            executor.open()
+            exit_code, stdout, stderr = executor.run(cmd)
+            if exit_code != 0 and not stdout:
+                return service_unknown_statuses(service_list, stderr or "Service status check failed"), stderr
+            target_map = {item["target_key"]: item for item in service_list}
+            now = int(time.time())
+            result = []
+            for line in str(stdout or "").splitlines():
+                parts = line.split("\t")
+                if len(parts) < 3:
+                    continue
+                key = parts[0].strip()
+                item = target_map.get(key)
+                if not item:
+                    continue
+                running = parts[1].strip() == "1"
+                result.append({
+                    "name": key,
+                    "target_key": key,
+                    "target_name": item["target_name"],
+                    "status": 1 if running else 0,
+                    "running": running,
+                    "state": parts[2].strip() or ("active" if running else "inactive"),
+                    "error_msg": "",
+                    "ts": now,
+                })
+            if not result:
+                return service_unknown_statuses(service_list, stderr or "Service status data is empty"), stderr
+            return result, ""
+        except RuntimeError:
+            return service_unknown_statuses(service_list, "SSH connection failed"), "SSH connection failed"
+        except Exception as e:
+            return service_unknown_statuses(service_list, str(e)), str(e)
+        finally:
+            if executor:
+                executor.close()
+
+    def get_service_options(self) -> Tuple[Dict[str, list], str]:
+        cmd = r"""
+check_status() {
+  svc="$1"
+  if command -v systemctl >/dev/null 2>&1; then
+    systemctl is-active --quiet "$svc"
+    rc=$?
+    state=$(systemctl is-active "$svc" 2>/dev/null || true)
+  else
+    service "$svc" status >/dev/null 2>&1
+    rc=$?
+    if [ "$rc" = "0" ]; then state="active"; else state="inactive"; fi
+  fi
+  if [ "$rc" = "0" ]; then running=1; else running=0; fi
+  printf '%s\t%s' "$running" "$state"
+}
+emit_service() {
+  family="$1"; key="$2"; label="$3"; paths="$4"
+  exists=0
+  old_ifs="$IFS"; IFS=":"
+  for p in $paths; do
+    if [ -e "$p" ]; then exists=1; break; fi
+  done
+  IFS="$old_ifs"
+  if [ "$exists" = "1" ]; then
+    status=$(check_status "$key")
+    printf 'SERVICE\t%s\t%s\t%s\t\t\t%s\n' "$family" "$key" "$label" "$status"
+  fi
+}
+emit_service nginx nginx Nginx "/www/server/nginx/sbin/nginx:/www/server/nginx/nginx/sbin/nginx:/etc/init.d/nginx"
+emit_service mysql mysqld MySQL "/www/server/mysql/bin/mysqld:/etc/init.d/mysqld"
+emit_service apache httpd Apache "/www/server/apache/bin/httpd:/etc/init.d/httpd"
+emit_service redis redis Redis "/www/server/redis/src/redis-server:/etc/init.d/redis"
+emit_service pure-ftpd pure-ftpd Pure-Ftpd "/www/server/pure-ftpd/sbin/pure-ftpd:/etc/init.d/pure-ftpd"
+emit_service memcached memcached Memcached "/usr/local/memcached/bin/memcached:/etc/init.d/memcached"
+emit_service mongodb mongodb MongoDB "/www/server/mongodb/bin/mongod:/etc/init.d/mongodb"
+emit_service pgsql pgsql PostgreSQL "/www/server/pgsql/bin/postgres:/etc/init.d/pgsql"
+if [ -d /www/server/php ]; then
+  for dir in /www/server/php/[0-9][0-9]; do
+    [ -d "$dir" ] || continue
+    ver=$(basename "$dir")
+    if [ -x "$dir/sbin/php-fpm" ] || [ -x "$dir/bin/php" ] || [ -e "/etc/init.d/php-fpm-$ver" ]; then
+      status=$(check_status "php-fpm-$ver")
+      major=$(printf '%s' "$ver" | cut -c1)
+      minor=$(printf '%s' "$ver" | cut -c2)
+      printf 'SERVICE\tphp\tphp-fpm-%s\tPHP %s.%s\t%s\t%s.%s\t%s\n' "$ver" "$major" "$minor" "$ver" "$major" "$minor" "$status"
+    fi
+  done
+fi
+"""
+        executor = None
+        try:
+            executor = self._get_ssh_executor()
+            executor.open()
+            exit_code, stdout, stderr = executor.run(cmd)
+            if exit_code != 0 and not stdout:
+                return {"services": [], "groups": []}, stderr or "Service discovery failed"
+            services = []
+            for line in str(stdout or "").splitlines():
+                parts = line.split("\t")
+                if len(parts) < 8 or parts[0] != "SERVICE":
+                    continue
+                running = parts[6].strip() == "1"
+                services.append({
+                    "target_type": "service",
+                    "service_family": parts[1].strip(),
+                    "target_key": parts[2].strip(),
+                    "target_name": parts[3].strip(),
+                    "version": parts[4].strip(),
+                    "version_name": parts[5].strip(),
+                    "enabled": 0,
+                    "installed": 1,
+                    "is_versioned": 1 if parts[1].strip() == "php" else 0,
+                    "status": 1 if running else 0,
+                    "running": running,
+                    "state": parts[7].strip() or ("active" if running else "inactive"),
+                    "sort": 0,
+                })
+            return {"services": services, "groups": build_service_groups(services)}, ""
+        except RuntimeError:
+            return {"services": [], "groups": []}, "SSH connection failed"
+        except Exception as e:
+            return {"services": [], "groups": []}, str(e)
+        finally:
+            if executor:
+                executor.close()
+
     @staticmethod
     def _tans_net_work_form_data(data: dict):
         data["mem"]["memAvailable"] = round(data["mem"]["memAvailable"] / 1024 / 1024, 2)