Coverage for src/secchi/services/intelligence.py: 75%
134 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-08-04 22:15 +0000
1"""Shared fetch, enrichment, caching, and signal-calculation pipeline."""
3from __future__ import annotations
5import asyncio
6from collections.abc import Awaitable, Callable
7from dataclasses import dataclass, field
8from datetime import datetime, timezone
9from pathlib import Path
10from typing import Any
12import httpx
14from secchi import derived as derive
15from secchi.aggregate import package_key
16from secchi.api.base import create_adapter
17from secchi.cache import load_package_cache, save_package_cache
18from secchi.history import append_snapshot, compute_delta, find_baseline, load_snapshots
19from secchi.http import HttpClientFactory
20from secchi.models import (
21 DerivedPackageData,
22 DownloadCounts,
23 FetchError,
24 GitHubStats,
25 HistorySnapshot,
26 MetricTimelinePoint,
27 PackageInfo,
28 PackageRef,
29)
30from secchi.utils import (
31 fetch_github_extended_stats_for_package,
32 fetch_release_notes_for_package,
33)
36@dataclass
37class IntelligenceResult:
38 """Data produced for one configured package reference."""
40 ref: PackageRef
41 info: PackageInfo | None = None
42 derived: DerivedPackageData | None = None
43 warnings: list[SignalWarning] = field(default_factory=list)
44 error: FetchError | None = None
45 fetched_at: datetime | None = None
48@dataclass
49class ProjectIntelligence:
50 """Results for all registry variants in a project."""
52 results: dict[str, IntelligenceResult] = field(default_factory=dict)
53 refreshed_at: datetime | None = None
56@dataclass(frozen=True)
57class SignalWarning:
58 """A non-fatal failure while enriching an otherwise valid package."""
60 source: str
61 message: str
64class PackageIntelligenceService:
65 """The single application pipeline used by show, dashboard, and reports."""
67 def __init__(
68 self,
69 *,
70 cache_dir: Path | None = None,
71 clock: Callable[[], datetime] | None = None,
72 http_factory: HttpClientFactory | None = None,
73 ) -> None:
74 self.cache_dir = cache_dir
75 self.clock = clock or (lambda: datetime.now(timezone.utc))
76 self.http_factory = http_factory or HttpClientFactory()
78 async def fetch_project(
79 self, refs: list[PackageRef], *, force_refresh: bool = False
80 ) -> ProjectIntelligence:
81 results = await asyncio.gather(
82 *(self.fetch_package(ref, force_refresh=force_refresh) for ref in refs)
83 )
84 fetched_times = [result.fetched_at for result in results if result.fetched_at]
85 return ProjectIntelligence(
86 results={package_key(result.ref): result for result in results},
87 refreshed_at=min(fetched_times) if fetched_times else None,
88 )
90 async def fetch_package(
91 self, ref: PackageRef, *, force_refresh: bool = False
92 ) -> IntelligenceResult:
93 key = package_key(ref)
94 try:
95 if not force_refresh:
96 cached = self._load_cache(key)
97 if cached is not None:
98 info, fetched_at = cached
99 return IntelligenceResult(
100 ref=ref,
101 info=info,
102 derived=derive.compute_all(info),
103 fetched_at=fetched_at,
104 )
106 async with self.http_factory.create() as client:
107 info, warnings = await self._fetch_fresh(ref, client)
108 self._apply_history_deltas(key, info)
109 derived = derive.compute_all(info)
110 fetched_at = self.clock()
111 self._save_cache(key, info, fetched_at)
112 return IntelligenceResult(
113 ref=ref,
114 info=info,
115 derived=derived,
116 warnings=warnings,
117 fetched_at=fetched_at,
118 )
119 except (httpx.HTTPError, OSError, ValueError, KeyError, TypeError) as exc:
120 return IntelligenceResult(
121 ref=ref,
122 error=FetchError(
123 package_name=ref.name, registry=ref.registry, message=str(exc)
124 ),
125 )
127 async def _fetch_fresh(
128 self, ref: PackageRef, client
129 ) -> tuple[PackageInfo, list[SignalWarning]]:
130 try:
131 adapter = create_adapter(ref.registry, client=client)
132 except TypeError:
133 # Keeps lightweight adapter test doubles compatible with the factory.
134 adapter = create_adapter(ref.registry)
135 info = await adapter.fetch_package(ref.name)
136 optional: list[tuple[Any, SignalWarning | None]]
137 optional = await asyncio.gather(
138 self._optional_signal(
139 "versions", lambda: adapter.fetch_versions(ref.name), []
140 ),
141 self._optional_signal(
142 "download trend",
143 lambda: adapter.fetch_download_trend(ref.name, days=730),
144 [],
145 ),
146 self._optional_signal(
147 "download counts",
148 lambda: adapter.fetch_download_counts(ref.name),
149 DownloadCounts(),
150 ),
151 self._optional_signal(
152 "GitHub extended stats",
153 lambda: fetch_github_extended_stats_for_package(
154 info.homepage, info.repository_url, client=client
155 ),
156 (GitHubStats(), []),
157 ),
158 self._optional_signal(
159 "version download breakdown",
160 lambda: adapter.fetch_version_download_breakdown(ref.name),
161 {},
162 ),
163 self._optional_signal(
164 "reverse dependencies",
165 lambda: adapter.fetch_reverse_dependencies(ref.name),
166 [],
167 ),
168 self._optional_signal(
169 "reverse dependency count",
170 lambda: adapter.fetch_reverse_dependency_count(ref.name),
171 None,
172 ),
173 )
174 values = [item[0] for item in optional]
175 warnings = [item[1] for item in optional if item[1] is not None]
176 (
177 versions,
178 trend,
179 counts,
180 gh_result,
181 version_downloads,
182 reverse_dependencies,
183 reverse_dependency_count,
184 ) = values
185 info.versions = versions
186 info.download_trend = trend
187 info.download_counts = counts
188 info.github_stats, info.github_issue_events = gh_result
189 info.version_downloads_recent = version_downloads
190 info.reverse_dependencies = reverse_dependencies
191 info.reverse_dependency_count = reverse_dependency_count
193 if info.latest_version:
194 dependencies, warning = await self._optional_signal(
195 "dependencies",
196 lambda: adapter.fetch_dependencies(ref.name, info.latest_version),
197 [],
198 )
199 info.dependencies = dependencies
200 if warning:
201 warnings.append(warning)
202 notes, warning = await self._optional_signal(
203 "release notes",
204 lambda: adapter.fetch_release_notes(ref.name, info.latest_version),
205 "",
206 )
207 if warning:
208 warnings.append(warning)
209 if not notes and (info.homepage or info.repository_url):
210 github_notes, warning = await self._optional_signal(
211 "GitHub release notes",
212 lambda: fetch_release_notes_for_package(
213 info.homepage,
214 info.repository_url,
215 info.latest_version,
216 client=client,
217 ),
218 "",
219 )
220 notes = github_notes
221 if warning:
222 warnings.append(warning)
223 info.release_notes = notes
224 return info, warnings
226 async def _optional_signal(
227 self,
228 source: str,
229 operation: Callable[[], Awaitable[Any]],
230 default: Any,
231 ) -> tuple[Any, SignalWarning | None]:
232 """Run one enrichment without making the package fetch fail."""
233 try:
234 return await operation(), None
235 except (httpx.HTTPError, OSError, ValueError, KeyError, TypeError) as exc:
236 return default, SignalWarning(source=source, message=str(exc))
238 def _apply_history_deltas(self, key: str, info: PackageInfo) -> None:
239 snapshots = load_snapshots(
240 key, path=self._history_path() if self.cache_dir is not None else None
241 )
242 github = info.github_stats
243 if github.resolved:
244 baseline = find_baseline(snapshots, now=self.clock)
245 github.stars_delta_7d = compute_delta(
246 github.stars, baseline.stars if baseline else None
247 )
248 github.open_issues_delta_7d = compute_delta(
249 github.open_issues, baseline.open_issues if baseline else None
250 )
252 health_total = derive.compute_health_score(info).total
253 monthly = find_baseline(
254 snapshots, min_age_days=28, max_age_days=35, now=self.clock
255 )
256 if info.reverse_dependency_count is not None:
257 info.reverse_dependency_monthly_growth = compute_delta(
258 info.reverse_dependency_count,
259 monthly.reverse_dependency_count if monthly else None,
260 )
261 info.health_history = health_history_points(
262 snapshots, health_total, now=self.clock
263 )
264 append_snapshot(
265 key,
266 HistorySnapshot(
267 timestamp=self.clock(),
268 stars=github.stars,
269 open_issues=github.open_issues,
270 health_score=health_total,
271 reverse_dependency_count=info.reverse_dependency_count,
272 ),
273 path=self._history_path() if self.cache_dir is not None else None,
274 )
276 def _load_cache(self, key: str):
277 if self.cache_dir is None:
278 return load_package_cache(key)
279 return load_package_cache(key, root=self.cache_dir, now=self.clock)
281 def _save_cache(self, key: str, info: PackageInfo, fetched_at: datetime) -> None:
282 if self.cache_dir is None:
283 save_package_cache(key, info, fetched_at)
284 else:
285 save_package_cache(key, info, fetched_at, root=self.cache_dir)
287 def _history_path(self) -> Path:
288 assert self.cache_dir is not None
289 return self.cache_dir / "history.json"
292def health_history_points(
293 snapshots: list[HistorySnapshot],
294 current_health: int,
295 *,
296 now: Callable[[], datetime] | None = None,
297) -> list[MetricTimelinePoint]:
298 latest_by_month: dict[str, tuple[datetime, int]] = {}
299 for snapshot in snapshots:
300 if snapshot.health_score is None:
301 continue
302 timestamp = snapshot.timestamp
303 if timestamp.tzinfo is None:
304 timestamp = timestamp.replace(tzinfo=timezone.utc)
305 key = timestamp.strftime("%Y-%m")
306 current = latest_by_month.get(key)
307 if current is None or timestamp > current[0]:
308 latest_by_month[key] = (timestamp, snapshot.health_score)
309 current_time = (now or (lambda: datetime.now(timezone.utc)))()
310 latest_by_month[current_time.strftime("%Y-%m")] = (current_time, current_health)
311 return [
312 MetricTimelinePoint(label=timestamp.strftime("%b"), value=value)
313 for _, (timestamp, value) in sorted(latest_by_month.items())
314 ][-12:]