Coverage for src/secchi/derived.py: 78%
210 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-08-04 22:15 +0000
1"""Derived metrics — pure functions over an already-fetched PackageInfo.
3No network or filesystem I/O lives here: everything is arithmetic over data the
4adapters/utils already fetched. This keeps the "real vs. derived" boundary sharp
5and makes the scoring logic unit-testable without mocking httpx.
6"""
8from __future__ import annotations
10from datetime import datetime, timezone
12from secchi.models import (
13 ActivityEvent,
14 ActivityEventKind,
15 DerivedPackageData,
16 HealthScore,
17 HealthSubScore,
18 InstallBreakdown,
19 InstallMethod,
20 PackageInfo,
21 Registry,
22 ReverseDependencySummary,
23)
26def _now() -> datetime:
27 return datetime.now(timezone.utc)
30def _aware(dt: datetime | None) -> datetime | None:
31 if dt is None:
32 return None
33 return dt if dt.tzinfo else dt.replace(tzinfo=timezone.utc)
36def _days_since(dt: datetime | None) -> float | None:
37 dt = _aware(dt)
38 if dt is None:
39 return None
40 return (_now() - dt).total_seconds() / 86400
43# ── Health score ─────────────────────────────────────────────────────────────
46def _score_maintained(info: PackageInfo) -> int:
47 candidates = [
48 _days_since(info.latest_release_date),
49 _days_since(info.github_stats.pushed_at),
50 ]
51 ages = [d for d in candidates if d is not None]
52 if not ages:
53 return 0
54 days = min(ages)
55 if days <= 30:
56 return 20
57 if days <= 90:
58 return 16
59 if days <= 180:
60 return 12
61 if days <= 365:
62 return 6
63 return 2
66def _score_documentation(info: PackageInfo) -> int:
67 score = 0
68 if info.homepage:
69 score += 5
70 if info.documentation_url:
71 score += 8
72 if info.github_stats.has_readme:
73 score += 7
74 return min(score, 20)
77def _score_testing(info: PackageInfo) -> int:
78 # Weak proxy: presence of CI workflows. Not test coverage.
79 return 20 if (info.github_stats.resolved and info.github_stats.has_ci) else 0
82def _issue_counts_90d(info: PackageInfo) -> tuple[int, int]:
83 now = _now()
84 opened = closed = 0
85 for ev in info.github_issue_events:
86 created = _aware(ev.created_at)
87 if created and (now - created).days <= 90:
88 opened += 1
89 closed_at = _aware(ev.closed_at)
90 if closed_at and (now - closed_at).days <= 90:
91 closed += 1
92 return opened, closed
95def _score_community(info: PackageInfo) -> int:
96 gh = info.github_stats
97 if not gh.resolved:
98 return 0
100 stars = gh.stars
101 if stars >= 100_000:
102 star_pts = 10
103 elif stars >= 10_000:
104 star_pts = 8
105 elif stars >= 1_000:
106 star_pts = 6
107 elif stars >= 100:
108 star_pts = 4
109 elif stars >= 1:
110 star_pts = 2
111 else:
112 star_pts = 0
114 forks = gh.forks
115 if forks >= 10_000:
116 fork_pts = 5
117 elif forks >= 1_000:
118 fork_pts = 4
119 elif forks >= 100:
120 fork_pts = 3
121 elif forks >= 10:
122 fork_pts = 2
123 else:
124 fork_pts = 0
126 opened, closed = _issue_counts_90d(info)
127 if opened == 0:
128 resp_pts = 3 # neutral — no signal
129 elif closed >= opened:
130 resp_pts = 5
131 elif closed >= 0.5 * opened:
132 resp_pts = 3
133 elif closed > 0:
134 resp_pts = 1
135 else:
136 resp_pts = 0
138 return min(star_pts + fork_pts + resp_pts, 20)
141def _score_activity(info: PackageInfo) -> int:
142 now = _now()
143 releases = 0
144 for v in info.versions:
145 rd = _aware(v.release_date)
146 if rd and (now - rd).days <= 365:
147 releases += 1
148 if releases == 0:
149 return 0
150 if releases <= 2:
151 return 6
152 if releases <= 5:
153 return 12
154 if releases <= 11:
155 return 16
156 return 20
159def _score_security(info: PackageInfo) -> int:
160 """Best-effort registry security signal from locally fetched metadata.
162 Secchi does not fetch advisory databases yet, so this category stays honest:
163 yanked recent releases and a missing source repository reduce confidence,
164 while the absence of those signals is treated as healthy.
165 """
166 score = 20
167 recent = info.versions[:5]
168 if any(v.is_yanked for v in recent):
169 score -= 8
170 if recent and recent[0].is_yanked:
171 score -= 6
172 if not (info.repository_url or info.github_stats.resolved):
173 score -= 4
174 return max(score, 0)
177def _grade(total: int) -> str:
178 if total >= 90:
179 return "A"
180 if total >= 75:
181 return "B"
182 if total >= 60:
183 return "C"
184 if total >= 40:
185 return "D"
186 return "F"
189def compute_health_score(info: PackageInfo) -> HealthScore:
190 def scaled(raw_20: int, max_score: int) -> int:
191 return round(max(0, min(raw_20, 20)) / 20 * max_score)
193 subs = [
194 HealthSubScore("Maintenance", _score_maintained(info), 20),
195 HealthSubScore("Community", scaled(_score_community(info), 15), 15),
196 HealthSubScore("Documentation", scaled(_score_documentation(info), 15), 15),
197 HealthSubScore("Releases", scaled(_score_activity(info), 15), 15),
198 HealthSubScore("Security", _score_security(info), 20),
199 HealthSubScore("Testing", scaled(_score_testing(info), 15), 15),
200 ]
201 total = sum(s.score for s in subs)
202 return HealthScore(sub_scores=subs, total=total, grade=_grade(total))
205# ── Release adoption ─────────────────────────────────────────────────────────
208def compute_release_adoption(
209 info: PackageInfo, limit: int = 5
210) -> tuple[dict[str, float], str]:
211 """Return (version -> percent, caption)."""
212 versions = info.versions[:limit]
213 if not versions:
214 return {}, ""
216 if info.registry is Registry.CRATES and info.version_downloads_recent:
217 raw = {
218 v.version: info.version_downloads_recent.get(v.external_id, 0)
219 for v in versions
220 }
221 caption = "Source: crates.io real per-version downloads (~90d)"
222 else:
223 raw = _adoption_from_trend(info, versions)
224 caption = "Estimated from 30-day download trend, sliced by release date"
226 total = sum(raw.values())
227 if total <= 0:
228 return {}, caption
229 return {ver: (count / total) * 100 for ver, count in raw.items()}, caption
232def _adoption_from_trend(info: PackageInfo, versions) -> dict[str, int]:
233 """Slice daily download totals into per-version windows by release date."""
234 trend = info.download_trend
235 if not trend:
236 return {v.version: 0 for v in versions}
238 def parse(d: str) -> datetime | None:
239 try:
240 return datetime.fromisoformat(d).replace(tzinfo=timezone.utc)
241 except (ValueError, TypeError):
242 return None
244 points = [(parse(p.date), p.count) for p in trend]
245 points = [(d, c) for d, c in points if d is not None]
247 raw: dict[str, int] = {}
248 # versions are newest-first; window for index i is [release[i], release[i-1])
249 for i, v in enumerate(versions):
250 start = _aware(v.release_date)
251 end = _aware(versions[i - 1].release_date) if i > 0 else None
252 total = 0
253 for d, c in points:
254 if start and d < start:
255 continue
256 if end and d >= end:
257 continue
258 total += c
259 raw[v.version] = total
260 return raw
263# ── Ecosystem distribution ───────────────────────────────────────────────────
266def compute_install_breakdown(info: PackageInfo) -> InstallBreakdown:
267 """Return download share by ecosystem.
269 The model name is kept for compatibility with existing views/export, but
270 the Overview dashboard renders this as ecosystem distribution rather than
271 installation commands.
272 """
273 count = _registry_activity_count(info)
274 label = info.registry.display_name
275 caption = "Download share by supported ecosystem."
276 return InstallBreakdown(
277 methods=[
278 InstallMethod(
279 label=label,
280 count=count,
281 percent=100.0 if count > 0 else 0.0,
282 )
283 ]
284 if count > 0
285 else [],
286 caption=caption,
287 is_estimate=False,
288 )
291def _registry_activity_count(info: PackageInfo) -> int:
292 count = info.download_counts.month or sum(
293 p.count for p in info.download_trend[-30:]
294 )
295 return count or info.total_downloads
298# ── Reverse dependencies + historical health ────────────────────────────────
301def compute_reverse_dependency_summary(info: PackageInfo) -> ReverseDependencySummary:
302 if info.reverse_dependency_count is None:
303 caption = f"{info.registry.display_name} reverse-dependency count unavailable."
304 else:
305 caption = "Projects depending on this package."
306 return ReverseDependencySummary(
307 count=info.reverse_dependency_count,
308 monthly_growth=info.reverse_dependency_monthly_growth,
309 caption=caption,
310 )
313# ── Activity timeline ────────────────────────────────────────────────────────
316def compute_activity_timeline(
317 info: PackageInfo, limit: int = 15
318) -> list[ActivityEvent]:
319 events: list[ActivityEvent] = []
320 for v in info.versions:
321 rd = _aware(v.release_date)
322 if rd:
323 events.append(
324 ActivityEvent(
325 kind=ActivityEventKind.RELEASE,
326 timestamp=rd,
327 title=f"v{v.version} published",
328 ref=v.version,
329 )
330 )
331 for ev in info.github_issue_events:
332 created = _aware(ev.created_at)
333 if created:
334 events.append(
335 ActivityEvent(
336 kind=ActivityEventKind.PR_OPENED
337 if ev.is_pull_request
338 else ActivityEventKind.ISSUE_OPENED,
339 timestamp=created,
340 title=ev.title,
341 ref=f"#{ev.number}",
342 url=ev.url,
343 )
344 )
345 closed = _aware(ev.closed_at)
346 if closed:
347 events.append(
348 ActivityEvent(
349 kind=ActivityEventKind.PR_CLOSED
350 if ev.is_pull_request
351 else ActivityEventKind.ISSUE_CLOSED,
352 timestamp=closed,
353 title=ev.title,
354 ref=f"#{ev.number}",
355 url=ev.url,
356 )
357 )
358 events.sort(key=lambda e: e.timestamp, reverse=True)
359 return events[:limit]
362# ── Downloads 30d + % change ─────────────────────────────────────────────────
365def compute_downloads_30d(info: PackageInfo) -> tuple[int, float | None]:
366 trend = info.download_trend
367 if not trend:
368 return info.download_counts.month, None
369 counts = [p.count for p in trend]
370 last30 = sum(counts[-30:])
371 prior30 = sum(counts[-60:-30])
372 if prior30 <= 0:
373 return last30, None
374 pct = (last30 - prior30) / prior30 * 100
375 return last30, pct
378# ── Aggregate ────────────────────────────────────────────────────────────────
381def compute_all(info: PackageInfo) -> DerivedPackageData:
382 adoption, adoption_caption = compute_release_adoption(info)
383 total_30d, pct = compute_downloads_30d(info)
384 return DerivedPackageData(
385 health_score=compute_health_score(info),
386 install_breakdown=compute_install_breakdown(info),
387 reverse_dependency_summary=compute_reverse_dependency_summary(info),
388 health_timeline=info.health_history,
389 activity=compute_activity_timeline(info),
390 release_adoption=adoption,
391 adoption_caption=adoption_caption,
392 downloads_30d_total=total_30d,
393 downloads_30d_pct_change=pct,
394 )