Coverage for src/secchi/aggregate.py: 20%

1"""Pure helpers for presenting one logical package across registries.""" 

2 

3from __future__ import annotations 

4 

5from collections.abc import Iterable 

6from dataclasses import replace 

7 

8from secchi.models import ( 

9 DownloadTrendPoint, 

10 InstallBreakdown, 

11 InstallMethod, 

12 PackageInfo, 

13 PackageRef, 

14 Registry, 

15) 

16 

17 

18def package_key(ref: PackageRef) -> str: 

19 project = f"{ref.project_name}:" if ref.project_name else "" 

20 return f"{project}{ref.registry.value}:{ref.name}" 

21 

22 

23def logical_package_refs(refs: list[PackageRef]) -> list[PackageRef]: 

24 """Collapse same-named registry variants into one navigable package.""" 

25 grouped: dict[str, PackageRef] = {} 

26 for ref in refs: 

27 key = ref.name.lower() 

28 current = grouped.get(key) 

29 if current is None: 

30 grouped[key] = replace(ref) 

31 elif ref.favorite and not current.favorite: 

32 current.favorite = True 

33 return list(grouped.values()) 

34 

35 

36def combine_package_infos(ref: PackageRef, infos: list[PackageInfo]) -> PackageInfo: 

37 """Combine compatible activity signals while retaining a primary profile.""" 

38 primary = pick_primary_info(infos) 

39 combined = replace(primary) 

40 combined.name = ref.name 

41 combined.source_registries = unique_registries(info.registry for info in infos) 

42 combined.total_downloads = sum(info.total_downloads for info in infos) 

43 combined.download_counts = replace(primary.download_counts) 

44 combined.download_counts.today = sum(info.download_counts.today for info in infos) 

45 combined.download_counts.week = sum(info.download_counts.week for info in infos) 

46 combined.download_counts.month = sum(info.download_counts.month for info in infos) 

47 combined.download_trend = combine_download_trends(infos) 

48 

49 best_github = next( 

50 (info.github_stats for info in infos if info.github_stats.resolved), None 

51 ) 

52 if best_github is not None: 

53 combined.github_stats = best_github 

54 

55 crates_info = next( 

56 (info for info in infos if info.registry is Registry.CRATES), None 

57 ) 

58 if crates_info is not None: 

59 combined.reverse_dependencies = crates_info.reverse_dependencies 

60 combined.reverse_dependency_count = crates_info.reverse_dependency_count 

61 combined.reverse_dependency_monthly_growth = ( 

62 crates_info.reverse_dependency_monthly_growth 

63 ) 

64 combined.health_history = primary.health_history 

65 return combined 

66 

67 

68def pick_primary_info(infos: list[PackageInfo]) -> PackageInfo: 

69 for registry in (Registry.CRATES, Registry.PYPI, Registry.NPM): 

70 for info in infos: 

71 if info.registry is registry and info.latest_version: 

72 return info 

73 return infos[0] 

74 

75 

76def unique_registries(registries: Iterable[Registry]) -> list[Registry]: 

77 seen: set[Registry] = set() 

78 return [ 

79 registry 

80 for registry in registries 

81 if not (registry in seen or seen.add(registry)) 

82 ] 

83 

84 

85def combine_download_trends(infos: list[PackageInfo]) -> list[DownloadTrendPoint]: 

86 counts: dict[str, int] = {} 

87 for info in infos: 

88 for point in info.download_trend: 

89 counts[point.date] = counts.get(point.date, 0) + point.count 

90 return [ 

91 DownloadTrendPoint(date=date, count=counts[date]) for date in sorted(counts) 

92 ] 

93 

94 

95def combine_install_breakdown(infos: list[PackageInfo]) -> InstallBreakdown: 

96 totals: dict[str, int] = {} 

97 for info in infos: 

98 label = info.registry.display_name 

99 count = info.download_counts.month or sum( 

100 p.count for p in info.download_trend[-30:] 

101 ) 

102 totals[label] = totals.get(label, 0) + (count or info.total_downloads) 

103 total = sum(totals.values()) 

104 if total <= 0: 

105 return InstallBreakdown( 

106 caption="No 30-day download data available across ecosystems." 

107 ) 

108 methods = [ 

109 InstallMethod(label=label, count=count, percent=count / total * 100) 

110 for label, count in sorted( 

111 totals.items(), key=lambda item: item[1], reverse=True 

112 ) 

113 ] 

114 return InstallBreakdown( 

115 methods=methods, caption="Combined from registry 30-day download totals." 

116 )