diff --git a/app/api/v1/endpoints/system.py b/app/api/v1/endpoints/system.py new file mode 100644 index 0000000..3958606 --- /dev/null +++ b/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)}" + ) diff --git a/app/core/app/router.py b/app/core/app/router.py index 6c1f226..796d03c 100644 --- a/app/core/app/router.py +++ b/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("路由注册配置完成") diff --git a/app/schemas/system.py b/app/schemas/system.py new file mode 100644 index 0000000..772e07c --- /dev/null +++ b/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="错误信息") diff --git a/app/services/system_service.py b/app/services/system_service.py new file mode 100644 index 0000000..406a4bf --- /dev/null +++ b/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() diff --git a/app/utils/system_utils.py b/app/utils/system_utils.py new file mode 100644 index 0000000..abf673b --- /dev/null +++ b/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) + } diff --git a/docs/modify.md b/docs/modify.md index 3bbce1c..90b51b1 100644 --- a/docs/modify.md +++ b/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使用 diff --git a/docs/system_api.md b/docs/system_api.md new file mode 100644 index 0000000..3e05bfb --- /dev/null +++ b/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. **安全性**：仅获取系统标识信息，不涉及敏感数据 diff --git a/test_system_api.py b/test_system_api.py index e7c9b15..cd95d58 100644 --- a/test_system_api.py +++ b/test_system_api.py @@ -3,166 +3,124 @@ 系统API测试脚本 """ import asyncio -import platform -import subprocess -import re -from typing import Dict, Any +import sys +import os -def get_windows_machine_code() -> str: - """获取Windows系统的机器码（UUID）""" - try: - # 使用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) - - # 如果没有找到标准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}") - except Exception as e: - print(f"获取Windows机器码失败: {e}") - raise Exception(f"获取Windows机器码失败: {e}") +# 添加项目根目录到Python路径 +sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) -def get_linux_machine_code() -> str: - """获取Linux系统的机器码""" - try: - # 尝试多种方法获取Linux机器码 - - # 方法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命令 +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 platform + + system_type = platform.system().lower() + print(f"当前系统: {system_type}") + + if system_type == "windows": try: + # 测试 wmic 命令 result = subprocess.run( - ['dmidecode', '-s', 'system-uuid'], + ["wmic", "csproduct", "get", "uuid"], capture_output=True, text=True, - check=True + timeout=10 ) - uuid = result.stdout.strip() - if uuid and uuid != '00000000-0000-0000-0000-000000000000': - return uuid - except Exception: - pass - - # 方法4: 从/etc/machine-id获取 + print(f"wmic命令输出: {result.stdout}") + print(f"wmic命令错误: {result.stderr}") + print(f"wmic命令返回码: {result.returncode}") + except Exception as e: + print(f"wmic命令执行失败: {e}") + + elif system_type == "linux": try: - with open('/etc/machine-id', 'r') as f: - machine_id = f.read().strip() - if machine_id: - return machine_id - except Exception: - pass + # 测试 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}") - # 方法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 - - # 如果所有方法都失败，返回系统信息组合 - system_info = f"{platform.node()}_{platform.machine()}_{platform.processor()}" - return system_info.replace(' ', '_').replace('-', '_') - - 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()}") + # 测试 hostid 命令 + result = subprocess.run( + ["hostid"], + capture_output=True, + text=True, + timeout=5 + ) + print(f"hostid命令输出: {result.stdout}") + print(f"hostid命令错误: {result.stderr}") + print(f"hostid命令返回码: {result.returncode}") + except Exception as e: + print(f"hostid命令执行失败: {e}") + print() + +async def main(): + """主测试函数""" + print("系统API功能测试") + print("=" * 50) - 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() - - # 显示完整的系统信息 - 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}") - - except Exception as e: - print(f"❌ 获取机器码失败: {e}") + # 测试系统工具类 + await test_system_utils() + + # 测试系统服务类 + await test_system_service() + + # 测试直接命令执行 + test_direct_commands() + + print("测试完成") if __name__ == "__main__": - test_machine_code() + asyncio.run(main()) diff --git a/test_system_api_endpoints.py b/test_system_api_endpoints.py new file mode 100644 index 0000000..5232d80 --- /dev/null +++ b/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()