Browse Source

系统接口

origin/hotfix/hlk-flight
root 4 months ago
parent
commit
29bb55d4c1
  1. 145
      app/api/v1/endpoints/system.py
  2. 9
      app/core/app/router.py
  3. 34
      app/schemas/system.py
  4. 150
      app/services/system_service.py
  5. 218
      app/utils/system_utils.py
  6. 137
      docs/modify.md
  7. 157
      docs/system_api.md
  8. 236
      test_system_api.py
  9. 44
      test_system_api_endpoints.py

145
app/api/v1/endpoints/system.py

@ -0,0 +1,145 @@
"""
系统管理API端点
"""
from fastapi import APIRouter, HTTPException, status
from app.services.system_service import SystemService
from app.schemas.system import MachineCodeResponse, SystemInfoResponse
from app.utils.api_decorators import handle_api_errors
from app.utils.structured_log import get_structured_logger
logger = get_structured_logger(__name__)
router = APIRouter()
# 创建全局system_service实例
system_service = SystemService()
@router.get("/system/machine-code",
summary="获取机器码",
description="获取当前系统的机器码,支持Windows和Linux系统",
response_model=MachineCodeResponse)
@handle_api_errors
async def get_machine_code():
"""获取机器码
Windows: 使用 wmic csproduct get uuid 命令
Linux: 使用 dmidecode -s system-serial-number 或读取系统文件
"""
try:
logger.info("API调用: 获取机器码")
result = await system_service.get_machine_code()
if result["success"]:
logger.info("机器码获取成功")
return result
else:
logger.warning("机器码获取失败")
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail=result["message"]
)
except HTTPException:
raise
except Exception as e:
logger.error(f"获取机器码API异常: {e}")
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail=f"获取机器码失败: {str(e)}"
)
@router.get("/system/info",
summary="获取系统信息",
description="获取当前系统的详细信息",
response_model=SystemInfoResponse)
@handle_api_errors
async def get_system_info():
"""获取系统信息"""
try:
logger.info("API调用: 获取系统信息")
result = await system_service.get_system_info()
if result["success"]:
logger.info("系统信息获取成功")
return result
else:
logger.warning("系统信息获取失败")
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail=result["message"]
)
except HTTPException:
raise
except Exception as e:
logger.error(f"获取系统信息API异常: {e}")
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail=f"获取系统信息失败: {str(e)}"
)
@router.get("/system/machine-code/detailed",
summary="获取详细机器码信息",
description="获取包含系统信息的详细机器码信息")
@handle_api_errors
async def get_detailed_machine_code_info():
"""获取详细的机器码信息"""
try:
logger.info("API调用: 获取详细机器码信息")
result = await system_service.get_detailed_machine_code_info()
if result["success"]:
logger.info("详细机器码信息获取成功")
return result
else:
logger.warning("详细机器码信息获取失败")
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail=result["message"]
)
except HTTPException:
raise
except Exception as e:
logger.error(f"获取详细机器码信息API异常: {e}")
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail=f"获取详细机器码信息失败: {str(e)}"
)
@router.get("/system/health",
summary="系统健康检查",
description="检查系统基本功能是否正常")
@handle_api_errors
async def system_health_check():
"""系统健康检查"""
try:
logger.info("API调用: 系统健康检查")
# 检查基本功能
system_type = system_service.system_utils.get_system_type()
machine_code = system_service.system_utils.get_machine_code()
health_status = {
"success": True,
"message": "系统健康检查通过",
"data": {
"system_type": system_type,
"machine_code_available": machine_code is not None,
"status": "healthy"
}
}
if not machine_code:
health_status["data"]["status"] = "warning"
health_status["data"]["warning"] = "机器码获取失败"
logger.info("系统健康检查完成")
return health_status
except Exception as e:
logger.error(f"系统健康检查异常: {e}")
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail=f"系统健康检查失败: {str(e)}"
)

9
app/core/app/router.py

@ -70,7 +70,7 @@ def register_routers():
try:
# 导入所有路由模块
from app.api.v1.endpoints import (
devices, ssh, at, plnk, websocket
devices, ssh, at, plnk, websocket, system
)
# 注册设备管理路由(统一入口)
@ -108,6 +108,13 @@ def register_routers():
tags=["PLNK协议"]
)
# 注册系统管理路由
router_manager.register_router(
system.router,
prefix="/api/v1",
tags=["系统管理"]
)
logger.info("路由注册配置完成")

34
app/schemas/system.py

@ -0,0 +1,34 @@
"""
系统相关数据模型
"""
from typing import Optional, Dict, Any
from pydantic import BaseModel, Field
class MachineCodeResponse(BaseModel):
"""机器码响应模型"""
success: bool = Field(..., description="操作是否成功")
message: str = Field(..., description="响应消息")
data: Optional[Dict[str, Any]] = Field(None, description="机器码数据")
class SystemInfoResponse(BaseModel):
"""系统信息响应模型"""
success: bool = Field(..., description="操作是否成功")
message: str = Field(..., description="响应消息")
data: Optional[Dict[str, Any]] = Field(None, description="系统信息数据")
class MachineCodeData(BaseModel):
"""机器码数据模型"""
machine_code: Optional[str] = Field(None, description="机器码")
system_type: str = Field(..., description="系统类型")
method: Optional[str] = Field(None, description="获取方法")
error: Optional[str] = Field(None, description="错误信息")
class SystemInfoData(BaseModel):
"""系统信息数据模型"""
system_type: str = Field(..., description="系统类型")
platform: str = Field(..., description="平台信息")
architecture: str = Field(..., description="架构")
processor: str = Field(..., description="处理器")
machine_code: Optional[str] = Field(None, description="机器码")
python_version: str = Field(..., description="Python版本")
error: Optional[str] = Field(None, description="错误信息")

150
app/services/system_service.py

@ -0,0 +1,150 @@
"""
系统服务模块
"""
from typing import Dict, Any, Optional
from app.utils.system_utils import SystemUtils
from app.utils.structured_log import get_structured_logger
logger = get_structured_logger(__name__)
class SystemService:
"""系统管理服务"""
def __init__(self):
self.system_utils = SystemUtils()
async def get_machine_code(self) -> Dict[str, Any]:
"""获取机器码"""
try:
logger.info("开始获取机器码")
system_type = SystemUtils.get_system_type()
machine_code = SystemUtils.get_machine_code()
if machine_code:
method = self._get_method_description(system_type)
logger.info(f"机器码获取成功: {machine_code}, 方法: {method}")
return {
"success": True,
"message": "机器码获取成功",
"data": {
"machine_code": machine_code,
"system_type": system_type,
"method": method
}
}
else:
logger.warning("机器码获取失败")
return {
"success": False,
"message": "机器码获取失败",
"data": {
"machine_code": None,
"system_type": system_type,
"method": None,
"error": "无法获取机器码"
}
}
except Exception as e:
logger.error(f"获取机器码异常: {e}")
return {
"success": False,
"message": f"获取机器码失败: {str(e)}",
"data": {
"machine_code": None,
"system_type": SystemUtils.get_system_type(),
"method": None,
"error": str(e)
}
}
async def get_system_info(self) -> Dict[str, Any]:
"""获取系统信息"""
try:
logger.info("开始获取系统信息")
system_info = SystemUtils.get_system_info()
if "error" not in system_info:
logger.info("系统信息获取成功")
return {
"success": True,
"message": "系统信息获取成功",
"data": system_info
}
else:
logger.warning("系统信息获取失败")
return {
"success": False,
"message": "系统信息获取失败",
"data": system_info
}
except Exception as e:
logger.error(f"获取系统信息异常: {e}")
return {
"success": False,
"message": f"获取系统信息失败: {str(e)}",
"data": {
"system_type": "unknown",
"error": str(e)
}
}
def _get_method_description(self, system_type: str) -> str:
"""获取方法描述"""
if system_type == "windows":
return "wmic csproduct get uuid"
elif system_type == "linux":
return "dmidecode -s system-serial-number / cat /sys/class/dmi/id/product_serial / hostid"
else:
return "unknown"
async def get_detailed_machine_code_info(self) -> Dict[str, Any]:
"""获取详细的机器码信息"""
try:
logger.info("开始获取详细机器码信息")
system_type = SystemUtils.get_system_type()
machine_code = SystemUtils.get_machine_code()
system_info = SystemUtils.get_system_info()
result = {
"success": True,
"message": "详细机器码信息获取成功",
"data": {
"machine_code": machine_code,
"system_type": system_type,
"method": self._get_method_description(system_type),
"system_info": system_info,
"timestamp": self._get_current_timestamp()
}
}
if not machine_code:
result["success"] = False
result["message"] = "机器码获取失败"
result["data"]["error"] = "无法获取机器码"
logger.info("详细机器码信息获取完成")
return result
except Exception as e:
logger.error(f"获取详细机器码信息异常: {e}")
return {
"success": False,
"message": f"获取详细机器码信息失败: {str(e)}",
"data": {
"machine_code": None,
"system_type": SystemUtils.get_system_type(),
"method": None,
"error": str(e)
}
}
def _get_current_timestamp(self) -> str:
"""获取当前时间戳"""
from datetime import datetime
return datetime.now().isoformat()

218
app/utils/system_utils.py

@ -0,0 +1,218 @@
"""
系统工具模块
"""
import platform
import subprocess
import re
from typing import Optional
from app.utils.structured_log import get_structured_logger
logger = get_structured_logger(__name__)
class SystemUtils:
"""系统工具类"""
@staticmethod
def get_system_type() -> str:
"""获取系统类型"""
return platform.system().lower()
@staticmethod
def get_machine_code() -> Optional[str]:
"""获取机器码
Windows: 使用 wmic csproduct get uuid
Linux: 使用 dmidecode -s system-serial-number cat /sys/class/dmi/id/product_serial
"""
try:
system_type = SystemUtils.get_system_type()
logger.info(f"获取机器码,系统类型: {system_type}")
if system_type == "windows":
return SystemUtils._get_windows_machine_code()
elif system_type == "linux":
return SystemUtils._get_linux_machine_code()
else:
logger.warning(f"不支持的系统类型: {system_type}")
return None
except Exception as e:
logger.error(f"获取机器码失败: {e}")
return None
@staticmethod
def _get_windows_machine_code() -> Optional[str]:
"""获取Windows系统机器码"""
try:
# 使用 wmic csproduct get uuid 命令
result = subprocess.run(
["wmic", "csproduct", "get", "uuid"],
capture_output=True,
text=True,
timeout=10
)
if result.returncode == 0:
# 解析输出,跳过标题行
lines = result.stdout.strip().split('\n')
if len(lines) >= 2:
uuid = lines[1].strip()
if uuid and uuid != "UUID":
logger.info(f"Windows机器码获取成功: {uuid}")
return uuid
logger.warning("Windows机器码获取失败,尝试备用方法")
return SystemUtils._get_windows_machine_code_fallback()
except subprocess.TimeoutExpired:
logger.error("Windows机器码获取超时")
return None
except Exception as e:
logger.error(f"Windows机器码获取异常: {e}")
return None
@staticmethod
def _get_windows_machine_code_fallback() -> Optional[str]:
"""Windows机器码获取备用方法"""
try:
# 尝试使用 PowerShell 命令
result = subprocess.run(
["powershell", "-Command", "Get-WmiObject -Class Win32_ComputerSystemProduct | Select-Object -ExpandProperty UUID"],
capture_output=True,
text=True,
timeout=10
)
if result.returncode == 0:
uuid = result.stdout.strip()
if uuid and len(uuid) > 0:
logger.info(f"Windows机器码获取成功(备用方法): {uuid}")
return uuid
return None
except Exception as e:
logger.error(f"Windows机器码备用方法失败: {e}")
return None
@staticmethod
def _get_linux_machine_code() -> Optional[str]:
"""获取Linux系统机器码"""
try:
# 首先尝试 dmidecode 命令
serial_number = SystemUtils._get_linux_serial_via_dmidecode()
if serial_number:
return serial_number
# 备用方法:读取系统文件
serial_number = SystemUtils._get_linux_serial_via_file()
if serial_number:
return serial_number
# 最后尝试:使用 hostid 命令
return SystemUtils._get_linux_serial_via_hostid()
except Exception as e:
logger.error(f"Linux机器码获取异常: {e}")
return None
@staticmethod
def _get_linux_serial_via_dmidecode() -> Optional[str]:
"""通过 dmidecode 获取Linux序列号"""
try:
result = subprocess.run(
["dmidecode", "-s", "system-serial-number"],
capture_output=True,
text=True,
timeout=10
)
if result.returncode == 0:
serial = result.stdout.strip()
if serial and serial != "0" and len(serial) > 0:
logger.info(f"Linux序列号获取成功(dmidecode): {serial}")
return serial
return None
except (subprocess.TimeoutExpired, FileNotFoundError):
logger.warning("dmidecode命令不可用或超时")
return None
except Exception as e:
logger.error(f"dmidecode获取序列号失败: {e}")
return None
@staticmethod
def _get_linux_serial_via_file() -> Optional[str]:
"""通过读取系统文件获取Linux序列号"""
try:
# 尝试读取 /sys/class/dmi/id/product_serial
result = subprocess.run(
["cat", "/sys/class/dmi/id/product_serial"],
capture_output=True,
text=True,
timeout=5
)
if result.returncode == 0:
serial = result.stdout.strip()
if serial and serial != "0" and len(serial) > 0:
logger.info(f"Linux序列号获取成功(文件): {serial}")
return serial
return None
except (subprocess.TimeoutExpired, FileNotFoundError):
logger.warning("系统文件读取失败")
return None
except Exception as e:
logger.error(f"文件读取序列号失败: {e}")
return None
@staticmethod
def _get_linux_serial_via_hostid() -> Optional[str]:
"""通过 hostid 命令获取Linux机器码"""
try:
result = subprocess.run(
["hostid"],
capture_output=True,
text=True,
timeout=5
)
if result.returncode == 0:
hostid = result.stdout.strip()
if hostid and len(hostid) > 0:
logger.info(f"Linux机器码获取成功(hostid): {hostid}")
return hostid
return None
except (subprocess.TimeoutExpired, FileNotFoundError):
logger.warning("hostid命令不可用")
return None
except Exception as e:
logger.error(f"hostid获取机器码失败: {e}")
return None
@staticmethod
def get_system_info() -> dict:
"""获取系统信息"""
try:
system_type = SystemUtils.get_system_type()
machine_code = SystemUtils.get_machine_code()
return {
"system_type": system_type,
"platform": platform.platform(),
"architecture": platform.architecture()[0],
"processor": platform.processor(),
"machine_code": machine_code,
"python_version": platform.python_version()
}
except Exception as e:
logger.error(f"获取系统信息失败: {e}")
return {
"system_type": "unknown",
"error": str(e)
}

137
docs/modify.md

@ -3,100 +3,53 @@
## 2025-08-15 实现系统接口获取机器码功能
**新增内容:**
实现系统接口获取机器码功能,支持Windows和Linux系统,提供统一的API接口。
按照分层架构实现获取机器码的系统接口,支持Windows和Linux系统,遵循当前项目架构设计。
**实现详情:**
### 1. 工具层 (app/utils/system_utils.py)
- **SystemUtils类**:实现跨平台的机器码获取功能
- **Windows支持**:使用 `wmic csproduct get uuid` 命令,备用PowerShell方法
- **Linux支持**:使用 `dmidecode -s system-serial-number`、读取系统文件、`hostid` 命令
- **错误处理**:完善的异常处理和日志记录
- **系统信息**:获取完整的系统信息(平台、架构、处理器等)
### 2. 服务层 (app/services/system_service.py)
- **SystemService类**:封装系统相关的业务逻辑
- **机器码获取**:提供统一的机器码获取接口
- **系统信息**:获取系统详细信息
- **详细信息**:提供包含时间戳的详细机器码信息
- **响应格式**:统一的成功/失败响应格式
### 3. 数据模型层 (app/schemas/system.py)
- **MachineCodeResponse**:机器码响应模型
- **SystemInfoResponse**:系统信息响应模型
- **MachineCodeData**:机器码数据模型
- **SystemInfoData**:系统信息数据模型
### 4. API端点层 (app/api/v1/endpoints/system.py)
- **GET /api/v1/system/machine-code**:获取机器码
- **GET /api/v1/system/info**:获取系统信息
- **GET /api/v1/system/machine-code/detailed**:获取详细机器码信息
- **GET /api/v1/system/health**:系统健康检查
### 5. 路由注册
- 更新 `app/api/v1/endpoints/__init__.py` 导入系统模块
- 更新 `app/core/app/router.py` 注册系统路由
**技术特点:**
- ✅ 跨平台支持:Windows使用wmic命令,Linux使用dmidecode
- ✅ 备用方案:多种获取方法确保兼容性
- ✅ 分层架构:严格遵循当前项目的分层设计
- ✅ 错误处理:完善的异常处理和日志记录
- ✅ 统一响应:标准化的API响应格式
- ✅ 健康检查:提供系统状态监控接口
**修改详情:**
1. **创建系统接口endpoint文件** (`app/api/v1/endpoints/system.py`):
```python
@router.get("/system/machine-code", summary="获取系统机器码", response_model=MachineCodeResponse)
async def get_machine_code():
"""获取当前系统的机器码(UUID)
支持Windows和Linux系统:
- Windows: 使用wmic csproduct get uuid命令
- Linux: 尝试多种方法获取唯一标识符
"""
```
2. **Windows系统机器码获取**
```python
def get_windows_machine_code() -> str:
"""获取Windows系统的机器码(UUID)"""
# 使用wmic命令获取UUID
result = subprocess.run(
['wmic', 'csproduct', 'get', 'uuid'],
capture_output=True,
text=True,
check=True
)
# 解析输出,提取UUID
output = result.stdout.strip()
lines = output.split('\n')
# 查找UUID行(跳过标题行)
for line in lines:
line = line.strip()
if line and line.lower() != 'uuid':
# 提取UUID(移除可能的空格和特殊字符)
uuid_match = re.search(r'[A-Fa-f0-9]{8}-[A-Fa-f0-9]{4}-[A-Fa-f0-9]{4}-[A-Fa-f0-9]{4}-[A-Fa-f0-9]{12}', line)
if uuid_match:
return uuid_match.group(0)
```
3. **Linux系统机器码获取**
```python
def get_linux_machine_code() -> str:
"""获取Linux系统的机器码"""
# 方法1: 从/proc/cpuinfo获取CPU信息
# 方法2: 从/sys/class/dmi/id/product_uuid获取
# 方法3: 使用dmidecode命令
# 方法4: 从/etc/machine-id获取
# 方法5: 从/proc/sys/kernel/random/uuid获取随机UUID
# 如果所有方法都失败,返回系统信息组合
```
4. **更新路由注册**
- 在 `app/api/v1/endpoints/__init__.py` 中添加system模块导入
- 在 `app/core/app/router.py` 中注册系统信息路由
5. **创建测试脚本** (`test_system_api.py`):
```python
def test_machine_code():
"""测试获取机器码功能"""
# 测试Windows和Linux系统的机器码获取功能
```
**API接口说明:**
- **GET /api/v1/system/machine-code**: 获取系统机器码
- **GET /api/v1/system/info**: 获取系统详细信息
**响应格式:**
```json
{
"success": true,
"message": "成功获取windows系统机器码",
"data": {
"machine_code": "03000200-0400-0500-0006-000700080009",
"system": "windows",
"method": "wmic csproduct get uuid",
"platform_info": {
"node": "DESKTOP-T6EU05A",
"machine": "AMD64",
"processor": "AMD64 Family 23 Model 24 Stepping 1, AuthenticAMD",
"platform": "Windows-10-10.0.17763-SP0"
}
}
}
```
**优化效果:**
- ✅ 支持Windows系统使用wmic命令获取UUID
- ✅ 支持Linux系统多种方法获取机器码
- ✅ 提供统一的API接口,自动识别操作系统
- ✅ 包含详细的系统信息和获取方法说明
- ✅ 完善的错误处理和日志记录
**测试验证:**
- ✅ 创建测试脚本验证功能正确性
- ✅ 符合项目代码规范和架构设计
- ✅ API端点测试确认接口正常工作
- ✅ Windows环境测试通过,成功获取机器码
- ✅ 所有端点返回正确的JSON格式响应
## 2025-08-15 规范化设备事件数据格式并修复EventType使用

157
docs/system_api.md

@ -0,0 +1,157 @@
# 系统API文档
## 概述
系统API提供了获取机器码和系统信息的功能,支持Windows和Linux系统,遵循项目的分层架构设计。
## API端点
### 1. 获取机器码
**端点:** `GET /api/v1/system/machine-code`
**描述:** 获取当前系统的机器码
**响应示例:**
```json
{
"success": true,
"message": "机器码获取成功",
"data": {
"machine_code": "03000200-0400-0500-0006-000700080009",
"system_type": "windows",
"method": "wmic csproduct get uuid"
}
}
```
### 2. 获取系统信息
**端点:** `GET /api/v1/system/info`
**描述:** 获取当前系统的详细信息
**响应示例:**
```json
{
"success": true,
"message": "系统信息获取成功",
"data": {
"system_type": "windows",
"platform": "Windows-10-10.0.17763-SP0",
"architecture": "64bit",
"processor": "AMD64 Family 23 Model 24 Stepping 1, AuthenticAMD",
"machine_code": "03000200-0400-0500-0006-000700080009",
"python_version": "3.10.5"
}
}
```
### 3. 获取详细机器码信息
**端点:** `GET /api/v1/system/machine-code/detailed`
**描述:** 获取包含系统信息的详细机器码信息
**响应示例:**
```json
{
"success": true,
"message": "详细机器码信息获取成功",
"data": {
"machine_code": "03000200-0400-0500-0006-000700080009",
"system_type": "windows",
"method": "wmic csproduct get uuid",
"system_info": {
"system_type": "windows",
"platform": "Windows-10-10.0.17763-SP0",
"architecture": "64bit",
"processor": "AMD64 Family 23 Model 24 Stepping 1, AuthenticAMD",
"machine_code": "03000200-0400-0500-0006-000700080009",
"python_version": "3.10.5"
},
"timestamp": "2025-08-15T15:21:54.205977"
}
}
```
### 4. 系统健康检查
**端点:** `GET /api/v1/system/health`
**描述:** 检查系统基本功能是否正常
**响应示例:**
```json
{
"success": true,
"message": "系统健康检查通过",
"data": {
"system_type": "windows",
"machine_code_available": true,
"status": "healthy"
}
}
```
## 技术实现
### Windows系统
- 主要方法:`wmic csproduct get uuid`
- 备用方法:PowerShell `Get-WmiObject -Class Win32_ComputerSystemProduct`
### Linux系统
- 主要方法:`dmidecode -s system-serial-number`
- 备用方法1:读取 `/sys/class/dmi/id/product_serial`
- 备用方法2:`hostid` 命令
### 错误处理
- 命令执行超时处理
- 多种备用方案确保兼容性
- 详细的错误日志记录
## 架构设计
```
API端点层 (app/api/v1/endpoints/system.py)
服务层 (app/services/system_service.py)
工具层 (app/utils/system_utils.py)
系统命令执行
```
## 使用示例
### Python客户端
```python
import requests
# 获取机器码
response = requests.get("http://localhost:8000/api/v1/system/machine-code")
machine_code = response.json()["data"]["machine_code"]
# 获取系统信息
response = requests.get("http://localhost:8000/api/v1/system/info")
system_info = response.json()["data"]
```
### curl命令
```bash
# 获取机器码
curl -X GET "http://localhost:8000/api/v1/system/machine-code"
# 获取系统信息
curl -X GET "http://localhost:8000/api/v1/system/info"
# 系统健康检查
curl -X GET "http://localhost:8000/api/v1/system/health"
```
## 注意事项
1. **权限要求**:某些系统命令可能需要管理员权限
2. **兼容性**:不同系统版本可能有差异,已提供多种备用方案
3. **性能**:命令执行有超时限制,避免长时间阻塞
4. **安全性**:仅获取系统标识信息,不涉及敏感数据

236
test_system_api.py

@ -3,166 +3,124 @@
系统API测试脚本
"""
import asyncio
import platform
import sys
import os
# 添加项目根目录到Python路径
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
from app.utils.system_utils import SystemUtils
from app.services.system_service import SystemService
async def test_system_utils():
"""测试系统工具类"""
print("=== 测试系统工具类 ===")
# 测试获取系统类型
system_type = SystemUtils.get_system_type()
print(f"系统类型: {system_type}")
# 测试获取机器码
machine_code = SystemUtils.get_machine_code()
print(f"机器码: {machine_code}")
# 测试获取系统信息
system_info = SystemUtils.get_system_info()
print(f"系统信息: {system_info}")
print()
async def test_system_service():
"""测试系统服务类"""
print("=== 测试系统服务类 ===")
service = SystemService()
# 测试获取机器码
result = await service.get_machine_code()
print(f"获取机器码结果: {result}")
# 测试获取系统信息
result = await service.get_system_info()
print(f"获取系统信息结果: {result}")
# 测试获取详细机器码信息
result = await service.get_detailed_machine_code_info()
print(f"获取详细机器码信息结果: {result}")
print()
def test_direct_commands():
"""测试直接命令执行"""
print("=== 测试直接命令执行 ===")
import subprocess
import re
from typing import Dict, Any
import platform
def get_windows_machine_code() -> str:
"""获取Windows系统的机器码(UUID)"""
system_type = platform.system().lower()
print(f"当前系统: {system_type}")
if system_type == "windows":
try:
# 使用wmic命令获取UUID
# 测试 wmic 命令
result = subprocess.run(
['wmic', 'csproduct', 'get', 'uuid'],
["wmic", "csproduct", "get", "uuid"],
capture_output=True,
text=True,
check=True
timeout=10
)
# 解析输出,提取UUID
output = result.stdout.strip()
lines = output.split('\n')
# 查找UUID行(跳过标题行)
for line in lines:
line = line.strip()
if line and line.lower() != 'uuid':
# 提取UUID(移除可能的空格和特殊字符)
uuid_match = re.search(r'[A-Fa-f0-9]{8}-[A-Fa-f0-9]{4}-[A-Fa-f0-9]{4}-[A-Fa-f0-9]{4}-[A-Fa-f0-9]{12}', line)
if uuid_match:
return uuid_match.group(0)
# 如果没有找到标准UUID格式,返回清理后的第一行非标题内容
for line in lines:
line = line.strip()
if line and line.lower() != 'uuid':
return line
raise Exception("无法从wmic输出中提取UUID")
except subprocess.CalledProcessError as e:
print(f"执行wmic命令失败: {e}")
raise Exception(f"执行wmic命令失败: {e}")
print(f"wmic命令输出: {result.stdout}")
print(f"wmic命令错误: {result.stderr}")
print(f"wmic命令返回码: {result.returncode}")
except Exception as e:
print(f"获取Windows机器码失败: {e}")
raise Exception(f"获取Windows机器码失败: {e}")
print(f"wmic命令执行失败: {e}")
def get_linux_machine_code() -> str:
"""获取Linux系统的机器码"""
elif system_type == "linux":
try:
# 尝试多种方法获取Linux机器码
# 测试 dmidecode 命令
result = subprocess.run(
["dmidecode", "-s", "system-serial-number"],
capture_output=True,
text=True,
timeout=10
)
print(f"dmidecode命令输出: {result.stdout}")
print(f"dmidecode命令错误: {result.stderr}")
print(f"dmidecode命令返回码: {result.returncode}")
except Exception as e:
print(f"dmidecode命令执行失败: {e}")
# 方法1: 从/proc/cpuinfo获取CPU信息
try:
with open('/proc/cpuinfo', 'r') as f:
cpu_info = f.read()
# 提取processor ID
processor_match = re.search(r'processor\s+:\s+(\d+)', cpu_info)
if processor_match:
return f"CPU_PROC_{processor_match.group(1)}"
except Exception:
pass
# 方法2: 从/sys/class/dmi/id/product_uuid获取
try:
with open('/sys/class/dmi/id/product_uuid', 'r') as f:
uuid = f.read().strip()
if uuid and uuid != '00000000-0000-0000-0000-000000000000':
return uuid
except Exception:
pass
# 方法3: 使用dmidecode命令
try:
# 测试 hostid 命令
result = subprocess.run(
['dmidecode', '-s', 'system-uuid'],
["hostid"],
capture_output=True,
text=True,
check=True
timeout=5
)
uuid = result.stdout.strip()
if uuid and uuid != '00000000-0000-0000-0000-000000000000':
return uuid
except Exception:
pass
print(f"hostid命令输出: {result.stdout}")
print(f"hostid命令错误: {result.stderr}")
print(f"hostid命令返回码: {result.returncode}")
except Exception as e:
print(f"hostid命令执行失败: {e}")
# 方法4: 从/etc/machine-id获取
try:
with open('/etc/machine-id', 'r') as f:
machine_id = f.read().strip()
if machine_id:
return machine_id
except Exception:
pass
# 方法5: 从/proc/sys/kernel/random/uuid获取随机UUID
try:
with open('/proc/sys/kernel/random/uuid', 'r') as f:
uuid = f.read().strip()
if uuid:
return uuid
except Exception:
pass
print()
# 如果所有方法都失败,返回系统信息组合
system_info = f"{platform.node()}_{platform.machine()}_{platform.processor()}"
return system_info.replace(' ', '_').replace('-', '_')
async def main():
"""主测试函数"""
print("系统API功能测试")
print("=" * 50)
except Exception as e:
print(f"获取Linux机器码失败: {e}")
raise Exception(f"获取Linux机器码失败: {e}")
def test_machine_code():
"""测试获取机器码功能"""
print("=== 系统机器码测试 ===")
print(f"当前系统: {platform.system()}")
print(f"系统版本: {platform.release()}")
print(f"机器架构: {platform.machine()}")
print(f"处理器: {platform.processor()}")
print(f"主机名: {platform.node()}")
print()
# 测试系统工具类
await test_system_utils()
try:
system = platform.system().lower()
if system == "windows":
print("正在获取Windows机器码...")
machine_code = get_windows_machine_code()
method = "wmic csproduct get uuid"
elif system == "linux":
print("正在获取Linux机器码...")
machine_code = get_linux_machine_code()
method = "multiple methods (dmi, machine-id, etc.)"
else:
print(f"正在获取{system}系统机器码...")
machine_code = f"{platform.node()}_{platform.machine()}_{platform.processor()}"
method = "system info combination"
machine_code = machine_code.replace(' ', '_').replace('-', '_')
print(f"✅ 成功获取机器码")
print(f"机器码: {machine_code}")
print(f"获取方法: {method}")
print()
# 测试系统服务类
await test_system_service()
# 显示完整的系统信息
system_info = {
"system": platform.system(),
"release": platform.release(),
"version": platform.version(),
"machine": platform.machine(),
"processor": platform.processor(),
"node": platform.node(),
"platform": platform.platform(),
"python_version": platform.python_version(),
"python_implementation": platform.python_implementation()
}
print("=== 完整系统信息 ===")
for key, value in system_info.items():
print(f"{key}: {value}")
# 测试直接命令执行
test_direct_commands()
except Exception as e:
print(f"❌ 获取机器码失败: {e}")
print("测试完成")
if __name__ == "__main__":
test_machine_code()
asyncio.run(main())

44
test_system_api_endpoints.py

@ -0,0 +1,44 @@
#!/usr/bin/env python3
"""
系统API端点测试脚本
"""
import requests
import json
import time
def test_system_endpoints():
"""测试系统API端点"""
base_url = "http://localhost:8000/api/v1"
# 等待服务器启动
print("等待服务器启动...")
time.sleep(3)
endpoints = [
"/system/machine-code",
"/system/info",
"/system/machine-code/detailed",
"/system/health"
]
for endpoint in endpoints:
try:
print(f"\n=== 测试端点: {endpoint} ===")
response = requests.get(f"{base_url}{endpoint}", timeout=10)
print(f"状态码: {response.status_code}")
if response.status_code == 200:
data = response.json()
print(f"响应: {json.dumps(data, indent=2, ensure_ascii=False)}")
else:
print(f"错误响应: {response.text}")
except requests.exceptions.ConnectionError:
print("连接失败,服务器可能未启动")
except requests.exceptions.Timeout:
print("请求超时")
except Exception as e:
print(f"请求异常: {e}")
if __name__ == "__main__":
test_system_endpoints()
Loading…
Cancel
Save