mirror of
https://github.com/fastapiadmin/FastapiAdmin.git
synced 2026-09-22 05:02:57 +00:00
zhangtao
7d2367e34e
本次提交包含多项优化： 1. 移除大量冗余的文件头注释与过时的from __future__导入 2. 将CRUD的list方法统一重命名为get_list保持接口一致 3. 修复前后端状态字段类型不匹配问题，将string类型status改为number 4. 修正前端文案错别字，将"代办事项"修正为标准写法 5. 更新sqlalchemy版本并调整依赖配置 6. 新增缓存工具类替代fastapi-cache2，重构缓存调用逻辑 7. 新增开源授权函生成相关工具与数据库字段支持 8. 为多个业务模块添加防重复提交loading状态 9. 修复邮件模型的外键关联缺失问题 10. 优化pdf生成工具的导入时机与文档注释
118 lines
3.8 KiB
Python
118 lines
3.8 KiB
Python
|
||
import platform
|
||
import socket
|
||
import time
|
||
from pathlib import Path
|
||
|
||
import psutil
|
||
|
||
from app.utils.common_util import bytes2human
|
||
|
||
from .schema import (
|
||
CpuInfoSchema,
|
||
DiskInfoSchema,
|
||
MemoryInfoSchema,
|
||
PyInfoSchema,
|
||
ServerMonitorSchema,
|
||
SysInfoSchema,
|
||
)
|
||
|
||
|
||
class ServerService:
|
||
"""服务监控模块服务层"""
|
||
|
||
@staticmethod
|
||
async def get_server_monitor_info() -> ServerMonitorSchema:
|
||
return ServerMonitorSchema(
|
||
cpu=ServerService._get_cpu_info(),
|
||
mem=ServerService._get_memory_info(),
|
||
sys=ServerService._get_system_info(),
|
||
py=ServerService._get_python_info(),
|
||
disks=ServerService._get_disk_info(),
|
||
)
|
||
|
||
@staticmethod
|
||
def _get_cpu_info() -> CpuInfoSchema:
|
||
cpu_times = psutil.cpu_times_percent()
|
||
cpu_num = psutil.cpu_count(logical=True)
|
||
if not cpu_num:
|
||
cpu_num = 1
|
||
return CpuInfoSchema(
|
||
cpu_num=cpu_num,
|
||
used=cpu_times.user,
|
||
sys=cpu_times.system,
|
||
free=cpu_times.idle,
|
||
)
|
||
|
||
@staticmethod
|
||
def _get_memory_info() -> MemoryInfoSchema:
|
||
memory = psutil.virtual_memory()
|
||
return MemoryInfoSchema(
|
||
total=bytes2human(memory.total),
|
||
used=bytes2human(memory.used),
|
||
free=bytes2human(memory.free),
|
||
usage=memory.percent,
|
||
)
|
||
|
||
@staticmethod
|
||
def _get_system_info() -> SysInfoSchema:
|
||
hostname = socket.gethostname()
|
||
return SysInfoSchema(
|
||
computer_ip=socket.gethostbyname(hostname),
|
||
computer_name=platform.node(),
|
||
os_arch=platform.machine(),
|
||
os_name=platform.platform(),
|
||
user_dir=str(Path.cwd()),
|
||
)
|
||
|
||
@staticmethod
|
||
def _get_python_info() -> PyInfoSchema:
|
||
current_process = psutil.Process()
|
||
memory = psutil.virtual_memory()
|
||
process_memory = current_process.memory_info()
|
||
|
||
start_time = current_process.create_time()
|
||
run_time = ServerService._calculate_run_time(start_time)
|
||
|
||
return PyInfoSchema(
|
||
name=current_process.name(),
|
||
version=platform.python_version(),
|
||
start_time=time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(start_time)),
|
||
run_time=run_time,
|
||
home=str(Path(current_process.exe())),
|
||
memory_total=bytes2human(memory.available),
|
||
memory_used=bytes2human(process_memory.rss),
|
||
memory_free=bytes2human(memory.available - process_memory.rss),
|
||
memory_usage=round((process_memory.rss / memory.available) * 100, 2),
|
||
)
|
||
|
||
@staticmethod
|
||
def _get_disk_info() -> list[DiskInfoSchema]:
|
||
disk_info = []
|
||
for partition in psutil.disk_partitions():
|
||
try:
|
||
usage = psutil.disk_usage(partition.mountpoint)
|
||
mount_point = str(Path(partition.mountpoint))
|
||
disk_info.append(
|
||
DiskInfoSchema(
|
||
dir_name=mount_point,
|
||
sys_type_name=partition.fstype,
|
||
type_name=f"本地固定磁盘（{mount_point}）",
|
||
total=bytes2human(usage.total),
|
||
used=bytes2human(usage.used),
|
||
free=bytes2human(usage.free),
|
||
usage=usage.percent,
|
||
)
|
||
)
|
||
except (PermissionError, FileNotFoundError):
|
||
continue
|
||
return disk_info
|
||
|
||
@staticmethod
|
||
def _calculate_run_time(start_time: float) -> str:
|
||
difference = time.time() - start_time
|
||
days = int(difference // (24 * 60 * 60))
|
||
hours = int((difference % (24 * 60 * 60)) // (60 * 60))
|
||
minutes = int((difference % (60 * 60)) // 60)
|
||
return f"{days}天{hours}小时{minutes}分钟"
|