Coverage for src/secchi/services/intelligence.py: 85%
174 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-08-14 23:28 +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 UTC, datetime
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 (
18 load_package_cache,
19 load_security_cache,
20 save_package_cache,
21 save_security_cache,
22)
23from secchi.diagnostics import (
24 DiagnosticLog,
25 DiagnosticStatus,
26 diagnostic_for_http_error,
27)
28from secchi.history import append_snapshot, compute_delta, find_baseline, load_snapshots
29from secchi.http import HttpClientFactory
30from secchi.models import (
31 DerivedPackageData,
32 DownloadCounts,
33 FetchError,
34 GitHubStats,
35 HistorySnapshot,
36 MetricTimelinePoint,
37 PackageInfo,
38 PackageRef,
39)
40from secchi.security import fetch_osv_advisories
41from secchi.utils import (
42 fetch_github_extended_stats_for_package,
43 fetch_release_notes_for_package,
44)
47@dataclass
48class IntelligenceResult:
49 """Data produced for one configured package reference."""
51 ref: PackageRef
52 info: PackageInfo | None = None
53 derived: DerivedPackageData | None = None
54 warnings: list[SignalWarning] = field(default_factory=list)
55 error: FetchError | None = None
56 fetched_at: datetime | None = None
59@dataclass
60class ProjectIntelligence:
61 """Results for all registry variants in a project."""
63 results: dict[str, IntelligenceResult] = field(default_factory=dict)
64 refreshed_at: datetime | None = None
67@dataclass(frozen=True)
68class SignalWarning:
69 """A non-fatal failure while enriching an otherwise valid package."""
71 source: str
72 message: str
75class PackageIntelligenceService:
76 """The single application pipeline used by show, dashboard, and reports."""
78 def __init__(
79 self,
80 *,
81 cache_dir: Path | None = None,
82 clock: Callable[[], datetime] | None = None,
83 http_factory: HttpClientFactory | None = None,
84 diagnostics: DiagnosticLog | None = None,
85 ) -> None:
86 self.cache_dir = cache_dir
87 self.clock = clock or (lambda: datetime.now(UTC))
88 self.diagnostics = diagnostics
89 self.http_factory = http_factory or HttpClientFactory(diagnostics=diagnostics)
91 async def fetch_project(
92 self,
93 refs: list[PackageRef],
94 *,
95 force_refresh: bool = False,
96 force_security_refresh: bool = False,
97 ) -> ProjectIntelligence:
98 results = await asyncio.gather(
99 *(
100 self.fetch_package(
101 ref,
102 force_refresh=force_refresh,
103 force_security_refresh=force_security_refresh,
104 )
105 for ref in refs
106 )
107 )
108 fetched_times = [result.fetched_at for result in results if result.fetched_at]
109 return ProjectIntelligence(
110 results={package_key(result.ref): result for result in results},
111 refreshed_at=min(fetched_times) if fetched_times else None,
112 )
114 async def fetch_package(
115 self,
116 ref: PackageRef,
117 *,
118 force_refresh: bool = False,
119 force_security_refresh: bool = False,
120 ) -> IntelligenceResult:
121 key = package_key(ref)
122 try:
123 if not force_refresh:
124 cached = self._load_cache(key)
125 if cached is not None:
126 if self.diagnostics is not None:
127 self.diagnostics.record(
128 DiagnosticStatus.SUCCESS,
129 "CACHE",
130 f"Using cached data for {ref.registry.display_name}:{ref.name}",
131 )
132 info, fetched_at = cached
133 security_warnings = await self._refresh_security_if_needed(
134 key,
135 info,
136 force=force_security_refresh,
137 )
138 return IntelligenceResult(
139 ref=ref,
140 info=info,
141 derived=derive.compute_all(info),
142 warnings=security_warnings,
143 fetched_at=fetched_at,
144 )
145 if self.diagnostics is not None:
146 self.diagnostics.record(
147 DiagnosticStatus.SUCCESS,
148 "CACHE",
149 f"No cached data for {ref.registry.display_name}:{ref.name}; fetching fresh data",
150 )
152 async with self.http_factory.create() as client:
153 info, warnings = await self._fetch_fresh(ref, client)
154 if self.diagnostics is not None:
155 self.diagnostics.record(
156 DiagnosticStatus.SUCCESS,
157 ref.registry.display_name,
158 f"Package metadata loaded: {ref.name} {info.latest_version}",
159 )
160 info.security_advisories, security_warning = await self._fetch_security(
161 key, info, client
162 )
163 if security_warning:
164 warnings.append(security_warning)
165 self._apply_history_deltas(key, info)
166 derived = derive.compute_all(info)
167 fetched_at = self.clock()
168 self._save_cache(key, info, fetched_at)
169 return IntelligenceResult(
170 ref=ref,
171 info=info,
172 derived=derived,
173 warnings=warnings,
174 fetched_at=fetched_at,
175 )
176 except (httpx.HTTPError, OSError, ValueError, KeyError, TypeError) as exc:
177 if self.diagnostics is not None:
178 self.diagnostics.record(
179 DiagnosticStatus.FAILURE,
180 ref.registry.display_name,
181 f"Package fetch failed for {ref.name}: {diagnostic_for_http_error(exc)}",
182 )
183 return IntelligenceResult(
184 ref=ref,
185 error=FetchError(
186 package_name=ref.name, registry=ref.registry, message=str(exc)
187 ),
188 )
190 async def _fetch_fresh(
191 self, ref: PackageRef, client
192 ) -> tuple[PackageInfo, list[SignalWarning]]:
193 try:
194 adapter = create_adapter(ref.registry, client=client)
195 except TypeError:
196 # Keeps lightweight adapter test doubles compatible with the factory.
197 adapter = create_adapter(ref.registry)
198 info = await adapter.fetch_package(ref.name)
199 optional: list[tuple[Any, SignalWarning | None]]
200 optional = await asyncio.gather(
201 self._optional_signal(
202 "versions", lambda: adapter.fetch_versions(ref.name), []
203 ),
204 self._optional_signal(
205 "download trend",
206 lambda: adapter.fetch_download_trend(ref.name, days=730),
207 [],
208 ),
209 self._optional_signal(
210 "download counts",
211 lambda: adapter.fetch_download_counts(ref.name),
212 DownloadCounts(),
213 ),
214 self._optional_signal(
215 "GitHub extended stats",
216 lambda: fetch_github_extended_stats_for_package(
217 info.homepage, info.repository_url, client=client
218 ),
219 (GitHubStats(), []),
220 ),
221 self._optional_signal(
222 "version download breakdown",
223 lambda: adapter.fetch_version_download_breakdown(ref.name),
224 {},
225 ),
226 self._optional_signal(
227 "reverse dependencies",
228 lambda: adapter.fetch_reverse_dependencies(ref.name),
229 [],
230 ),
231 self._optional_signal(
232 "reverse dependency count",
233 lambda: adapter.fetch_reverse_dependency_count(ref.name),
234 None,
235 ),
236 )
237 values = [item[0] for item in optional]
238 warnings = [item[1] for item in optional if item[1] is not None]
239 (
240 versions,
241 trend,
242 counts,
243 gh_result,
244 version_downloads,
245 reverse_dependencies,
246 reverse_dependency_count,
247 ) = values
248 info.versions = versions
249 info.download_trend = trend
250 info.download_counts = counts
251 info.github_stats, info.github_issue_events = gh_result
252 info.version_downloads_recent = version_downloads
253 info.reverse_dependencies = reverse_dependencies
254 info.reverse_dependency_count = reverse_dependency_count
256 if info.latest_version:
257 dependencies, warning = await self._optional_signal(
258 "dependencies",
259 lambda: adapter.fetch_dependencies(ref.name, info.latest_version),
260 [],
261 )
262 info.dependencies = dependencies
263 if warning:
264 warnings.append(warning)
265 notes, warning = await self._optional_signal(
266 "release notes",
267 lambda: adapter.fetch_release_notes(ref.name, info.latest_version),
268 "",
269 )
270 if warning:
271 warnings.append(warning)
272 if not notes and (info.homepage or info.repository_url):
273 github_notes, warning = await self._optional_signal(
274 "GitHub release notes",
275 lambda: fetch_release_notes_for_package(
276 info.homepage,
277 info.repository_url,
278 info.latest_version,
279 client=client,
280 ),
281 "",
282 )
283 notes = github_notes
284 if warning:
285 warnings.append(warning)
286 info.release_notes = notes
287 return info, warnings
289 async def _refresh_security_if_needed(
290 self,
291 key: str,
292 info: PackageInfo,
293 *,
294 force: bool,
295 ) -> list[SignalWarning]:
296 if not force:
297 cached = self._load_security_cache(key, info.latest_version)
298 if cached is not None:
299 info.security_advisories = cached[0]
300 return []
302 async with self.http_factory.create() as client:
303 advisories, warning = await self._fetch_security(
304 key, info, client, fallback=info.security_advisories
305 )
306 info.security_advisories = advisories
307 return [warning] if warning else []
309 async def _fetch_security(
310 self,
311 key: str,
312 info: PackageInfo,
313 client,
314 *,
315 fallback: list | None = None,
316 ) -> tuple[list, SignalWarning | None]:
317 advisories, warning = await self._optional_signal(
318 "security advisories",
319 lambda: fetch_osv_advisories(info, client=client),
320 fallback if fallback is not None else [],
321 )
322 if warning is None:
323 self._save_security_cache(
324 key, info.latest_version, advisories, self.clock()
325 )
326 return advisories, warning
328 async def _optional_signal(
329 self,
330 source: str,
331 operation: Callable[[], Awaitable[Any]],
332 default: Any,
333 ) -> tuple[Any, SignalWarning | None]:
334 """Run one enrichment without making the package fetch fail."""
335 try:
336 return await operation(), None
337 except (httpx.HTTPError, OSError, ValueError, KeyError, TypeError) as exc:
338 if self.diagnostics is not None:
339 self.diagnostics.record(
340 DiagnosticStatus.WARN,
341 "SIGNAL",
342 f"{source} unavailable: {diagnostic_for_http_error(exc)}",
343 )
344 return default, SignalWarning(source=source, message=str(exc))
346 def _apply_history_deltas(self, key: str, info: PackageInfo) -> None:
347 snapshots = load_snapshots(
348 key, path=self._history_path() if self.cache_dir is not None else None
349 )
350 github = info.github_stats
351 if github.resolved:
352 baseline = find_baseline(snapshots, now=self.clock)
353 github.stars_delta_7d = compute_delta(
354 github.stars, baseline.stars if baseline else None
355 )
356 github.open_issues_delta_7d = compute_delta(
357 github.open_issues, baseline.open_issues if baseline else None
358 )
360 health_total = derive.compute_health_score(info).total
361 monthly = find_baseline(
362 snapshots, min_age_days=28, max_age_days=35, now=self.clock
363 )
364 if info.reverse_dependency_count is not None:
365 info.reverse_dependency_monthly_growth = compute_delta(
366 info.reverse_dependency_count,
367 monthly.reverse_dependency_count if monthly else None,
368 )
369 info.health_history = health_history_points(
370 snapshots, health_total, now=self.clock
371 )
372 append_snapshot(
373 key,
374 HistorySnapshot(
375 timestamp=self.clock(),
376 stars=github.stars,
377 open_issues=github.open_issues,
378 health_score=health_total,
379 reverse_dependency_count=info.reverse_dependency_count,
380 ),
381 path=self._history_path() if self.cache_dir is not None else None,
382 )
384 def _load_cache(self, key: str):
385 if self.cache_dir is None:
386 return load_package_cache(key)
387 return load_package_cache(key, root=self.cache_dir, now=self.clock)
389 def _load_security_cache(self, key: str, package_version: str):
390 if self.cache_dir is None:
391 return load_security_cache(key, package_version, now=self.clock)
392 return load_security_cache(
393 key, package_version, root=self.cache_dir, now=self.clock
394 )
396 def _save_cache(self, key: str, info: PackageInfo, fetched_at: datetime) -> None:
397 if self.cache_dir is None:
398 save_package_cache(key, info, fetched_at)
399 else:
400 save_package_cache(key, info, fetched_at, root=self.cache_dir)
402 def _save_security_cache(
403 self,
404 key: str,
405 package_version: str,
406 advisories: list,
407 fetched_at: datetime,
408 ) -> None:
409 if self.cache_dir is None:
410 save_security_cache(key, package_version, advisories, fetched_at)
411 else:
412 save_security_cache(
413 key,
414 package_version,
415 advisories,
416 fetched_at,
417 root=self.cache_dir,
418 )
420 def _history_path(self) -> Path:
421 assert self.cache_dir is not None
422 return self.cache_dir / "history.json"
425def health_history_points(
426 snapshots: list[HistorySnapshot],
427 current_health: int,
428 *,
429 now: Callable[[], datetime] | None = None,
430) -> list[MetricTimelinePoint]:
431 latest_by_month: dict[str, tuple[datetime, int]] = {}
432 for snapshot in snapshots:
433 if snapshot.health_score is None:
434 continue
435 timestamp = snapshot.timestamp
436 if timestamp.tzinfo is None:
437 timestamp = timestamp.replace(tzinfo=UTC)
438 key = timestamp.strftime("%Y-%m")
439 current = latest_by_month.get(key)
440 if current is None or timestamp > current[0]:
441 latest_by_month[key] = (timestamp, snapshot.health_score)
442 current_time = (now or (lambda: datetime.now(UTC)))()
443 latest_by_month[current_time.strftime("%Y-%m")] = (current_time, current_health)
444 return [
445 MetricTimelinePoint(label=timestamp.strftime("%b"), value=value)
446 for _, (timestamp, value) in sorted(latest_by_month.items())
447 ][-12:]