Coverage for src/somesy/pom_xml/writer.py: 92%
171 statements
« prev ^ index » next coverage.py v7.16.0, created at 2026-11-03 13:13 +0000
1"""Writer adapter for pom.xml files."""
3import logging
4import xml.etree.ElementTree as ET
5from pathlib import Path
6from typing import Any
8from somesy.core.models import Entity, Person
9from somesy.core.types import ContributionTypeEnum
10from somesy.core.writer import FieldKeyMapping, IgnoreKey, ProjectMetadataWriter
12from . import POM_ROOT_ATRS, POM_URL
13from .xmlproxy import XMLProxy
15logger = logging.getLogger("somesy")
18class POM(ProjectMetadataWriter):
19 """Java Maven pom.xml parser and saver."""
21 # TODO: write a wrapper for ElementTree that behaves like a dict
22 # TODO: set up correct field name mappings
24 def __init__(
25 self,
26 path: Path,
27 create_if_not_exists: bool = True,
28 pass_validation: bool | None = False,
29 ):
30 """Java Maven pom.xml parser.
32 See [somesy.core.writer.ProjectMetadataWriter.__init__][].
33 """
34 mappings: FieldKeyMapping = {
35 # "year": ["inceptionYear"], # not supported by somesy + does not really change
36 # "project_slug": ["artifactId"], # not supported by somesy for sync
37 "license": ["licenses", "license"],
38 "homepage": ["url"],
39 "repository": ["scm"],
40 "documentation": ["distributionManagement", "site"],
41 "authors": ["developers", "developer"],
42 "contributors": ["contributors", "contributor"],
43 }
44 super().__init__(
45 path,
46 create_if_not_exists=create_if_not_exists,
47 direct_mappings=mappings,
48 pass_validation=pass_validation,
49 )
51 def _init_new_file(self):
52 """Initialize new pom.xml file."""
53 pom = XMLProxy(ET.Element("project", POM_ROOT_ATRS))
54 pom["properties"] = {"info.versionScheme": "semver-spec"}
55 pom.write(self.path)
57 def _load(self):
58 """Load the POM file."""
59 ET.register_namespace("", POM_URL) # register POM as default xml namespace
60 self._data = XMLProxy.parse(self.path, default_namespace=POM_URL)
62 def _validate(self) -> None:
63 """Validate the POM file."""
64 logger.info("Cannot validate POM file, skipping validation.")
66 def save(self, path: Path | None = None) -> None:
67 """Save the POM DOM to a file."""
68 self._data.write(path or self.path, default_namespace=None)
70 def _get_property(
71 self,
72 key: str | list[str] | IgnoreKey,
73 *,
74 remove: bool = False,
75 ) -> Any | None:
76 """Get (a) property by key."""
77 if isinstance(key, IgnoreKey):
78 return None
79 elem = super()._get_property(key, remove=remove)
80 if elem is not None:
81 if isinstance(elem, list):
82 return [e.to_jsonlike() for e in elem]
83 else:
84 return elem.to_jsonlike()
85 return None
87 @staticmethod
88 def _from_person(person: Entity | Person):
89 """Convert person object to dict for POM XML person format."""
90 ret: dict[str, Any] = {}
91 if isinstance(person, Person):
92 person_id = person.to_name_email_string()
93 if person.orcid:
94 person_id = str(person.orcid)
95 ret["url"] = str(person.orcid)
96 else:
97 person_id = person.to_name_email_string()
98 if person.website:
99 person_id = str(person.website)
100 ret["url"] = str(person.website)
101 ret["id"] = person_id
102 ret["name"] = person.full_name
103 if person.email:
104 ret["email"] = person.email
105 if person.contribution_types:
106 ret["roles"] = {"role": [c.value for c in person.contribution_types]}
107 return ret
109 @staticmethod
110 def _parse_roles(person_obj: dict) -> list[ContributionTypeEnum] | None:
111 """Return the POM roles that are valid contribution types (roles are free text in POM)."""
112 roles = (person_obj.get("roles") or {}).get("role")
113 if not roles:
114 return None
115 if not isinstance(roles, list):
116 roles = [roles]
117 known = {c.value for c in ContributionTypeEnum}
118 valid = [ContributionTypeEnum(r) for r in roles if r in known]
119 if len(valid) < len(roles):
120 logger.debug(f"Ignoring POM roles that are not contribution types: {roles}")
121 return valid or None
123 def _records_keeping_roles(
124 self, people: list[Entity | Person], existing: list[dict]
125 ) -> list[dict[str, Any]]:
126 """Convert people to POM records, keeping free-text roles of matching existing entries."""
127 known = {c.value for c in ContributionTypeEnum}
128 old_people = [(self._to_person(e), e) for e in existing]
129 records = []
130 for person in people:
131 record = self._from_person(person)
132 for old_person, old in old_people:
133 if old_person is None or not old_person.same_person(person):
134 continue
135 old_roles = (old.get("roles") or {}).get("role") or []
136 if not isinstance(old_roles, list):
137 old_roles = [old_roles]
138 roles = record.get("roles", {}).get("role", [])
139 roles += [r for r in dict.fromkeys(old_roles) if r not in known]
140 if roles:
141 record["roles"] = {"role": roles}
142 break
143 records.append(record)
144 return records
146 @staticmethod
147 def _to_person(person_obj: dict) -> Entity | Person | None:
148 """Parse POM XML person to a somesy Person.
150 POM has no person/entity distinction, so this is a heuristic: a name
151 containing a space is a Person (last word is the family name),
152 anything else is an Entity.
153 """
154 if "name" not in person_obj:
155 logger.warning(f"POM person without name, omitting: {person_obj}")
156 return None
157 if " " in person_obj["name"]:
158 names = person_obj["name"].split()
159 gnames = " ".join(names[:-1])
160 fname = names[-1]
161 email = person_obj.get("email")
162 url = person_obj.get("url")
163 maybe_orcid = url if url and "orcid.org" in url else None
164 contr = POM._parse_roles(person_obj)
166 return Person(
167 given_names=gnames,
168 family_names=fname,
169 email=email,
170 orcid=maybe_orcid,
171 contribution_types=contr,
172 )
173 else:
174 name = person_obj["name"]
175 email = person_obj.get("email")
176 url = person_obj.get("url")
177 contr = POM._parse_roles(person_obj)
179 return Entity(
180 name=name,
181 email=email,
182 website=url,
183 contribution_types=contr,
184 )
186 # no search keywords supported in POM
187 @property
188 def keywords(self) -> list[str] | None:
189 """Return the keywords of the project."""
191 @keywords.setter
192 def keywords(self, keywords: list[str]) -> None:
193 """Set the keywords of the project."""
195 # authors must be a list
196 @property
197 def authors(self):
198 """Return the authors of the project."""
199 authors = self._get_property(self._get_key("authors"))
200 if authors is None:
201 return []
202 return authors if isinstance(authors, list) else [authors]
204 @authors.setter
205 def authors(self, authors: list[Entity | Person]) -> None:
206 """Set the authors of the project."""
207 author_records = self._records_keeping_roles(authors, self.authors)
208 self._set_property(self._get_key("authors"), author_records)
210 # contributors must be a list
211 @property
212 def contributors(self):
213 """Return the contributors of the project."""
214 contr = self._get_property(self._get_key("contributors"))
215 if contr is None:
216 return []
217 return contr if isinstance(contr, list) else [contr]
219 @contributors.setter
220 def contributors(self, contributors: list[Entity | Person]) -> None:
221 """Set the contributors of the project."""
222 contr = self._records_keeping_roles(contributors, self.contributors)
223 self._set_property(self._get_key("contributors"), contr)
225 # no maintainers supported im POM, only developers and contributors
226 @property
227 def maintainers(self):
228 """Return the maintainers of the project."""
229 return []
231 @maintainers.setter
232 def maintainers(self, maintainers: list[Person | Entity]) -> None:
233 """Set the maintainers of the project."""
235 @property
236 def license(self) -> str | list[str] | None:
237 """Return the license of the project."""
238 licenses = self._get_property(self._get_key("license"))
239 if licenses is None:
240 return None
241 licenses = licenses if isinstance(licenses, list) else [licenses]
242 names = [license["name"] for license in licenses]
243 return names[0] if len(names) == 1 else names
245 @license.setter
246 def license(self, license: str | list[str] | None) -> None:
247 """Set the license of the project."""
248 licenses = license if isinstance(license, list) else [license]
249 self._set_property(
250 self._get_key("license"),
251 [{"name": license, "distribution": "repo"} for license in licenses],
252 )
254 @property
255 def repository(self) -> str | dict | None:
256 """Return the repository url of the project."""
257 repo = super().repository
258 if isinstance(repo, str):
259 return repo
260 return repo.get("url") if repo is not None else None
262 @repository.setter
263 def repository(self, value: str | dict | None) -> None:
264 """Set the repository url of the project."""
265 if value is None:
266 return
267 # only touch <scm><url>, keep e.g. <connection> intact
268 self._set_property(["scm", "url"], value)
270 @property
271 def documentation(self) -> str | dict | None:
272 """Return the documentation url of the project."""
273 docs = super().documentation
274 if isinstance(docs, str):
275 return docs
276 return docs.get("url") if docs is not None else None
278 @documentation.setter
279 def documentation(self, value: str | dict | None) -> None:
280 """Set the documentation url of the project."""
281 if value is None:
282 return
283 self._set_property(
284 self._get_key("documentation"), {"name": "documentation site", "url": value}
285 )
287 def sync(self, metadata) -> None:
288 """Sync codemeta.json with project metadata.
290 Use existing sync function from ProjectMetadataWriter but update repository and contributors.
291 """
292 super().sync(metadata)
293 if licenses := metadata.license:
294 self.license = (
295 [license.value for license in licenses]
296 if isinstance(licenses, list)
297 else licenses.value
298 )
299 self.contributors = self._sync_person_list(
300 self.contributors, metadata.contributors()
301 )