Coverage for src/secchi/export.py: 55%

1"""Report file helpers for package and project exports.""" 

2 

3from __future__ import annotations 

4 

5from datetime import datetime, timezone 

6from pathlib import Path 

7from typing import Any 

8 

9from secchi.models import ( 

10 DerivedPackageData, 

11 PackageInfo, 

12 PackageRef, 

13) 

14from secchi.schemas import PackageExport, SignalWarningSchema 

15 

16 

17def export_package_json( 

18 info: PackageInfo, 

19 derived: DerivedPackageData | None, 

20 ref: PackageRef, 

21 project_name: str, 

22 warnings: list[object] | None = None, 

23) -> str: 

24 document = PackageExport( 

25 project=project_name, 

26 package=ref.name, 

27 registry=ref.registry.value, 

28 exported_at=datetime.now(timezone.utc), 

29 package_info=_serialize_package_info(info) if info else None, 

30 derived=_serialize_derived(derived) if derived else None, 

31 warnings=[ 

32 SignalWarningSchema(source=warning.source, message=warning.message) 

33 for warning in (warnings or []) 

34 ], 

35 ) 

36 return document.model_dump_json(indent=2, by_alias=True) 

37 

38 

39def save_export(json_str: str, project_name: str, pkg_name: str) -> Path: 

40 return save_report(json_str, project_name, pkg_name, "json") 

41 

42 

43def save_report( 

44 content: str, 

45 project_name: str, 

46 subject: str, 

47 format_name: str, 

48 directory: Path | None = None, 

49) -> Path: 

50 """Save a report in the target directory, creating it when necessary.""" 

51 date = datetime.now(timezone.utc).strftime("%Y-%m-%d") 

52 safe_project = project_name.replace("/", "_").replace(" ", "_") 

53 safe_subject = subject.replace("/", "_").replace(" ", "_") 

54 extension = "md" if format_name == "markdown" else format_name 

55 filename = f"secchi-{safe_project}-{safe_subject}-{date}.{extension}" 

56 path = (directory or Path.cwd()) / filename 

57 path.parent.mkdir(parents=True, exist_ok=True) 

58 path.write_text(content) 

59 return path 

60 

61 

62def _serialize_package_info(info: PackageInfo) -> dict[str, Any]: 

63 return { 

64 "name": info.name, 

65 "registry": info.registry.value, 

66 "source_registries": [r.value for r in info.source_registries], 

67 "description": info.description, 

68 "author": info.author, 

69 "license": info.license, 

70 "homepage": info.homepage, 

71 "repository_url": info.repository_url, 

72 "documentation_url": info.documentation_url, 

73 "latest_version": info.latest_version, 

74 "latest_release_date": info.latest_release_date.isoformat() 

75 if info.latest_release_date 

76 else None, 

77 "total_downloads": info.total_downloads, 

78 "download_counts": { 

79 "today": info.download_counts.today, 

80 "week": info.download_counts.week, 

81 "month": info.download_counts.month, 

82 }, 

83 "github_stats": { 

84 "stars": info.github_stats.stars, 

85 "forks": info.github_stats.forks, 

86 "open_issues": info.github_stats.open_issues, 

87 "has_ci": info.github_stats.has_ci, 

88 "has_readme": info.github_stats.has_readme, 

89 "stars_delta_7d": info.github_stats.stars_delta_7d, 

90 "open_issues_delta_7d": info.github_stats.open_issues_delta_7d, 

91 "created_at": info.github_stats.created_at.isoformat() 

92 if info.github_stats.created_at 

93 else None, 

94 "pushed_at": info.github_stats.pushed_at.isoformat() 

95 if info.github_stats.pushed_at 

96 else None, 

97 }, 

98 "versions": [ 

99 { 

100 "version": v.version, 

101 "release_date": v.release_date.isoformat() if v.release_date else None, 

102 "downloads": v.downloads, 

103 "is_yanked": v.is_yanked, 

104 "size_bytes": v.size_bytes, 

105 } 

106 for v in info.versions 

107 ], 

108 "dependencies": [ 

109 { 

110 "name": d.name, 

111 "requirement": d.requirement, 

112 "optional": d.optional, 

113 } 

114 for d in info.dependencies 

115 ], 

116 "download_trend": [ 

117 {"date": p.date, "count": p.count} for p in info.download_trend 

118 ], 

119 "reverse_dependency_count": info.reverse_dependency_count, 

120 "reverse_dependency_monthly_growth": info.reverse_dependency_monthly_growth, 

121 "health_history": [ 

122 {"label": p.label, "value": p.value} for p in info.health_history 

123 ], 

124 } 

125 

126 

127def _serialize_derived(derived: DerivedPackageData) -> dict[str, Any]: 

128 return { 

129 "health_score": { 

130 "total": derived.health_score.total, 

131 "grade": derived.health_score.grade, 

132 "sub_scores": [ 

133 {"label": s.label, "score": s.score, "max_score": s.max_score} 

134 for s in derived.health_score.sub_scores 

135 ], 

136 }, 

137 "downloads_30d": { 

138 "total": derived.downloads_30d_total, 

139 "pct_change": derived.downloads_30d_pct_change, 

140 }, 

141 "install_breakdown": [ 

142 {"label": m.label, "count": m.count, "percent": m.percent} 

143 for m in derived.install_breakdown.methods 

144 ], 

145 "reverse_dependencies": { 

146 "count": derived.reverse_dependency_summary.count, 

147 "monthly_growth": derived.reverse_dependency_summary.monthly_growth, 

148 }, 

149 "health_timeline": [ 

150 {"label": p.label, "value": p.value} for p in derived.health_timeline 

151 ], 

152 "activity": [ 

153 { 

154 "kind": ev.kind.value, 

155 "timestamp": ev.timestamp.isoformat(), 

156 "title": ev.title, 

157 "ref": ev.ref, 

158 "url": ev.url, 

159 } 

160 for ev in derived.activity 

161 ], 

162 "release_adoption": derived.release_adoption, 

163 } 

164 

165 

166def _json_default(obj: Any) -> Any: 

167 if isinstance(obj, datetime): 

168 return obj.isoformat() 

169 raise TypeError(f"Object of type {type(obj)} is not JSON serializable")