Coverage for src/somesy/codemeta/writer.py: 87%
193 statements
« prev ^ index » next coverage.py v7.16.0, created at 2026-11-03 13:31 +0000
1"""codemeta.json creation module."""
3import json
4import logging
5import uuid
6from collections import OrderedDict
7from collections.abc import Sequence
8from pathlib import Path
9from typing import Any
11from somesy.codemeta.utils import V3_CONTEXT, validate_codemeta
12from somesy.core.core import write_json
13from somesy.core.models import Entity, Person, ProjectMetadata
14from somesy.core.writer import FieldKeyMapping, ProjectMetadataWriter
16logger = logging.getLogger("somesy")
18V2_CONTEXT = "https://doi.org/10.5063/schema/codemeta-2.0"
21class CodeMeta(ProjectMetadataWriter):
22 """Codemeta.json parser and saver."""
24 def __init__(
25 self,
26 path: Path,
27 merge: bool | None = False,
28 pass_validation: bool | None = False,
29 ):
30 """Codemeta.json parser.
32 See [somesy.core.writer.ProjectMetadataWriter.__init__][].
33 """
34 self.merge = merge
35 self._default_context = V3_CONTEXT
36 mappings: FieldKeyMapping = {
37 "repository": ["codeRepository"],
38 "homepage": ["softwareHelp"],
39 "documentation": ["buildInstructions"],
40 "keywords": ["keywords"],
41 "authors": ["author"],
42 "maintainers": ["maintainer"],
43 "contributors": ["contributor"],
44 }
45 super().__init__(
46 path,
47 create_if_not_exists=True,
48 direct_mappings=mappings,
49 merge=merge,
50 pass_validation=pass_validation,
51 )
53 @property
54 def authors(self):
55 """Return the authors of the codemeta.json file."""
56 return self._get_property(self._get_key("authors")) or []
58 @authors.setter
59 def authors(self, authors: list[Person | Entity]) -> None:
60 """Set the authors of the project."""
61 authors_dict = self._people_with_roles(authors)
62 self._set_property(self._get_key("authors"), authors_dict)
64 @property
65 def maintainers(self):
66 """Return the maintainers of the codemeta.json file."""
67 return self._get_property(self._get_key("maintainers"))
69 @maintainers.setter
70 def maintainers(self, maintainers: list[Person | Entity]) -> None:
71 """Set the maintainers of the project."""
72 maintainers_dict = [self._from_person(m) for m in maintainers]
73 self._set_property(self._get_key("maintainers"), maintainers_dict)
75 @property
76 def contributors(self):
77 """Return the contributors of the codemeta.json file."""
78 return self._get_property(self._get_key("contributors"))
80 @contributors.setter
81 def contributors(self, contributors: list[Person | Entity]) -> None:
82 """Set the contributors of the project."""
83 contributors_dict = self._people_with_roles(contributors)
84 self._set_property(self._get_key("contributors"), contributors_dict)
86 def _load(self) -> None:
87 """Load codemeta.json file."""
88 with self.path.open(encoding="utf-8") as f:
89 self._data = json.load(f, object_pairs_hook=OrderedDict)
91 def _upgrade_to_v3(self) -> None:
92 """Normalize an existing CodeMeta file before v3.1 validation."""
93 context = self._data.get("@context", [])
94 context = context if isinstance(context, list) else [context]
95 context = [item for item in context if item != V2_CONTEXT]
96 if V3_CONTEXT not in context:
97 context.insert(0, V3_CONTEXT)
98 self._data["@context"] = context
100 for old, new in (
101 ("contIntegration", "continuousIntegration"),
102 ("embargoDate", "embargoEndDate"),
103 ):
104 if old in self._data and new in self._data:
105 logger.warning(
106 f"Both '{old}' and '{new}' found in codemeta.json, "
107 f"dropping '{old}' in favor of '{new}'."
108 )
109 elif old in self._data:
110 self._data[new] = self._data[old]
111 self._data.pop(old, None)
113 def _validate(self) -> None:
114 """Validate codemeta.json content using pydantic class."""
115 if self.pass_validation:
116 return
117 if not self.merge:
118 return
119 self._upgrade_to_v3()
120 invalid_fields = validate_codemeta(self._data)
121 if invalid_fields:
122 raise ValueError(
123 f"Invalid fields in codemeta.json: {invalid_fields}. Cannot merge with invalid fields."
124 )
126 def _init_new_file(self) -> None:
127 """Create a new codemeta.json file with bare minimum generic data."""
128 data = self._new_data()
129 # dump to file
130 write_json(data, self.path)
132 def _new_data(self) -> dict[str, Any]:
133 """Return the bare minimum generic CodeMeta data."""
134 return {
135 "@context": self._default_context,
136 "@type": "SoftwareSourceCode",
137 "author": [],
138 }
140 def save(self, path: Path | None = None) -> None:
141 """Save the codemeta.json file."""
142 path = path or self.path
143 logger.debug(f"Saving codemeta.json to {path}")
145 # copy the _data
146 data = self._data.copy()
148 # set license
149 if "license" in data:
150 licenses = data["license"]
151 licenses = licenses if isinstance(licenses, list) else [licenses]
152 data["license"] = [
153 license
154 if license.startswith("https://spdx.org/licenses/")
155 else f"https://spdx.org/licenses/{license}"
156 for license in licenses
157 ]
159 write_json(data, path)
161 @staticmethod
162 def _from_person(person: Person | Entity) -> dict:
163 """Convert project metadata person object to codemeta.json dict for person format."""
164 if isinstance(person, Person):
165 person_dict = {
166 "@type": "Person",
167 }
168 if person.given_names:
169 person_dict["givenName"] = person.given_names
170 if person.family_names:
171 person_dict["familyName"] = person.family_names
172 if person.email:
173 person_dict["email"] = person.email
174 if person.orcid:
175 person_dict["@id"] = str(person.orcid)
176 person_dict["identifier"] = str(person.orcid)
177 if person.address:
178 person_dict["address"] = person.address
179 if person.affiliation:
180 person_dict["affiliation"] = person.affiliation
181 return person_dict
182 else:
183 entity_dict = {"@type": "Organization", "name": person.name}
184 if person.address:
185 entity_dict["address"] = person.address
186 if person.email:
187 entity_dict["email"] = person.email
188 if person.date_start:
189 entity_dict["startDate"] = person.date_start.isoformat()
190 if person.date_end:
191 entity_dict["endDate"] = person.date_end.isoformat()
192 if person.website:
193 entity_dict["@id"] = str(person.website)
194 entity_dict["identifier"] = str(person.website)
195 if person.rorid:
196 entity_dict["@id"] = str(person.rorid)
197 entity_dict["identifier"] = str(person.rorid)
198 return entity_dict
200 @staticmethod
201 def _role_identifier(person: Person | Entity, person_dict: dict) -> str:
202 """Return an identifier that can link a CodeMeta role to its person."""
203 if identifier := person_dict.get("@id"):
204 return identifier
205 identity = person.email or person.full_name
206 return f"urn:uuid:{uuid.uuid5(uuid.NAMESPACE_URL, f'{type(person).__name__}:{identity}')}"
208 @staticmethod
209 def _has_contribution_metadata(person: Person | Entity) -> bool:
210 """Return whether a person has metadata representable by a CodeMeta role."""
211 return any(
212 (
213 person.contribution,
214 person.contribution_types,
215 person.contribution_begin,
216 person.contribution_end,
217 )
218 )
220 def _people_with_roles(self, people: Sequence[Person | Entity]) -> list[dict]:
221 """Serialize people and their granular contribution roles."""
222 result = []
223 for person in people:
224 person_dict = self._from_person(person)
225 if not self._has_contribution_metadata(person):
226 result.append(person_dict)
227 continue
229 identifier = self._role_identifier(person, person_dict)
230 person_dict["@id"] = identifier
231 result.append(person_dict)
233 role_names: list[str | None] = []
234 if person.contribution:
235 role_names.append(person.contribution)
236 role_names.extend(
237 contribution_type.value
238 for contribution_type in person.contribution_types or []
239 )
240 if not role_names:
241 role_names.append(None)
242 for role_name in role_names:
243 role = {"@type": "Role", "schema:author": identifier}
244 if role_name:
245 role["roleName"] = role_name
246 if person.contribution_begin:
247 role["startDate"] = person.contribution_begin.isoformat()
248 if person.contribution_end:
249 role["endDate"] = person.contribution_end.isoformat()
250 result.append(role)
251 return result
253 @staticmethod
254 def _to_person(person_obj) -> Person | Entity | None:
255 """Convert codemeta.json dict or str for person/entity format to project metadata person object."""
256 if person_obj.get("@type") == "Role":
257 return None
258 if "name" in person_obj:
259 entity_obj = {"name": person_obj["name"]}
260 return Entity(**entity_obj)
261 else:
262 person_data = {}
263 if "givenName" in person_obj:
264 person_data["given_names"] = person_obj["givenName"].strip()
265 if "familyName" in person_obj:
266 person_data["family_names"] = person_obj["familyName"].strip()
267 if "email" in person_obj:
268 person_data["email"] = person_obj["email"].strip()
269 if "orcid.org/" in person_obj.get("@id", ""):
270 person_data["orcid"] = person_obj["@id"].strip()
271 if "address" in person_obj:
272 person_data["address"] = person_obj["address"].strip()
274 return Person(**person_data)
276 def _sync_person_list(
277 self, old: list[Any], new: Sequence[Person | Entity]
278 ) -> list[Any]:
279 """Override the _sync_person_list function from ProjectMetadataWriter.
281 This method wont care about existing persons in codemeta.json file.
283 Args:
284 old (List[Any]): existing persons in codemeta.json file, in this case ignored in the output. However, it is necessary to make the function compatible with the parent class.
285 new (List[Person]): new persons to add to codemeta.json file
287 Returns:
288 List[Any]: list of new persons to add to codemeta.json file
290 """
291 return list(new)
293 def sync(self, metadata: ProjectMetadata) -> None:
294 """Sync codemeta.json with project metadata.
296 Use existing sync function from ProjectMetadataWriter but update repository and contributors.
297 """
298 if not self.merge:
299 self._data = self._new_data()
300 else:
301 self._upgrade_to_v3()
302 self._data["@type"] = "SoftwareSourceCode"
303 if metadata.authors():
304 self._data["author"] = []
305 self._data["maintainer"] = []
306 self._data["contributor"] = []
308 super().sync(metadata)
309 if metadata.doi:
310 self._data["identifier"] = f"https://doi.org/{metadata.doi}"
311 elif "identifier" in self._data:
312 identifiers = self._data["identifier"]
313 identifiers = (
314 identifiers if isinstance(identifiers, list) else [identifiers]
315 )
316 kept = [
317 i
318 for i in identifiers
319 if not (isinstance(i, str) and i.startswith("https://doi.org/"))
320 ]
321 if kept:
322 self._data["identifier"] = kept if len(kept) > 1 else kept[0]
323 else:
324 del self._data["identifier"]
325 if licenses := metadata.license:
326 self.license = [
327 f"https://spdx.org/licenses/{license.value}"
328 for license in (licenses if isinstance(licenses, list) else [licenses])
329 ]
330 self.contributors = metadata.contributors()
332 if "softwareHelp" in self._data:
333 self._data["url"] = self._data["softwareHelp"]