Coverage for src/secchi/workspace/aggregate.py: 94%

1"""Pure aggregation logic for multi-source and multi-project workspaces.""" 

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 logical_package_refs(refs: list[PackageRef]) -> list[PackageRef]: 

19 """Collapse same-name registry refs for single-project navigation.""" 

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

21 for ref in refs: 

22 key = ref.name.lower() 

23 current = grouped.get(key) 

24 if current is None: 

25 grouped[key] = replace(ref) 

26 elif ref.favorite and not current.favorite: 

27 current.favorite = True 

28 return list(grouped.values()) 

29 

30 

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

32 """Combine the same logical package across supported registries.""" 

33 if not infos: 

34 raise ValueError("at least one package info value is required") 

35 

36 primary = pick_primary_info(infos) 

37 combined = replace(primary) 

38 combined.name = ref.name 

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

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

41 combined.download_counts = replace(primary.download_counts) 

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

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

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

45 combined.download_trend = combine_download_trends(infos) 

46 

47 best_github = next( 

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

49 ) 

50 if best_github is not None: 

51 combined.github_stats = best_github 

52 

53 crates_info = next( 

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

55 ) 

56 if crates_info is not None: 

57 combined.reverse_dependencies = crates_info.reverse_dependencies 

58 combined.reverse_dependency_count = crates_info.reverse_dependency_count 

59 combined.reverse_dependency_monthly_growth = ( 

60 crates_info.reverse_dependency_monthly_growth 

61 ) 

62 

63 combined.health_history = primary.health_history 

64 return combined 

65 

66 

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

68 """Choose the preferred source for fields that cannot be combined.""" 

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 out: list[Registry] = [] 

79 for registry in registries: 

80 if registry not in seen: 

81 seen.add(registry) 

82 out.append(registry) 

83 return out 

84 

85 

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

87 """Sum activity for equal periods across package registries.""" 

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

89 for info in infos: 

90 for point in info.download_trend: 

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

92 return [ 

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

94 ] 

95 

96 

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

98 """Build ecosystem distribution from each source's best available total.""" 

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

100 for info in infos: 

101 label = info.registry.display_name 

102 count = info.download_counts.month or sum( 

103 point.count for point in info.download_trend[-30:] 

104 ) 

105 if count == 0: 

106 count = info.total_downloads 

107 totals[label] = totals.get(label, 0) + count 

108 

109 total = sum(totals.values()) 

110 if total <= 0: 

111 return InstallBreakdown( 

112 methods=[], 

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

114 ) 

115 

116 methods = [ 

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

118 for label, count in sorted( 

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

120 ) 

121 ] 

122 return InstallBreakdown( 

123 methods=methods, 

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

125 )