Coverage for src/somesy/codemeta/enrich.py: 94%

1"""Fill optional CodeMeta fields from configured project files and Git.""" 

2 

3from __future__ import annotations 

4 

5import json 

6import re 

7from collections.abc import Iterable 

8from pathlib import Path 

9from typing import Any 

10from urllib.parse import urlparse 

11 

12import tomlkit 

13from defusedxml import ElementTree 

14from packaging.requirements import Requirement 

15from tomlkit.exceptions import ParseError 

16 

17from somesy.git import harvest as harvest_git 

18 

19 

20def enrich( 

21 codemeta: dict[str, Any], 

22 sources: dict[str, Path | list[Path] | None], 

23 root: Path, 

24) -> None: 

25 """Fill CodeMeta fields absent after canonical Somesy synchronization. 

26 

27 ``codemeta`` already contains data written from somesy.toml. This function 

28 therefore only adds optional fields and never replaces an existing value. 

29 """ 

30 values: dict[str, Any] = {"requirements": [], "languages": [], "runtimes": []} 

31 for source_type, paths in sources.items(): 

32 for path in _paths(paths): 

33 if path.is_file(): 

34 _read_source(source_type, path, values) 

35 

36 _add(codemeta, "readme", values.get("readme")) 

37 _add(codemeta, "issueTracker", values.get("issueTracker")) 

38 _add(codemeta, "releaseNotes", values.get("releaseNotes")) 

39 _add(codemeta, "softwareRequirements", _unique(values["requirements"])) 

40 _add(codemeta, "programmingLanguage", ", ".join(_unique(values["languages"]))) 

41 _add(codemeta, "runtimePlatform", ", ".join(_unique(values["runtimes"]))) 

42 

43 _read_git(root, values) 

44 for key in ( 

45 "codeRepository", 

46 "issueTracker", 

47 "dateCreated", 

48 "dateModified", 

49 "version", 

50 ): 

51 _add(codemeta, key, values.get(key)) 

52 

53 

54def _paths(paths: Path | list[Path] | None) -> list[Path]: 

55 if paths is None: 

56 return [] 

57 return paths if isinstance(paths, list) else [paths] 

58 

59 

60def _read_source(source_type: str, path: Path, values: dict[str, Any]) -> None: 

61 if source_type == "pyproject": 

62 _read_pyproject(path, values) 

63 elif source_type == "package_json": 

64 _read_package_json(path, values) 

65 elif source_type in {"julia", "fortran", "rust"}: 

66 _read_toml_project(source_type, path, values) 

67 elif source_type == "pom_xml": 

68 _read_pom(path, values) 

69 

70 

71def _read_pyproject(path: Path, values: dict[str, Any]) -> None: 

72 data = tomlkit.parse(path.read_text()) 

73 project = data.get("project") or data.get("tool", {}).get("poetry", {}) 

74 if not project: 

75 return 

76 values["languages"].append("Python") 

77 requires_python = project.get("requires-python") 

78 if requires_python: 

79 values["runtimes"].append(f"Python {requires_python}") 

80 urls = project.get("urls", {}) 

81 _urls(urls, values) 

82 _url(values, "readme", project.get("readme")) 

83 

84 dependencies = project.get("dependencies", []) 

85 if isinstance(dependencies, dict): # Poetry v1 

86 python = dependencies.pop("python", None) 

87 if python: 

88 values["runtimes"].append(f"Python {python}") 

89 locked = _poetry_lock_versions(path.with_name("poetry.lock")) 

90 for name, spec in dependencies.items(): 

91 _named_requirement(values, name, _poetry_version(spec), "Python", locked) 

92 return 

93 locked = _poetry_lock_versions(path.with_name("poetry.lock")) 

94 for dependency in dependencies: 

95 _requirement(values, dependency, "Python", locked) 

96 

97 

98def _poetry_version(spec: Any) -> str: 

99 return spec if isinstance(spec, str) else spec.get("version", "") 

100 

101 

102def _poetry_lock_versions(path: Path) -> dict[str, str]: 

103 if not path.is_file(): 

104 return {} 

105 try: 

106 packages = tomlkit.parse(path.read_text()).get("package", []) 

107 return {_normalise(package["name"]): package["version"] for package in packages} 

108 except (KeyError, TypeError, ParseError): 

109 return {} 

110 

111 

112def _read_package_json(path: Path, values: dict[str, Any]) -> None: 

113 data = json.loads(path.read_text()) 

114 values["languages"].append("JavaScript") 

115 if node := data.get("engines", {}).get("node"): 

116 values["runtimes"].append(f"Node.js {node}") 

117 bugs = data.get("bugs") 

118 _url(values, "issueTracker", bugs.get("url") if isinstance(bugs, dict) else bugs) 

119 for section in ("dependencies", "peerDependencies", "optionalDependencies"): 

120 for name, version in data.get(section, {}).items(): 

121 _named_requirement(values, name, version, "Node.js") 

122 

123 

124def _read_toml_project(source_type: str, path: Path, values: dict[str, Any]) -> None: 

125 data = tomlkit.parse(path.read_text()) 

126 if source_type == "rust": 

127 package = data.get("package", {}) 

128 values["languages"].append("Rust") 

129 if version := package.get("rust-version"): 

130 values["runtimes"].append(f"Rust {version}") 

131 dependencies = data.get("dependencies", {}) 

132 runtime = "Rust" 

133 elif source_type == "julia": 

134 values["languages"].append("Julia") 

135 compat = data.get("compat", {}) 

136 if version := compat.get("julia"): 

137 values["runtimes"].append(f"Julia {version}") 

138 dependencies = {name: compat.get(name, "") for name in data.get("deps", {})} 

139 runtime = "Julia" 

140 else: 

141 values["languages"].append("Fortran") 

142 dependencies = data.get("dependencies", {}) 

143 runtime = "Fortran" 

144 for name, version in dependencies.items(): 

145 _named_requirement(values, name, _poetry_version(version), runtime) 

146 

147 

148def _read_pom(path: Path, values: dict[str, Any]) -> None: 

149 root = ElementTree.parse(path).getroot() 

150 if root is None: 

151 return 

152 namespace = root.tag[1:].partition("}")[0] if root.tag.startswith("{") else "" 

153 prefix = f"{{{namespace}}}" if namespace else "" 

154 values["languages"].append("Java") 

155 properties = root.find(f"{prefix}properties") 

156 if properties is not None: 

157 for key in ( 

158 "maven.compiler.release", 

159 "maven.compiler.target", 

160 "maven.compiler.source", 

161 ): 

162 if version := properties.findtext(f"{prefix}{key}"): 

163 values["runtimes"].append(f"Java {version}") 

164 break 

165 for dependency in root.findall(f".//{prefix}dependency"): 

166 name = dependency.findtext(f"{prefix}artifactId") 

167 if name: 

168 version = dependency.findtext(f"{prefix}version") or "" 

169 _named_requirement(values, name, version, "Java") 

170 

171 

172def _urls(urls: Any, values: dict[str, Any]) -> None: 

173 if not isinstance(urls, dict): 

174 return 

175 for name, url in urls.items(): 

176 lowered = name.lower() 

177 if lowered == "issues": 

178 _url(values, "issueTracker", url) 

179 elif lowered == "changelog": 

180 _url(values, "releaseNotes", url) 

181 

182 

183def _url(values: dict[str, Any], key: str, value: Any) -> None: 

184 if ( 

185 key not in values 

186 and isinstance(value, str) 

187 and urlparse(value).scheme in {"http", "https"} 

188 ): 

189 values[key] = value 

190 

191 

192def _requirement( 

193 values: dict[str, Any], 

194 value: str, 

195 runtime: str, 

196 locked: dict[str, str] | None = None, 

197) -> None: 

198 try: 

199 requirement = Requirement(value) 

200 except (TypeError, ValueError): 

201 match = re.fullmatch(r"((?:@[^/]+/)?[A-Za-z0-9_.-]+)(.*)", value) 

202 if match is None: 

203 return 

204 name, version = match.groups() 

205 else: 

206 name, version = requirement.name, str(requirement.specifier) 

207 _named_requirement(values, name, version, runtime, locked) 

208 

209 

210def _named_requirement( 

211 values: dict[str, Any], 

212 name: str, 

213 version: Any, 

214 runtime: str, 

215 locked: dict[str, str] | None = None, 

216) -> None: 

217 version = (locked or {}).get(_normalise(name), str(version)) 

218 values["requirements"].append( 

219 { 

220 "@type": "SoftwareApplication", 

221 "identifier": name, 

222 "name": name, 

223 "runtimePlatform": runtime, 

224 **({"version": version} if version else {}), 

225 } 

226 ) 

227 

228 

229def _normalise(name: str) -> str: 

230 return re.sub(r"[-_.]+", "-", name).lower() 

231 

232 

233def _unique(values: Iterable[Any]) -> list[Any]: 

234 seen: set[str] = set() 

235 result = [] 

236 for value in values: 

237 key = ( 

238 json.dumps(value, sort_keys=True) if isinstance(value, dict) else str(value) 

239 ) 

240 if value and key not in seen: 

241 seen.add(key) 

242 result.append(value) 

243 return result 

244 

245 

246def _add(codemeta: dict[str, Any], key: str, value: Any) -> None: 

247 if key not in codemeta and value not in (None, "", []): 

248 codemeta[key] = value 

249 

250 

251def _read_git(root: Path, values: dict[str, Any]) -> None: 

252 metadata = harvest_git(root) 

253 if metadata is None: 

254 return 

255 repository = metadata.repository 

256 if repository: 

257 if match := re.fullmatch(r"git@([^:]+):(.+)", repository): 

258 repository = f"https://{match[1]}/{match[2]}" 

259 values["codeRepository"] = repository.removesuffix(".git") 

260 parsed = urlparse(values["codeRepository"]) 

261 if parsed.netloc in {"github.com", "gitlab.com"}: 

262 _url(values, "issueTracker", values["codeRepository"] + "/issues") 

263 values["dateCreated"] = ( 

264 metadata.date_created.isoformat() if metadata.date_created else None 

265 ) 

266 values["dateModified"] = ( 

267 metadata.date_modified.isoformat() if metadata.date_modified else None 

268 ) 

269 values["version"] = metadata.version