Optimize the server information retrieval

Author: wu-clan

backend/utils/server_info.py

@@ -1,12 +1,15 @@
 #!/usr/bin/env python3
 # -*- coding: utf-8 -*-
+from __future__ import annotations
+
 import os
 import platform
 import socket
 import sys
 
 from datetime import datetime, timedelta
 from datetime import timezone as tz
+from typing import Dict, List
 
 import psutil
 
@@ -61,94 +64,111 @@ def fmt_timedelta(td: timedelta) -> str:
         :param td: 时间差对象
         :return:
         """
-        total_seconds = round(td.total_seconds())
-        return ServerInfo.fmt_seconds(total_seconds)
+        return ServerInfo.fmt_seconds(round(td.total_seconds()))
 
     @staticmethod
     def get_cpu_info() -> dict[str, float | int]:
         """获取 CPU 信息"""
-        cpu_info = {'usage': round(psutil.cpu_percent(percpu=False), 2)}  # %
+        cpu_info = {
+            'usage': round(psutil.cpu_percent(interval=0.1), 2),  # %
+            'logical_num': psutil.cpu_count(logical=True) or 0,
+            'physical_num': psutil.cpu_count(logical=False) or 0,
+            'max_freq': 0.0,
+            'min_freq': 0.0,
+            'current_freq': 0.0,
+        }
 
         try:
-            # CPU 频率信息，最大、最小和当前频率
-            cpu_freq = psutil.cpu_freq()
-            cpu_info.update({
-                'max_freq': round(cpu_freq.max, 2),  # MHz
-                'min_freq': round(cpu_freq.min, 2),  # MHz
-                'current_freq': round(cpu_freq.current, 2),  # MHz
-            })
+            if hasattr(psutil, 'cpu_freq'):
+                cpu_freq = psutil.cpu_freq()
+                if cpu_freq:  # Some systems return None
+                    cpu_info.update({
+                        'max_freq': round(cpu_freq.max, 2),
+                        'min_freq': round(cpu_freq.min, 2),
+                        'current_freq': round(cpu_freq.current, 2),
+                    })
         except Exception:
-            cpu_info.update({'max_freq': 0, 'min_freq': 0, 'current_freq': 0})
+            pass
 
-        # CPU 逻辑核心数，物理核心数
-        cpu_info.update({
-            'logical_num': psutil.cpu_count(logical=True),
-            'physical_num': psutil.cpu_count(logical=False),
-        })
         return cpu_info
 
     @staticmethod
     def get_mem_info() -> dict[str, float]:
         """获取内存信息"""
         mem = psutil.virtual_memory()
+        gb_factor = 1024**3
         return {
-            'total': round(mem.total / 1024 / 1024 / 1024, 2),  # GB
-            'used': round(mem.used / 1024 / 1024 / 1024, 2),  # GB
-            'free': round(mem.available / 1024 / 1024 / 1024, 2),  # GB
-            'usage': round(mem.percent, 2),  # %
+            'total': round(mem.total / gb_factor, 2),
+            'used': round(mem.used / gb_factor, 2),
+            'free': round(mem.available / gb_factor, 2),
+            'usage': round(mem.percent, 2),
         }
 
     @staticmethod
-    def get_sys_info() -> dict[str, str]:
+    def get_sys_info() -> Dict[str, str]:
         """获取服务器信息"""
+        hostname = socket.gethostname()
+        ip = '127.0.0.1'
+
         try:
-            with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as sk:
-                sk.connect(('8.8.8.8', 80))
-                ip = sk.getsockname()[0]
-        except socket.gaierror:
-            ip = '127.0.0.1'
+            with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as s:
+                s.settimeout(0.5)
+                s.connect(('8.8.8.8', 80))
+                ip = s.getsockname()[0]
+        except (socket.gaierror, socket.timeout, OSError):
+            pass
 
         return {
-            'name': socket.gethostname(),
+            'name': hostname,
             'ip': ip,
             'os': platform.system(),
             'arch': platform.machine(),
         }
 
     @staticmethod
-    def get_disk_info() -> list[dict[str, str]]:
+    def get_disk_info() -> List[Dict[str, str]]:
         """获取磁盘信息"""
         disk_info = []
-        for disk in psutil.disk_partitions():
-            usage = psutil.disk_usage(disk.mountpoint)
-            disk_info.append({
-                'dir': disk.mountpoint,
-                'type': disk.fstype,
-                'device': disk.device,
-                'total': ServerInfo.format_bytes(usage.total),
-                'free': ServerInfo.format_bytes(usage.free),
-                'used': ServerInfo.format_bytes(usage.used),
-                'usage': f'{round(usage.percent, 2)} %',
-            })
+        for partition in psutil.disk_partitions(all=False):
+            try:
+                usage = psutil.disk_usage(partition.mountpoint)
+                disk_info.append({
+                    'dir': partition.mountpoint,
+                    'type': partition.fstype,
+                    'device': partition.device,
+                    'total': ServerInfo.format_bytes(usage.total),
+                    'free': ServerInfo.format_bytes(usage.free),
+                    'used': ServerInfo.format_bytes(usage.used),
+                    'usage': f'{usage.percent:.2f}%',
+                })
+            except (PermissionError, psutil.AccessDenied):
+                continue
         return disk_info
 
     @staticmethod
     def get_service_info() -> dict[str, str | datetime]:
         """获取服务信息"""
         process = psutil.Process(os.getpid())
         mem_info = process.memory_info()
-        start_time = timezone.f_datetime(datetime.utcfromtimestamp(process.create_time()).replace(tzinfo=tz.utc))
+
+        try:
+            create_time = datetime.fromtimestamp(process.create_time(), tz=tz.utc)
+            start_time = timezone.f_datetime(create_time)
+        except (psutil.NoSuchProcess, OSError):
+            start_time = timezone.now()
+
+        elapsed = ServerInfo.fmt_timedelta(timezone.now() - start_time)
 
         return {
             'name': 'Python3',
             'version': platform.python_version(),
             'home': sys.executable,
-            'cpu_usage': f'{round(process.cpu_percent(interval=1), 2)} %',
-            'mem_vms': ServerInfo.format_bytes(mem_info.vms),  # 虚拟内存, 即当前进程申请的虚拟内存
-            'mem_rss': ServerInfo.format_bytes(mem_info.rss),  # 常驻内存, 即当前进程实际使用的物理内存
-            'mem_free': ServerInfo.format_bytes(mem_info.vms - mem_info.rss),  # 空闲内存
+            'cpu_usage': f'{process.cpu_percent(interval=0.1):.2f}%',
+            'mem_vms': ServerInfo.format_bytes(mem_info.vms),
+            'mem_rss': ServerInfo.format_bytes(mem_info.rss),
+            'mem_free': ServerInfo.format_bytes(mem_info.vms - mem_info.rss),
             'startup': start_time,
-            'elapsed': ServerInfo.fmt_timedelta(timezone.now() - start_time),
+            'elapsed': elapsed,
         }