Coverage for src\performance.py: 0%

1""" 

2PAO系统性能优化模块 

3 

4提供： 

5- 数据库查询优化 

6- 网络通信压缩 

7- 内存使用优化 

8- 缓存管理 

9""" 

10 

11import time 

12import asyncio 

13import psutil 

14import logging 

15from typing import Dict, Any, Optional, Callable 

16from dataclasses import dataclass 

17from functools import wraps 

18import hashlib 

19import json 

20 

21logger = logging.getLogger(__name__) 

22 

23 

24@dataclass 

25class PerformanceMetrics: 

26 """性能指标""" 

27 operation: str 

28 duration_ms: float 

29 memory_before_mb: float 

30 memory_after_mb: float 

31 timestamp: float 

32 

33 

34class PerformanceMonitor: 

35 """性能监控器""" 

36 

37 def __init__(self): 

38 self.metrics: list[PerformanceMetrics] = [] 

39 self.process = psutil.Process() 

40 

41 def record(self, operation: str, duration_ms: float, 

42 memory_before_mb: float, memory_after_mb: float): 

43 """记录性能指标""" 

44 metric = PerformanceMetrics( 

45 operation=operation, 

46 duration_ms=duration_ms, 

47 memory_before_mb=memory_before_mb, 

48 memory_after_mb=memory_after_mb, 

49 timestamp=time.time() 

50 ) 

51 self.metrics.append(metric) 

52 

53 def get_metrics(self, operation: Optional[str] = None) -> list[PerformanceMetrics]: 

54 """获取性能指标""" 

55 if operation: 

56 return [m for m in self.metrics if m.operation == operation] 

57 return self.metrics 

58 

59 def get_average_duration(self, operation: str) -> float: 

60 """获取平均执行时间""" 

61 metrics = self.get_metrics(operation) 

62 if not metrics: 

63 return 0.0 

64 return sum(m.duration_ms for m in metrics) / len(metrics) 

65 

66 def clear(self): 

67 """清空指标""" 

68 self.metrics.clear() 

69 

70 

71class CacheManager: 

72 """缓存管理器""" 

73 

74 def __init__(self, max_size_mb: int = 100, ttl_seconds: int = 300): 

75 self.max_size_mb = max_size_mb 

76 self.ttl_seconds = ttl_seconds 

77 self._cache: Dict[str, tuple[Any, float]] = {} 

78 self._hits = 0 

79 self._misses = 0 

80 

81 def get(self, key: str) -> Optional[Any]: 

82 """获取缓存""" 

83 if key in self._cache: 

84 value, timestamp = self._cache[key] 

85 if time.time() - timestamp < self.ttl_seconds: 

86 self._hits += 1 

87 return value 

88 else: 

89 del self._cache[key] 

90 self._misses += 1 

91 return None 

92 

93 def set(self, key: str, value: Any): 

94 """设置缓存""" 

95 self._cache[key] = (value, time.time()) 

96 

97 def invalidate(self, key: str): 

98 """使缓存失效""" 

99 if key in self._cache: 

100 del self._cache[key] 

101 

102 def clear(self): 

103 """清空缓存""" 

104 self._cache.clear() 

105 self._hits = 0 

106 self._misses = 0 

107 

108 def get_stats(self) -> Dict[str, Any]: 

109 """获取缓存统计""" 

110 total = self._hits + self._misses 

111 hit_rate = self._hits / total if total > 0 else 0 

112 return { 

113 "size": len(self._cache), 

114 "hits": self._hits, 

115 "misses": self._misses, 

116 "hit_rate": hit_rate 

117 } 

118 

119 

120def timed(metric_name: str = "operation"): 

121 """性能计时装饰器""" 

122 def decorator(func: Callable): 

123 @wraps(func) 

124 async def async_wrapper(*args, **kwargs): 

125 start = time.perf_counter() 

126 mem_before = psutil.Process().memory_info().rss / 1024 / 1024 

127 try: 

128 result = await func(*args, **kwargs) 

129 return result 

130 finally: 

131 duration = (time.perf_counter() - start) * 1000 

132 mem_after = psutil.Process().memory_info().rss / 1024 / 1024 

133 logger.debug(f"{metric_name}: {duration:.2f}ms, mem: {mem_before:.1f}->{mem_after:.1f}MB") 

134 @wraps(func) 

135 def sync_wrapper(*args, **kwargs): 

136 start = time.perf_counter() 

137 mem_before = psutil.Process().memory_info().rss / 1024 / 1024 

138 try: 

139 result = func(*args, **kwargs) 

140 return result 

141 finally: 

142 duration = (time.perf_counter() - start) * 1000 

143 mem_after = psutil.Process().memory_info().rss / 1024 / 1024 

144 logger.debug(f"{metric_name}: {duration:.2f}ms, mem: {mem_before:.1f}->{mem_after:.1f}MB") 

145 if asyncio.iscoroutinefunction(func): 

146 return async_wrapper 

147 return sync_wrapper 

148 return decorator 

149 

150 

151class DataCompressor: 

152 """数据压缩器""" 

153 

154 @staticmethod 

155 def compress(data: str) -> bytes: 

156 """压缩数据（使用简单编码）""" 

157 if not data: 

158 return b"" 

159 # 简单压缩：移除多余空白 

160 compressed = " ".join(data.split()) 

161 return compressed.encode('utf-8') 

162 

163 @staticmethod 

164 def decompress(data: bytes) -> str: 

165 """解压数据""" 

166 if not data: 

167 return "" 

168 return data.decode('utf-8') 

169 

170 @staticmethod 

171 def should_compress(data: str, threshold: int = 1024) -> bool: 

172 """判断是否应该压缩""" 

173 return len(data) > threshold 

174 

175 

176class QueryOptimizer: 

177 """查询优化器""" 

178 

179 def __init__(self): 

180 self._query_cache = CacheManager(max_size_mb=50, ttl_seconds=60) 

181 

182 def get_cached_query(self, query_key: str) -> Optional[Any]: 

183 """获取缓存的查询结果""" 

184 return self._query_cache.get(query_key) 

185 

186 def cache_query_result(self, query_key: str, result: Any): 

187 """缓存查询结果""" 

188 self._query_cache.set(query_key, result) 

189 

190 @staticmethod 

191 def generate_query_key(query: str, params: Dict[str, Any]) -> str: 

192 """生成查询缓存键""" 

193 key_data = json.dumps({"query": query, "params": params}, sort_keys=True) 

194 return hashlib.md5(key_data.encode()).hexdigest() 

195 

196 

197class MemoryOptimizer: 

198 """内存优化器""" 

199 

200 def __init__(self): 

201 self.process = psutil.Process() 

202 self._peak_memory_mb = 0 

203 

204 def get_current_memory_mb(self) -> float: 

205 """获取当前内存使用（MB）""" 

206 return self.process.memory_info().rss / 1024 / 1024 

207 

208 def get_memory_percent(self) -> float: 

209 """获取内存使用百分比""" 

210 return self.process.memory_percent() 

211 

212 def track_peak(self): 

213 """追踪峰值内存""" 

214 current = self.get_current_memory_mb() 

215 if current > self._peak_memory_mb: 

216 self._peak_memory_mb = current 

217 

218 def get_peak_memory_mb(self) -> float: 

219 """获取峰值内存""" 

220 return self._peak_memory_mb 

221 

222 def suggest_gc(self, threshold_mb: int = 500) -> bool: 

223 """建议是否进行垃圾回收""" 

224 current = self.get_current_memory_mb() 

225 if current > threshold_mb: 

226 import gc 

227 gc.collect() 

228 return True 

229 return False 

230 

231 

232# 全局性能监控实例 

233performance_monitor = PerformanceMonitor() 

234query_optimizer = QueryOptimizer() 

235memory_optimizer = MemoryOptimizer()