Coverage for session_buddy / utils / quality_utils.py: 12.88%

109 statements  

« prev     ^ index     » next       coverage.py v7.13.1, created at 2026-01-04 00:43 -0800

1#!/usr/bin/env python3 

2"""Quality analysis utilities for session management. 

3 

4This module provides quality assessment and analysis functionality following 

5crackerjack architecture patterns with single responsibility principle. 

6""" 

7 

8from __future__ import annotations 

9 

10from contextlib import suppress 

11from typing import Any 

12 

13 

14def _extract_score_from_content(content: str) -> float | None: 

15 """Extract score from reflection content.""" 

16 with suppress(ValueError, TypeError, AttributeError): 

17 # Parse common quality score formats 

18 if "quality score:" in content: 

19 # Extract score after "quality score:" 

20 parts = content.split("quality score:") 

21 if len(parts) > 1: 

22 score_text = parts[1].split()[0] # Get first word after 

23 return _parse_score_text(score_text) 

24 return None 

25 

26 

27def _extract_score_from_metadata(reflection: dict[str, Any]) -> float | None: 

28 """Extract score from reflection metadata.""" 

29 with suppress(ValueError, TypeError, AttributeError): 

30 metadata = reflection.get("metadata", {}) 

31 if "quality_score" in metadata: 

32 score = float(metadata["quality_score"]) 

33 if 0 <= score <= 100: 

34 return score 

35 return None 

36 

37 

38def _parse_score_text(score_text: str) -> float | None: 

39 """Parse various score text formats into normalized 0-100 score.""" 

40 with suppress(ValueError, TypeError, IndexError): 

41 # Handle formats like "85/100", "0.85", "85" 

42 if "/" in score_text: 

43 numerator = float(score_text.split("/", maxsplit=1)[0]) 

44 denominator = float(score_text.split("/")[1]) 

45 score = (numerator / denominator) * 100 

46 elif "." in score_text and float(score_text) <= 1.0: 

47 score = float(score_text) * 100 

48 else: 

49 score = float(score_text) 

50 

51 if 0 <= score <= 100: 

52 return score 

53 return None 

54 

55 

56def _extract_quality_scores(reflections: list[dict[str, Any]]) -> list[float]: 

57 """Extract quality scores from reflection data.""" 

58 scores = [] 

59 

60 for reflection in reflections: 

61 try: 

62 # Look for quality score in reflection content 

63 content = reflection.get("content", "").lower() 

64 score = _extract_score_from_content(content) 

65 if score is not None: 

66 scores.append(score) 

67 continue 

68 

69 # Check metadata for score 

70 score = _extract_score_from_metadata(reflection) 

71 if score is not None: 

72 scores.append(score) 

73 

74 except (ValueError, TypeError, AttributeError): 

75 # Skip malformed scores 

76 continue 

77 

78 return scores 

79 

80 

81def _analyze_quality_trend(quality_scores: list[float]) -> tuple[str, list[str], bool]: 

82 """Analyze quality trend from historical scores.""" 

83 if len(quality_scores) < 2: 

84 return "insufficient_data", ["Not enough data to analyze trend"], False 

85 

86 # Calculate trend 

87 recent_scores = quality_scores[-5:] # Last 5 scores 

88 older_scores = ( 

89 quality_scores[-10:-5] if len(quality_scores) >= 10 else quality_scores[:-5] 

90 ) 

91 

92 if not older_scores: 

93 return "stable", ["Initial quality baseline established"], True 

94 

95 recent_avg = sum(recent_scores) / len(recent_scores) 

96 older_avg = sum(older_scores) / len(older_scores) 

97 

98 difference = recent_avg - older_avg 

99 insights = [] 

100 improving = False 

101 

102 if difference > 5: 

103 trend = "improving" 

104 improving = True 

105 insights.extend( 

106 [ 

107 f"📈 Quality improving: +{difference:.1f} points", 

108 "🎯 Continue current development practices", 

109 ], 

110 ) 

111 elif difference < -5: 

112 trend = "declining" 

113 insights.extend( 

114 [ 

115 f"📉 Quality declining: {difference:.1f} points", 

116 "⚠️ Review recent changes and processes", 

117 ], 

118 ) 

119 else: 

120 trend = "stable" 

121 improving = True 

122 insights.extend( 

123 [ 

124 f"📊 Quality stable: {difference:+.1f} points variation", 

125 "✅ Maintaining consistent development standards", 

126 ], 

127 ) 

128 

129 # Add specific recommendations based on score level 

130 current_score = recent_scores[-1] if recent_scores else 0 

131 if current_score < 70: 

132 insights.append("🔧 Focus on code quality improvements") 

133 elif current_score > 90: 

134 insights.append("⭐ Excellent quality standards maintained") 

135 

136 return trend, insights, improving 

137 

138 

139# Remove the duplicate function 

140 

141 

142def _generate_quality_trend_recommendations(scores: list[float]) -> list[str]: 

143 """Generate specific recommendations based on quality trend analysis.""" 

144 if not scores: 

145 return ["📊 Start tracking quality metrics for trend analysis"] 

146 

147 recommendations = [] 

148 current_score = scores[-1] 

149 

150 # Score-based recommendations 

151 if current_score < 60: 

152 recommendations.extend( 

153 [ 

154 "🚨 Critical: Immediate quality improvement needed", 

155 "• Run comprehensive code review and testing", 

156 "• Focus on reducing technical debt", 

157 "• Consider pair programming for complex changes", 

158 ], 

159 ) 

160 elif current_score < 75: 

161 recommendations.extend( 

162 [ 

163 "⚠️ Quality below target: Focus on improvement", 

164 "• Increase test coverage and documentation", 

165 "• Review and refactor complex code sections", 

166 ], 

167 ) 

168 elif current_score < 90: 

169 recommendations.extend( 

170 [ 

171 "✅ Good quality: Minor optimizations available", 

172 "• Fine-tune linting and formatting rules", 

173 "• Enhance error handling and logging", 

174 ], 

175 ) 

176 else: 

177 recommendations.extend( 

178 [ 

179 "⭐ Excellent quality: Maintain current standards", 

180 "• Share best practices with team", 

181 "• Document successful patterns for reuse", 

182 ], 

183 ) 

184 

185 # Trend-based recommendations 

186 if len(scores) >= 3: 

187 recent_trend = scores[-3:] 

188 if all( 

189 recent_trend[i] < recent_trend[i + 1] for i in range(len(recent_trend) - 1) 

190 ): 

191 recommendations.append("📈 Positive trend: Continue current practices") 

192 elif all( 

193 recent_trend[i] > recent_trend[i + 1] for i in range(len(recent_trend) - 1) 

194 ): 

195 recommendations.append("📉 Declining trend: Review recent changes") 

196 

197 return recommendations 

198 

199 

200def _get_time_based_recommendations(hour: int) -> list[str]: 

201 """Generate recommendations based on current time of day.""" 

202 recommendations = [] 

203 

204 if 6 <= hour < 12: # Morning 204 ↛ 205line 204 didn't jump to line 205 because the condition on line 204 was never true

205 recommendations.extend( 

206 [ 

207 "🌅 Morning session: Good time for complex problem-solving", 

208 "• Focus on architecture and design decisions", 

209 "• Plan day's development priorities", 

210 ], 

211 ) 

212 elif 12 <= hour < 17: # Afternoon 212 ↛ 213line 212 didn't jump to line 213 because the condition on line 212 was never true

213 recommendations.extend( 

214 [ 

215 "☀️ Afternoon session: Peak productivity time", 

216 "• Implement planned features and fixes", 

217 "• Conduct code reviews and testing", 

218 ], 

219 ) 

220 elif 17 <= hour < 21: # Evening 220 ↛ 221line 220 didn't jump to line 221 because the condition on line 220 was never true

221 recommendations.extend( 

222 [ 

223 "🌆 Evening session: Good for documentation and cleanup", 

224 "• Update documentation and comments", 

225 "• Refactor and optimize existing code", 

226 ], 

227 ) 

228 else: # Late night/early morning 

229 recommendations.extend( 

230 [ 

231 "🌙 Late session: Focus on simple, well-tested changes", 

232 "• Avoid complex architectural changes", 

233 "• Consider shorter development sessions", 

234 ], 

235 ) 

236 

237 return recommendations 

238 

239 

240def _ensure_default_recommendations(priority_actions: list[str]) -> list[str]: 

241 """Ensure there are always some recommendations available.""" 

242 if not priority_actions: 

243 return [ 

244 "🎯 Focus on current development goals", 

245 "📝 Keep documentation updated", 

246 "🧪 Maintain test coverage", 

247 "🔍 Regular code quality checks", 

248 ] 

249 return priority_actions 

250 

251 

252def _get_intelligence_error_result(error: Exception) -> dict[str, Any]: 

253 """Generate error result for intelligence system failures.""" 

254 return { 

255 "success": False, 

256 "error": f"Intelligence system error: {error}", 

257 "recommendations": [ 

258 "⚠️ Intelligence features temporarily unavailable", 

259 "• Basic session management tools still functional", 

260 "• Manual quality assessment recommended", 

261 "• Check system dependencies and configuration", 

262 ], 

263 "fallback_mode": True, 

264 }