248 lines
7.4 KiB
Python
248 lines
7.4 KiB
Python
"""
|
|
通信工具类模块
|
|
提供实用工具函数
|
|
"""
|
|
|
|
import time
|
|
import hashlib
|
|
import json
|
|
from typing import Dict, Any, List, Optional
|
|
|
|
class CommUtils:
|
|
"""
|
|
通信工具类
|
|
提供实用工具函数
|
|
"""
|
|
|
|
@staticmethod
|
|
def generate_unique_id(prefix: str = "") -> str:
|
|
"""
|
|
生成唯一ID
|
|
|
|
Args:
|
|
prefix: ID前缀
|
|
|
|
Returns:
|
|
生成的唯一ID
|
|
"""
|
|
try:
|
|
timestamp = int(time.time() * 1000000) # 微秒时间戳
|
|
unique_id = f"{prefix}{timestamp}" if prefix else str(timestamp)
|
|
return unique_id
|
|
except Exception as e:
|
|
print(f"✗ 唯一ID生成失败: {e}")
|
|
return f"error_id_{int(time.time() * 1000000)}"
|
|
|
|
@staticmethod
|
|
def calculate_checksum(data: bytes) -> str:
|
|
"""
|
|
计算数据校验和
|
|
|
|
Args:
|
|
data: 要计算校验和的数据
|
|
|
|
Returns:
|
|
校验和字符串
|
|
"""
|
|
try:
|
|
checksum = hashlib.md5(data).hexdigest()
|
|
return checksum
|
|
except Exception as e:
|
|
print(f"✗ 校验和计算失败: {e}")
|
|
return ""
|
|
|
|
@staticmethod
|
|
def format_bytes(bytes_count: int) -> str:
|
|
"""
|
|
格式化字节数
|
|
|
|
Args:
|
|
bytes_count: 字节数
|
|
|
|
Returns:
|
|
格式化后的字符串
|
|
"""
|
|
try:
|
|
if bytes_count < 1024:
|
|
return f"{bytes_count} B"
|
|
elif bytes_count < 1024 * 1024:
|
|
return f"{bytes_count / 1024:.2f} KB"
|
|
elif bytes_count < 1024 * 1024 * 1024:
|
|
return f"{bytes_count / (1024 * 1024):.2f} MB"
|
|
else:
|
|
return f"{bytes_count / (1024 * 1024 * 1024):.2f} GB"
|
|
except Exception as e:
|
|
print(f"✗ 字节格式化失败: {e}")
|
|
return f"{bytes_count} B"
|
|
|
|
@staticmethod
|
|
def validate_json(data: str) -> bool:
|
|
"""
|
|
验证JSON数据
|
|
|
|
Args:
|
|
data: JSON字符串
|
|
|
|
Returns:
|
|
是否为有效JSON
|
|
"""
|
|
try:
|
|
json.loads(data)
|
|
return True
|
|
except Exception:
|
|
return False
|
|
|
|
@staticmethod
|
|
def sanitize_input(input_str: str) -> str:
|
|
"""
|
|
清理输入字符串
|
|
|
|
Args:
|
|
input_str: 输入字符串
|
|
|
|
Returns:
|
|
清理后的字符串
|
|
"""
|
|
try:
|
|
# 移除潜在的危险字符
|
|
sanitized = input_str.replace("<", "<").replace(">", ">")
|
|
sanitized = sanitized.replace("\"", """).replace("'", "'")
|
|
return sanitized
|
|
except Exception as e:
|
|
print(f"✗ 输入清理失败: {e}")
|
|
return input_str
|
|
|
|
@staticmethod
|
|
def create_response(success: bool, message: str = "", data: Any = None) -> Dict[str, Any]:
|
|
"""
|
|
创建响应数据
|
|
|
|
Args:
|
|
success: 是否成功
|
|
message: 消息
|
|
data: 数据
|
|
|
|
Returns:
|
|
响应数据字典
|
|
"""
|
|
try:
|
|
response = {
|
|
"success": success,
|
|
"message": message,
|
|
"timestamp": time.time()
|
|
}
|
|
|
|
if data is not None:
|
|
response["data"] = data
|
|
|
|
return response
|
|
except Exception as e:
|
|
print(f"✗ 响应创建失败: {e}")
|
|
return {
|
|
"success": False,
|
|
"message": "响应创建失败",
|
|
"timestamp": time.time()
|
|
}
|
|
|
|
@staticmethod
|
|
def merge_dicts(dict1: Dict[str, Any], dict2: Dict[str, Any]) -> Dict[str, Any]:
|
|
"""
|
|
合并字典（递归）
|
|
|
|
Args:
|
|
dict1: 第一个字典
|
|
dict2: 第二个字典
|
|
|
|
Returns:
|
|
合并后的字典
|
|
"""
|
|
try:
|
|
result = dict1.copy()
|
|
|
|
for key, value in dict2.items():
|
|
if key in result and isinstance(result[key], dict) and isinstance(value, dict):
|
|
# 递归合并嵌套字典
|
|
result[key] = CommUtils.merge_dicts(result[key], value)
|
|
else:
|
|
# 直接覆盖
|
|
result[key] = value
|
|
|
|
return result
|
|
except Exception as e:
|
|
print(f"✗ 字典合并失败: {e}")
|
|
return dict1
|
|
|
|
@staticmethod
|
|
def get_system_info() -> Dict[str, Any]:
|
|
"""
|
|
获取系统信息
|
|
|
|
Returns:
|
|
系统信息字典
|
|
"""
|
|
try:
|
|
import platform
|
|
import psutil
|
|
|
|
# 系统信息
|
|
system_info = {
|
|
"platform": platform.system(),
|
|
"platform_version": platform.version(),
|
|
"architecture": platform.architecture()[0],
|
|
"processor": platform.processor(),
|
|
"python_version": platform.python_version()
|
|
}
|
|
|
|
# CPU信息
|
|
cpu_info = {
|
|
"physical_cores": psutil.cpu_count(logical=False),
|
|
"total_cores": psutil.cpu_count(logical=True),
|
|
"max_frequency": psutil.cpu_freq().max if psutil.cpu_freq() else 0,
|
|
"current_frequency": psutil.cpu_freq().current if psutil.cpu_freq() else 0
|
|
}
|
|
|
|
# 内存信息
|
|
memory_info = psutil.virtual_memory()
|
|
memory_stats = {
|
|
"total": memory_info.total,
|
|
"available": memory_info.available,
|
|
"used": memory_info.used,
|
|
"percentage": memory_info.percent
|
|
}
|
|
|
|
# 网络信息
|
|
net_info = psutil.net_io_counters()
|
|
network_stats = {
|
|
"bytes_sent": net_info.bytes_sent,
|
|
"bytes_recv": net_info.bytes_recv,
|
|
"packets_sent": net_info.packets_sent,
|
|
"packets_recv": net_info.packets_recv
|
|
}
|
|
|
|
return {
|
|
"system": system_info,
|
|
"cpu": cpu_info,
|
|
"memory": memory_stats,
|
|
"network": network_stats,
|
|
"timestamp": time.time()
|
|
}
|
|
|
|
except Exception as e:
|
|
print(f"✗ 系统信息获取失败: {e}")
|
|
return {
|
|
"system": {},
|
|
"cpu": {},
|
|
"memory": {},
|
|
"network": {},
|
|
"timestamp": time.time()
|
|
}
|
|
|
|
# 工具函数别名
|
|
generate_id = CommUtils.generate_unique_id
|
|
calculate_checksum = CommUtils.calculate_checksum
|
|
format_bytes = CommUtils.format_bytes
|
|
validate_json = CommUtils.validate_json
|
|
sanitize_input = CommUtils.sanitize_input
|
|
create_response = CommUtils.create_response
|
|
merge_dicts = CommUtils.merge_dicts
|
|
get_system_info = CommUtils.get_system_info |