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

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 combined.security_advisories = _combine_security_advisories(infos) 

66 return combined 

67 

68 

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

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

71 for info in infos: 

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

73 return info 

74 return infos[0] 

75 

76 

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

78 seen: set[Registry] = set() 

79 return [ 

80 registry 

81 for registry in registries 

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

83 ] 

84 

85 

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

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

88 for info in infos: 

89 for point in info.download_trend: 

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

91 return [ 

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

93 ] 

94 

95 

96def _combine_security_advisories(infos: list[PackageInfo]): 

97 seen: set[str] = set() 

98 advisories = [] 

99 for info in infos: 

100 for advisory in info.security_advisories: 

101 if advisory.id not in seen: 

102 seen.add(advisory.id) 

103 advisories.append(advisory) 

104 return advisories 

105 

106 

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

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

109 for info in infos: 

110 label = info.registry.display_name 

111 count = info.download_counts.month or sum( 

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

113 ) 

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

115 total = sum(totals.values()) 

116 if total <= 0: 

117 return InstallBreakdown( 

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

119 ) 

120 methods = [ 

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

122 for label, count in sorted( 

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

124 ) 

125 ] 

126 return InstallBreakdown( 

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

128 )