Coverage for src/secchi/services/comparison.py: 76%
151 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-08-04 22:15 +0000
1"""Evidence-based package comparison and agent recommendation logic.
3This module is deliberately read-only and deterministic. It consumes the same
4``IntelligenceResult`` objects as the CLI, dashboard, reports, and MCP server;
5it does not fetch data or make installation decisions on a user's behalf.
6"""
8from __future__ import annotations
10import math
11from dataclasses import dataclass, field
12from datetime import datetime, timezone
13from enum import Enum
15from secchi.models import PackageInfo, PackageRef
16from secchi.schema import COMPARISON_SCHEMA_VERSION
17from secchi.schemas import ComparisonExport
18from secchi.services.intelligence import IntelligenceResult
21class Recommendation(str, Enum):
22 RECOMMENDED = "Recommended"
23 ACCEPTABLE = "Acceptable"
24 CAUTION = "Use with caution"
25 AVOID = "Avoid"
28@dataclass(frozen=True)
29class ComparisonCandidate:
30 ref: PackageRef
31 recommendation: Recommendation
32 score: float | None
33 confidence: float
34 health_score: int | None
35 latest_version: str | None
36 adoption_change_pct: float | None
37 github_stars: int | None
38 strengths: list[str] = field(default_factory=list)
39 concerns: list[str] = field(default_factory=list)
40 evidence: list[str] = field(default_factory=list)
41 warnings: list[dict[str, str]] = field(default_factory=list)
42 error: str | None = None
45@dataclass(frozen=True)
46class ComparisonResult:
47 candidates: list[ComparisonCandidate]
48 recommendation_basis: str = (
49 "Health, momentum, community, release recency, and data completeness; "
50 "unknown signals reduce confidence."
51 )
53 @property
54 def winner(self) -> ComparisonCandidate | None:
55 usable = [
56 candidate for candidate in self.candidates if candidate.score is not None
57 ]
58 return usable[0] if usable else None
60 def as_dict(self) -> dict:
61 payload = {
62 "schema_version": COMPARISON_SCHEMA_VERSION,
63 "schema": "secchi.package-comparison",
64 "generated_by": "Secchi",
65 "recommendation_basis": self.recommendation_basis,
66 "winner": _candidate_dict(self.winner) if self.winner else None,
67 "candidates": [_candidate_dict(candidate) for candidate in self.candidates],
68 }
69 return ComparisonExport.model_validate(payload).model_dump(
70 mode="json", by_alias=True
71 )
74def compare_intelligence(results: list[IntelligenceResult]) -> ComparisonResult:
75 """Rank fetched package intelligence for an agent-readable decision."""
76 candidates = [evaluate_candidate(result) for result in results]
77 candidates.sort(
78 key=lambda candidate: (
79 candidate.score is not None,
80 candidate.score if candidate.score is not None else -1,
81 ),
82 reverse=True,
83 )
84 return ComparisonResult(candidates=candidates)
87def evaluate_candidate(result: IntelligenceResult) -> ComparisonCandidate:
88 """Turn one intelligence result into a recommendation with evidence."""
89 ref = result.ref
90 if result.error or result.info is None or result.derived is None:
91 return ComparisonCandidate(
92 ref=ref,
93 recommendation=Recommendation.AVOID,
94 score=None,
95 confidence=0.0,
96 health_score=None,
97 latest_version=None,
98 adoption_change_pct=None,
99 github_stars=None,
100 concerns=["Package intelligence could not be fetched."],
101 error=(getattr(result.error, "message", None) or str(result.error))
102 if result.error
103 else "No package data returned.",
104 )
106 info = result.info
107 derived = result.derived
108 health = derived.health_score.total
109 momentum = _momentum_score(derived.downloads_30d_pct_change)
110 community = _community_score(info.github_stats.stars, info.github_stats.resolved)
111 recency = _recency_score(info)
112 completeness = _completeness_score(info)
113 score = round(
114 health * 0.55
115 + momentum * 0.15
116 + community * 0.10
117 + recency * 0.10
118 + completeness * 0.10,
119 1,
120 )
121 confidence = round(completeness / 100, 2)
123 strengths: list[str] = []
124 concerns: list[str] = []
125 evidence: list[str] = [f"Health score: {health}/100"]
126 if health >= 80:
127 strengths.append("Strong overall health")
128 elif health < 60:
129 concerns.append("Health score is below the preferred range")
130 if derived.downloads_30d_pct_change is not None:
131 change = derived.downloads_30d_pct_change
132 evidence.append(f"30-day adoption change: {change:+.1f}%")
133 if change >= 10:
134 strengths.append("Adoption is growing")
135 elif change < -10:
136 concerns.append("Adoption is declining")
137 else:
138 concerns.append("Adoption trend has no comparison baseline")
139 if info.github_stats.resolved and info.github_stats.stars > 0:
140 evidence.append(f"GitHub stars: {info.github_stats.stars:,}")
141 if info.github_stats.stars >= 1000:
142 strengths.append("Established community signal")
143 else:
144 concerns.append("Repository or community data is unavailable")
145 if info.github_stats.has_ci:
146 strengths.append("Repository CI detected")
147 else:
148 concerns.append("Repository CI was not detected")
149 if info.latest_version:
150 evidence.append(f"Latest version: {info.latest_version}")
151 else:
152 concerns.append("Latest version is unavailable")
153 if info.versions and info.versions[0].is_yanked:
154 concerns.append("Latest release is yanked")
156 warnings = [
157 {"source": warning.source, "message": warning.message}
158 for warning in result.warnings
159 ]
160 if warnings:
161 concerns.append(f"{len(warnings)} enrichment signal(s) unavailable")
163 recommendation = _recommendation(score, confidence, info)
164 return ComparisonCandidate(
165 ref=ref,
166 recommendation=recommendation,
167 score=score,
168 confidence=confidence,
169 health_score=health,
170 latest_version=info.latest_version or None,
171 adoption_change_pct=derived.downloads_30d_pct_change,
172 github_stars=info.github_stats.stars if info.github_stats.resolved else None,
173 strengths=strengths,
174 concerns=concerns,
175 evidence=evidence,
176 warnings=warnings,
177 )
180def _recommendation(
181 score: float, confidence: float, info: PackageInfo
182) -> Recommendation:
183 if (
184 not info.latest_version
185 or (info.versions and info.versions[0].is_yanked)
186 or score < 40
187 ):
188 return Recommendation.AVOID
189 if score >= 80 and confidence >= 0.65:
190 return Recommendation.RECOMMENDED
191 if score >= 65 and confidence >= 0.45:
192 return Recommendation.ACCEPTABLE
193 return Recommendation.CAUTION
196def _momentum_score(change: float | None) -> float:
197 if change is None:
198 return 50
199 return max(0, min(100, 50 + change * 2))
202def _community_score(stars: int, resolved: bool) -> float:
203 if not resolved:
204 return 0
205 if stars <= 0:
206 return 20
207 return min(100, 20 + math.log10(stars) * 25)
210def _recency_score(info: PackageInfo) -> float:
211 date = info.latest_release_date
212 if date is None:
213 return 0
214 if date.tzinfo is None:
215 date = date.replace(tzinfo=timezone.utc)
216 age_days = max(0, (datetime.now(timezone.utc) - date).days)
217 if age_days <= 90:
218 return 100
219 if age_days <= 180:
220 return 80
221 if age_days <= 365:
222 return 55
223 if age_days <= 730:
224 return 25
225 return 0
228def _completeness_score(info: PackageInfo) -> float:
229 signals = [
230 bool(info.latest_version),
231 bool(info.latest_release_date),
232 bool(info.download_trend or info.download_counts.month),
233 info.github_stats.resolved,
234 bool(info.repository_url or info.homepage),
235 bool(info.versions),
236 info.reverse_dependency_count is not None,
237 bool(info.dependencies),
238 ]
239 return round(sum(signals) / len(signals) * 100, 1)
242def _candidate_dict(candidate: ComparisonCandidate | None) -> dict | None:
243 if candidate is None:
244 return None
245 return {
246 "package": candidate.ref.name,
247 "registry": candidate.ref.registry.value,
248 "recommendation": candidate.recommendation.value,
249 "score": candidate.score,
250 "confidence": candidate.confidence,
251 "health_score": candidate.health_score,
252 "latest_version": candidate.latest_version,
253 "adoption_change_pct": candidate.adoption_change_pct,
254 "github_stars": candidate.github_stars,
255 "strengths": candidate.strengths,
256 "concerns": candidate.concerns,
257 "evidence": candidate.evidence,
258 "warnings": candidate.warnings,
259 "error": candidate.error,
260 }
263def render_comparison(result: ComparisonResult) -> str:
264 """Render a compact terminal comparison while retaining agent JSON support."""
265 lines = ["Package comparison", "=" * 72]
266 for index, candidate in enumerate(result.candidates, start=1):
267 ref = f"{candidate.ref.name} ({candidate.ref.registry.value})"
268 if candidate.score is None:
269 lines.append(f"{index}. {ref} — {candidate.recommendation.value}")
270 lines.append(f" Error: {candidate.error}")
271 continue
272 change = (
273 "—"
274 if candidate.adoption_change_pct is None
275 else f"{candidate.adoption_change_pct:+.1f}%"
276 )
277 stars = "—" if candidate.github_stars is None else f"{candidate.github_stars:,}"
278 lines.extend(
279 [
280 f"{index}. {ref} — {candidate.recommendation.value}",
281 f" Score {candidate.score:.1f} Confidence {candidate.confidence:.0%} Health {candidate.health_score}/100",
282 f" Latest {candidate.latest_version or '—'} Adoption {change} GitHub stars {stars}",
283 ]
284 )
285 if candidate.strengths:
286 lines.append(f" Strengths: {', '.join(candidate.strengths)}")
287 if candidate.concerns:
288 lines.append(f" Concerns: {', '.join(candidate.concerns)}")
289 if result.winner:
290 lines.append("")
291 lines.append(
292 f"Recommendation: {result.winner.ref.name} ({result.winner.ref.registry.value}) "
293 f"— {result.winner.recommendation.value}."
294 )
295 lines.append(
296 "Note: Secchi provides advisory evidence; review compatibility, license, and security requirements before adopting."
297 )
298 return "\n".join(lines)